diff --git a/pkg/storage/canonical.go b/pkg/storage/canonical.go new file mode 100644 index 000000000..3cd960951 --- /dev/null +++ b/pkg/storage/canonical.go @@ -0,0 +1,24 @@ +package storage + +import "strings" + +// CanonicalKey folds an identity string — repository full name, database +// name, database type, environment — to its canonical single-spelling form. +// Identity strings are the cross-dialect row-identity keys: MySQL's +// utf8mb4_0900_ai_ci storage collation forgives case drift while PostgreSQL +// compares bytes, so every boundary that accepts an identity string folds it +// before matching or persisting. Folding is lowercase-only; accent folding is +// deliberately excluded because identity strings are ASCII in practice. +// +// Deliberately not folded here: lock owner strings ("org/repo#42") are the +// ownership predicate on lock acquire, release, and intent verification, but +// the repository they are derived from is folded at ingress, so owners are +// canonical by construction rather than re-folded at this boundary. +// Deployment names are identity keys too, but they are operator-controlled +// configuration that doubles as routing and schema-directory path components, +// so config validation rejects non-canonical spellings instead of silently +// rewriting them. Table names, SHAs, and GitHub node IDs are case-significant +// or opaque values, not row-identity keys. +func CanonicalKey(s string) string { + return strings.ToLower(s) +} diff --git a/pkg/storage/internal/sqlstore/applies.go b/pkg/storage/internal/sqlstore/applies.go index ffa9e289e..1e86153e5 100644 --- a/pkg/storage/internal/sqlstore/applies.go +++ b/pkg/storage/internal/sqlstore/applies.go @@ -229,6 +229,10 @@ func (w *applyWriteTx) commit() error { } func applyTargetLockName(database, dbType, environment string) string { + // Fold every component so caller spelling cannot change the derived lock name. + database = storage.CanonicalKey(database) + dbType = storage.CanonicalKey(dbType) + environment = storage.CanonicalKey(environment) sum := sha256.Sum256([]byte(database + "\x00" + dbType + "\x00" + environment)) return "schemabot_apply_" + hex.EncodeToString(sum[:16]) } @@ -772,13 +776,15 @@ func verifyExpectedLockIntent(ctx context.Context, tx *rebindTx, apply *storage. // one: it matches only a lock whose pending_plan_id is unset (the column is // NOT NULL DEFAULT ''), so an unpinned lock that a rollback re-pins mid-flight // still fails this check. + database := storage.CanonicalKey(apply.Database) + databaseType := storage.CanonicalKey(apply.DatabaseType) var lockID int64 err := tx.QueryRowContext(ctx, ` SELECT id FROM locks WHERE database_name = ? AND database_type = ? AND owner = ? AND pending_plan_id = ? FOR UPDATE - `, apply.Database, apply.DatabaseType, apply.ExpectedLockOwner, apply.ExpectedPendingPlanID).Scan(&lockID) + `, database, databaseType, apply.ExpectedLockOwner, apply.ExpectedPendingPlanID).Scan(&lockID) if errors.Is(err, sql.ErrNoRows) { return storage.ErrLockIntentChanged } diff --git a/pkg/storage/internal/sqlstore/apply_target_lock_test.go b/pkg/storage/internal/sqlstore/apply_target_lock_test.go index 4f5f6a5f4..f524824d3 100644 --- a/pkg/storage/internal/sqlstore/apply_target_lock_test.go +++ b/pkg/storage/internal/sqlstore/apply_target_lock_test.go @@ -3,9 +3,17 @@ package sqlstore import ( "testing" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) +func TestApplyTargetLockName_CanonicalComponents(t *testing.T) { + lowercase := applyTargetLockName("orders_db", "mysql", "staging") + mixedCase := applyTargetLockName("Orders_DB", "MySQL", "Staging") + + assert.Equal(t, lowercase, mixedCase) +} + // TestAcquireApplyTargetLockConn_NilLockerFailsClosed verifies an apply // target without an advisory-lock implementation is rejected before any // connection is opened: applies must not proceed unserialized across diff --git a/pkg/storage/internal/sqlstore/checks.go b/pkg/storage/internal/sqlstore/checks.go index 4687784b7..0f954a1e0 100644 --- a/pkg/storage/internal/sqlstore/checks.go +++ b/pkg/storage/internal/sqlstore/checks.go @@ -30,8 +30,19 @@ type checkStore struct { classifier ErrorClassifier } +func canonicalizeCheck(check *storage.Check) { + if check == nil { + return + } + check.Repository = storage.CanonicalKey(check.Repository) + check.Environment = storage.CanonicalKey(check.Environment) + check.DatabaseType = storage.CanonicalKey(check.DatabaseType) + check.DatabaseName = storage.CanonicalKey(check.DatabaseName) +} + // Upsert creates or updates stored check state. func (s *checkStore) Upsert(ctx context.Context, check *storage.Check) error { + canonicalizeCheck(check) // Convert CheckRunID=0 to NULL (0 is Go's zero value, not a valid check run ID) var checkRunID any if check.CheckRunID != 0 { @@ -82,6 +93,7 @@ func (s *checkStore) Upsert(ctx context.Context, check *storage.Check) error { // only a write that re-ran the rollup and found the deployments clean may clear // it. See storage.PlanDriftState. func (s *checkStore) UpsertPlanResult(ctx context.Context, check *storage.Check, drift storage.PlanDriftState) error { + canonicalizeCheck(check) var checkRunID any if check.CheckRunID != 0 { checkRunID = check.CheckRunID @@ -182,6 +194,7 @@ func (s *checkStore) UpsertPlanResult(ctx context.Context, check *storage.Check, // RecoverApplyOwnedCheckWithNoOpPlan updates same-head apply-owned stored check // state when a successful no-op plan proves the target already matches the PR schema. func (s *checkStore) RecoverApplyOwnedCheckWithNoOpPlan(ctx context.Context, check *storage.Check) (bool, error) { + canonicalizeCheck(check) if !successfulNoOpPlanResult(check) { return false, nil } @@ -240,6 +253,7 @@ func successfulNoOpPlanResult(check *storage.Check) bool { // plan is no longer part of the merge gate, so the plan-only drift block should // stop blocking. A started apply still owns the row and is left untouched. func (s *checkStore) MarkStalePlanSuccessful(ctx context.Context, check *storage.Check) (bool, error) { + canonicalizeCheck(check) var checkRunID any if check.CheckRunID != 0 { checkRunID = check.CheckRunID @@ -300,6 +314,7 @@ func (s *checkStore) MarkStalePlanSuccessful(ctx context.Context, check *storage // commit) does not match and is preserved. Returns true when the row was // cleared. func (s *checkStore) ClearAggregateBlock(ctx context.Context, check *storage.Check) (bool, error) { + canonicalizeCheck(check) result, err := s.db.ExecContext(ctx, ` UPDATE checks SET blocking_reason = '', @@ -335,6 +350,7 @@ func isPlanOnlySuccessful(check *storage.Check) bool { // CompleteForApply updates stored check state to a terminal state only if it // still belongs to the apply being completed. func (s *checkStore) CompleteForApply(ctx context.Context, check *storage.Check, apply *storage.Apply) (bool, error) { + canonicalizeCheck(check) var checkRunID any if check.CheckRunID != 0 { checkRunID = check.CheckRunID @@ -407,6 +423,7 @@ func (s *checkStore) CompleteForApply(ctx context.Context, check *storage.Check, // Cancelled forward applies additionally require that no completed forward // task exists under this or an earlier apply for the target. func (s *checkStore) MarkActionRequiredForApply(ctx context.Context, check *storage.Check, apply *storage.Apply) (bool, error) { + canonicalizeCheck(check) var checkRunID any if check.CheckRunID != 0 { checkRunID = check.CheckRunID @@ -483,6 +500,7 @@ func (s *checkStore) MarkActionRequiredForApply(ctx context.Context, check *stor // after the cancellation, and the completed-task predicate keeps a // cancellation that is safe to release from being retained here. func (s *checkStore) MarkCancelledApplyFailed(ctx context.Context, check *storage.Check, apply *storage.Apply) (bool, error) { + canonicalizeCheck(check) var checkRunID any if check.CheckRunID != 0 { checkRunID = check.CheckRunID @@ -568,6 +586,10 @@ func (s *checkStore) completedForwardTaskPredicate(required bool) string { // Get returns a check by its unique key (PR + env + database), or nil if not found. func (s *checkStore) Get(ctx context.Context, repo string, pr int, environment, dbType, database string) (*storage.Check, error) { + repo = storage.CanonicalKey(repo) + environment = storage.CanonicalKey(environment) + dbType = storage.CanonicalKey(dbType) + database = storage.CanonicalKey(database) row := s.db.QueryRowContext(ctx, ` SELECT `+checkColumns+` FROM checks @@ -591,6 +613,7 @@ func (s *checkStore) GetByCheckRunID(ctx context.Context, checkRunID int64) (*st // GetByPR returns all checks for a PR. func (s *checkStore) GetByPR(ctx context.Context, repo string, pr int) ([]*storage.Check, error) { + repo = storage.CanonicalKey(repo) rows, err := s.db.QueryContext(ctx, ` SELECT `+checkColumns+` FROM checks @@ -608,6 +631,10 @@ func (s *checkStore) GetByPR(ctx context.Context, repo string, pr int) ([]*stora // GetByDatabase returns all checks for a database across all PRs. // Used for cross-PR coordination (blocking other PRs when one is applying). func (s *checkStore) GetByDatabase(ctx context.Context, repo, environment, dbType, database string) ([]*storage.Check, error) { + repo = storage.CanonicalKey(repo) + environment = storage.CanonicalKey(environment) + dbType = storage.CanonicalKey(dbType) + database = storage.CanonicalKey(database) rows, err := s.db.QueryContext(ctx, ` SELECT `+checkColumns+` FROM checks @@ -665,6 +692,7 @@ func (s *checkStore) Delete(ctx context.Context, id int64) error { // action_required when the schema change is gone from the PR, or a fresh // plan result replaces it when the change is still present. func (s *checkStore) DeleteByPRRetainingBlockingApplyOwned(ctx context.Context, repo string, pr int, merged bool) error { + repo = storage.CanonicalKey(repo) if merged { _, err := s.db.ExecContext(ctx, ` DELETE FROM checks diff --git a/pkg/storage/internal/sqlstore/dialect.go b/pkg/storage/internal/sqlstore/dialect.go index 233c1f77f..632fcb858 100644 --- a/pkg/storage/internal/sqlstore/dialect.go +++ b/pkg/storage/internal/sqlstore/dialect.go @@ -53,6 +53,12 @@ type Dialect interface { // placed immediately after a table name or alias; dialects without index-hint // syntax return an empty string. IndexHint(index string) string + // BinaryEquals returns a predicate that is true only when column's stored + // value equals the single bound placeholder byte-for-byte. Opaque identity + // values such as lock owners compare byte-wise by contract, so dialects + // whose default column collation folds case or accents must force a binary + // comparison rather than inherit the column's collation. + BinaryEquals(column string) string // JoinedUpdate returns an UPDATE that changes target rows selected through a // join. Aliases qualify join and predicate expressions, while assignments to // target columns must be unqualified so the statement is valid across @@ -252,6 +258,13 @@ func (MySQLDialect) IndexHint(index string) string { return " FORCE INDEX (`" + strings.ReplaceAll(index, "`", "``") + "`)" } +// BinaryEquals compares under the binary collation: the schema's table default +// is accent- and case-insensitive utf8mb4_0900_ai_ci, under which +// differently-spelled opaque values would match. +func (MySQLDialect) BinaryEquals(column string) string { + return column + " COLLATE utf8mb4_0900_bin = ?" +} + // JoinedUpdate builds a MySQL multi-table UPDATE statement. func (MySQLDialect) JoinedUpdate(targetTable, targetAlias, joinTable, joinAlias, joinCondition string, assignments []JoinedUpdateAssignment, predicate string) string { if len(assignments) == 0 { @@ -445,6 +458,12 @@ func (PostgresDialect) JSONBooleanIsTrue(expression string, path []string) strin // relies on the planner to choose the access path. func (PostgresDialect) IndexHint(string) string { return "" } +// BinaryEquals returns plain equality: PostgreSQL's deterministic collations +// already compare equality byte-for-byte. +func (PostgresDialect) BinaryEquals(column string) string { + return column + " = ?" +} + // JoinedUpdate builds a PostgreSQL UPDATE … FROM statement. The join condition // moves into the WHERE clause alongside the residual predicate; SET // placeholders still precede predicate placeholders, so the placeholder-free diff --git a/pkg/storage/internal/sqlstore/dialect_test.go b/pkg/storage/internal/sqlstore/dialect_test.go index da774bcaa..e0ebaa841 100644 --- a/pkg/storage/internal/sqlstore/dialect_test.go +++ b/pkg/storage/internal/sqlstore/dialect_test.go @@ -33,6 +33,10 @@ func TestMySQLDialectIndexHint(t *testing.T) { assert.Equal(t, " FORCE INDEX (`idx_database_env_deployment`)", MySQLDialect{}.IndexHint("idx_database_env_deployment")) } +func TestMySQLDialectBinaryEquals(t *testing.T) { + assert.Equal(t, "owner COLLATE utf8mb4_0900_bin = ?", MySQLDialect{}.BinaryEquals("owner")) +} + func TestMySQLDialectJoinedUpdate(t *testing.T) { assert.Equal(t, "UPDATE apply_comments c JOIN applies a ON a.id = c.apply_id SET c.edit_count = c.edit_count + 1, c.updated_at = NOW() WHERE c.apply_id = ? AND a.lease_token = ?", @@ -293,6 +297,10 @@ func TestPostgresDialect(t *testing.T) { d.RelativeTime(TimestampPrecisionDefault, AfterCurrentTime, LiteralIntervalAmount(2), IntervalDay)) } +func TestPostgresDialectBinaryEquals(t *testing.T) { + assert.Equal(t, "owner = ?", PostgresDialect{}.BinaryEquals("owner")) +} + func TestPostgresDialectInsertIfAbsent(t *testing.T) { assert.Equal(t, InsertIfAbsentSyntax{Suffix: " ON CONFLICT (apply_id, comment_state) DO NOTHING"}, diff --git a/pkg/storage/internal/sqlstore/locks.go b/pkg/storage/internal/sqlstore/locks.go index 571d46978..d8c649664 100644 --- a/pkg/storage/internal/sqlstore/locks.go +++ b/pkg/storage/internal/sqlstore/locks.go @@ -19,9 +19,19 @@ const lockColumns = `id, database_name, database_type, repository, pull_request, // lockStore implements storage.LockStore using MySQL. type lockStore struct { db *rebindDB + dialect Dialect classifier ErrorClassifier } +func canonicalizeLock(lock *storage.Lock) { + if lock == nil { + return + } + lock.DatabaseName = storage.CanonicalKey(lock.DatabaseName) + lock.DatabaseType = storage.CanonicalKey(lock.DatabaseType) + lock.Repository = storage.CanonicalKey(lock.Repository) +} + // Acquire attempts to acquire a lock. Returns ErrLockHeld if held by another owner. // Acquiring a lock the same owner already holds is a success (idempotent): two // concurrent applies for the same PR and database (e.g. staging and production @@ -31,6 +41,7 @@ type lockStore struct { // confirm command loads, and its disclosure record travels with it. A re-acquire // that passes an empty PendingPlanID (CLI) leaves the existing values intact. func (s *lockStore) Acquire(ctx context.Context, lock *storage.Lock) error { + canonicalizeLock(lock) op := fmt.Sprintf("acquire lock for %s/%s owner=%s", lock.DatabaseName, lock.DatabaseType, lock.Owner) return withLockRetry(ctx, s.classifier, op, func() error { return s.acquireOnce(ctx, lock) @@ -39,7 +50,7 @@ func (s *lockStore) Acquire(ctx context.Context, lock *storage.Lock) error { // acquireOnce performs a single claim attempt. Concurrent same-owner callers // racing to claim the same key can hit a transient InnoDB lock conflict on the -// INSERT below; Acquire retries those. +// INSERT below; Acquire retries those. Acquire canonicalizes the lock first. func (s *lockStore) acquireOnce(ctx context.Context, lock *storage.Lock) error { existing, err := s.Get(ctx, lock.DatabaseName, lock.DatabaseType) if err != nil { @@ -103,6 +114,7 @@ func (s *lockStore) acquireOnce(ctx context.Context, lock *storage.Lock) error { // between this caller's read and its write. The owner still holds the lock in that // case, so the refresh has succeeded. To distinguish that from a genuine ownership // change, re-read the lock and branch on its actual state. +// Acquire canonicalizes the lock before reaching this helper. func (s *lockStore) refreshPendingConfirmation(ctx context.Context, lock, existing *storage.Lock) error { if lock.PendingPlanID == "" || lock.PendingPlanID == existing.PendingPlanID { return nil @@ -110,7 +122,7 @@ func (s *lockStore) refreshPendingConfirmation(ctx context.Context, lock, existi result, err := s.db.ExecContext(ctx, ` UPDATE locks SET pending_plan_id = ?, disclosed_copy_discard = ?, updated_at = NOW() - WHERE database_name = ? AND database_type = ? AND owner = ? + WHERE database_name = ? AND database_type = ? AND `+s.dialect.BinaryEquals("owner")+` `, lock.PendingPlanID, lock.DisclosedCopyDiscard, lock.DatabaseName, lock.DatabaseType, lock.Owner) if err != nil { return fmt.Errorf("refresh pending confirmation for %s/%s owner=%s: %w", @@ -143,9 +155,11 @@ func (s *lockStore) refreshPendingConfirmation(ctx context.Context, lock, existi // Release releases a lock. Only succeeds if caller is the owner. func (s *lockStore) Release(ctx context.Context, database, dbType, owner string) error { + database = storage.CanonicalKey(database) + dbType = storage.CanonicalKey(dbType) result, err := s.db.ExecContext(ctx, ` DELETE FROM locks - WHERE database_name = ? AND database_type = ? AND owner = ? + WHERE database_name = ? AND database_type = ? AND `+s.dialect.BinaryEquals("owner")+` `, database, dbType, owner) if err != nil { return err @@ -172,9 +186,11 @@ func (s *lockStore) Release(ctx context.Context, database, dbType, owner string) // apply lock can become a rollback lock), so owner-only release is insufficient // after a network call or other long-running operation. func (s *lockStore) ReleaseIfPendingPlanID(ctx context.Context, database, dbType, owner, pendingPlanID string) (bool, error) { + database = storage.CanonicalKey(database) + dbType = storage.CanonicalKey(dbType) result, err := s.db.ExecContext(ctx, ` DELETE FROM locks - WHERE database_name = ? AND database_type = ? AND owner = ? AND pending_plan_id = ? + WHERE database_name = ? AND database_type = ? AND `+s.dialect.BinaryEquals("owner")+` AND pending_plan_id = ? `, database, dbType, owner, pendingPlanID) if err != nil { return false, err @@ -189,6 +205,8 @@ func (s *lockStore) ReleaseIfPendingPlanID(ctx context.Context, database, dbType // ForceRelease releases a lock regardless of owner (admin override). func (s *lockStore) ForceRelease(ctx context.Context, database, dbType string) error { + database = storage.CanonicalKey(database) + dbType = storage.CanonicalKey(dbType) result, err := s.db.ExecContext(ctx, ` DELETE FROM locks WHERE database_name = ? AND database_type = ? @@ -202,6 +220,8 @@ func (s *lockStore) ForceRelease(ctx context.Context, database, dbType string) e // Get returns a lock by database name and type, or nil if not found. func (s *lockStore) Get(ctx context.Context, database, dbType string) (*storage.Lock, error) { + database = storage.CanonicalKey(database) + dbType = storage.CanonicalKey(dbType) row := s.db.QueryRowContext(ctx, ` SELECT `+lockColumns+` FROM locks @@ -240,10 +260,11 @@ func (s *lockStore) List(ctx context.Context) ([]*storage.Lock, error) { // ErrLockNotFound when it is gone or ErrLockNotOwned when another owner holds // it. func (s *lockStore) Update(ctx context.Context, lock *storage.Lock) error { + canonicalizeLock(lock) result, err := s.db.ExecContext(ctx, ` UPDATE locks SET updated_at = NOW() - WHERE database_name = ? AND database_type = ? AND owner = ? + WHERE database_name = ? AND database_type = ? AND `+s.dialect.BinaryEquals("owner")+` `, lock.DatabaseName, lock.DatabaseType, lock.Owner) if err != nil { return fmt.Errorf("touch lock for %s/%s: %w", lock.DatabaseName, lock.DatabaseType, err) @@ -273,6 +294,7 @@ func (s *lockStore) Update(ctx context.Context, lock *storage.Lock) error { // GetByPR returns all locks associated with a PR. func (s *lockStore) GetByPR(ctx context.Context, repo string, pr int) ([]*storage.Lock, error) { + repo = storage.CanonicalKey(repo) rows, err := s.db.QueryContext(ctx, ` SELECT `+lockColumns+` FROM locks diff --git a/pkg/storage/internal/sqlstore/locks_test.go b/pkg/storage/internal/sqlstore/locks_test.go index 10d567036..471922071 100644 --- a/pkg/storage/internal/sqlstore/locks_test.go +++ b/pkg/storage/internal/sqlstore/locks_test.go @@ -89,7 +89,7 @@ func TestLockStore_Acquire_RefreshSameOwnerValueAlreadyMatches(t *testing.T) { require.NoError(t, db.Close()) }) require.NoError(t, db.PingContext(ctx)) - store := &lockStore{db: newRebindDB(db, MySQLDialect{}), classifier: NewMySQLErrorClassifier()} + store := &lockStore{db: newRebindDB(db, MySQLDialect{}), dialect: MySQLDialect{}, classifier: NewMySQLErrorClassifier()} require.NoError(t, store.Acquire(ctx, &storage.Lock{ DatabaseName: "testdb", @@ -150,7 +150,7 @@ func TestLockStore_Acquire_RefreshSameOwnerValueAlreadyMatches(t *testing.T) { func TestLockStore_Acquire_RefreshOwnerNoLongerMatches(t *testing.T) { clearTables(t) ctx := t.Context() - store := &lockStore{db: newRebindDB(testDB, MySQLDialect{}), classifier: NewMySQLErrorClassifier()} + store := &lockStore{db: newRebindDB(testDB, MySQLDialect{}), dialect: MySQLDialect{}, classifier: NewMySQLErrorClassifier()} require.NoError(t, store.Acquire(ctx, &storage.Lock{ DatabaseName: "testdb", @@ -280,7 +280,7 @@ func TestLockStore_UpdateSameSecondSucceeds(t *testing.T) { _, err = db.ExecContext(ctx, "SET TIMESTAMP = 1700000000") require.NoError(t, err) - store := &lockStore{db: newRebindDB(db, MySQLDialect{}), classifier: NewMySQLErrorClassifier()} + store := &lockStore{db: newRebindDB(db, MySQLDialect{}), dialect: MySQLDialect{}, classifier: NewMySQLErrorClassifier()} // Acquire seeds the row via the locks table's DEFAULT CURRENT_TIMESTAMP, // which resolves to the frozen NOW(). diff --git a/pkg/storage/internal/sqlstore/storage.go b/pkg/storage/internal/sqlstore/storage.go index 64b15c87c..a7c107439 100644 --- a/pkg/storage/internal/sqlstore/storage.go +++ b/pkg/storage/internal/sqlstore/storage.go @@ -85,7 +85,7 @@ func NewWithDependencies(deps Dependencies) *Storage { } return &Storage{ db: rdb, - locks: &lockStore{db: rdb, classifier: deps.Classifier}, + locks: &lockStore{db: rdb, dialect: deps.Dialect, classifier: deps.Classifier}, plans: &planStore{db: rdb, identity: deps.Identity, classifier: deps.Classifier}, applies: &applyStore{db: rdb, dialect: deps.Dialect, identity: deps.Identity, locker: deps.Locker, classifier: deps.Classifier}, tasks: &taskStore{db: rdb, dialect: deps.Dialect, identity: deps.Identity, locker: deps.Locker}, diff --git a/pkg/storage/storage.go b/pkg/storage/storage.go index e2e1cf4b2..30829c82c 100644 --- a/pkg/storage/storage.go +++ b/pkg/storage/storage.go @@ -112,6 +112,7 @@ type Storage interface { // Locks prevent concurrent schema changes to the same database. // Lock key is database:type (not per-environment) to block concurrent changes // across environments and PRs. +// Methods accepting a *Lock canonicalize its repository, database name, and database type in place before persisting. type LockStore interface { // Acquire attempts to acquire a lock. Returns ErrLockHeld if already held by another owner. // If the same owner already holds the lock, this is a no-op (idempotent). @@ -150,6 +151,7 @@ type LockStore interface { // CheckStore manages SchemaBot's stored check state. // Per-database rows track internal status for a PR/environment/database. // Aggregate rows store the GitHub check_run_id for the visible GitHub Check Run. +// Methods accepting a *Check canonicalize its repository, database name, database type, and environment in place before persisting. type CheckStore interface { // Upsert creates or updates stored check state. Upsert(ctx context.Context, check *Check) error diff --git a/pkg/storage/storagetest/checks.go b/pkg/storage/storagetest/checks.go index 6fa9e3627..e4e464b10 100644 --- a/pkg/storage/storagetest/checks.go +++ b/pkg/storage/storagetest/checks.go @@ -15,6 +15,45 @@ import ( // upserts, aggregate and per-database row isolation, apply ownership, and // review-time deployment drift disposition. func TestChecks(t *testing.T, h Harness) { + t.Run("CanonicalIdentityKey", func(t *testing.T) { + ctx := t.Context() + store := h.NewStorage(t) + check := &storage.Check{ + Repository: "MixedCase/Sample-Repo", PullRequest: 123, HeadSHA: "abc123", + Environment: "Staging", DatabaseType: "MySQL", DatabaseName: "Orders_DB", + Status: "completed", Conclusion: "success", + } + require.NoError(t, store.Checks().Upsert(ctx, check)) + + stored, err := store.Checks().Get(ctx, "MIXEDCASE/SAMPLE-REPO", 123, "STAGING", "MYSQL", "ORDERS_DB") + require.NoError(t, err) + require.NotNil(t, stored) + assert.Equal(t, "mixedcase/sample-repo", stored.Repository) + assert.Equal(t, "staging", stored.Environment) + assert.Equal(t, "mysql", stored.DatabaseType) + assert.Equal(t, "orders_db", stored.DatabaseName) + + check.Repository = "MIXEDCASE/SAMPLE-REPO" + check.Environment = "STAGING" + check.DatabaseType = "MYSQL" + check.DatabaseName = "ORDERS_DB" + check.HeadSHA = "def456" + require.NoError(t, store.Checks().Upsert(ctx, check)) + + checks, err := store.Checks().GetByDatabase(ctx, "MixedCase/Sample-Repo", "Staging", "MySQL", "Orders_DB") + require.NoError(t, err) + require.Len(t, checks, 1) + assert.Equal(t, "def456", checks[0].HeadSHA) + + checks, err = store.Checks().GetByPR(ctx, "MIXEDCASE/SAMPLE-REPO", 123) + require.NoError(t, err) + require.Len(t, checks, 1) + require.NoError(t, store.Checks().DeleteByPRRetainingBlockingApplyOwned(ctx, "MixedCase/Sample-Repo", 123, false)) + checks, err = store.Checks().GetByPR(ctx, "mixedcase/sample-repo", 123) + require.NoError(t, err) + assert.Empty(t, checks) + }) + t.Run("Upsert", func(t *testing.T) { ctx := t.Context() store := h.NewStorage(t) diff --git a/pkg/storage/storagetest/locks.go b/pkg/storage/storagetest/locks.go index edbc4d220..1eb000dd0 100644 --- a/pkg/storage/storagetest/locks.go +++ b/pkg/storage/storagetest/locks.go @@ -16,6 +16,56 @@ import ( // requires backdating the stored row via SQL, which is dialect-specific and // lives in each implementation's own integration tests. func TestLocks(t *testing.T, h Harness) { + t.Run("CanonicalIdentityKey", func(t *testing.T) { + ctx := t.Context() + store := h.NewStorage(t) + lock := &storage.Lock{ + DatabaseName: "Orders_DB", + DatabaseType: "MySQL", + Repository: "Org/Repo", + PullRequest: 42, + Owner: "Org/Repo#42", + PendingPlanID: "plan-1", + } + require.NoError(t, store.Locks().Acquire(ctx, lock)) + + stored, err := store.Locks().Get(ctx, "ORDERS_DB", "MYSQL") + require.NoError(t, err) + require.NotNil(t, stored) + assert.Equal(t, "orders_db", stored.DatabaseName) + assert.Equal(t, "mysql", stored.DatabaseType) + assert.Equal(t, "org/repo", stored.Repository) + assert.Equal(t, "Org/Repo#42", stored.Owner) + + locks, err := store.Locks().GetByPR(ctx, "ORG/REPO", 42) + require.NoError(t, err) + require.Len(t, locks, 1) + + require.NoError(t, store.Locks().Acquire(ctx, &storage.Lock{ + DatabaseName: "orders_db", DatabaseType: "mysql", + Repository: "org/repo", PullRequest: 42, + Owner: "Org/Repo#42", + })) + require.ErrorIs(t, store.Locks().Acquire(ctx, &storage.Lock{ + DatabaseName: "ORDERS_DB", DatabaseType: "MYSQL", + Repository: "ORG/REPO", PullRequest: 42, + Owner: "org/repo#42", + }), storage.ErrLockHeld) + require.ErrorIs(t, store.Locks().Release(ctx, "ORDERS_DB", "MYSQL", "org/repo#42"), storage.ErrLockNotOwned) + require.ErrorIs(t, store.Locks().Update(ctx, &storage.Lock{ + DatabaseName: "ORDERS_DB", DatabaseType: "MYSQL", + Owner: "org/repo#42", + }), storage.ErrLockNotOwned) + + released, err := store.Locks().ReleaseIfPendingPlanID(ctx, "ORDERS_DB", "MYSQL", "org/repo#42", "plan-1") + require.NoError(t, err) + assert.False(t, released) + + released, err = store.Locks().ReleaseIfPendingPlanID(ctx, "ORDERS_DB", "MYSQL", "Org/Repo#42", "plan-1") + require.NoError(t, err) + assert.True(t, released) + }) + t.Run("Acquire", func(t *testing.T) { ctx := t.Context() store := h.NewStorage(t)