diff --git a/pkg/storage/internal/sqlstore/plan_comments.go b/pkg/storage/internal/sqlstore/plan_comments.go index 4c7c70fbe..2842160d1 100644 --- a/pkg/storage/internal/sqlstore/plan_comments.go +++ b/pkg/storage/internal/sqlstore/plan_comments.go @@ -24,6 +24,8 @@ type planCommentStore struct { // Insert stores a newly posted plan comment and sets comment.ID. func (s *planCommentStore) Insert(ctx context.Context, comment *storage.PlanComment) error { + canonicalizePlanCommentIdentity(comment) + id, err := s.identity.InsertID(ctx, s.db, ` INSERT INTO plan_comments (repository, pull_request, database_name, database_type, environment_scope, head_sha, github_comment_id, github_node_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?) @@ -39,6 +41,9 @@ func (s *planCommentStore) Insert(ctx context.Context, comment *storage.PlanComm // ListUnminimizedForSlot returns the not-yet-minimized comments for a // (repository, pull_request, database) slot, ordered by id ascending. func (s *planCommentStore) ListUnminimizedForSlot(ctx context.Context, repo string, pr int, database, databaseType string) ([]*storage.PlanComment, error) { + repo = storage.CanonicalKey(repo) + database = storage.CanonicalKey(database) + databaseType = storage.CanonicalKey(databaseType) rows, err := s.db.QueryContext(ctx, ` SELECT `+planCommentColumns+` FROM plan_comments @@ -58,6 +63,7 @@ func (s *planCommentStore) ListUnminimizedForSlot(ctx context.Context, repo stri // database has no other way to ask. The (repository, pull_request) prefix of // the slot index serves it. func (s *planCommentStore) ListUnminimizedForRepoPR(ctx context.Context, repo string, pr int) ([]*storage.PlanComment, error) { + repo = storage.CanonicalKey(repo) rows, err := s.db.QueryContext(ctx, ` SELECT `+planCommentColumns+` FROM plan_comments @@ -101,6 +107,17 @@ func (s *planCommentStore) MarkMinimized(ctx context.Context, id int64) error { return nil } +// canonicalizePlanCommentIdentity folds the identity keys that appear in this +// store's SQL predicates. EnvironmentScope is deliberately not folded: no +// query filters on it, and its consumers compare it in Go against values built +// fresh from configured environment names, so folding only the stored side +// would break those comparisons. +func canonicalizePlanCommentIdentity(comment *storage.PlanComment) { + comment.Repository = storage.CanonicalKey(comment.Repository) + comment.DatabaseName = storage.CanonicalKey(comment.DatabaseName) + comment.DatabaseType = storage.CanonicalKey(comment.DatabaseType) +} + // scanPlanComment scans plan comment data from any scanner (Row or Rows). func scanPlanComment(s scanner) (*storage.PlanComment, error) { var comment storage.PlanComment diff --git a/pkg/storage/internal/sqlstore/plans.go b/pkg/storage/internal/sqlstore/plans.go index 81c73aaae..f4412de47 100644 --- a/pkg/storage/internal/sqlstore/plans.go +++ b/pkg/storage/internal/sqlstore/plans.go @@ -36,6 +36,8 @@ type planStore struct { // Create stores a new plan and returns its ID. func (s *planStore) Create(ctx context.Context, plan *storage.Plan) (int64, error) { + canonicalizePlanIdentity(plan) + planDataJSON, err := json.Marshal(namespacesWithShardPlans(plan)) if err != nil { return 0, fmt.Errorf("marshal plan data: %w", err) @@ -98,6 +100,7 @@ func (s *planStore) GetByLock(ctx context.Context, lockID int64) ([]*storage.Pla // latestPlanForTarget take the first matching row as "the newest plan" and rely // on this deterministic order to avoid picking an older SHA on ties. func (s *planStore) GetByPR(ctx context.Context, repo string, pr int) ([]*storage.Plan, error) { + repo = storage.CanonicalKey(repo) rows, err := s.db.QueryContext(ctx, ` SELECT `+planColumns+` FROM plans @@ -119,6 +122,10 @@ func (s *planStore) GetByPR(ctx context.Context, repo string, pr int) ([]*storag // plans.created_at is datetime (second precision), so the id tiebreaker keeps // same-second plans in a deterministic newest-first order. func (s *planStore) List(ctx context.Context, opts storage.ListPlansOptions) ([]*storage.Plan, error) { + opts.Database = storage.CanonicalKey(opts.Database) + opts.Environment = storage.CanonicalKey(opts.Environment) + opts.Repository = storage.CanonicalKey(opts.Repository) + if opts.Limit <= 0 { return nil, fmt.Errorf("list plans for database %q environment %q: limit must be positive, got %d", opts.Database, opts.Environment, opts.Limit) } @@ -176,10 +183,18 @@ func (s *planStore) Delete(ctx context.Context, id int64) error { // DeleteByPR removes all plans for a PR. func (s *planStore) DeleteByPR(ctx context.Context, repo string, pr int) error { + repo = storage.CanonicalKey(repo) _, err := s.db.ExecContext(ctx, `DELETE FROM plans WHERE repository = ? AND pull_request = ?`, repo, pr) return err } +func canonicalizePlanIdentity(plan *storage.Plan) { + plan.Database = storage.CanonicalKey(plan.Database) + plan.DatabaseType = storage.CanonicalKey(plan.DatabaseType) + plan.Repository = storage.CanonicalKey(plan.Repository) + plan.Environment = storage.CanonicalKey(plan.Environment) +} + // scanPlan scans a single plan row, returning nil if not found. func scanPlan(row *sql.Row) (*storage.Plan, error) { plan, err := scanPlanInto(row) diff --git a/pkg/storage/internal/sqlstore/webhook_events.go b/pkg/storage/internal/sqlstore/webhook_events.go index 1b61e700b..04ec83d59 100644 --- a/pkg/storage/internal/sqlstore/webhook_events.go +++ b/pkg/storage/internal/sqlstore/webhook_events.go @@ -33,6 +33,8 @@ func (s *webhookEventStore) Create(ctx context.Context, event *storage.WebhookEv if event.Event == "" { return false, fmt.Errorf("webhook event type is required") } + event.Repository = storage.CanonicalKey(event.Repository) + provider := event.Provider if provider == "" { provider = storage.WebhookProviderGitHub @@ -182,6 +184,7 @@ func webhookClaimableArgs() []any { } func (s *webhookEventStore) HasEventForHead(ctx context.Context, provider, repository string, pullRequest int, headSHA string) (bool, error) { + repository = storage.CanonicalKey(repository) if provider == "" { provider = storage.WebhookProviderGitHub } @@ -267,21 +270,23 @@ func (s *webhookEventStore) coveringSuccessorQuery(provider string, event *stora // the live PR is worth a GitHub call at all — the common no-successor claim // stays storage-only. func (s *webhookEventStore) HasCoveringSuccessor(ctx context.Context, event *storage.WebhookEvent) (bool, error) { - if event.Repository == "" || event.PullRequest == 0 { + queryEvent := *event + queryEvent.Repository = storage.CanonicalKey(queryEvent.Repository) + if queryEvent.Repository == "" || queryEvent.PullRequest == 0 { return false, fmt.Errorf("check covering successor for webhook event %d: repository and pull request are required for coalescing", event.ID) } provider := event.Provider if provider == "" { provider = storage.WebhookProviderGitHub } - query, args := s.coveringSuccessorQuery(provider, event) + query, args := s.coveringSuccessorQuery(provider, &queryEvent) var one int err := s.db.QueryRowContext(ctx, query, args...).Scan(&one) if errors.Is(err, sql.ErrNoRows) { return false, nil } if err != nil { - return false, fmt.Errorf("check covering successor for webhook event %d (repo=%s, pr=%d): %w", event.ID, event.Repository, event.PullRequest, err) + return false, fmt.Errorf("check covering successor for webhook event %d (repo=%s, pr=%d): %w", event.ID, queryEvent.Repository, event.PullRequest, err) } return true, nil } @@ -311,10 +316,11 @@ func (s *webhookEventStore) HasCoveringSuccessor(ctx context.Context, event *sto // terminally failed / superseded rows never run — none of those may justify // discarding older work. func (s *webhookEventStore) SupersedeIfCovered(ctx context.Context, event *storage.WebhookEvent) (bool, error) { + repository := storage.CanonicalKey(event.Repository) if event.LeaseToken == "" { return false, fmt.Errorf("webhook event lease token is required") } - if event.Repository == "" || event.PullRequest == 0 { + if repository == "" || event.PullRequest == 0 { return false, fmt.Errorf("supersede webhook event %d: repository and pull request are required for coalescing", event.ID) } provider := event.Provider @@ -322,7 +328,9 @@ func (s *webhookEventStore) SupersedeIfCovered(ctx context.Context, event *stora provider = storage.WebhookProviderGitHub } autoPlanIn := placeholders(len(storage.AutoPlanPullRequestActions)) - successorQuery, successorArgs := s.coveringSuccessorQuery(provider, event) + queryEvent := *event + queryEvent.Repository = repository + successorQuery, successorArgs := s.coveringSuccessorQuery(provider, &queryEvent) args := []any{storage.WebhookEventSuperseded, event.ID, event.LeaseToken, storage.WebhookEventProcessing} args = append(args, stringArgs(storage.AutoPlanPullRequestActions)...) args = append(args, successorArgs...) @@ -336,7 +344,7 @@ func (s *webhookEventStore) SupersedeIfCovered(ctx context.Context, event *stora ) `, args...) if err != nil { - return false, fmt.Errorf("supersede webhook event %d (repo=%s, pr=%d): %w", event.ID, event.Repository, event.PullRequest, err) + return false, fmt.Errorf("supersede webhook event %d (repo=%s, pr=%d): %w", event.ID, repository, event.PullRequest, err) } rows, err := result.RowsAffected() if err != nil { diff --git a/pkg/storage/storage.go b/pkg/storage/storage.go index 30829c82c..618887b27 100644 --- a/pkg/storage/storage.go +++ b/pkg/storage/storage.go @@ -276,6 +276,10 @@ type SettingsStore interface { // primitive behind fast webhook acknowledgement: handlers can persist a delivery // before returning 2xx, and drivers can claim/retry the stored event after the // HTTP request has finished. +// Create canonicalizes the provided event's repository in place before +// persisting. The coalescing reads (HasCoveringSuccessor, SupersedeIfCovered) +// fold the repository only inside their SQL predicates and leave the caller's +// event untouched. type WebhookEventStore interface { // Create records a webhook delivery in the pending state. Returns // inserted=false when provider + delivery GUID already exists, so callers @@ -477,6 +481,7 @@ type ListPlansOptions struct { // PlanStore manages schema change plans. // Plans are created by Plan() and stored for Apply() and staleness detection. // Both GRPCClient and LocalClient are stateless - SchemaBot owns plan storage. +// Create canonicalizes the provided plan's repository, environment, database, and database type in place before persisting. type PlanStore interface { // Create stores a new plan and returns its ID. Returns error if plan_identifier already exists. Create(ctx context.Context, plan *Plan) (int64, error) @@ -1082,6 +1087,10 @@ const SummaryClaimStaleAfter = 2 * time.Minute // for comments actually posted; minimized_at is set only after the GitHub // minimize call succeeded, so an unminimized row is always retried by the next // supersede. +// Insert canonicalizes the provided comment's repository, database, and +// database type in place before persisting. EnvironmentScope is stored as +// given: no query predicate filters on it, and its consumers compare it in Go +// against a scope built from the configured environment names. type PlanCommentStore interface { // Insert stores a newly posted plan comment and sets comment.ID. Insert(ctx context.Context, comment *PlanComment) error diff --git a/pkg/storage/storagetest/plan_comments.go b/pkg/storage/storagetest/plan_comments.go index 523e3433f..9359ad067 100644 --- a/pkg/storage/storagetest/plan_comments.go +++ b/pkg/storage/storagetest/plan_comments.go @@ -40,6 +40,30 @@ func InsertPlanComment(t *testing.T, store storage.Storage, repo string, pr int, // behavior — a minimized comment drops out of the unminimized listings and // repeat or missing-id marks are no-ops. func TestPlanComments(t *testing.T, h Harness) { + t.Run("CanonicalizesIdentityKeys", func(t *testing.T) { + ctx := t.Context() + store := h.NewStorage(t) + + inserted := InsertPlanComment(t, store, "MixedCase/Sample-Repo", 42, "OrdersDB", "MySQL", "Production,Staging", "sha1", 100) + // Stored-value equality is the cross-dialect check that identity keys are canonicalized before persistence. + assert.Equal(t, "mixedcase/sample-repo", inserted.Repository) + assert.Equal(t, "ordersdb", inserted.DatabaseName) + assert.Equal(t, "mysql", inserted.DatabaseType) + // EnvironmentScope is not a query predicate and is compared in Go against + // a scope built from configured environment names, so it is stored as given. + assert.Equal(t, "Production,Staging", inserted.EnvironmentScope) + + comments, err := store.PlanComments().ListUnminimizedForSlot(ctx, "MIXEDCASE/SAMPLE-REPO", 42, "ORDERSDB", "MYSQL") + require.NoError(t, err) + require.Len(t, comments, 1) + assert.Equal(t, inserted.ID, comments[0].ID) + + comments, err = store.PlanComments().ListUnminimizedForRepoPR(ctx, "MIXEDCASE/SAMPLE-REPO", 42) + require.NoError(t, err) + require.Len(t, comments, 1) + assert.Equal(t, inserted.ID, comments[0].ID) + }) + t.Run("Insert_And_ListUnminimizedForSlot", func(t *testing.T) { ctx := t.Context() store := h.NewStorage(t) diff --git a/pkg/storage/storagetest/plans.go b/pkg/storage/storagetest/plans.go index 3ca60d7b8..0023a1267 100644 --- a/pkg/storage/storagetest/plans.go +++ b/pkg/storage/storagetest/plans.go @@ -17,6 +17,46 @@ import ( // storage.ErrNotImplemented. When an implementation lands, it joins this // family. func TestPlans(t *testing.T, h Harness) { + t.Run("CanonicalizesIdentityKeys", func(t *testing.T) { + ctx := t.Context() + store := h.NewStorage(t) + plan := &storage.Plan{ + PlanIdentifier: "plan_mixed_case", + Database: "OrdersDB", + DatabaseType: "MySQL", + Repository: "MixedCase/Sample-Repo", + PullRequest: 42, + Environment: "Staging", + CreatedAt: time.Now().UTC().Truncate(time.Second), + } + _, err := store.Plans().Create(ctx, plan) + require.NoError(t, err) + + // Stored-value equality is the cross-dialect check that identity keys are canonicalized before persistence. + assert.Equal(t, "ordersdb", plan.Database) + assert.Equal(t, "mysql", plan.DatabaseType) + assert.Equal(t, "mixedcase/sample-repo", plan.Repository) + assert.Equal(t, "staging", plan.Environment) + + byPR, err := store.Plans().GetByPR(ctx, "MIXEDCASE/SAMPLE-REPO", 42) + require.NoError(t, err) + require.Len(t, byPR, 1) + assert.Equal(t, "plan_mixed_case", byPR[0].PlanIdentifier) + + listed, err := store.Plans().List(ctx, storage.ListPlansOptions{ + Database: "ORDERSDB", Environment: "STAGING", + Repository: "MIXEDCASE/SAMPLE-REPO", PullRequest: 42, Limit: 10, + }) + require.NoError(t, err) + require.Len(t, listed, 1) + assert.Equal(t, "plan_mixed_case", listed[0].PlanIdentifier) + + require.NoError(t, store.Plans().DeleteByPR(ctx, "MIXEDCASE/SAMPLE-REPO", 42)) + deleted, err := store.Plans().Get(ctx, "plan_mixed_case") + require.NoError(t, err) + assert.Nil(t, deleted) + }) + t.Run("Create_And_Get", func(t *testing.T) { ctx := t.Context() store := h.NewStorage(t) diff --git a/pkg/storage/storagetest/webhook_events.go b/pkg/storage/storagetest/webhook_events.go index 61844da36..bf714e33f 100644 --- a/pkg/storage/storagetest/webhook_events.go +++ b/pkg/storage/storagetest/webhook_events.go @@ -107,6 +107,36 @@ func TestWebhookEvents(t *testing.T, h Harness) { return claimExpecting(t, store, "old") } + t.Run("CanonicalizesRepositoryKey", func(t *testing.T) { + ctx := t.Context() + store := h.NewStorage(t) + + event := newPullRequestEvent("delivery-mixed-case", "synchronize", 42, "mixed-head", time.Time{}) + event.Repository = "MixedCase/Sample-Repo" + createEvent(t, store, event) + // Stored-value equality is the cross-dialect check that the repository is canonicalized before persistence. + assert.Equal(t, "mixedcase/sample-repo", event.Repository) + + found, err := store.WebhookEvents().HasEventForHead(ctx, storage.WebhookProviderGitHub, "MIXEDCASE/SAMPLE-REPO", 42, "mixed-head") + require.NoError(t, err) + assert.True(t, found) + + claimed := claimExpecting(t, store, "delivery-mixed-case") + claimed.Repository = "MIXEDCASE/SAMPLE-REPO" + successor := newPullRequestEvent("delivery-mixed-successor", "synchronize", 42, "new-head", claimed.ReceivedAt.Add(time.Second)) + successor.Repository = "mixedcase/sample-repo" + createEvent(t, store, successor) + + covered, err := store.WebhookEvents().HasCoveringSuccessor(ctx, claimed) + require.NoError(t, err) + assert.True(t, covered) + assert.Equal(t, "MIXEDCASE/SAMPLE-REPO", claimed.Repository) + superseded, err := store.WebhookEvents().SupersedeIfCovered(ctx, claimed) + require.NoError(t, err) + assert.True(t, superseded) + assert.Equal(t, "MIXEDCASE/SAMPLE-REPO", claimed.Repository) + }) + t.Run("Create_DeduplicatesDeliveryID", func(t *testing.T) { ctx := t.Context() store := h.NewStorage(t)