fix(storage): canonicalize lock and check identity keys

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)
+}

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
 	}

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

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

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

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"},

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,14 +114,15 @@ 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
 	}
 	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

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().