From 6d589c8b61d3955118fa1b01eb339ae40fc8dd78 Mon Sep 17 00:00:00 2001
From: Kiran Muddukrishna <kmuddukrishna@squareup.com>
Date: Thu, 6 Aug 2026 14:55:18 +1000
Subject: [PATCH 1/7] refactor(storage): depend on storage.Storage instead of
 *mysqlstore.Storage

Callers only use the interface method set; widening the server field,
buildGRPCTernClient, and test-helper signatures removes the concrete-type
coupling so a second storage backend can be wired without touching callers.
---
 integration/operator_test.go | 10 +++++-----
 integration/workflow_test.go |  3 ++-
 pkg/serve/serve.go           |  4 ++--
 3 files changed, 9 insertions(+), 8 deletions(-)

diff --git a/integration/operator_test.go b/integration/operator_test.go
index 2afe1b19f..3a985b9a6 100644
--- a/integration/operator_test.go
+++ b/integration/operator_test.go
@@ -35,7 +35,7 @@ import (
 type operatorClaimFixture struct {
 	appDBName string
 	storageDB *sql.DB
-	store     *mysqlstore.Storage
+	store     storage.Storage
 }
 
 type blockingResumeClient struct {
@@ -1164,7 +1164,7 @@ func TestOperator_ClaimOrdering(t *testing.T) {
 // seedStartedOperation inserts a running apply_operations row for the apply so
 // claim-policy tests can exercise FindNextApplyOperation against work that
 // looks like a drive already started it.
-func seedStartedOperation(t *testing.T, stor *mysqlstore.Storage, applyID int64, deployment string) int64 {
+func seedStartedOperation(t *testing.T, stor storage.Storage, applyID int64, deployment string) int64 {
 	t.Helper()
 
 	opID, err := stor.ApplyOperations().Insert(t.Context(), &storage.ApplyOperation{
@@ -1457,7 +1457,7 @@ func TestOperator_MultipleWorkersResumeDifferentTargets(t *testing.T) {
 	waitForOperatorAppliesCompleted(t, stor, []int64{apply1ID}, 5*time.Second)
 }
 
-func planCreateTableForOperator(t *testing.T, client tern.Client, stor *mysqlstore.Storage, dbName, tableName string) *storage.Plan {
+func planCreateTableForOperator(t *testing.T, client tern.Client, stor storage.Storage, dbName, tableName string) *storage.Plan {
 	t.Helper()
 
 	resp, err := client.Plan(t.Context(), &ternv1.PlanRequest{
@@ -1489,7 +1489,7 @@ CREATE TABLE %s (
 
 func seedStaleOperatorApply(
 	t *testing.T,
-	stor *mysqlstore.Storage,
+	stor storage.Storage,
 	db *sql.DB,
 	dbName string,
 	plan *storage.Plan,
@@ -1553,7 +1553,7 @@ func seedStaleOperatorApply(
 	return applyID
 }
 
-func waitForOperatorAppliesCompleted(t *testing.T, stor *mysqlstore.Storage, applyIDs []int64, timeout time.Duration) {
+func waitForOperatorAppliesCompleted(t *testing.T, stor storage.Storage, applyIDs []int64, timeout time.Duration) {
 	t.Helper()
 
 	completed := make(map[int64]bool, len(applyIDs))
diff --git a/integration/workflow_test.go b/integration/workflow_test.go
index 46371f65d..921d0659d 100644
--- a/integration/workflow_test.go
+++ b/integration/workflow_test.go
@@ -22,6 +22,7 @@ import (
 
 	schemabotapi "github.com/block/schemabot/pkg/api"
 	"github.com/block/schemabot/pkg/state"
+	"github.com/block/schemabot/pkg/storage"
 	"github.com/block/schemabot/pkg/storage/mysqlstore"
 	"github.com/block/schemabot/pkg/tern"
 )
@@ -33,7 +34,7 @@ import (
 // testServer holds the address and storage of a running SchemaBot HTTP server.
 type testServer struct {
 	Addr    string
-	Storage *mysqlstore.Storage
+	Storage storage.Storage
 	Service *schemabotapi.Service
 }
 
diff --git a/pkg/serve/serve.go b/pkg/serve/serve.go
index 04700ae02..e1ec00762 100644
--- a/pkg/serve/serve.go
+++ b/pkg/serve/serve.go
@@ -234,7 +234,7 @@ func Run(ctx context.Context, cfg *api.ServerConfig, opts ...Option) error {
 type Server struct {
 	cfg             *api.ServerConfig
 	svc             *api.Service
-	storage         *mysqlstore.Storage
+	storage         storage.Storage
 	logger          *slog.Logger
 	dataPlaneClient tern.Client
 	// grpcClient is the single-database client RegisterGRPC builds when no
@@ -598,7 +598,7 @@ func (s *Server) Close() error {
 // environment is unused in this mode because each request carries its own.
 // Otherwise it falls back to a single LocalClient bound to the one database
 // configured for env.
-func buildGRPCTernClient(ctx context.Context, config *api.ServerConfig, st *mysqlstore.Storage, logger *slog.Logger, env string, engineFactories map[string]tern.EngineFactory, wakeOperator ...func(applyIdentifier, database, environment string)) (tern.Client, error) {
+func buildGRPCTernClient(ctx context.Context, config *api.ServerConfig, st storage.Storage, logger *slog.Logger, env string, engineFactories map[string]tern.EngineFactory, wakeOperator ...func(applyIdentifier, database, environment string)) (tern.Client, error) {
 	var wake func(applyIdentifier, database, environment string)
 	if len(wakeOperator) > 0 {
 		wake = wakeOperator[0]

From 35c004e15bcc5cebc79dc8f1a8755f157f14f1f8 Mon Sep 17 00:00:00 2001
From: Kiran Muddukrishna <kmuddukrishna@squareup.com>
Date: Thu, 6 Aug 2026 16:55:43 +1000
Subject: [PATCH 2/7] refactor(storage): address review suggestions on storage
 interface decoupling

Rename locals that shadowed the storage package import and add a
compile-time storage.Storage conformance assertion in mysqlstore.
---
 integration/workflow_test.go      | 8 ++++----
 pkg/serve/serve.go                | 8 ++++----
 pkg/storage/mysqlstore/storage.go | 2 ++
 3 files changed, 10 insertions(+), 8 deletions(-)

diff --git a/integration/workflow_test.go b/integration/workflow_test.go
index 921d0659d..654306d0d 100644
--- a/integration/workflow_test.go
+++ b/integration/workflow_test.go
@@ -76,13 +76,13 @@ func startTestServerWithOperatorInterval(t *testing.T, appDBName, appDSN string,
 	schemabotDB, err := sql.Open("mysql", schemabotDSN)
 	require.NoError(t, err, "open schemabot db")
 	clearStorageDB(t, schemabotDB)
-	storage := mysqlstore.New(schemabotDB)
+	store := mysqlstore.New(schemabotDB)
 
 	localClient, err := tern.NewLocalClient(tern.LocalConfig{
 		Database:  appDBName,
 		Type:      "mysql",
 		TargetDSN: appDSN,
-	}, storage, logger)
+	}, store, logger)
 	require.NoError(t, err, "create local client")
 
 	serverConfig := &schemabotapi.ServerConfig{
@@ -95,7 +95,7 @@ func startTestServerWithOperatorInterval(t *testing.T, appDBName, appDSN string,
 			},
 		},
 	}
-	svc := schemabotapi.New(storage, serverConfig, map[string]tern.Client{
+	svc := schemabotapi.New(store, serverConfig, map[string]tern.Client{
 		appDBName + "/staging": localClient,
 	}, logger)
 	startTestOperatorWithInterval(t, svc, operatorInterval)
@@ -119,7 +119,7 @@ func startTestServerWithOperatorInterval(t *testing.T, appDBName, appDSN string,
 		_ = schemabotDB.Close()
 	})
 
-	return testServer{Addr: addr, Storage: storage, Service: svc}
+	return testServer{Addr: addr, Storage: store, Service: svc}
 }
 
 func startTestOperator(t *testing.T, svc *schemabotapi.Service) {
diff --git a/pkg/serve/serve.go b/pkg/serve/serve.go
index e1ec00762..56ef72024 100644
--- a/pkg/serve/serve.go
+++ b/pkg/serve/serve.go
@@ -322,8 +322,8 @@ func Build(ctx context.Context, cfg *api.ServerConfig, opts ...Option) (*Server,
 	}
 
 	// Create service with dependencies
-	storage := mysqlstore.New(db)
-	svc := api.New(storage, cfg, nil, logger)
+	store := mysqlstore.New(db)
+	svc := api.New(store, cfg, nil, logger)
 	defer func() {
 		if !success {
 			utils.CloseAndLog(svc)
@@ -359,7 +359,7 @@ func Build(ctx context.Context, cfg *api.ServerConfig, opts ...Option) (*Server,
 	// gRPC transport reuses the same instance.
 	var dataPlaneClient tern.Client
 	if cfg.TargetResolver.Enabled() {
-		dataPlaneClient, err = buildGRPCTernClient(ctx, cfg, storage, logger, os.Getenv("TERN_ENVIRONMENT"), o.engines, svc.WakeOperator)
+		dataPlaneClient, err = buildGRPCTernClient(ctx, cfg, store, logger, os.Getenv("TERN_ENVIRONMENT"), o.engines, svc.WakeOperator)
 		if err != nil {
 			return nil, fmt.Errorf("build data-plane target router: %w", err)
 		}
@@ -385,7 +385,7 @@ func Build(ctx context.Context, cfg *api.ServerConfig, opts ...Option) (*Server,
 	return &Server{
 		cfg:             cfg,
 		svc:             svc,
-		storage:         storage,
+		storage:         store,
 		logger:          logger,
 		dataPlaneClient: dataPlaneClient,
 		webhook:         webhookRuntime,
diff --git a/pkg/storage/mysqlstore/storage.go b/pkg/storage/mysqlstore/storage.go
index 880cfa941..cea324318 100644
--- a/pkg/storage/mysqlstore/storage.go
+++ b/pkg/storage/mysqlstore/storage.go
@@ -26,6 +26,8 @@ type Storage struct {
 	webhookEvents   *webhookEventStore
 }
 
+var _ storage.Storage = (*Storage)(nil)
+
 // New creates a new MySQL storage instance.
 func New(db *sql.DB) *Storage {
 	return &Storage{

From 285a27f852f14233897a08e7ad3386ccf3c29b21 Mon Sep 17 00:00:00 2001
From: Kiran Muddukrishna <kmuddukrishna@squareup.com>
Date: Thu, 6 Aug 2026 17:14:19 +1000
Subject: [PATCH 3/7] refactor(storage): move mysqlstore internals to shared
 internal/sqlstore core

Mechanical move (rename-detected) of the store implementation and its
white-box tests; mysqlstore becomes a thin public constructor over the
shared core so a second dialect backend can assemble the same store
logic with its own dependencies. No SQL or behavior changes.
---
 .../sqlstore}/applies.go                      |  2 +-
 .../sqlstore}/applies_test.go                 |  2 +-
 .../sqlstore}/apply_comments.go               |  2 +-
 .../sqlstore}/apply_comments_test.go          |  2 +-
 .../sqlstore}/apply_logs.go                   |  2 +-
 .../sqlstore}/apply_operations.go             |  2 +-
 .../sqlstore}/apply_operations_test.go        |  2 +-
 .../sqlstore}/apply_target_lock_test.go       |  2 +-
 .../sqlstore}/checks.go                       |  2 +-
 .../sqlstore}/checks_test.go                  |  2 +-
 .../sqlstore}/claimable_states_test.go        |  2 +-
 .../sqlstore}/control_requests.go             |  2 +-
 .../sqlstore}/control_requests_test.go        |  2 +-
 .../sqlstore}/dialect.go                      | 17 +++++++--------
 .../sqlstore}/dialect_test.go                 |  2 +-
 .../sqlstore}/identity.go                     |  2 +-
 .../sqlstore}/identity_test.go                |  2 +-
 .../sqlstore}/locks.go                        |  2 +-
 .../sqlstore}/locks_test.go                   |  2 +-
 .../sqlstore}/mysql_test.go                   |  2 +-
 .../sqlstore}/parity_test.go                  |  2 +-
 .../sqlstore}/plan_comments.go                |  2 +-
 .../sqlstore}/plan_comments_test.go           |  2 +-
 .../sqlstore}/plans.go                        |  2 +-
 .../sqlstore}/plans_test.go                   |  2 +-
 .../sqlstore}/retry.go                        |  2 +-
 .../sqlstore}/retry_test.go                   |  2 +-
 .../sqlstore}/settings.go                     |  2 +-
 .../sqlstore}/sql_helpers.go                  |  2 +-
 .../sqlstore}/storage.go                      |  7 +++++--
 .../sqlstore}/tasks.go                        |  2 +-
 .../sqlstore}/tasks_test.go                   |  2 +-
 .../sqlstore}/webhook_events.go               |  2 +-
 .../sqlstore}/webhook_events_test.go          |  2 +-
 pkg/storage/mysqlstore/mysqlstore.go          | 21 +++++++++++++++++++
 35 files changed, 66 insertions(+), 43 deletions(-)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/applies.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/applies_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/apply_comments.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/apply_comments_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/apply_logs.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/apply_operations.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/apply_operations_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/apply_target_lock_test.go (97%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/checks.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/checks_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/claimable_states_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/control_requests.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/control_requests_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/dialect.go (90%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/dialect_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/identity.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/identity_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/locks.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/locks_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/mysql_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/parity_test.go (97%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/plan_comments.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/plan_comments_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/plans.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/plans_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/retry.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/retry_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/settings.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/sql_helpers.go (98%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/storage.go (91%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/tasks.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/tasks_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/webhook_events.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/webhook_events_test.go (99%)
 create mode 100644 pkg/storage/mysqlstore/mysqlstore.go

diff --git a/pkg/storage/mysqlstore/applies.go b/pkg/storage/internal/sqlstore/applies.go
similarity index 99%
rename from pkg/storage/mysqlstore/applies.go
rename to pkg/storage/internal/sqlstore/applies.go
index dacf6f7c9..2ae610127 100644
--- a/pkg/storage/mysqlstore/applies.go
+++ b/pkg/storage/internal/sqlstore/applies.go
@@ -1,6 +1,6 @@
 // applies.go implements ApplyStore for tracking schema change executions.
 // Each apply is a top-level container that holds one or more tasks.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/applies_test.go b/pkg/storage/internal/sqlstore/applies_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/applies_test.go
rename to pkg/storage/internal/sqlstore/applies_test.go
index 2d20e21e4..dc04e464b 100644
--- a/pkg/storage/mysqlstore/applies_test.go
+++ b/pkg/storage/internal/sqlstore/applies_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/apply_comments.go b/pkg/storage/internal/sqlstore/apply_comments.go
similarity index 99%
rename from pkg/storage/mysqlstore/apply_comments.go
rename to pkg/storage/internal/sqlstore/apply_comments.go
index 762766ff4..88733be7a 100644
--- a/pkg/storage/mysqlstore/apply_comments.go
+++ b/pkg/storage/internal/sqlstore/apply_comments.go
@@ -1,6 +1,6 @@
 // apply_comments.go implements ApplyCommentStore for tracking GitHub PR comment IDs.
 // One comment per (apply_id, comment_state) combination.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/apply_comments_test.go b/pkg/storage/internal/sqlstore/apply_comments_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/apply_comments_test.go
rename to pkg/storage/internal/sqlstore/apply_comments_test.go
index 7a1798683..b5ee37931 100644
--- a/pkg/storage/mysqlstore/apply_comments_test.go
+++ b/pkg/storage/internal/sqlstore/apply_comments_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"database/sql"
diff --git a/pkg/storage/mysqlstore/apply_logs.go b/pkg/storage/internal/sqlstore/apply_logs.go
similarity index 99%
rename from pkg/storage/mysqlstore/apply_logs.go
rename to pkg/storage/internal/sqlstore/apply_logs.go
index 44cc358ae..e718e03e1 100644
--- a/pkg/storage/mysqlstore/apply_logs.go
+++ b/pkg/storage/internal/sqlstore/apply_logs.go
@@ -1,6 +1,6 @@
 // apply_logs.go implements ApplyLogStore for audit and debugging log entries.
 // Captures state transitions, errors, and progress events during applies.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/apply_operations.go b/pkg/storage/internal/sqlstore/apply_operations.go
similarity index 99%
rename from pkg/storage/mysqlstore/apply_operations.go
rename to pkg/storage/internal/sqlstore/apply_operations.go
index a5c73159d..69b6daf64 100644
--- a/pkg/storage/mysqlstore/apply_operations.go
+++ b/pkg/storage/internal/sqlstore/apply_operations.go
@@ -1,7 +1,7 @@
 // apply_operations.go implements ApplyOperationStore for per-(apply,
 // deployment, operation_key) child rows under a multi-operation apply — the
 // unit of work the driver claims.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/apply_operations_test.go b/pkg/storage/internal/sqlstore/apply_operations_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/apply_operations_test.go
rename to pkg/storage/internal/sqlstore/apply_operations_test.go
index 4d7a378ac..483a78cdf 100644
--- a/pkg/storage/mysqlstore/apply_operations_test.go
+++ b/pkg/storage/internal/sqlstore/apply_operations_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/apply_target_lock_test.go b/pkg/storage/internal/sqlstore/apply_target_lock_test.go
similarity index 97%
rename from pkg/storage/mysqlstore/apply_target_lock_test.go
rename to pkg/storage/internal/sqlstore/apply_target_lock_test.go
index 17117ea14..4f5f6a5f4 100644
--- a/pkg/storage/mysqlstore/apply_target_lock_test.go
+++ b/pkg/storage/internal/sqlstore/apply_target_lock_test.go
@@ -1,4 +1,4 @@
-package mysqlstore
+package sqlstore
 
 import (
 	"testing"
diff --git a/pkg/storage/mysqlstore/checks.go b/pkg/storage/internal/sqlstore/checks.go
similarity index 99%
rename from pkg/storage/mysqlstore/checks.go
rename to pkg/storage/internal/sqlstore/checks.go
index fcf12ebb3..f7a3e7fe1 100644
--- a/pkg/storage/mysqlstore/checks.go
+++ b/pkg/storage/internal/sqlstore/checks.go
@@ -1,5 +1,5 @@
 // checks.go implements CheckStore for SchemaBot's stored check state.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/checks_test.go b/pkg/storage/internal/sqlstore/checks_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/checks_test.go
rename to pkg/storage/internal/sqlstore/checks_test.go
index c1c9414a8..63776a631 100644
--- a/pkg/storage/mysqlstore/checks_test.go
+++ b/pkg/storage/internal/sqlstore/checks_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"database/sql"
diff --git a/pkg/storage/mysqlstore/claimable_states_test.go b/pkg/storage/internal/sqlstore/claimable_states_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/claimable_states_test.go
rename to pkg/storage/internal/sqlstore/claimable_states_test.go
index e1dcf1190..e67213e21 100644
--- a/pkg/storage/mysqlstore/claimable_states_test.go
+++ b/pkg/storage/internal/sqlstore/claimable_states_test.go
@@ -1,4 +1,4 @@
-package mysqlstore
+package sqlstore
 
 import (
 	"reflect"
diff --git a/pkg/storage/mysqlstore/control_requests.go b/pkg/storage/internal/sqlstore/control_requests.go
similarity index 99%
rename from pkg/storage/mysqlstore/control_requests.go
rename to pkg/storage/internal/sqlstore/control_requests.go
index 5b800117f..bb83b8dcc 100644
--- a/pkg/storage/mysqlstore/control_requests.go
+++ b/pkg/storage/internal/sqlstore/control_requests.go
@@ -1,4 +1,4 @@
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/control_requests_test.go b/pkg/storage/internal/sqlstore/control_requests_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/control_requests_test.go
rename to pkg/storage/internal/sqlstore/control_requests_test.go
index b12e237bf..664d2e0e7 100644
--- a/pkg/storage/mysqlstore/control_requests_test.go
+++ b/pkg/storage/internal/sqlstore/control_requests_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"sync"
diff --git a/pkg/storage/mysqlstore/dialect.go b/pkg/storage/internal/sqlstore/dialect.go
similarity index 90%
rename from pkg/storage/mysqlstore/dialect.go
rename to pkg/storage/internal/sqlstore/dialect.go
index d53dd5ecc..bd1cdd796 100644
--- a/pkg/storage/mysqlstore/dialect.go
+++ b/pkg/storage/internal/sqlstore/dialect.go
@@ -1,4 +1,4 @@
-package mysqlstore
+package sqlstore
 
 import (
 	"fmt"
@@ -8,11 +8,10 @@ import (
 
 // Dialect abstracts the SQL-syntax differences between database families (MySQL
 // and, in the future, Postgres) that the state store depends on. This is an
-// incremental seam: the store lives in mysqlstore and still emits MySQL-style
-// "?" placeholders, and only the family-varying syntax (upsert clause,
-// current-time and relative-time expressions) routes through here. When a
-// Postgres store is introduced this interface likely moves to a shared package,
-// and its parameterized SQL will need a "?"-to-"$n" rebind at the store
+// incremental seam: the store still emits MySQL-style "?" placeholders, and
+// only the family-varying syntax (upsert clause, current-time and
+// relative-time expressions) routes through here. When a Postgres backend is
+// introduced its parameterized SQL will need a "?"-to-"$n" rebind at the store
 // boundary.
 type Dialect interface {
 	// UpsertClause returns the trailing conflict-resolution clause that turns an
@@ -131,7 +130,7 @@ func (MySQLDialect) CurrentTimestamp(precision TimestampPrecision) string {
 	case TimestampPrecisionMicrosecond:
 		return "NOW(6)"
 	default:
-		panic(fmt.Sprintf("mysqlstore: unknown timestamp precision %d", precision))
+		panic(fmt.Sprintf("sqlstore: unknown timestamp precision %d", precision))
 	}
 }
 
@@ -153,7 +152,7 @@ func (d MySQLDialect) RelativeTime(precision TimestampPrecision, direction Relat
 	case AfterCurrentTime:
 		return "DATE_ADD(" + now + ", " + interval + ")"
 	default:
-		panic(fmt.Sprintf("mysqlstore: unknown relative-time direction %d", direction))
+		panic(fmt.Sprintf("sqlstore: unknown relative-time direction %d", direction))
 	}
 }
 
@@ -170,6 +169,6 @@ func mysqlIntervalUnit(unit IntervalUnit) string {
 	case IntervalDay:
 		return "DAY"
 	default:
-		panic(fmt.Sprintf("mysqlstore: unknown interval unit %d", unit))
+		panic(fmt.Sprintf("sqlstore: unknown interval unit %d", unit))
 	}
 }
diff --git a/pkg/storage/mysqlstore/dialect_test.go b/pkg/storage/internal/sqlstore/dialect_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/dialect_test.go
rename to pkg/storage/internal/sqlstore/dialect_test.go
index 636b84b98..a588f3620 100644
--- a/pkg/storage/mysqlstore/dialect_test.go
+++ b/pkg/storage/internal/sqlstore/dialect_test.go
@@ -1,4 +1,4 @@
-package mysqlstore
+package sqlstore
 
 import (
 	"testing"
diff --git a/pkg/storage/mysqlstore/identity.go b/pkg/storage/internal/sqlstore/identity.go
similarity index 99%
rename from pkg/storage/mysqlstore/identity.go
rename to pkg/storage/internal/sqlstore/identity.go
index 9b44cf650..b52a11f9b 100644
--- a/pkg/storage/mysqlstore/identity.go
+++ b/pkg/storage/internal/sqlstore/identity.go
@@ -1,4 +1,4 @@
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/identity_test.go b/pkg/storage/internal/sqlstore/identity_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/identity_test.go
rename to pkg/storage/internal/sqlstore/identity_test.go
index 1b2d94f08..18fc6f1b6 100644
--- a/pkg/storage/mysqlstore/identity_test.go
+++ b/pkg/storage/internal/sqlstore/identity_test.go
@@ -1,4 +1,4 @@
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/locks.go b/pkg/storage/internal/sqlstore/locks.go
similarity index 99%
rename from pkg/storage/mysqlstore/locks.go
rename to pkg/storage/internal/sqlstore/locks.go
index 8c5eb70e9..42a71b12e 100644
--- a/pkg/storage/mysqlstore/locks.go
+++ b/pkg/storage/internal/sqlstore/locks.go
@@ -1,6 +1,6 @@
 // locks.go implements LockStore for database-level deployment locks.
 // Locks prevent concurrent schema changes to the same database.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/locks_test.go b/pkg/storage/internal/sqlstore/locks_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/locks_test.go
rename to pkg/storage/internal/sqlstore/locks_test.go
index 29b484671..1cb87b2f4 100644
--- a/pkg/storage/mysqlstore/locks_test.go
+++ b/pkg/storage/internal/sqlstore/locks_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"database/sql"
diff --git a/pkg/storage/mysqlstore/mysql_test.go b/pkg/storage/internal/sqlstore/mysql_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/mysql_test.go
rename to pkg/storage/internal/sqlstore/mysql_test.go
index 4f4c0e2d2..1b7034d6b 100644
--- a/pkg/storage/mysqlstore/mysql_test.go
+++ b/pkg/storage/internal/sqlstore/mysql_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/parity_test.go b/pkg/storage/internal/sqlstore/parity_test.go
similarity index 97%
rename from pkg/storage/mysqlstore/parity_test.go
rename to pkg/storage/internal/sqlstore/parity_test.go
index 4eee68542..768267fc0 100644
--- a/pkg/storage/mysqlstore/parity_test.go
+++ b/pkg/storage/internal/sqlstore/parity_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"database/sql"
diff --git a/pkg/storage/mysqlstore/plan_comments.go b/pkg/storage/internal/sqlstore/plan_comments.go
similarity index 99%
rename from pkg/storage/mysqlstore/plan_comments.go
rename to pkg/storage/internal/sqlstore/plan_comments.go
index e967d5bee..897b1a8a1 100644
--- a/pkg/storage/mysqlstore/plan_comments.go
+++ b/pkg/storage/internal/sqlstore/plan_comments.go
@@ -1,7 +1,7 @@
 // plan_comments.go implements PlanCommentStore for tracking posted plan
 // comments so a newer plan comment for the same database can minimize the
 // ones it supersedes on GitHub.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/plan_comments_test.go b/pkg/storage/internal/sqlstore/plan_comments_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/plan_comments_test.go
rename to pkg/storage/internal/sqlstore/plan_comments_test.go
index 303bcee27..666e96f31 100644
--- a/pkg/storage/mysqlstore/plan_comments_test.go
+++ b/pkg/storage/internal/sqlstore/plan_comments_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"database/sql"
diff --git a/pkg/storage/mysqlstore/plans.go b/pkg/storage/internal/sqlstore/plans.go
similarity index 99%
rename from pkg/storage/mysqlstore/plans.go
rename to pkg/storage/internal/sqlstore/plans.go
index 802ea1e0a..04a408d44 100644
--- a/pkg/storage/mysqlstore/plans.go
+++ b/pkg/storage/internal/sqlstore/plans.go
@@ -1,6 +1,6 @@
 // plans.go implements PlanStore for schema change plans.
 // Plans are generated by diffing desired vs. live schema and stored for later execution.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/plans_test.go b/pkg/storage/internal/sqlstore/plans_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/plans_test.go
rename to pkg/storage/internal/sqlstore/plans_test.go
index 35aafc0e3..3fc9d29db 100644
--- a/pkg/storage/mysqlstore/plans_test.go
+++ b/pkg/storage/internal/sqlstore/plans_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"testing"
diff --git a/pkg/storage/mysqlstore/retry.go b/pkg/storage/internal/sqlstore/retry.go
similarity index 99%
rename from pkg/storage/mysqlstore/retry.go
rename to pkg/storage/internal/sqlstore/retry.go
index 31ca9134a..d90ca0940 100644
--- a/pkg/storage/mysqlstore/retry.go
+++ b/pkg/storage/internal/sqlstore/retry.go
@@ -3,7 +3,7 @@
 // or time out waiting for a lock; both roll the failed work back — the offending
 // statement, or the whole transaction when the work runs inside one — so the
 // operation can be safely re-run.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/retry_test.go b/pkg/storage/internal/sqlstore/retry_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/retry_test.go
rename to pkg/storage/internal/sqlstore/retry_test.go
index 2a25bf049..a3851e2b1 100644
--- a/pkg/storage/mysqlstore/retry_test.go
+++ b/pkg/storage/internal/sqlstore/retry_test.go
@@ -1,4 +1,4 @@
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/settings.go b/pkg/storage/internal/sqlstore/settings.go
similarity index 99%
rename from pkg/storage/mysqlstore/settings.go
rename to pkg/storage/internal/sqlstore/settings.go
index 509e0f524..d04d0f0a3 100644
--- a/pkg/storage/mysqlstore/settings.go
+++ b/pkg/storage/internal/sqlstore/settings.go
@@ -1,5 +1,5 @@
 // settings.go implements SettingsStore for admin-level runtime configuration.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/sql_helpers.go b/pkg/storage/internal/sqlstore/sql_helpers.go
similarity index 98%
rename from pkg/storage/mysqlstore/sql_helpers.go
rename to pkg/storage/internal/sqlstore/sql_helpers.go
index a5625da5b..3c9b66eef 100644
--- a/pkg/storage/mysqlstore/sql_helpers.go
+++ b/pkg/storage/internal/sqlstore/sql_helpers.go
@@ -1,5 +1,5 @@
 // sql_helpers.go provides shared utilities for MySQL store implementations.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/storage.go b/pkg/storage/internal/sqlstore/storage.go
similarity index 91%
rename from pkg/storage/mysqlstore/storage.go
rename to pkg/storage/internal/sqlstore/storage.go
index cea324318..b9d5b918a 100644
--- a/pkg/storage/mysqlstore/storage.go
+++ b/pkg/storage/internal/sqlstore/storage.go
@@ -1,5 +1,8 @@
-// Package mysql implements the storage interface using MySQL.
-package mysqlstore
+// Package sqlstore implements the storage interface over database/sql. It is
+// the shared dialect-parameterized core behind the public per-dialect
+// constructors (mysqlstore, and in the future postgresstore); it currently
+// emits MySQL-style SQL and is wired with MySQL dependencies.
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/tasks.go b/pkg/storage/internal/sqlstore/tasks.go
similarity index 99%
rename from pkg/storage/mysqlstore/tasks.go
rename to pkg/storage/internal/sqlstore/tasks.go
index 26e772d8b..d823c8df6 100644
--- a/pkg/storage/mysqlstore/tasks.go
+++ b/pkg/storage/internal/sqlstore/tasks.go
@@ -1,6 +1,6 @@
 // tasks.go implements TaskStore for individual DDL operations within an apply.
 // Each task represents one table's schema change.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/tasks_test.go b/pkg/storage/internal/sqlstore/tasks_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/tasks_test.go
rename to pkg/storage/internal/sqlstore/tasks_test.go
index de5689594..744b999d5 100644
--- a/pkg/storage/mysqlstore/tasks_test.go
+++ b/pkg/storage/internal/sqlstore/tasks_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/webhook_events.go b/pkg/storage/internal/sqlstore/webhook_events.go
similarity index 99%
rename from pkg/storage/mysqlstore/webhook_events.go
rename to pkg/storage/internal/sqlstore/webhook_events.go
index 61cc3e4f9..940dc68e6 100644
--- a/pkg/storage/mysqlstore/webhook_events.go
+++ b/pkg/storage/internal/sqlstore/webhook_events.go
@@ -1,5 +1,5 @@
 // webhook_events.go implements WebhookEventStore for durable webhook ingestion.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/webhook_events_test.go b/pkg/storage/internal/sqlstore/webhook_events_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/webhook_events_test.go
rename to pkg/storage/internal/sqlstore/webhook_events_test.go
index 1547eb99c..a9c09e1fe 100644
--- a/pkg/storage/mysqlstore/webhook_events_test.go
+++ b/pkg/storage/internal/sqlstore/webhook_events_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"database/sql"
diff --git a/pkg/storage/mysqlstore/mysqlstore.go b/pkg/storage/mysqlstore/mysqlstore.go
new file mode 100644
index 000000000..039c7e5f2
--- /dev/null
+++ b/pkg/storage/mysqlstore/mysqlstore.go
@@ -0,0 +1,21 @@
+// Package mysqlstore provides the MySQL-backed storage.Storage implementation.
+// The store logic lives in the shared internal sqlstore core; this package is
+// the public constructor that assembles it with MySQL dependencies.
+package mysqlstore
+
+import (
+	"database/sql"
+
+	"github.com/block/schemabot/pkg/storage"
+	"github.com/block/schemabot/pkg/storage/internal/sqlstore"
+)
+
+// Storage is the MySQL-backed storage implementation.
+type Storage = sqlstore.Storage
+
+var _ storage.Storage = (*Storage)(nil)
+
+// New creates a new MySQL storage instance.
+func New(db *sql.DB) *Storage {
+	return sqlstore.New(db)
+}

From 79dc872788d084dba4ba0b977667caa6f2df029c Mon Sep 17 00:00:00 2001
From: Kiran Muddukrishna <kmuddukrishna@squareup.com>
Date: Thu, 6 Aug 2026 19:13:19 +1000
Subject: [PATCH 4/7] feat(storage): route direct pool execution through
 rebind-aware DB wrapper

Stores hold a rebindDB instead of a raw *sql.DB, so every statement
executed directly on the pool passes through the dialect's placeholder
binder exactly once (identity for MySQL). This is the execution seam a
Postgres backend needs to rewrite "?" placeholders to "$n" without
touching store SQL. Transaction and pinned-connection handles remain
raw passthroughs for a follow-up.
---
 pkg/storage/internal/sqlstore/applies.go      |  10 +-
 pkg/storage/internal/sqlstore/applies_test.go |   2 +-
 .../internal/sqlstore/apply_comments.go       |   2 +-
 pkg/storage/internal/sqlstore/apply_logs.go   |   2 +-
 .../internal/sqlstore/apply_operations.go     |   2 +-
 pkg/storage/internal/sqlstore/checks.go       |   2 +-
 .../internal/sqlstore/control_requests.go     |   2 +-
 pkg/storage/internal/sqlstore/db.go           |  78 +++++++++
 pkg/storage/internal/sqlstore/db_test.go      | 151 ++++++++++++++++++
 pkg/storage/internal/sqlstore/locks.go        |   2 +-
 pkg/storage/internal/sqlstore/locks_test.go   |   6 +-
 .../internal/sqlstore/plan_comments.go        |   3 +-
 pkg/storage/internal/sqlstore/plans.go        |   2 +-
 pkg/storage/internal/sqlstore/settings.go     |   2 +-
 pkg/storage/internal/sqlstore/storage.go      |  29 ++--
 pkg/storage/internal/sqlstore/tasks.go        |   2 +-
 .../internal/sqlstore/webhook_events.go       |   2 +-
 17 files changed, 264 insertions(+), 35 deletions(-)
 create mode 100644 pkg/storage/internal/sqlstore/db.go
 create mode 100644 pkg/storage/internal/sqlstore/db_test.go

diff --git a/pkg/storage/internal/sqlstore/applies.go b/pkg/storage/internal/sqlstore/applies.go
index 2ae610127..89ace4796 100644
--- a/pkg/storage/internal/sqlstore/applies.go
+++ b/pkg/storage/internal/sqlstore/applies.go
@@ -54,7 +54,7 @@ const (
 
 // applyStore implements storage.ApplyStore using MySQL.
 type applyStore struct {
-	db       *sql.DB
+	db       *rebindDB
 	dialect  Dialect
 	identity identityInserter
 	// locker serializes concurrent applies against the same target with a
@@ -188,7 +188,7 @@ func beginApplyWriteTx(ctx context.Context, beginner txBeginner, operation strin
 	return &applyWriteTx{tx: tx}, nil
 }
 
-func beginApplyTargetWriteTx(ctx context.Context, db *sql.DB, locker namedlock.Locker, operation, database, dbType, environment string) (*applyWriteTx, error) {
+func beginApplyTargetWriteTx(ctx context.Context, db *rebindDB, locker namedlock.Locker, operation, database, dbType, environment string) (*applyWriteTx, error) {
 	conn, lockName, err := acquireApplyTargetLockConn(ctx, db, locker, database, dbType, environment)
 	if err != nil {
 		return nil, err
@@ -229,7 +229,7 @@ func applyTargetLockName(database, dbType, environment string) string {
 	return "schemabot_apply_" + hex.EncodeToString(sum[:16])
 }
 
-func acquireApplyTargetLockConn(ctx context.Context, db *sql.DB, locker namedlock.Locker, database, dbType, environment string) (*sql.Conn, string, error) {
+func acquireApplyTargetLockConn(ctx context.Context, db *rebindDB, locker namedlock.Locker, database, dbType, environment string) (*sql.Conn, string, error) {
 	if !hasApplyTarget(database, dbType, environment) {
 		return nil, "", fmt.Errorf("active apply target is required for %s/%s/%s", database, dbType, environment)
 	}
@@ -1746,7 +1746,7 @@ func (o claimOutcome) claimedAndComplete() bool {
 // window between the re-check and the commit. When another active apply owns the
 // target the claim is refused and the pending start control request is failed so
 // the operator sees why.
-func persistApplyClaim(ctx context.Context, db *sql.DB, locker namedlock.Locker, tx *sql.Tx, apply *storage.Apply, owner string) (claimOutcome, error) {
+func persistApplyClaim(ctx context.Context, db *rebindDB, locker namedlock.Locker, tx *sql.Tx, apply *storage.Apply, owner string) (claimOutcome, error) {
 	leaseToken := uuid.NewString()
 	leaseAcquiredAt := time.Now()
 	apply.LeaseOwner = owner
@@ -1784,7 +1784,7 @@ func persistApplyClaim(ctx context.Context, db *sql.DB, locker namedlock.Locker,
 // invariant, then either transition+commit or refuse+commit. It commits inside
 // the lock so a concurrent create cannot add a second active apply between the
 // re-check and the commit. The lock is released only after the commit.
-func claimStoppedApplyUnderTargetLock(ctx context.Context, db *sql.DB, locker namedlock.Locker, tx *sql.Tx, apply *storage.Apply, owner, leaseToken string) (claimOutcome, error) {
+func claimStoppedApplyUnderTargetLock(ctx context.Context, db *rebindDB, locker namedlock.Locker, tx *sql.Tx, apply *storage.Apply, owner, leaseToken string) (claimOutcome, error) {
 	database, dbType, environment, deployment, err := applyTargetForUpdate(ctx, tx, apply)
 	if err != nil {
 		return claimLostRace, err
diff --git a/pkg/storage/internal/sqlstore/applies_test.go b/pkg/storage/internal/sqlstore/applies_test.go
index dc04e464b..68a8264bb 100644
--- a/pkg/storage/internal/sqlstore/applies_test.go
+++ b/pkg/storage/internal/sqlstore/applies_test.go
@@ -1025,7 +1025,7 @@ func TestApplyStore_CreateWaitsForApplyTargetLock(t *testing.T) {
 	// Hold the same target lock that the create path must acquire. The creates
 	// below use the public store API; a result before release means active
 	// apply writes are not serialized by the per-target lock.
-	guardConn, guardLockName, err := acquireApplyTargetLockConn(ctx, testDB, namedlock.MySQL{}, "testdb", "mysql", "staging")
+	guardConn, guardLockName, err := acquireApplyTargetLockConn(ctx, newRebindDB(testDB, MySQLDialect{}), namedlock.MySQL{}, "testdb", "mysql", "staging")
 	require.NoError(t, err)
 	releaseGuard := func() {
 		if guardConn == nil {
diff --git a/pkg/storage/internal/sqlstore/apply_comments.go b/pkg/storage/internal/sqlstore/apply_comments.go
index 88733be7a..b3d28d978 100644
--- a/pkg/storage/internal/sqlstore/apply_comments.go
+++ b/pkg/storage/internal/sqlstore/apply_comments.go
@@ -18,7 +18,7 @@ const applyCommentColumns = `id, apply_id, comment_state, github_comment_id, pos
 
 // applyCommentStore implements storage.ApplyCommentStore using MySQL.
 type applyCommentStore struct {
-	db      *sql.DB
+	db      *rebindDB
 	dialect Dialect
 }
 
diff --git a/pkg/storage/internal/sqlstore/apply_logs.go b/pkg/storage/internal/sqlstore/apply_logs.go
index e718e03e1..db2c90576 100644
--- a/pkg/storage/internal/sqlstore/apply_logs.go
+++ b/pkg/storage/internal/sqlstore/apply_logs.go
@@ -18,7 +18,7 @@ const applyLogColumns = `id, apply_id, task_id, level, event_type, source, messa
 
 // applyLogStore implements storage.ApplyLogStore using MySQL.
 type applyLogStore struct {
-	db       *sql.DB
+	db       *rebindDB
 	identity identityInserter
 }
 
diff --git a/pkg/storage/internal/sqlstore/apply_operations.go b/pkg/storage/internal/sqlstore/apply_operations.go
index 69b6daf64..cc3b10053 100644
--- a/pkg/storage/internal/sqlstore/apply_operations.go
+++ b/pkg/storage/internal/sqlstore/apply_operations.go
@@ -27,7 +27,7 @@ const applyOperationColumns = `id, apply_id, deployment, operation_key, operatio
 
 // applyOperationStore implements storage.ApplyOperationStore using MySQL.
 type applyOperationStore struct {
-	db       *sql.DB
+	db       *rebindDB
 	dialect  Dialect
 	identity identityInserter
 	locker   namedlock.Locker
diff --git a/pkg/storage/internal/sqlstore/checks.go b/pkg/storage/internal/sqlstore/checks.go
index f7a3e7fe1..757785051 100644
--- a/pkg/storage/internal/sqlstore/checks.go
+++ b/pkg/storage/internal/sqlstore/checks.go
@@ -24,7 +24,7 @@ const (
 
 // checkStore implements storage.CheckStore using MySQL.
 type checkStore struct {
-	db      *sql.DB
+	db      *rebindDB
 	dialect Dialect
 }
 
diff --git a/pkg/storage/internal/sqlstore/control_requests.go b/pkg/storage/internal/sqlstore/control_requests.go
index bb83b8dcc..0d9cd326b 100644
--- a/pkg/storage/internal/sqlstore/control_requests.go
+++ b/pkg/storage/internal/sqlstore/control_requests.go
@@ -14,7 +14,7 @@ const controlRequestColumns = `id, apply_id, operation, status,
 	requested_by, error_message, metadata, completed_at, created_at, updated_at`
 
 type controlRequestStore struct {
-	db       *sql.DB
+	db       *rebindDB
 	identity identityInserter
 }
 
diff --git a/pkg/storage/internal/sqlstore/db.go b/pkg/storage/internal/sqlstore/db.go
new file mode 100644
index 000000000..4bec372a2
--- /dev/null
+++ b/pkg/storage/internal/sqlstore/db.go
@@ -0,0 +1,78 @@
+package sqlstore
+
+import (
+	"context"
+	"database/sql"
+)
+
+// binder rewrites a statement's parameter placeholders into the wire syntax
+// the target engine expects. The store builds all SQL with MySQL-style "?"
+// placeholders; engines that use a different syntax (Postgres "$n") rebind at
+// the execution boundary, while MySQL's binder is the identity.
+type binder interface {
+	// Rebind returns the query with its placeholders rewritten for the target
+	// engine. It must be a pure string transformation: same input, same output,
+	// no side effects.
+	Rebind(query string) string
+}
+
+// Rebind returns the query unchanged: MySQL consumes the store's native "?"
+// placeholders directly.
+func (MySQLDialect) Rebind(query string) string {
+	return query
+}
+
+// rebindDB wraps the connection pool and owns placeholder rebinding: every
+// statement executed directly on the pool passes through the dialect's binder
+// exactly once before reaching the SQL driver. Stores hold a *rebindDB rather
+// than a raw *sql.DB so no store can bypass the rebind boundary.
+//
+// Transactions and pinned connections obtained via BeginTx / Conn are handed
+// back as raw handles: statements executed on them do not pass through the
+// binder yet, so those paths remain MySQL-placeholder only.
+type rebindDB struct {
+	pool   *sql.DB
+	binder binder
+}
+
+// newRebindDB wraps pool so all direct execution rebinds placeholders with b.
+func newRebindDB(pool *sql.DB, b binder) *rebindDB {
+	return &rebindDB{pool: pool, binder: b}
+}
+
+// ExecContext rebinds the query's placeholders and executes it on the pool.
+func (d *rebindDB) ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) {
+	return d.pool.ExecContext(ctx, d.binder.Rebind(query), args...)
+}
+
+// QueryContext rebinds the query's placeholders and runs it on the pool.
+func (d *rebindDB) QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) {
+	return d.pool.QueryContext(ctx, d.binder.Rebind(query), args...)
+}
+
+// QueryRowContext rebinds the query's placeholders and runs it on the pool.
+func (d *rebindDB) QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row {
+	return d.pool.QueryRowContext(ctx, d.binder.Rebind(query), args...)
+}
+
+// BeginTx starts a transaction on the underlying pool. The returned *sql.Tx is
+// a raw handle: statements executed on it are not rebound.
+func (d *rebindDB) BeginTx(ctx context.Context, opts *sql.TxOptions) (*sql.Tx, error) {
+	return d.pool.BeginTx(ctx, opts)
+}
+
+// Conn pins a single connection from the underlying pool. The returned
+// *sql.Conn is a raw handle: statements executed on it are not rebound.
+func (d *rebindDB) Conn(ctx context.Context) (*sql.Conn, error) {
+	return d.pool.Conn(ctx)
+}
+
+// PingContext verifies the underlying pool's connectivity.
+func (d *rebindDB) PingContext(ctx context.Context) error {
+	return d.pool.PingContext(ctx)
+}
+
+// Close closes the underlying pool.
+func (d *rebindDB) Close() error {
+	return d.pool.Close()
+}
diff --git a/pkg/storage/internal/sqlstore/db_test.go b/pkg/storage/internal/sqlstore/db_test.go
new file mode 100644
index 000000000..b961f7bac
--- /dev/null
+++ b/pkg/storage/internal/sqlstore/db_test.go
@@ -0,0 +1,151 @@
+package sqlstore
+
+import (
+	"context"
+	"database/sql"
+	"database/sql/driver"
+	"io"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// recordingConnector is a database/sql driver stub that records every SQL
+// statement text the pool hands to the driver, so tests can assert exactly
+// what reached the wire after rebinding.
+type recordingConnector struct {
+	queries []string
+	args    [][]driver.Value
+}
+
+func (c *recordingConnector) Connect(context.Context) (driver.Conn, error) { return c, nil }
+func (c *recordingConnector) Driver() driver.Driver                        { return nil }
+
+func (c *recordingConnector) Prepare(query string) (driver.Stmt, error) {
+	c.queries = append(c.queries, query)
+	return &recordingStmt{connector: c}, nil
+}
+func (c *recordingConnector) Close() error              { return nil }
+func (c *recordingConnector) Begin() (driver.Tx, error) { return noopTx{}, nil }
+
+type recordingStmt struct {
+	connector *recordingConnector
+}
+
+func (s *recordingStmt) Close() error  { return nil }
+func (s *recordingStmt) NumInput() int { return -1 }
+
+func (s *recordingStmt) Exec(args []driver.Value) (driver.Result, error) {
+	s.connector.args = append(s.connector.args, args)
+	return driver.RowsAffected(0), nil
+}
+
+func (s *recordingStmt) Query(args []driver.Value) (driver.Rows, error) {
+	s.connector.args = append(s.connector.args, args)
+	return emptyRows{}, nil
+}
+
+type emptyRows struct{}
+
+func (emptyRows) Columns() []string              { return nil }
+func (emptyRows) Close() error                   { return nil }
+func (emptyRows) Next(dest []driver.Value) error { return io.EOF }
+
+type noopTx struct{}
+
+func (noopTx) Commit() error   { return nil }
+func (noopTx) Rollback() error { return nil }
+
+// countingBinder rewrites the query to a recognizable form and counts how many
+// times it was invoked, so tests can prove each execution rebinds exactly once
+// and forwards the rebound SQL downstream.
+type countingBinder struct {
+	calls int
+}
+
+func (b *countingBinder) Rebind(query string) string {
+	b.calls++
+	return query + " /* rebound */"
+}
+
+// newRecordingRebindDB returns a rebindDB whose pool is backed by the
+// recording driver stub, so tests observe the exact SQL reaching the driver.
+func newRecordingRebindDB(t *testing.T, b binder) (*rebindDB, *recordingConnector) {
+	connector := &recordingConnector{}
+	pool := sql.OpenDB(connector)
+	t.Cleanup(func() {
+		require.NoError(t, pool.Close())
+	})
+	return newRebindDB(pool, b), connector
+}
+
+func TestRebindDBExecContextRebindsOnce(t *testing.T) {
+	b := &countingBinder{}
+	rdb, connector := newRecordingRebindDB(t, b)
+
+	_, err := rdb.ExecContext(t.Context(), "UPDATE applies SET state = ? WHERE id = ?", "running", int64(7))
+	require.NoError(t, err)
+
+	assert.Equal(t, 1, b.calls)
+	assert.Equal(t, []string{"UPDATE applies SET state = ? WHERE id = ? /* rebound */"}, connector.queries)
+	assert.Equal(t, [][]driver.Value{{"running", int64(7)}}, connector.args)
+}
+
+func TestRebindDBQueryContextRebindsOnce(t *testing.T) {
+	b := &countingBinder{}
+	rdb, connector := newRecordingRebindDB(t, b)
+
+	rows, err := rdb.QueryContext(t.Context(), "SELECT id FROM applies WHERE state = ?", "running")
+	require.NoError(t, err)
+	require.NoError(t, rows.Close())
+
+	assert.Equal(t, 1, b.calls)
+	assert.Equal(t, []string{"SELECT id FROM applies WHERE state = ? /* rebound */"}, connector.queries)
+	assert.Equal(t, [][]driver.Value{{"running"}}, connector.args)
+}
+
+func TestRebindDBQueryRowContextRebindsOnce(t *testing.T) {
+	b := &countingBinder{}
+	rdb, connector := newRecordingRebindDB(t, b)
+
+	var id int64
+	err := rdb.QueryRowContext(t.Context(), "SELECT id FROM applies WHERE id = ?", int64(7)).Scan(&id)
+	require.ErrorIs(t, err, sql.ErrNoRows)
+
+	assert.Equal(t, 1, b.calls)
+	assert.Equal(t, []string{"SELECT id FROM applies WHERE id = ? /* rebound */"}, connector.queries)
+	assert.Equal(t, [][]driver.Value{{int64(7)}}, connector.args)
+}
+
+// Transactions and pinned connections are raw passthrough handles: obtaining
+// them must not invoke the binder, and statements executed on them reach the
+// driver with the store's native placeholders unchanged.
+func TestRebindDBTransactionAndConnBypassBinder(t *testing.T) {
+	b := &countingBinder{}
+	rdb, connector := newRecordingRebindDB(t, b)
+	ctx := t.Context()
+
+	tx, err := rdb.BeginTx(ctx, nil)
+	require.NoError(t, err)
+	_, err = tx.ExecContext(ctx, "UPDATE applies SET state = ? WHERE id = ?", "stopped", int64(7))
+	require.NoError(t, err)
+	require.NoError(t, tx.Rollback())
+
+	conn, err := rdb.Conn(ctx)
+	require.NoError(t, err)
+	_, err = conn.ExecContext(ctx, "SELECT GET_LOCK(?, ?)", "lock", int64(0))
+	require.NoError(t, err)
+	require.NoError(t, conn.Close())
+
+	assert.Equal(t, 0, b.calls)
+	assert.Equal(t, []string{
+		"UPDATE applies SET state = ? WHERE id = ?",
+		"SELECT GET_LOCK(?, ?)",
+	}, connector.queries)
+}
+
+func TestMySQLDialectRebindIsIdentity(t *testing.T) {
+	query := "SELECT id FROM applies WHERE state = ? AND database_name = ?"
+	assert.Equal(t, query, MySQLDialect{}.Rebind(query))
+}
diff --git a/pkg/storage/internal/sqlstore/locks.go b/pkg/storage/internal/sqlstore/locks.go
index 42a71b12e..1d078cc81 100644
--- a/pkg/storage/internal/sqlstore/locks.go
+++ b/pkg/storage/internal/sqlstore/locks.go
@@ -20,7 +20,7 @@ const lockColumns = `id, database_name, database_type, repository, pull_request,
 
 // lockStore implements storage.LockStore using MySQL.
 type lockStore struct {
-	db *sql.DB
+	db *rebindDB
 }
 
 // Acquire attempts to acquire a lock. Returns ErrLockHeld if held by another owner.
diff --git a/pkg/storage/internal/sqlstore/locks_test.go b/pkg/storage/internal/sqlstore/locks_test.go
index 1cb87b2f4..e3776717e 100644
--- a/pkg/storage/internal/sqlstore/locks_test.go
+++ b/pkg/storage/internal/sqlstore/locks_test.go
@@ -166,7 +166,7 @@ func TestLockStore_Acquire_RefreshSameOwnerValueAlreadyMatches(t *testing.T) {
 		require.NoError(t, db.Close())
 	})
 	require.NoError(t, db.PingContext(ctx))
-	store := &lockStore{db: db}
+	store := &lockStore{db: newRebindDB(db, MySQLDialect{})}
 
 	require.NoError(t, store.Acquire(ctx, &storage.Lock{
 		DatabaseName:  "testdb",
@@ -227,7 +227,7 @@ func TestLockStore_Acquire_RefreshSameOwnerValueAlreadyMatches(t *testing.T) {
 func TestLockStore_Acquire_RefreshOwnerNoLongerMatches(t *testing.T) {
 	clearTables(t)
 	ctx := t.Context()
-	store := &lockStore{db: testDB}
+	store := &lockStore{db: newRebindDB(testDB, MySQLDialect{})}
 
 	require.NoError(t, store.Acquire(ctx, &storage.Lock{
 		DatabaseName:  "testdb",
@@ -554,7 +554,7 @@ func TestLockStore_UpdateSameSecondSucceeds(t *testing.T) {
 	_, err = db.ExecContext(ctx, "SET TIMESTAMP = 1700000000")
 	require.NoError(t, err)
 
-	store := &lockStore{db: db}
+	store := &lockStore{db: newRebindDB(db, MySQLDialect{})}
 
 	// 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/plan_comments.go b/pkg/storage/internal/sqlstore/plan_comments.go
index 897b1a8a1..cccfbfa2a 100644
--- a/pkg/storage/internal/sqlstore/plan_comments.go
+++ b/pkg/storage/internal/sqlstore/plan_comments.go
@@ -5,7 +5,6 @@ package sqlstore
 
 import (
 	"context"
-	"database/sql"
 	"fmt"
 
 	"github.com/block/schemabot/pkg/storage"
@@ -17,7 +16,7 @@ const planCommentColumns = `id, repository, pull_request, database_name, databas
 
 // planCommentStore implements storage.PlanCommentStore using MySQL.
 type planCommentStore struct {
-	db       *sql.DB
+	db       *rebindDB
 	identity identityInserter
 }
 
diff --git a/pkg/storage/internal/sqlstore/plans.go b/pkg/storage/internal/sqlstore/plans.go
index 04a408d44..929cde448 100644
--- a/pkg/storage/internal/sqlstore/plans.go
+++ b/pkg/storage/internal/sqlstore/plans.go
@@ -20,7 +20,7 @@ const planColumns = `id, plan_identifier, database_name, database_type,
 
 // planStore implements storage.PlanStore using MySQL.
 type planStore struct {
-	db       *sql.DB
+	db       *rebindDB
 	identity identityInserter
 }
 
diff --git a/pkg/storage/internal/sqlstore/settings.go b/pkg/storage/internal/sqlstore/settings.go
index d04d0f0a3..7f3edd2a3 100644
--- a/pkg/storage/internal/sqlstore/settings.go
+++ b/pkg/storage/internal/sqlstore/settings.go
@@ -15,7 +15,7 @@ const settingColumns = `id, setting_key, setting_value, created_at, updated_at`
 
 // settingsStore implements storage.SettingsStore using MySQL.
 type settingsStore struct {
-	db      *sql.DB
+	db      *rebindDB
 	dialect Dialect
 }
 
diff --git a/pkg/storage/internal/sqlstore/storage.go b/pkg/storage/internal/sqlstore/storage.go
index b9d5b918a..5c2797a1b 100644
--- a/pkg/storage/internal/sqlstore/storage.go
+++ b/pkg/storage/internal/sqlstore/storage.go
@@ -14,7 +14,7 @@ import (
 
 // Storage implements the storage.Storage interface using MySQL.
 type Storage struct {
-	db              *sql.DB
+	db              *rebindDB
 	locks           *lockStore
 	plans           *planStore
 	applies         *applyStore
@@ -33,20 +33,21 @@ var _ storage.Storage = (*Storage)(nil)
 
 // New creates a new MySQL storage instance.
 func New(db *sql.DB) *Storage {
+	rdb := newRebindDB(db, MySQLDialect{})
 	return &Storage{
-		db:              db,
-		locks:           &lockStore{db: db},
-		plans:           &planStore{db: db, identity: MySQLDialect{}},
-		applies:         &applyStore{db: db, dialect: MySQLDialect{}, identity: MySQLDialect{}, locker: namedlock.MySQL{}},
-		tasks:           &taskStore{db: db, identity: MySQLDialect{}},
-		applyLogs:       &applyLogStore{db: db, identity: MySQLDialect{}},
-		controlRequests: &controlRequestStore{db: db, identity: MySQLDialect{}},
-		applyComments:   &applyCommentStore{db: db, dialect: MySQLDialect{}},
-		planComments:    &planCommentStore{db: db, identity: MySQLDialect{}},
-		applyOperations: &applyOperationStore{db: db, dialect: MySQLDialect{}, identity: MySQLDialect{}, locker: namedlock.MySQL{}},
-		checks:          &checkStore{db: db, dialect: MySQLDialect{}},
-		settings:        &settingsStore{db: db, dialect: MySQLDialect{}},
-		webhookEvents:   &webhookEventStore{db: db, dialect: MySQLDialect{}, identity: MySQLDialect{}},
+		db:              rdb,
+		locks:           &lockStore{db: rdb},
+		plans:           &planStore{db: rdb, identity: MySQLDialect{}},
+		applies:         &applyStore{db: rdb, dialect: MySQLDialect{}, identity: MySQLDialect{}, locker: namedlock.MySQL{}},
+		tasks:           &taskStore{db: rdb, identity: MySQLDialect{}},
+		applyLogs:       &applyLogStore{db: rdb, identity: MySQLDialect{}},
+		controlRequests: &controlRequestStore{db: rdb, identity: MySQLDialect{}},
+		applyComments:   &applyCommentStore{db: rdb, dialect: MySQLDialect{}},
+		planComments:    &planCommentStore{db: rdb, identity: MySQLDialect{}},
+		applyOperations: &applyOperationStore{db: rdb, dialect: MySQLDialect{}, identity: MySQLDialect{}, locker: namedlock.MySQL{}},
+		checks:          &checkStore{db: rdb, dialect: MySQLDialect{}},
+		settings:        &settingsStore{db: rdb, dialect: MySQLDialect{}},
+		webhookEvents:   &webhookEventStore{db: rdb, dialect: MySQLDialect{}, identity: MySQLDialect{}},
 	}
 }
 
diff --git a/pkg/storage/internal/sqlstore/tasks.go b/pkg/storage/internal/sqlstore/tasks.go
index d823c8df6..660fae083 100644
--- a/pkg/storage/internal/sqlstore/tasks.go
+++ b/pkg/storage/internal/sqlstore/tasks.go
@@ -41,7 +41,7 @@ var terminalTaskStatesSQL = func() string {
 
 // taskStore implements storage.TaskStore using MySQL.
 type taskStore struct {
-	db       *sql.DB
+	db       *rebindDB
 	identity identityInserter
 }
 
diff --git a/pkg/storage/internal/sqlstore/webhook_events.go b/pkg/storage/internal/sqlstore/webhook_events.go
index 940dc68e6..7e7d06806 100644
--- a/pkg/storage/internal/sqlstore/webhook_events.go
+++ b/pkg/storage/internal/sqlstore/webhook_events.go
@@ -20,7 +20,7 @@ const webhookEventColumns = `id, provider, delivery_id, event, action, repositor
 	received_at, started_at, completed_at, created_at, updated_at`
 
 type webhookEventStore struct {
-	db       *sql.DB
+	db       *rebindDB
 	dialect  Dialect
 	identity identityInserter
 }

From 4a76fb1810ec2b02db3ada436c3d4a05baaa8886 Mon Sep 17 00:00:00 2001
From: Kiran Muddukrishna <kmuddukrishna@squareup.com>
Date: Thu, 6 Aug 2026 19:58:14 +1000
Subject: [PATCH 5/7] feat(storage): rebind transaction and pinned-connection
 SQL exactly once

Extend the rebind boundary from direct pool execution to transactions
and pinned connections: BeginTx/Conn now return rebind-aware wrappers,
so every store statement rebinds its placeholders exactly once at
execution time. The advisory-lock flow keeps a sanctioned raw()
escape because namedlock.Locker emits engine-native SQL on *sql.Conn.
---
 pkg/storage/internal/sqlstore/applies.go      |  46 ++++----
 .../internal/sqlstore/apply_operations.go     |   4 +-
 .../sqlstore/apply_operations_test.go         |   4 +-
 .../internal/sqlstore/control_requests.go     |   4 +-
 pkg/storage/internal/sqlstore/db.go           | 105 ++++++++++++++++--
 pkg/storage/internal/sqlstore/db_test.go      |  93 ++++++++++++++--
 pkg/storage/internal/sqlstore/identity.go     |   9 +-
 pkg/storage/internal/sqlstore/sql_helpers.go  |   2 +-
 8 files changed, 210 insertions(+), 57 deletions(-)

diff --git a/pkg/storage/internal/sqlstore/applies.go b/pkg/storage/internal/sqlstore/applies.go
index 89ace4796..120ebf9c8 100644
--- a/pkg/storage/internal/sqlstore/applies.go
+++ b/pkg/storage/internal/sqlstore/applies.go
@@ -64,9 +64,9 @@ type applyStore struct {
 }
 
 type applyWriteTx struct {
-	tx             *sql.Tx
+	tx             *rebindTx
 	locker         namedlock.Locker
-	targetLockConn *sql.Conn
+	targetLockConn *rebindConn
 	targetLockName string
 }
 
@@ -75,7 +75,7 @@ type queryRower interface {
 }
 
 type txBeginner interface {
-	BeginTx(ctx context.Context, opts *sql.TxOptions) (*sql.Tx, error)
+	BeginTx(ctx context.Context, opts *sql.TxOptions) (*rebindTx, error)
 }
 
 // claimableApplyStates returns active apply states where operator recovery can
@@ -229,7 +229,7 @@ func applyTargetLockName(database, dbType, environment string) string {
 	return "schemabot_apply_" + hex.EncodeToString(sum[:16])
 }
 
-func acquireApplyTargetLockConn(ctx context.Context, db *rebindDB, locker namedlock.Locker, database, dbType, environment string) (*sql.Conn, string, error) {
+func acquireApplyTargetLockConn(ctx context.Context, db *rebindDB, locker namedlock.Locker, database, dbType, environment string) (*rebindConn, string, error) {
 	if !hasApplyTarget(database, dbType, environment) {
 		return nil, "", fmt.Errorf("active apply target is required for %s/%s/%s", database, dbType, environment)
 	}
@@ -243,7 +243,7 @@ func acquireApplyTargetLockConn(ctx context.Context, db *rebindDB, locker namedl
 	}
 
 	lockName := applyTargetLockName(database, dbType, environment)
-	acquired, err := locker.Acquire(ctx, conn, lockName, applyTargetLockWait)
+	acquired, err := locker.Acquire(ctx, conn.raw(), lockName, applyTargetLockWait)
 	if err != nil {
 		slog.WarnContext(ctx, "failed to acquire apply target lock",
 			"database", database,
@@ -268,7 +268,7 @@ func acquireApplyTargetLockConn(ctx context.Context, db *rebindDB, locker namedl
 	return conn, lockName, nil
 }
 
-func releaseApplyTargetLockConn(ctx context.Context, locker namedlock.Locker, conn *sql.Conn, lockName, operation string) {
+func releaseApplyTargetLockConn(ctx context.Context, locker namedlock.Locker, conn *rebindConn, lockName, operation string) {
 	if conn == nil {
 		return
 	}
@@ -276,7 +276,7 @@ func releaseApplyTargetLockConn(ctx context.Context, locker namedlock.Locker, co
 	releaseCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), applyTargetLockReleaseTimeout)
 	defer cancel()
 
-	released, err := locker.Release(releaseCtx, conn, lockName)
+	released, err := locker.Release(releaseCtx, conn.raw(), lockName)
 	switch {
 	case err != nil:
 		slog.WarnContext(releaseCtx, "failed to release apply target lock; discarding connection",
@@ -297,7 +297,7 @@ func releaseApplyTargetLockConn(ctx context.Context, locker namedlock.Locker, co
 // discardApplyTargetLockConn marks the pinned connection as bad so the pool
 // destroys it instead of reusing a session whose advisory-lock state is
 // uncertain.
-func discardApplyTargetLockConn(ctx context.Context, conn *sql.Conn, lockName, operation string) {
+func discardApplyTargetLockConn(ctx context.Context, conn *rebindConn, lockName, operation string) {
 	if rawErr := conn.Raw(func(any) error { return driver.ErrBadConn }); rawErr != nil && !errors.Is(rawErr, driver.ErrBadConn) {
 		slog.WarnContext(ctx, "failed to discard apply target lock connection",
 			"operation", operation,
@@ -306,7 +306,7 @@ func discardApplyTargetLockConn(ctx context.Context, conn *sql.Conn, lockName, o
 	}
 }
 
-func closeApplyTargetLockConn(ctx context.Context, conn *sql.Conn, lockName, operation string) {
+func closeApplyTargetLockConn(ctx context.Context, conn *rebindConn, lockName, operation string) {
 	if err := conn.Close(); err != nil {
 		slog.WarnContext(ctx, "failed to close apply target lock connection",
 			"operation", operation,
@@ -339,7 +339,7 @@ func dedupeDeployments(deployments []string) []string {
 // apply_operations rows. It is used to build the full target set of an existing
 // apply being started/resumed/reclaimed so the overlap check covers every
 // deployment the apply owns, not just the parent's primary deployment.
-func operationDeploymentsForApply(ctx context.Context, tx *sql.Tx, applyID int64) ([]string, error) {
+func operationDeploymentsForApply(ctx context.Context, tx *rebindTx, applyID int64) ([]string, error) {
 	rows, err := tx.QueryContext(ctx, `SELECT deployment FROM apply_operations WHERE apply_id = ?`, applyID)
 	if err != nil {
 		return nil, fmt.Errorf("list operation deployments for apply %d: %w", applyID, err)
@@ -374,7 +374,7 @@ func operationDeploymentsForApply(ctx context.Context, tx *sql.Tx, applyID int64
 // is matched against both the parent applies.deployment (the primary) and the
 // apply_operations.deployment rows, so single-operation applies (where the two
 // are equal) behave exactly as before.
-func checkNoActiveApplyForTargets(ctx context.Context, tx *sql.Tx, database, dbType, environment string, deployments []string, excludeApplyID int64) error {
+func checkNoActiveApplyForTargets(ctx context.Context, tx *rebindTx, database, dbType, environment string, deployments []string, excludeApplyID int64) error {
 	deployments = dedupeDeployments(deployments)
 	if len(deployments) == 0 {
 		return fmt.Errorf("active apply target requires at least one deployment for %s/%s/%s", database, dbType, environment)
@@ -569,7 +569,7 @@ func (s *applyStore) CreateWithTasks(ctx context.Context, apply *storage.Apply,
 	return s.CreateWithTasksAndOperations(ctx, apply, tasks, nil)
 }
 
-type applyCreateWriter func(ctx context.Context, tx *sql.Tx, apply *storage.Apply, applyID int64) error
+type applyCreateWriter func(ctx context.Context, tx *rebindTx, apply *storage.Apply, applyID int64) error
 
 // CreateWithTasksAndOperations is the unified atomic apply-create path: it
 // inserts the applies row, the initial tasks, and (optionally) the
@@ -582,7 +582,7 @@ func (s *applyStore) CreateWithTasksAndOperations(ctx context.Context, apply *st
 	for _, op := range operations {
 		deployments = append(deployments, op.Deployment)
 	}
-	return s.createWithRows(ctx, apply, opName, deployments, func(ctx context.Context, tx *sql.Tx, apply *storage.Apply, applyID int64) error {
+	return s.createWithRows(ctx, apply, opName, deployments, func(ctx context.Context, tx *rebindTx, apply *storage.Apply, applyID int64) error {
 		return insertApplyTasksAndOperations(ctx, tx, s.identity, apply, applyID, tasks, operations)
 	})
 }
@@ -596,7 +596,7 @@ func (s *applyStore) CreateWithGroupedOperations(ctx context.Context, apply *sto
 			deployments = append(deployments, group.Operation.Deployment)
 		}
 	}
-	return s.createWithRows(ctx, apply, opName, deployments, func(ctx context.Context, tx *sql.Tx, apply *storage.Apply, applyID int64) error {
+	return s.createWithRows(ctx, apply, opName, deployments, func(ctx context.Context, tx *rebindTx, apply *storage.Apply, applyID int64) error {
 		return insertApplyGroupedOperations(ctx, tx, s.identity, apply, applyID, groups)
 	})
 }
@@ -666,7 +666,7 @@ func (s *applyStore) createWithRows(ctx context.Context, apply *storage.Apply, o
 // same transaction that inserts the apply. This closes the gap where a
 // same-owner rollback can replace pending_plan_id after freshness validation
 // but before the forward apply becomes durable.
-func verifyExpectedLockIntent(ctx context.Context, tx *sql.Tx, apply *storage.Apply) error {
+func verifyExpectedLockIntent(ctx context.Context, tx *rebindTx, apply *storage.Apply) error {
 	if apply.ExpectedLockOwner == "" {
 		if apply.ExpectedPendingPlanID != "" {
 			return fmt.Errorf("verify lock intent for %s/%s: expected pending plan ID set without an expected lock owner", apply.Database, apply.DatabaseType)
@@ -695,7 +695,7 @@ func verifyExpectedLockIntent(ctx context.Context, tx *sql.Tx, apply *storage.Ap
 	return nil
 }
 
-func insertApplyTasksAndOperations(ctx context.Context, tx *sql.Tx, identity identityInserter, apply *storage.Apply, applyID int64, tasks []*storage.Task, operations []*storage.ApplyOperation) error {
+func insertApplyTasksAndOperations(ctx context.Context, tx *rebindTx, identity identityInserter, apply *storage.Apply, applyID int64, tasks []*storage.Task, operations []*storage.ApplyOperation) error {
 	// Operations are inserted BEFORE tasks so each task can be persisted
 	// with the apply_operation_id of the operation it belongs to. Today
 	// the config layer hard-blocks multi-entry deployments so there is
@@ -771,7 +771,7 @@ func insertApplyTasksAndOperations(ctx context.Context, tx *sql.Tx, identity ide
 	return nil
 }
 
-func insertApplyGroupedOperations(ctx context.Context, tx *sql.Tx, identity identityInserter, apply *storage.Apply, applyID int64, groups []*storage.ApplyOperationWithTasks) error {
+func insertApplyGroupedOperations(ctx context.Context, tx *rebindTx, identity identityInserter, apply *storage.Apply, applyID int64, groups []*storage.ApplyOperationWithTasks) error {
 	if len(groups) == 0 {
 		return fmt.Errorf("create apply %s: grouped operations are empty", apply.ApplyIdentifier)
 	}
@@ -1746,7 +1746,7 @@ func (o claimOutcome) claimedAndComplete() bool {
 // window between the re-check and the commit. When another active apply owns the
 // target the claim is refused and the pending start control request is failed so
 // the operator sees why.
-func persistApplyClaim(ctx context.Context, db *rebindDB, locker namedlock.Locker, tx *sql.Tx, apply *storage.Apply, owner string) (claimOutcome, error) {
+func persistApplyClaim(ctx context.Context, db *rebindDB, locker namedlock.Locker, tx *rebindTx, apply *storage.Apply, owner string) (claimOutcome, error) {
 	leaseToken := uuid.NewString()
 	leaseAcquiredAt := time.Now()
 	apply.LeaseOwner = owner
@@ -1784,7 +1784,7 @@ func persistApplyClaim(ctx context.Context, db *rebindDB, locker namedlock.Locke
 // invariant, then either transition+commit or refuse+commit. It commits inside
 // the lock so a concurrent create cannot add a second active apply between the
 // re-check and the commit. The lock is released only after the commit.
-func claimStoppedApplyUnderTargetLock(ctx context.Context, db *rebindDB, locker namedlock.Locker, tx *sql.Tx, apply *storage.Apply, owner, leaseToken string) (claimOutcome, error) {
+func claimStoppedApplyUnderTargetLock(ctx context.Context, db *rebindDB, locker namedlock.Locker, tx *rebindTx, apply *storage.Apply, owner, leaseToken string) (claimOutcome, error) {
 	database, dbType, environment, deployment, err := applyTargetForUpdate(ctx, tx, apply)
 	if err != nil {
 		return claimLostRace, err
@@ -1839,7 +1839,7 @@ func isStartingClaim(applyState string) bool {
 // landing between the SELECT and this UPDATE; a zero rows-affected result means
 // another driver already moved the row, so the caller backs off cleanly. Reports
 // false on that lost race.
-func transitionClaimToState(ctx context.Context, tx *sql.Tx, apply *storage.Apply, targetState, owner, leaseToken string) (bool, error) {
+func transitionClaimToState(ctx context.Context, tx *rebindTx, apply *storage.Apply, targetState, owner, leaseToken string) (bool, error) {
 	result, err := tx.ExecContext(ctx, `
 		UPDATE applies
 		SET state = ?, updated_at = NOW(),
@@ -1871,7 +1871,7 @@ func transitionClaimToState(ctx context.Context, tx *sql.Tx, apply *storage.Appl
 // target, and commits that failure inside tx. The start was accepted but cannot
 // be honored, so leaving the request pending would silently strand it; failing
 // it surfaces the reason to the operator. Returns claimRefusedActiveTarget.
-func refuseStoppedClaimForActiveTarget(ctx context.Context, tx *sql.Tx, apply *storage.Apply, owner, database, dbType, environment string) (claimOutcome, error) {
+func refuseStoppedClaimForActiveTarget(ctx context.Context, tx *rebindTx, apply *storage.Apply, owner, database, dbType, environment string) (claimOutcome, error) {
 	reason := fmt.Sprintf("start refused: another active apply exists for %s/%s/%s", database, dbType, environment)
 	if err := failPendingStartControlRequestTx(ctx, tx, apply.ID, reason); err != nil {
 		return claimLostRace, fmt.Errorf("fail pending start control request for apply %d (%s) on %s/%s/%s: %w", apply.ID, apply.ApplyIdentifier, database, dbType, environment, err)
@@ -1893,7 +1893,7 @@ func refuseStoppedClaimForActiveTarget(ctx context.Context, tx *sql.Tx, apply *s
 // an apply failed inside the supplied claim transaction so the refusal and the
 // failed request commit atomically. Mirrors controlRequestStore.FailPending's
 // SQL; this path holds no apply lease, so it is unguarded by lease token.
-func failPendingStartControlRequestTx(ctx context.Context, tx *sql.Tx, applyID int64, reason string) error {
+func failPendingStartControlRequestTx(ctx context.Context, tx *rebindTx, applyID int64, reason string) error {
 	_, err := tx.ExecContext(ctx, `
 		UPDATE apply_control_requests
 		SET status = ?, error_message = ?, completed_at = COALESCE(completed_at, NOW()), updated_at = NOW()
@@ -2199,7 +2199,7 @@ func (s *applyStore) ReapplyFailed(ctx context.Context, applyID int64) (*storage
 	return apply, nil
 }
 
-func rejectNonReappliableOperations(ctx context.Context, tx *sql.Tx, apply *storage.Apply) error {
+func rejectNonReappliableOperations(ctx context.Context, tx *rebindTx, apply *storage.Apply) error {
 	var count int
 	if err := tx.QueryRowContext(ctx, `
 		SELECT COUNT(*)
diff --git a/pkg/storage/internal/sqlstore/apply_operations.go b/pkg/storage/internal/sqlstore/apply_operations.go
index cc3b10053..db3b51b67 100644
--- a/pkg/storage/internal/sqlstore/apply_operations.go
+++ b/pkg/storage/internal/sqlstore/apply_operations.go
@@ -1563,7 +1563,7 @@ func (s *applyOperationStore) ReapStranded(ctx context.Context, limit int) ([]*s
 
 	// Do not wait for the lock: whoever holds it is doing this pass's work, and
 	// this instance's next tick is soon enough.
-	acquired, err := s.locker.Acquire(ctx, conn, strandedReaperLockName, 0)
+	acquired, err := s.locker.Acquire(ctx, conn.raw(), strandedReaperLockName, 0)
 	if err != nil {
 		return nil, fmt.Errorf("acquire stranded reaper lock: %w", err)
 	}
@@ -1574,7 +1574,7 @@ func (s *applyOperationStore) ReapStranded(ctx context.Context, limit int) ([]*s
 		// A held lock parks every instance's reaper until this session is
 		// retired, so the two ways it can survive the pass are reported apart:
 		// the release errored, or it ran and reported the lock was not held.
-		released, err := s.locker.Release(context.WithoutCancel(ctx), conn, strandedReaperLockName)
+		released, err := s.locker.Release(context.WithoutCancel(ctx), conn.raw(), strandedReaperLockName)
 		if err != nil {
 			slog.WarnContext(ctx, "failed to release the stranded reaper lock; reapers stay blocked until this session is retired",
 				"lock", strandedReaperLockName, "error", err)
diff --git a/pkg/storage/internal/sqlstore/apply_operations_test.go b/pkg/storage/internal/sqlstore/apply_operations_test.go
index 483a78cdf..cf7dd02a7 100644
--- a/pkg/storage/internal/sqlstore/apply_operations_test.go
+++ b/pkg/storage/internal/sqlstore/apply_operations_test.go
@@ -4847,7 +4847,7 @@ func TestApplyOperationStore_ReapStranded_ElectsOneReaperPerPass(t *testing.T) {
 	other := New(testDB)
 	held, err := other.applyOperations.db.Conn(ctx)
 	require.NoError(t, err)
-	acquired, err := namedlock.MySQL{}.Acquire(ctx, held, strandedReaperLockName, 0)
+	acquired, err := namedlock.MySQL{}.Acquire(ctx, held.raw(), strandedReaperLockName, 0)
 	require.NoError(t, err)
 	require.True(t, acquired, "the stand-in instance should take the reaper lock")
 
@@ -4855,7 +4855,7 @@ func TestApplyOperationStore_ReapStranded_ElectsOneReaperPerPass(t *testing.T) {
 	require.ErrorIs(t, err, storage.ErrStrandedReaperBusy, "a second reaper steps aside rather than re-scanning")
 	assertApplyOperationState(t, store, op.ID, state.ApplyOperation.Pending)
 
-	released, err := namedlock.MySQL{}.Release(ctx, held, strandedReaperLockName)
+	released, err := namedlock.MySQL{}.Release(ctx, held.raw(), strandedReaperLockName)
 	require.NoError(t, err)
 	require.True(t, released)
 	require.NoError(t, held.Close())
diff --git a/pkg/storage/internal/sqlstore/control_requests.go b/pkg/storage/internal/sqlstore/control_requests.go
index 0d9cd326b..abaee4b98 100644
--- a/pkg/storage/internal/sqlstore/control_requests.go
+++ b/pkg/storage/internal/sqlstore/control_requests.go
@@ -193,7 +193,7 @@ func (s *controlRequestStore) FailPending(ctx context.Context, applyID int64, op
 	return nil
 }
 
-func (s *controlRequestStore) getByIDForUpdate(ctx context.Context, tx *sql.Tx, id int64) (*storage.ApplyControlRequest, error) {
+func (s *controlRequestStore) getByIDForUpdate(ctx context.Context, tx *rebindTx, id int64) (*storage.ApplyControlRequest, error) {
 	row := tx.QueryRowContext(ctx, `
 		SELECT `+controlRequestColumns+`
 		FROM apply_control_requests
@@ -209,7 +209,7 @@ func (s *controlRequestStore) getByIDForUpdate(ctx context.Context, tx *sql.Tx,
 
 func (s *controlRequestStore) getByApplyOperationForUpdate(
 	ctx context.Context,
-	tx *sql.Tx,
+	tx *rebindTx,
 	applyID int64,
 	operation storage.ControlOperation,
 ) (*storage.ApplyControlRequest, error) {
diff --git a/pkg/storage/internal/sqlstore/db.go b/pkg/storage/internal/sqlstore/db.go
index 4bec372a2..10745ded9 100644
--- a/pkg/storage/internal/sqlstore/db.go
+++ b/pkg/storage/internal/sqlstore/db.go
@@ -27,9 +27,10 @@ func (MySQLDialect) Rebind(query string) string {
 // exactly once before reaching the SQL driver. Stores hold a *rebindDB rather
 // than a raw *sql.DB so no store can bypass the rebind boundary.
 //
-// Transactions and pinned connections obtained via BeginTx / Conn are handed
-// back as raw handles: statements executed on them do not pass through the
-// binder yet, so those paths remain MySQL-placeholder only.
+// Transactions and pinned connections obtained via BeginTx / Conn are wrapped
+// the same way, so every execution path — pool, transaction, or pinned
+// connection — rebinds exactly once on the final assembled SQL. The only
+// sanctioned escape is rebindConn.raw() for the advisory-lock boundary.
 type rebindDB struct {
 	pool   *sql.DB
 	binder binder
@@ -55,16 +56,24 @@ func (d *rebindDB) QueryRowContext(ctx context.Context, query string, args ...an
 	return d.pool.QueryRowContext(ctx, d.binder.Rebind(query), args...)
 }
 
-// BeginTx starts a transaction on the underlying pool. The returned *sql.Tx is
-// a raw handle: statements executed on it are not rebound.
-func (d *rebindDB) BeginTx(ctx context.Context, opts *sql.TxOptions) (*sql.Tx, error) {
-	return d.pool.BeginTx(ctx, opts)
+// BeginTx starts a transaction on the underlying pool, wrapped so statements
+// executed on it rebind their placeholders like direct pool execution.
+func (d *rebindDB) BeginTx(ctx context.Context, opts *sql.TxOptions) (*rebindTx, error) {
+	tx, err := d.pool.BeginTx(ctx, opts)
+	if err != nil {
+		return nil, err
+	}
+	return &rebindTx{tx: tx, binder: d.binder}, nil
 }
 
-// Conn pins a single connection from the underlying pool. The returned
-// *sql.Conn is a raw handle: statements executed on it are not rebound.
-func (d *rebindDB) Conn(ctx context.Context) (*sql.Conn, error) {
-	return d.pool.Conn(ctx)
+// Conn pins a single connection from the underlying pool, wrapped so
+// transactions begun on it rebind their placeholders.
+func (d *rebindDB) Conn(ctx context.Context) (*rebindConn, error) {
+	conn, err := d.pool.Conn(ctx)
+	if err != nil {
+		return nil, err
+	}
+	return &rebindConn{conn: conn, binder: d.binder}, nil
 }
 
 // PingContext verifies the underlying pool's connectivity.
@@ -76,3 +85,77 @@ func (d *rebindDB) PingContext(ctx context.Context) error {
 func (d *rebindDB) Close() error {
 	return d.pool.Close()
 }
+
+// rebindTx wraps an in-flight transaction so every statement executed on it
+// passes through the dialect's binder exactly once, matching direct pool
+// execution. Statements must reach the binder as final assembled SQL: a query
+// is rebound at the moment it executes, never earlier and never twice.
+type rebindTx struct {
+	tx     *sql.Tx
+	binder binder
+}
+
+// ExecContext rebinds the query's placeholders and executes it in the
+// transaction.
+func (t *rebindTx) ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) {
+	return t.tx.ExecContext(ctx, t.binder.Rebind(query), args...)
+}
+
+// QueryContext rebinds the query's placeholders and runs it in the
+// transaction.
+func (t *rebindTx) QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) {
+	return t.tx.QueryContext(ctx, t.binder.Rebind(query), args...)
+}
+
+// QueryRowContext rebinds the query's placeholders and runs it in the
+// transaction.
+func (t *rebindTx) QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row {
+	return t.tx.QueryRowContext(ctx, t.binder.Rebind(query), args...)
+}
+
+// Commit commits the transaction.
+func (t *rebindTx) Commit() error {
+	return t.tx.Commit()
+}
+
+// Rollback aborts the transaction.
+func (t *rebindTx) Rollback() error {
+	return t.tx.Rollback()
+}
+
+// rebindConn wraps a pinned pool connection. It exists for the advisory-lock
+// flow, which holds a session-scoped lock on the pinned session and runs its
+// apply writes in transactions begun on that same session.
+type rebindConn struct {
+	conn   *sql.Conn
+	binder binder
+}
+
+// BeginTx starts a transaction on the pinned connection, wrapped so statements
+// executed on it rebind their placeholders.
+func (c *rebindConn) BeginTx(ctx context.Context, opts *sql.TxOptions) (*rebindTx, error) {
+	tx, err := c.conn.BeginTx(ctx, opts)
+	if err != nil {
+		return nil, err
+	}
+	return &rebindTx{tx: tx, binder: c.binder}, nil
+}
+
+// Raw runs f against the pinned driver connection, for lifecycle control such
+// as discarding a session whose advisory-lock state is uncertain.
+func (c *rebindConn) Raw(f func(driverConn any) error) error {
+	return c.conn.Raw(f)
+}
+
+// Close returns the pinned session to the pool.
+func (c *rebindConn) Close() error {
+	return c.conn.Close()
+}
+
+// raw exposes the pinned *sql.Conn for the advisory-lock boundary. It is the
+// only sanctioned binder escape: namedlock.Locker implementations emit their
+// engine's native placeholders and must execute on the pinned session that
+// holds the lock. No store SQL may execute through it.
+func (c *rebindConn) raw() *sql.Conn {
+	return c.conn
+}
diff --git a/pkg/storage/internal/sqlstore/db_test.go b/pkg/storage/internal/sqlstore/db_test.go
index b961f7bac..bec1e1d34 100644
--- a/pkg/storage/internal/sqlstore/db_test.go
+++ b/pkg/storage/internal/sqlstore/db_test.go
@@ -5,6 +5,7 @@ import (
 	"database/sql"
 	"database/sql/driver"
 	"io"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -57,16 +58,36 @@ type noopTx struct{}
 func (noopTx) Commit() error   { return nil }
 func (noopTx) Rollback() error { return nil }
 
+// rebindMarker is appended by countingBinder so tests can assert, on the SQL
+// the driver actually received, that a statement was rebound exactly once.
+const rebindMarker = " /* rebound */"
+
 // countingBinder rewrites the query to a recognizable form and counts how many
 // times it was invoked, so tests can prove each execution rebinds exactly once
-// and forwards the rebound SQL downstream.
+// and forwards the rebound SQL downstream. It also records whether it was ever
+// handed a query that was already rebound, which would mean a double rebind.
 type countingBinder struct {
-	calls int
+	calls         int
+	doubleRebound bool
 }
 
 func (b *countingBinder) Rebind(query string) string {
 	b.calls++
-	return query + " /* rebound */"
+	if strings.Contains(query, rebindMarker) {
+		b.doubleRebound = true
+	}
+	return query + rebindMarker
+}
+
+// assertReboundExactlyOnce verifies every statement the driver received was
+// rebound exactly once: the binder never saw already-rebound SQL, and each
+// recorded query carries exactly one marker.
+func assertReboundExactlyOnce(t *testing.T, b *countingBinder, connector *recordingConnector) {
+	t.Helper()
+	assert.False(t, b.doubleRebound, "binder received already-rebound SQL")
+	for _, q := range connector.queries {
+		assert.Equal(t, 1, strings.Count(q, rebindMarker), "query not rebound exactly once: %s", q)
+	}
 }
 
 // newRecordingRebindDB returns a rebindDB whose pool is backed by the
@@ -118,31 +139,79 @@ func TestRebindDBQueryRowContextRebindsOnce(t *testing.T) {
 	assert.Equal(t, [][]driver.Value{{int64(7)}}, connector.args)
 }
 
-// Transactions and pinned connections are raw passthrough handles: obtaining
-// them must not invoke the binder, and statements executed on them reach the
-// driver with the store's native placeholders unchanged.
-func TestRebindDBTransactionAndConnBypassBinder(t *testing.T) {
+// Statements executed inside a pool transaction pass through the binder
+// exactly once, the same as direct pool execution.
+func TestRebindTxRebindsOnce(t *testing.T) {
 	b := &countingBinder{}
 	rdb, connector := newRecordingRebindDB(t, b)
 	ctx := t.Context()
 
 	tx, err := rdb.BeginTx(ctx, nil)
 	require.NoError(t, err)
+
 	_, err = tx.ExecContext(ctx, "UPDATE applies SET state = ? WHERE id = ?", "stopped", int64(7))
 	require.NoError(t, err)
+
+	rows, err := tx.QueryContext(ctx, "SELECT id FROM tasks WHERE apply_id = ?", int64(7))
+	require.NoError(t, err)
+	require.NoError(t, rows.Close())
+
+	var id int64
+	err = tx.QueryRowContext(ctx, "SELECT id FROM applies WHERE id = ?", int64(7)).Scan(&id)
+	require.ErrorIs(t, err, sql.ErrNoRows)
+
 	require.NoError(t, tx.Rollback())
 
+	assert.Equal(t, 3, b.calls)
+	assert.Equal(t, []string{
+		"UPDATE applies SET state = ? WHERE id = ?" + rebindMarker,
+		"SELECT id FROM tasks WHERE apply_id = ?" + rebindMarker,
+		"SELECT id FROM applies WHERE id = ?" + rebindMarker,
+	}, connector.queries)
+	assertReboundExactlyOnce(t, b, connector)
+}
+
+// A transaction begun on a pinned connection rebinds its statements the same
+// way as one begun on the pool, so the advisory-lock apply-write path cannot
+// leak native placeholders past the binder.
+func TestRebindConnTxRebindsOnce(t *testing.T) {
+	b := &countingBinder{}
+	rdb, connector := newRecordingRebindDB(t, b)
+	ctx := t.Context()
+
 	conn, err := rdb.Conn(ctx)
 	require.NoError(t, err)
-	_, err = conn.ExecContext(ctx, "SELECT GET_LOCK(?, ?)", "lock", int64(0))
+
+	tx, err := conn.BeginTx(ctx, nil)
+	require.NoError(t, err)
+	_, err = tx.ExecContext(ctx, "UPDATE applies SET state = ? WHERE id = ?", "running", int64(7))
+	require.NoError(t, err)
+	require.NoError(t, tx.Commit())
+	require.NoError(t, conn.Close())
+
+	assert.Equal(t, 1, b.calls)
+	assert.Equal(t, []string{"UPDATE applies SET state = ? WHERE id = ?" + rebindMarker}, connector.queries)
+	assertReboundExactlyOnce(t, b, connector)
+}
+
+// The raw() escape hands the advisory locker the pinned session without the
+// binder in the way: locker SQL reaches the driver with its engine-native
+// placeholders untouched, and the binder is never invoked.
+func TestRebindConnRawEscapeBypassesBinder(t *testing.T) {
+	b := &countingBinder{}
+	rdb, connector := newRecordingRebindDB(t, b)
+	ctx := t.Context()
+
+	conn, err := rdb.Conn(ctx)
 	require.NoError(t, err)
+
+	var result sql.NullInt64
+	err = conn.raw().QueryRowContext(ctx, "SELECT GET_LOCK(?, ?)", "lock", int64(0)).Scan(&result)
+	require.ErrorIs(t, err, sql.ErrNoRows)
 	require.NoError(t, conn.Close())
 
 	assert.Equal(t, 0, b.calls)
-	assert.Equal(t, []string{
-		"UPDATE applies SET state = ? WHERE id = ?",
-		"SELECT GET_LOCK(?, ?)",
-	}, connector.queries)
+	assert.Equal(t, []string{"SELECT GET_LOCK(?, ?)"}, connector.queries)
 }
 
 func TestMySQLDialectRebindIsIdentity(t *testing.T) {
diff --git a/pkg/storage/internal/sqlstore/identity.go b/pkg/storage/internal/sqlstore/identity.go
index b52a11f9b..e5d7c92fc 100644
--- a/pkg/storage/internal/sqlstore/identity.go
+++ b/pkg/storage/internal/sqlstore/identity.go
@@ -5,10 +5,11 @@ import (
 	"database/sql"
 )
 
-// queryExecer is the subset of *sql.DB / *sql.Tx needed to run an INSERT and
-// read back the generated identity, either from a driver Result (MySQL) or a
-// RETURNING clause (Postgres). Both the connection pool and an in-flight
-// transaction satisfy it.
+// queryExecer is the execution subset needed to run an INSERT and read back
+// the generated identity, either from a driver Result (MySQL) or a RETURNING
+// clause (Postgres). Both the rebind-aware pool (*rebindDB) and an in-flight
+// transaction (*rebindTx) satisfy it, so identity inserts always execute
+// behind the placeholder-rebind boundary.
 type queryExecer interface {
 	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
 	QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
diff --git a/pkg/storage/internal/sqlstore/sql_helpers.go b/pkg/storage/internal/sqlstore/sql_helpers.go
index 3c9b66eef..0e4345d50 100644
--- a/pkg/storage/internal/sqlstore/sql_helpers.go
+++ b/pkg/storage/internal/sqlstore/sql_helpers.go
@@ -11,7 +11,7 @@ import (
 // rollbackTx rolls back tx, logging a warning if the rollback fails for a
 // reason other than the transaction already being finished. operation is
 // included in the log to identify the originating call site.
-func rollbackTx(ctx context.Context, tx *sql.Tx, operation string) {
+func rollbackTx(ctx context.Context, tx *rebindTx, operation string) {
 	if err := tx.Rollback(); err != nil && !errors.Is(err, sql.ErrTxDone) {
 		slog.WarnContext(ctx, "failed to roll back transaction", "operation", operation, "error", err)
 	}

From 14a19fa092f6bac9bd19d2242d66731d0b972ba5 Mon Sep 17 00:00:00 2001
From: kiran01bm <17925757+Kiran01bm@users.noreply.github.com>
Date: Fri, 7 Aug 2026 06:33:34 +1000
Subject: [PATCH 6/7] refactor(storage): move mysqlstore internals to shared
 internal/sqlstore core (#948)

Mechanical move (rename-detected) of the store implementation and its
white-box tests; mysqlstore becomes a thin public constructor over the
shared core so a second dialect backend can assemble the same store
logic with its own dependencies. No SQL or behavior changes.
---
 .../sqlstore}/applies.go                      |  2 +-
 .../sqlstore}/applies_test.go                 |  2 +-
 .../sqlstore}/apply_comments.go               |  2 +-
 .../sqlstore}/apply_comments_test.go          |  2 +-
 .../sqlstore}/apply_logs.go                   |  2 +-
 .../sqlstore}/apply_operations.go             |  2 +-
 .../sqlstore}/apply_operations_test.go        |  2 +-
 .../sqlstore}/apply_target_lock_test.go       |  2 +-
 .../sqlstore}/checks.go                       |  2 +-
 .../sqlstore}/checks_test.go                  |  2 +-
 .../sqlstore}/claimable_states_test.go        |  2 +-
 .../sqlstore}/control_requests.go             |  2 +-
 .../sqlstore}/control_requests_test.go        |  2 +-
 .../sqlstore}/dialect.go                      | 17 +++++++--------
 .../sqlstore}/dialect_test.go                 |  2 +-
 .../sqlstore}/identity.go                     |  2 +-
 .../sqlstore}/identity_test.go                |  2 +-
 .../sqlstore}/locks.go                        |  2 +-
 .../sqlstore}/locks_test.go                   |  2 +-
 .../sqlstore}/mysql_test.go                   |  2 +-
 .../sqlstore}/parity_test.go                  |  2 +-
 .../sqlstore}/plan_comments.go                |  2 +-
 .../sqlstore}/plan_comments_test.go           |  2 +-
 .../sqlstore}/plans.go                        |  2 +-
 .../sqlstore}/plans_test.go                   |  2 +-
 .../sqlstore}/retry.go                        |  2 +-
 .../sqlstore}/retry_test.go                   |  2 +-
 .../sqlstore}/settings.go                     |  2 +-
 .../sqlstore}/sql_helpers.go                  |  2 +-
 .../sqlstore}/storage.go                      |  7 +++++--
 .../sqlstore}/tasks.go                        |  2 +-
 .../sqlstore}/tasks_test.go                   |  2 +-
 .../sqlstore}/webhook_events.go               |  2 +-
 .../sqlstore}/webhook_events_test.go          |  2 +-
 pkg/storage/mysqlstore/mysqlstore.go          | 21 +++++++++++++++++++
 35 files changed, 66 insertions(+), 43 deletions(-)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/applies.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/applies_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/apply_comments.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/apply_comments_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/apply_logs.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/apply_operations.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/apply_operations_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/apply_target_lock_test.go (97%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/checks.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/checks_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/claimable_states_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/control_requests.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/control_requests_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/dialect.go (90%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/dialect_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/identity.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/identity_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/locks.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/locks_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/mysql_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/parity_test.go (97%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/plan_comments.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/plan_comments_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/plans.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/plans_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/retry.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/retry_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/settings.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/sql_helpers.go (98%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/storage.go (91%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/tasks.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/tasks_test.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/webhook_events.go (99%)
 rename pkg/storage/{mysqlstore => internal/sqlstore}/webhook_events_test.go (99%)
 create mode 100644 pkg/storage/mysqlstore/mysqlstore.go

diff --git a/pkg/storage/mysqlstore/applies.go b/pkg/storage/internal/sqlstore/applies.go
similarity index 99%
rename from pkg/storage/mysqlstore/applies.go
rename to pkg/storage/internal/sqlstore/applies.go
index dacf6f7c9..2ae610127 100644
--- a/pkg/storage/mysqlstore/applies.go
+++ b/pkg/storage/internal/sqlstore/applies.go
@@ -1,6 +1,6 @@
 // applies.go implements ApplyStore for tracking schema change executions.
 // Each apply is a top-level container that holds one or more tasks.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/applies_test.go b/pkg/storage/internal/sqlstore/applies_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/applies_test.go
rename to pkg/storage/internal/sqlstore/applies_test.go
index 2d20e21e4..dc04e464b 100644
--- a/pkg/storage/mysqlstore/applies_test.go
+++ b/pkg/storage/internal/sqlstore/applies_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/apply_comments.go b/pkg/storage/internal/sqlstore/apply_comments.go
similarity index 99%
rename from pkg/storage/mysqlstore/apply_comments.go
rename to pkg/storage/internal/sqlstore/apply_comments.go
index 762766ff4..88733be7a 100644
--- a/pkg/storage/mysqlstore/apply_comments.go
+++ b/pkg/storage/internal/sqlstore/apply_comments.go
@@ -1,6 +1,6 @@
 // apply_comments.go implements ApplyCommentStore for tracking GitHub PR comment IDs.
 // One comment per (apply_id, comment_state) combination.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/apply_comments_test.go b/pkg/storage/internal/sqlstore/apply_comments_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/apply_comments_test.go
rename to pkg/storage/internal/sqlstore/apply_comments_test.go
index 7a1798683..b5ee37931 100644
--- a/pkg/storage/mysqlstore/apply_comments_test.go
+++ b/pkg/storage/internal/sqlstore/apply_comments_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"database/sql"
diff --git a/pkg/storage/mysqlstore/apply_logs.go b/pkg/storage/internal/sqlstore/apply_logs.go
similarity index 99%
rename from pkg/storage/mysqlstore/apply_logs.go
rename to pkg/storage/internal/sqlstore/apply_logs.go
index 44cc358ae..e718e03e1 100644
--- a/pkg/storage/mysqlstore/apply_logs.go
+++ b/pkg/storage/internal/sqlstore/apply_logs.go
@@ -1,6 +1,6 @@
 // apply_logs.go implements ApplyLogStore for audit and debugging log entries.
 // Captures state transitions, errors, and progress events during applies.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/apply_operations.go b/pkg/storage/internal/sqlstore/apply_operations.go
similarity index 99%
rename from pkg/storage/mysqlstore/apply_operations.go
rename to pkg/storage/internal/sqlstore/apply_operations.go
index a5c73159d..69b6daf64 100644
--- a/pkg/storage/mysqlstore/apply_operations.go
+++ b/pkg/storage/internal/sqlstore/apply_operations.go
@@ -1,7 +1,7 @@
 // apply_operations.go implements ApplyOperationStore for per-(apply,
 // deployment, operation_key) child rows under a multi-operation apply — the
 // unit of work the driver claims.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/apply_operations_test.go b/pkg/storage/internal/sqlstore/apply_operations_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/apply_operations_test.go
rename to pkg/storage/internal/sqlstore/apply_operations_test.go
index 4d7a378ac..483a78cdf 100644
--- a/pkg/storage/mysqlstore/apply_operations_test.go
+++ b/pkg/storage/internal/sqlstore/apply_operations_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/apply_target_lock_test.go b/pkg/storage/internal/sqlstore/apply_target_lock_test.go
similarity index 97%
rename from pkg/storage/mysqlstore/apply_target_lock_test.go
rename to pkg/storage/internal/sqlstore/apply_target_lock_test.go
index 17117ea14..4f5f6a5f4 100644
--- a/pkg/storage/mysqlstore/apply_target_lock_test.go
+++ b/pkg/storage/internal/sqlstore/apply_target_lock_test.go
@@ -1,4 +1,4 @@
-package mysqlstore
+package sqlstore
 
 import (
 	"testing"
diff --git a/pkg/storage/mysqlstore/checks.go b/pkg/storage/internal/sqlstore/checks.go
similarity index 99%
rename from pkg/storage/mysqlstore/checks.go
rename to pkg/storage/internal/sqlstore/checks.go
index fcf12ebb3..f7a3e7fe1 100644
--- a/pkg/storage/mysqlstore/checks.go
+++ b/pkg/storage/internal/sqlstore/checks.go
@@ -1,5 +1,5 @@
 // checks.go implements CheckStore for SchemaBot's stored check state.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/checks_test.go b/pkg/storage/internal/sqlstore/checks_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/checks_test.go
rename to pkg/storage/internal/sqlstore/checks_test.go
index c1c9414a8..63776a631 100644
--- a/pkg/storage/mysqlstore/checks_test.go
+++ b/pkg/storage/internal/sqlstore/checks_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"database/sql"
diff --git a/pkg/storage/mysqlstore/claimable_states_test.go b/pkg/storage/internal/sqlstore/claimable_states_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/claimable_states_test.go
rename to pkg/storage/internal/sqlstore/claimable_states_test.go
index e1dcf1190..e67213e21 100644
--- a/pkg/storage/mysqlstore/claimable_states_test.go
+++ b/pkg/storage/internal/sqlstore/claimable_states_test.go
@@ -1,4 +1,4 @@
-package mysqlstore
+package sqlstore
 
 import (
 	"reflect"
diff --git a/pkg/storage/mysqlstore/control_requests.go b/pkg/storage/internal/sqlstore/control_requests.go
similarity index 99%
rename from pkg/storage/mysqlstore/control_requests.go
rename to pkg/storage/internal/sqlstore/control_requests.go
index 5b800117f..bb83b8dcc 100644
--- a/pkg/storage/mysqlstore/control_requests.go
+++ b/pkg/storage/internal/sqlstore/control_requests.go
@@ -1,4 +1,4 @@
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/control_requests_test.go b/pkg/storage/internal/sqlstore/control_requests_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/control_requests_test.go
rename to pkg/storage/internal/sqlstore/control_requests_test.go
index b12e237bf..664d2e0e7 100644
--- a/pkg/storage/mysqlstore/control_requests_test.go
+++ b/pkg/storage/internal/sqlstore/control_requests_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"sync"
diff --git a/pkg/storage/mysqlstore/dialect.go b/pkg/storage/internal/sqlstore/dialect.go
similarity index 90%
rename from pkg/storage/mysqlstore/dialect.go
rename to pkg/storage/internal/sqlstore/dialect.go
index d53dd5ecc..bd1cdd796 100644
--- a/pkg/storage/mysqlstore/dialect.go
+++ b/pkg/storage/internal/sqlstore/dialect.go
@@ -1,4 +1,4 @@
-package mysqlstore
+package sqlstore
 
 import (
 	"fmt"
@@ -8,11 +8,10 @@ import (
 
 // Dialect abstracts the SQL-syntax differences between database families (MySQL
 // and, in the future, Postgres) that the state store depends on. This is an
-// incremental seam: the store lives in mysqlstore and still emits MySQL-style
-// "?" placeholders, and only the family-varying syntax (upsert clause,
-// current-time and relative-time expressions) routes through here. When a
-// Postgres store is introduced this interface likely moves to a shared package,
-// and its parameterized SQL will need a "?"-to-"$n" rebind at the store
+// incremental seam: the store still emits MySQL-style "?" placeholders, and
+// only the family-varying syntax (upsert clause, current-time and
+// relative-time expressions) routes through here. When a Postgres backend is
+// introduced its parameterized SQL will need a "?"-to-"$n" rebind at the store
 // boundary.
 type Dialect interface {
 	// UpsertClause returns the trailing conflict-resolution clause that turns an
@@ -131,7 +130,7 @@ func (MySQLDialect) CurrentTimestamp(precision TimestampPrecision) string {
 	case TimestampPrecisionMicrosecond:
 		return "NOW(6)"
 	default:
-		panic(fmt.Sprintf("mysqlstore: unknown timestamp precision %d", precision))
+		panic(fmt.Sprintf("sqlstore: unknown timestamp precision %d", precision))
 	}
 }
 
@@ -153,7 +152,7 @@ func (d MySQLDialect) RelativeTime(precision TimestampPrecision, direction Relat
 	case AfterCurrentTime:
 		return "DATE_ADD(" + now + ", " + interval + ")"
 	default:
-		panic(fmt.Sprintf("mysqlstore: unknown relative-time direction %d", direction))
+		panic(fmt.Sprintf("sqlstore: unknown relative-time direction %d", direction))
 	}
 }
 
@@ -170,6 +169,6 @@ func mysqlIntervalUnit(unit IntervalUnit) string {
 	case IntervalDay:
 		return "DAY"
 	default:
-		panic(fmt.Sprintf("mysqlstore: unknown interval unit %d", unit))
+		panic(fmt.Sprintf("sqlstore: unknown interval unit %d", unit))
 	}
 }
diff --git a/pkg/storage/mysqlstore/dialect_test.go b/pkg/storage/internal/sqlstore/dialect_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/dialect_test.go
rename to pkg/storage/internal/sqlstore/dialect_test.go
index 636b84b98..a588f3620 100644
--- a/pkg/storage/mysqlstore/dialect_test.go
+++ b/pkg/storage/internal/sqlstore/dialect_test.go
@@ -1,4 +1,4 @@
-package mysqlstore
+package sqlstore
 
 import (
 	"testing"
diff --git a/pkg/storage/mysqlstore/identity.go b/pkg/storage/internal/sqlstore/identity.go
similarity index 99%
rename from pkg/storage/mysqlstore/identity.go
rename to pkg/storage/internal/sqlstore/identity.go
index 9b44cf650..b52a11f9b 100644
--- a/pkg/storage/mysqlstore/identity.go
+++ b/pkg/storage/internal/sqlstore/identity.go
@@ -1,4 +1,4 @@
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/identity_test.go b/pkg/storage/internal/sqlstore/identity_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/identity_test.go
rename to pkg/storage/internal/sqlstore/identity_test.go
index 1b2d94f08..18fc6f1b6 100644
--- a/pkg/storage/mysqlstore/identity_test.go
+++ b/pkg/storage/internal/sqlstore/identity_test.go
@@ -1,4 +1,4 @@
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/locks.go b/pkg/storage/internal/sqlstore/locks.go
similarity index 99%
rename from pkg/storage/mysqlstore/locks.go
rename to pkg/storage/internal/sqlstore/locks.go
index 8c5eb70e9..42a71b12e 100644
--- a/pkg/storage/mysqlstore/locks.go
+++ b/pkg/storage/internal/sqlstore/locks.go
@@ -1,6 +1,6 @@
 // locks.go implements LockStore for database-level deployment locks.
 // Locks prevent concurrent schema changes to the same database.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/locks_test.go b/pkg/storage/internal/sqlstore/locks_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/locks_test.go
rename to pkg/storage/internal/sqlstore/locks_test.go
index 29b484671..1cb87b2f4 100644
--- a/pkg/storage/mysqlstore/locks_test.go
+++ b/pkg/storage/internal/sqlstore/locks_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"database/sql"
diff --git a/pkg/storage/mysqlstore/mysql_test.go b/pkg/storage/internal/sqlstore/mysql_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/mysql_test.go
rename to pkg/storage/internal/sqlstore/mysql_test.go
index 4f4c0e2d2..1b7034d6b 100644
--- a/pkg/storage/mysqlstore/mysql_test.go
+++ b/pkg/storage/internal/sqlstore/mysql_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/parity_test.go b/pkg/storage/internal/sqlstore/parity_test.go
similarity index 97%
rename from pkg/storage/mysqlstore/parity_test.go
rename to pkg/storage/internal/sqlstore/parity_test.go
index 4eee68542..768267fc0 100644
--- a/pkg/storage/mysqlstore/parity_test.go
+++ b/pkg/storage/internal/sqlstore/parity_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"database/sql"
diff --git a/pkg/storage/mysqlstore/plan_comments.go b/pkg/storage/internal/sqlstore/plan_comments.go
similarity index 99%
rename from pkg/storage/mysqlstore/plan_comments.go
rename to pkg/storage/internal/sqlstore/plan_comments.go
index e967d5bee..897b1a8a1 100644
--- a/pkg/storage/mysqlstore/plan_comments.go
+++ b/pkg/storage/internal/sqlstore/plan_comments.go
@@ -1,7 +1,7 @@
 // plan_comments.go implements PlanCommentStore for tracking posted plan
 // comments so a newer plan comment for the same database can minimize the
 // ones it supersedes on GitHub.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/plan_comments_test.go b/pkg/storage/internal/sqlstore/plan_comments_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/plan_comments_test.go
rename to pkg/storage/internal/sqlstore/plan_comments_test.go
index 303bcee27..666e96f31 100644
--- a/pkg/storage/mysqlstore/plan_comments_test.go
+++ b/pkg/storage/internal/sqlstore/plan_comments_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"database/sql"
diff --git a/pkg/storage/mysqlstore/plans.go b/pkg/storage/internal/sqlstore/plans.go
similarity index 99%
rename from pkg/storage/mysqlstore/plans.go
rename to pkg/storage/internal/sqlstore/plans.go
index 802ea1e0a..04a408d44 100644
--- a/pkg/storage/mysqlstore/plans.go
+++ b/pkg/storage/internal/sqlstore/plans.go
@@ -1,6 +1,6 @@
 // plans.go implements PlanStore for schema change plans.
 // Plans are generated by diffing desired vs. live schema and stored for later execution.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/plans_test.go b/pkg/storage/internal/sqlstore/plans_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/plans_test.go
rename to pkg/storage/internal/sqlstore/plans_test.go
index 35aafc0e3..3fc9d29db 100644
--- a/pkg/storage/mysqlstore/plans_test.go
+++ b/pkg/storage/internal/sqlstore/plans_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"testing"
diff --git a/pkg/storage/mysqlstore/retry.go b/pkg/storage/internal/sqlstore/retry.go
similarity index 99%
rename from pkg/storage/mysqlstore/retry.go
rename to pkg/storage/internal/sqlstore/retry.go
index 31ca9134a..d90ca0940 100644
--- a/pkg/storage/mysqlstore/retry.go
+++ b/pkg/storage/internal/sqlstore/retry.go
@@ -3,7 +3,7 @@
 // or time out waiting for a lock; both roll the failed work back — the offending
 // statement, or the whole transaction when the work runs inside one — so the
 // operation can be safely re-run.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/retry_test.go b/pkg/storage/internal/sqlstore/retry_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/retry_test.go
rename to pkg/storage/internal/sqlstore/retry_test.go
index 2a25bf049..a3851e2b1 100644
--- a/pkg/storage/mysqlstore/retry_test.go
+++ b/pkg/storage/internal/sqlstore/retry_test.go
@@ -1,4 +1,4 @@
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/settings.go b/pkg/storage/internal/sqlstore/settings.go
similarity index 99%
rename from pkg/storage/mysqlstore/settings.go
rename to pkg/storage/internal/sqlstore/settings.go
index 509e0f524..d04d0f0a3 100644
--- a/pkg/storage/mysqlstore/settings.go
+++ b/pkg/storage/internal/sqlstore/settings.go
@@ -1,5 +1,5 @@
 // settings.go implements SettingsStore for admin-level runtime configuration.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/sql_helpers.go b/pkg/storage/internal/sqlstore/sql_helpers.go
similarity index 98%
rename from pkg/storage/mysqlstore/sql_helpers.go
rename to pkg/storage/internal/sqlstore/sql_helpers.go
index a5625da5b..3c9b66eef 100644
--- a/pkg/storage/mysqlstore/sql_helpers.go
+++ b/pkg/storage/internal/sqlstore/sql_helpers.go
@@ -1,5 +1,5 @@
 // sql_helpers.go provides shared utilities for MySQL store implementations.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/storage.go b/pkg/storage/internal/sqlstore/storage.go
similarity index 91%
rename from pkg/storage/mysqlstore/storage.go
rename to pkg/storage/internal/sqlstore/storage.go
index cea324318..b9d5b918a 100644
--- a/pkg/storage/mysqlstore/storage.go
+++ b/pkg/storage/internal/sqlstore/storage.go
@@ -1,5 +1,8 @@
-// Package mysql implements the storage interface using MySQL.
-package mysqlstore
+// Package sqlstore implements the storage interface over database/sql. It is
+// the shared dialect-parameterized core behind the public per-dialect
+// constructors (mysqlstore, and in the future postgresstore); it currently
+// emits MySQL-style SQL and is wired with MySQL dependencies.
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/tasks.go b/pkg/storage/internal/sqlstore/tasks.go
similarity index 99%
rename from pkg/storage/mysqlstore/tasks.go
rename to pkg/storage/internal/sqlstore/tasks.go
index 26e772d8b..d823c8df6 100644
--- a/pkg/storage/mysqlstore/tasks.go
+++ b/pkg/storage/internal/sqlstore/tasks.go
@@ -1,6 +1,6 @@
 // tasks.go implements TaskStore for individual DDL operations within an apply.
 // Each task represents one table's schema change.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/tasks_test.go b/pkg/storage/internal/sqlstore/tasks_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/tasks_test.go
rename to pkg/storage/internal/sqlstore/tasks_test.go
index de5689594..744b999d5 100644
--- a/pkg/storage/mysqlstore/tasks_test.go
+++ b/pkg/storage/internal/sqlstore/tasks_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/webhook_events.go b/pkg/storage/internal/sqlstore/webhook_events.go
similarity index 99%
rename from pkg/storage/mysqlstore/webhook_events.go
rename to pkg/storage/internal/sqlstore/webhook_events.go
index 61cc3e4f9..940dc68e6 100644
--- a/pkg/storage/mysqlstore/webhook_events.go
+++ b/pkg/storage/internal/sqlstore/webhook_events.go
@@ -1,5 +1,5 @@
 // webhook_events.go implements WebhookEventStore for durable webhook ingestion.
-package mysqlstore
+package sqlstore
 
 import (
 	"context"
diff --git a/pkg/storage/mysqlstore/webhook_events_test.go b/pkg/storage/internal/sqlstore/webhook_events_test.go
similarity index 99%
rename from pkg/storage/mysqlstore/webhook_events_test.go
rename to pkg/storage/internal/sqlstore/webhook_events_test.go
index 1547eb99c..a9c09e1fe 100644
--- a/pkg/storage/mysqlstore/webhook_events_test.go
+++ b/pkg/storage/internal/sqlstore/webhook_events_test.go
@@ -1,6 +1,6 @@
 //go:build integration
 
-package mysqlstore
+package sqlstore
 
 import (
 	"database/sql"
diff --git a/pkg/storage/mysqlstore/mysqlstore.go b/pkg/storage/mysqlstore/mysqlstore.go
new file mode 100644
index 000000000..039c7e5f2
--- /dev/null
+++ b/pkg/storage/mysqlstore/mysqlstore.go
@@ -0,0 +1,21 @@
+// Package mysqlstore provides the MySQL-backed storage.Storage implementation.
+// The store logic lives in the shared internal sqlstore core; this package is
+// the public constructor that assembles it with MySQL dependencies.
+package mysqlstore
+
+import (
+	"database/sql"
+
+	"github.com/block/schemabot/pkg/storage"
+	"github.com/block/schemabot/pkg/storage/internal/sqlstore"
+)
+
+// Storage is the MySQL-backed storage implementation.
+type Storage = sqlstore.Storage
+
+var _ storage.Storage = (*Storage)(nil)
+
+// New creates a new MySQL storage instance.
+func New(db *sql.DB) *Storage {
+	return sqlstore.New(db)
+}

From 54515e52a2734833744ebc97d8a4edb7609403e7 Mon Sep 17 00:00:00 2001
From: kiran01bm <17925757+Kiran01bm@users.noreply.github.com>
Date: Fri, 7 Aug 2026 09:33:11 +1000
Subject: [PATCH 7/7] feat(storage): route direct pool execution through
 rebind-aware DB wrapper (#949)

* refactor(storage): move mysqlstore internals to shared internal/sqlstore core

Mechanical move (rename-detected) of the store implementation and its
white-box tests; mysqlstore becomes a thin public constructor over the
shared core so a second dialect backend can assemble the same store
logic with its own dependencies. No SQL or behavior changes.

* feat(storage): route direct pool execution through rebind-aware DB wrapper

Stores hold a rebindDB instead of a raw *sql.DB, so every statement
executed directly on the pool passes through the dialect's placeholder
binder exactly once (identity for MySQL). This is the execution seam a
Postgres backend needs to rewrite "?" placeholders to "$n" without
touching store SQL. Transaction and pinned-connection handles remain
raw passthroughs for a follow-up.
---
 pkg/storage/internal/sqlstore/applies.go      |  10 +-
 pkg/storage/internal/sqlstore/applies_test.go |   2 +-
 .../internal/sqlstore/apply_comments.go       |   2 +-
 pkg/storage/internal/sqlstore/apply_logs.go   |   2 +-
 .../internal/sqlstore/apply_operations.go     |   2 +-
 pkg/storage/internal/sqlstore/checks.go       |   2 +-
 .../internal/sqlstore/control_requests.go     |   2 +-
 pkg/storage/internal/sqlstore/db.go           |  78 +++++++++
 pkg/storage/internal/sqlstore/db_test.go      | 151 ++++++++++++++++++
 pkg/storage/internal/sqlstore/locks.go        |   2 +-
 pkg/storage/internal/sqlstore/locks_test.go   |   6 +-
 .../internal/sqlstore/plan_comments.go        |   3 +-
 pkg/storage/internal/sqlstore/plans.go        |   2 +-
 pkg/storage/internal/sqlstore/settings.go     |   2 +-
 pkg/storage/internal/sqlstore/storage.go      |  29 ++--
 pkg/storage/internal/sqlstore/tasks.go        |   2 +-
 .../internal/sqlstore/webhook_events.go       |   2 +-
 17 files changed, 264 insertions(+), 35 deletions(-)
 create mode 100644 pkg/storage/internal/sqlstore/db.go
 create mode 100644 pkg/storage/internal/sqlstore/db_test.go

diff --git a/pkg/storage/internal/sqlstore/applies.go b/pkg/storage/internal/sqlstore/applies.go
index 2ae610127..89ace4796 100644
--- a/pkg/storage/internal/sqlstore/applies.go
+++ b/pkg/storage/internal/sqlstore/applies.go
@@ -54,7 +54,7 @@ const (
 
 // applyStore implements storage.ApplyStore using MySQL.
 type applyStore struct {
-	db       *sql.DB
+	db       *rebindDB
 	dialect  Dialect
 	identity identityInserter
 	// locker serializes concurrent applies against the same target with a
@@ -188,7 +188,7 @@ func beginApplyWriteTx(ctx context.Context, beginner txBeginner, operation strin
 	return &applyWriteTx{tx: tx}, nil
 }
 
-func beginApplyTargetWriteTx(ctx context.Context, db *sql.DB, locker namedlock.Locker, operation, database, dbType, environment string) (*applyWriteTx, error) {
+func beginApplyTargetWriteTx(ctx context.Context, db *rebindDB, locker namedlock.Locker, operation, database, dbType, environment string) (*applyWriteTx, error) {
 	conn, lockName, err := acquireApplyTargetLockConn(ctx, db, locker, database, dbType, environment)
 	if err != nil {
 		return nil, err
@@ -229,7 +229,7 @@ func applyTargetLockName(database, dbType, environment string) string {
 	return "schemabot_apply_" + hex.EncodeToString(sum[:16])
 }
 
-func acquireApplyTargetLockConn(ctx context.Context, db *sql.DB, locker namedlock.Locker, database, dbType, environment string) (*sql.Conn, string, error) {
+func acquireApplyTargetLockConn(ctx context.Context, db *rebindDB, locker namedlock.Locker, database, dbType, environment string) (*sql.Conn, string, error) {
 	if !hasApplyTarget(database, dbType, environment) {
 		return nil, "", fmt.Errorf("active apply target is required for %s/%s/%s", database, dbType, environment)
 	}
@@ -1746,7 +1746,7 @@ func (o claimOutcome) claimedAndComplete() bool {
 // window between the re-check and the commit. When another active apply owns the
 // target the claim is refused and the pending start control request is failed so
 // the operator sees why.
-func persistApplyClaim(ctx context.Context, db *sql.DB, locker namedlock.Locker, tx *sql.Tx, apply *storage.Apply, owner string) (claimOutcome, error) {
+func persistApplyClaim(ctx context.Context, db *rebindDB, locker namedlock.Locker, tx *sql.Tx, apply *storage.Apply, owner string) (claimOutcome, error) {
 	leaseToken := uuid.NewString()
 	leaseAcquiredAt := time.Now()
 	apply.LeaseOwner = owner
@@ -1784,7 +1784,7 @@ func persistApplyClaim(ctx context.Context, db *sql.DB, locker namedlock.Locker,
 // invariant, then either transition+commit or refuse+commit. It commits inside
 // the lock so a concurrent create cannot add a second active apply between the
 // re-check and the commit. The lock is released only after the commit.
-func claimStoppedApplyUnderTargetLock(ctx context.Context, db *sql.DB, locker namedlock.Locker, tx *sql.Tx, apply *storage.Apply, owner, leaseToken string) (claimOutcome, error) {
+func claimStoppedApplyUnderTargetLock(ctx context.Context, db *rebindDB, locker namedlock.Locker, tx *sql.Tx, apply *storage.Apply, owner, leaseToken string) (claimOutcome, error) {
 	database, dbType, environment, deployment, err := applyTargetForUpdate(ctx, tx, apply)
 	if err != nil {
 		return claimLostRace, err
diff --git a/pkg/storage/internal/sqlstore/applies_test.go b/pkg/storage/internal/sqlstore/applies_test.go
index dc04e464b..68a8264bb 100644
--- a/pkg/storage/internal/sqlstore/applies_test.go
+++ b/pkg/storage/internal/sqlstore/applies_test.go
@@ -1025,7 +1025,7 @@ func TestApplyStore_CreateWaitsForApplyTargetLock(t *testing.T) {
 	// Hold the same target lock that the create path must acquire. The creates
 	// below use the public store API; a result before release means active
 	// apply writes are not serialized by the per-target lock.
-	guardConn, guardLockName, err := acquireApplyTargetLockConn(ctx, testDB, namedlock.MySQL{}, "testdb", "mysql", "staging")
+	guardConn, guardLockName, err := acquireApplyTargetLockConn(ctx, newRebindDB(testDB, MySQLDialect{}), namedlock.MySQL{}, "testdb", "mysql", "staging")
 	require.NoError(t, err)
 	releaseGuard := func() {
 		if guardConn == nil {
diff --git a/pkg/storage/internal/sqlstore/apply_comments.go b/pkg/storage/internal/sqlstore/apply_comments.go
index 88733be7a..b3d28d978 100644
--- a/pkg/storage/internal/sqlstore/apply_comments.go
+++ b/pkg/storage/internal/sqlstore/apply_comments.go
@@ -18,7 +18,7 @@ const applyCommentColumns = `id, apply_id, comment_state, github_comment_id, pos
 
 // applyCommentStore implements storage.ApplyCommentStore using MySQL.
 type applyCommentStore struct {
-	db      *sql.DB
+	db      *rebindDB
 	dialect Dialect
 }
 
diff --git a/pkg/storage/internal/sqlstore/apply_logs.go b/pkg/storage/internal/sqlstore/apply_logs.go
index e718e03e1..db2c90576 100644
--- a/pkg/storage/internal/sqlstore/apply_logs.go
+++ b/pkg/storage/internal/sqlstore/apply_logs.go
@@ -18,7 +18,7 @@ const applyLogColumns = `id, apply_id, task_id, level, event_type, source, messa
 
 // applyLogStore implements storage.ApplyLogStore using MySQL.
 type applyLogStore struct {
-	db       *sql.DB
+	db       *rebindDB
 	identity identityInserter
 }
 
diff --git a/pkg/storage/internal/sqlstore/apply_operations.go b/pkg/storage/internal/sqlstore/apply_operations.go
index 69b6daf64..cc3b10053 100644
--- a/pkg/storage/internal/sqlstore/apply_operations.go
+++ b/pkg/storage/internal/sqlstore/apply_operations.go
@@ -27,7 +27,7 @@ const applyOperationColumns = `id, apply_id, deployment, operation_key, operatio
 
 // applyOperationStore implements storage.ApplyOperationStore using MySQL.
 type applyOperationStore struct {
-	db       *sql.DB
+	db       *rebindDB
 	dialect  Dialect
 	identity identityInserter
 	locker   namedlock.Locker
diff --git a/pkg/storage/internal/sqlstore/checks.go b/pkg/storage/internal/sqlstore/checks.go
index f7a3e7fe1..757785051 100644
--- a/pkg/storage/internal/sqlstore/checks.go
+++ b/pkg/storage/internal/sqlstore/checks.go
@@ -24,7 +24,7 @@ const (
 
 // checkStore implements storage.CheckStore using MySQL.
 type checkStore struct {
-	db      *sql.DB
+	db      *rebindDB
 	dialect Dialect
 }
 
diff --git a/pkg/storage/internal/sqlstore/control_requests.go b/pkg/storage/internal/sqlstore/control_requests.go
index bb83b8dcc..0d9cd326b 100644
--- a/pkg/storage/internal/sqlstore/control_requests.go
+++ b/pkg/storage/internal/sqlstore/control_requests.go
@@ -14,7 +14,7 @@ const controlRequestColumns = `id, apply_id, operation, status,
 	requested_by, error_message, metadata, completed_at, created_at, updated_at`
 
 type controlRequestStore struct {
-	db       *sql.DB
+	db       *rebindDB
 	identity identityInserter
 }
 
diff --git a/pkg/storage/internal/sqlstore/db.go b/pkg/storage/internal/sqlstore/db.go
new file mode 100644
index 000000000..4bec372a2
--- /dev/null
+++ b/pkg/storage/internal/sqlstore/db.go
@@ -0,0 +1,78 @@
+package sqlstore
+
+import (
+	"context"
+	"database/sql"
+)
+
+// binder rewrites a statement's parameter placeholders into the wire syntax
+// the target engine expects. The store builds all SQL with MySQL-style "?"
+// placeholders; engines that use a different syntax (Postgres "$n") rebind at
+// the execution boundary, while MySQL's binder is the identity.
+type binder interface {
+	// Rebind returns the query with its placeholders rewritten for the target
+	// engine. It must be a pure string transformation: same input, same output,
+	// no side effects.
+	Rebind(query string) string
+}
+
+// Rebind returns the query unchanged: MySQL consumes the store's native "?"
+// placeholders directly.
+func (MySQLDialect) Rebind(query string) string {
+	return query
+}
+
+// rebindDB wraps the connection pool and owns placeholder rebinding: every
+// statement executed directly on the pool passes through the dialect's binder
+// exactly once before reaching the SQL driver. Stores hold a *rebindDB rather
+// than a raw *sql.DB so no store can bypass the rebind boundary.
+//
+// Transactions and pinned connections obtained via BeginTx / Conn are handed
+// back as raw handles: statements executed on them do not pass through the
+// binder yet, so those paths remain MySQL-placeholder only.
+type rebindDB struct {
+	pool   *sql.DB
+	binder binder
+}
+
+// newRebindDB wraps pool so all direct execution rebinds placeholders with b.
+func newRebindDB(pool *sql.DB, b binder) *rebindDB {
+	return &rebindDB{pool: pool, binder: b}
+}
+
+// ExecContext rebinds the query's placeholders and executes it on the pool.
+func (d *rebindDB) ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) {
+	return d.pool.ExecContext(ctx, d.binder.Rebind(query), args...)
+}
+
+// QueryContext rebinds the query's placeholders and runs it on the pool.
+func (d *rebindDB) QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) {
+	return d.pool.QueryContext(ctx, d.binder.Rebind(query), args...)
+}
+
+// QueryRowContext rebinds the query's placeholders and runs it on the pool.
+func (d *rebindDB) QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row {
+	return d.pool.QueryRowContext(ctx, d.binder.Rebind(query), args...)
+}
+
+// BeginTx starts a transaction on the underlying pool. The returned *sql.Tx is
+// a raw handle: statements executed on it are not rebound.
+func (d *rebindDB) BeginTx(ctx context.Context, opts *sql.TxOptions) (*sql.Tx, error) {
+	return d.pool.BeginTx(ctx, opts)
+}
+
+// Conn pins a single connection from the underlying pool. The returned
+// *sql.Conn is a raw handle: statements executed on it are not rebound.
+func (d *rebindDB) Conn(ctx context.Context) (*sql.Conn, error) {
+	return d.pool.Conn(ctx)
+}
+
+// PingContext verifies the underlying pool's connectivity.
+func (d *rebindDB) PingContext(ctx context.Context) error {
+	return d.pool.PingContext(ctx)
+}
+
+// Close closes the underlying pool.
+func (d *rebindDB) Close() error {
+	return d.pool.Close()
+}
diff --git a/pkg/storage/internal/sqlstore/db_test.go b/pkg/storage/internal/sqlstore/db_test.go
new file mode 100644
index 000000000..b961f7bac
--- /dev/null
+++ b/pkg/storage/internal/sqlstore/db_test.go
@@ -0,0 +1,151 @@
+package sqlstore
+
+import (
+	"context"
+	"database/sql"
+	"database/sql/driver"
+	"io"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// recordingConnector is a database/sql driver stub that records every SQL
+// statement text the pool hands to the driver, so tests can assert exactly
+// what reached the wire after rebinding.
+type recordingConnector struct {
+	queries []string
+	args    [][]driver.Value
+}
+
+func (c *recordingConnector) Connect(context.Context) (driver.Conn, error) { return c, nil }
+func (c *recordingConnector) Driver() driver.Driver                        { return nil }
+
+func (c *recordingConnector) Prepare(query string) (driver.Stmt, error) {
+	c.queries = append(c.queries, query)
+	return &recordingStmt{connector: c}, nil
+}
+func (c *recordingConnector) Close() error              { return nil }
+func (c *recordingConnector) Begin() (driver.Tx, error) { return noopTx{}, nil }
+
+type recordingStmt struct {
+	connector *recordingConnector
+}
+
+func (s *recordingStmt) Close() error  { return nil }
+func (s *recordingStmt) NumInput() int { return -1 }
+
+func (s *recordingStmt) Exec(args []driver.Value) (driver.Result, error) {
+	s.connector.args = append(s.connector.args, args)
+	return driver.RowsAffected(0), nil
+}
+
+func (s *recordingStmt) Query(args []driver.Value) (driver.Rows, error) {
+	s.connector.args = append(s.connector.args, args)
+	return emptyRows{}, nil
+}
+
+type emptyRows struct{}
+
+func (emptyRows) Columns() []string              { return nil }
+func (emptyRows) Close() error                   { return nil }
+func (emptyRows) Next(dest []driver.Value) error { return io.EOF }
+
+type noopTx struct{}
+
+func (noopTx) Commit() error   { return nil }
+func (noopTx) Rollback() error { return nil }
+
+// countingBinder rewrites the query to a recognizable form and counts how many
+// times it was invoked, so tests can prove each execution rebinds exactly once
+// and forwards the rebound SQL downstream.
+type countingBinder struct {
+	calls int
+}
+
+func (b *countingBinder) Rebind(query string) string {
+	b.calls++
+	return query + " /* rebound */"
+}
+
+// newRecordingRebindDB returns a rebindDB whose pool is backed by the
+// recording driver stub, so tests observe the exact SQL reaching the driver.
+func newRecordingRebindDB(t *testing.T, b binder) (*rebindDB, *recordingConnector) {
+	connector := &recordingConnector{}
+	pool := sql.OpenDB(connector)
+	t.Cleanup(func() {
+		require.NoError(t, pool.Close())
+	})
+	return newRebindDB(pool, b), connector
+}
+
+func TestRebindDBExecContextRebindsOnce(t *testing.T) {
+	b := &countingBinder{}
+	rdb, connector := newRecordingRebindDB(t, b)
+
+	_, err := rdb.ExecContext(t.Context(), "UPDATE applies SET state = ? WHERE id = ?", "running", int64(7))
+	require.NoError(t, err)
+
+	assert.Equal(t, 1, b.calls)
+	assert.Equal(t, []string{"UPDATE applies SET state = ? WHERE id = ? /* rebound */"}, connector.queries)
+	assert.Equal(t, [][]driver.Value{{"running", int64(7)}}, connector.args)
+}
+
+func TestRebindDBQueryContextRebindsOnce(t *testing.T) {
+	b := &countingBinder{}
+	rdb, connector := newRecordingRebindDB(t, b)
+
+	rows, err := rdb.QueryContext(t.Context(), "SELECT id FROM applies WHERE state = ?", "running")
+	require.NoError(t, err)
+	require.NoError(t, rows.Close())
+
+	assert.Equal(t, 1, b.calls)
+	assert.Equal(t, []string{"SELECT id FROM applies WHERE state = ? /* rebound */"}, connector.queries)
+	assert.Equal(t, [][]driver.Value{{"running"}}, connector.args)
+}
+
+func TestRebindDBQueryRowContextRebindsOnce(t *testing.T) {
+	b := &countingBinder{}
+	rdb, connector := newRecordingRebindDB(t, b)
+
+	var id int64
+	err := rdb.QueryRowContext(t.Context(), "SELECT id FROM applies WHERE id = ?", int64(7)).Scan(&id)
+	require.ErrorIs(t, err, sql.ErrNoRows)
+
+	assert.Equal(t, 1, b.calls)
+	assert.Equal(t, []string{"SELECT id FROM applies WHERE id = ? /* rebound */"}, connector.queries)
+	assert.Equal(t, [][]driver.Value{{int64(7)}}, connector.args)
+}
+
+// Transactions and pinned connections are raw passthrough handles: obtaining
+// them must not invoke the binder, and statements executed on them reach the
+// driver with the store's native placeholders unchanged.
+func TestRebindDBTransactionAndConnBypassBinder(t *testing.T) {
+	b := &countingBinder{}
+	rdb, connector := newRecordingRebindDB(t, b)
+	ctx := t.Context()
+
+	tx, err := rdb.BeginTx(ctx, nil)
+	require.NoError(t, err)
+	_, err = tx.ExecContext(ctx, "UPDATE applies SET state = ? WHERE id = ?", "stopped", int64(7))
+	require.NoError(t, err)
+	require.NoError(t, tx.Rollback())
+
+	conn, err := rdb.Conn(ctx)
+	require.NoError(t, err)
+	_, err = conn.ExecContext(ctx, "SELECT GET_LOCK(?, ?)", "lock", int64(0))
+	require.NoError(t, err)
+	require.NoError(t, conn.Close())
+
+	assert.Equal(t, 0, b.calls)
+	assert.Equal(t, []string{
+		"UPDATE applies SET state = ? WHERE id = ?",
+		"SELECT GET_LOCK(?, ?)",
+	}, connector.queries)
+}
+
+func TestMySQLDialectRebindIsIdentity(t *testing.T) {
+	query := "SELECT id FROM applies WHERE state = ? AND database_name = ?"
+	assert.Equal(t, query, MySQLDialect{}.Rebind(query))
+}
diff --git a/pkg/storage/internal/sqlstore/locks.go b/pkg/storage/internal/sqlstore/locks.go
index 42a71b12e..1d078cc81 100644
--- a/pkg/storage/internal/sqlstore/locks.go
+++ b/pkg/storage/internal/sqlstore/locks.go
@@ -20,7 +20,7 @@ const lockColumns = `id, database_name, database_type, repository, pull_request,
 
 // lockStore implements storage.LockStore using MySQL.
 type lockStore struct {
-	db *sql.DB
+	db *rebindDB
 }
 
 // Acquire attempts to acquire a lock. Returns ErrLockHeld if held by another owner.
diff --git a/pkg/storage/internal/sqlstore/locks_test.go b/pkg/storage/internal/sqlstore/locks_test.go
index 1cb87b2f4..e3776717e 100644
--- a/pkg/storage/internal/sqlstore/locks_test.go
+++ b/pkg/storage/internal/sqlstore/locks_test.go
@@ -166,7 +166,7 @@ func TestLockStore_Acquire_RefreshSameOwnerValueAlreadyMatches(t *testing.T) {
 		require.NoError(t, db.Close())
 	})
 	require.NoError(t, db.PingContext(ctx))
-	store := &lockStore{db: db}
+	store := &lockStore{db: newRebindDB(db, MySQLDialect{})}
 
 	require.NoError(t, store.Acquire(ctx, &storage.Lock{
 		DatabaseName:  "testdb",
@@ -227,7 +227,7 @@ func TestLockStore_Acquire_RefreshSameOwnerValueAlreadyMatches(t *testing.T) {
 func TestLockStore_Acquire_RefreshOwnerNoLongerMatches(t *testing.T) {
 	clearTables(t)
 	ctx := t.Context()
-	store := &lockStore{db: testDB}
+	store := &lockStore{db: newRebindDB(testDB, MySQLDialect{})}
 
 	require.NoError(t, store.Acquire(ctx, &storage.Lock{
 		DatabaseName:  "testdb",
@@ -554,7 +554,7 @@ func TestLockStore_UpdateSameSecondSucceeds(t *testing.T) {
 	_, err = db.ExecContext(ctx, "SET TIMESTAMP = 1700000000")
 	require.NoError(t, err)
 
-	store := &lockStore{db: db}
+	store := &lockStore{db: newRebindDB(db, MySQLDialect{})}
 
 	// 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/plan_comments.go b/pkg/storage/internal/sqlstore/plan_comments.go
index 897b1a8a1..cccfbfa2a 100644
--- a/pkg/storage/internal/sqlstore/plan_comments.go
+++ b/pkg/storage/internal/sqlstore/plan_comments.go
@@ -5,7 +5,6 @@ package sqlstore
 
 import (
 	"context"
-	"database/sql"
 	"fmt"
 
 	"github.com/block/schemabot/pkg/storage"
@@ -17,7 +16,7 @@ const planCommentColumns = `id, repository, pull_request, database_name, databas
 
 // planCommentStore implements storage.PlanCommentStore using MySQL.
 type planCommentStore struct {
-	db       *sql.DB
+	db       *rebindDB
 	identity identityInserter
 }
 
diff --git a/pkg/storage/internal/sqlstore/plans.go b/pkg/storage/internal/sqlstore/plans.go
index 04a408d44..929cde448 100644
--- a/pkg/storage/internal/sqlstore/plans.go
+++ b/pkg/storage/internal/sqlstore/plans.go
@@ -20,7 +20,7 @@ const planColumns = `id, plan_identifier, database_name, database_type,
 
 // planStore implements storage.PlanStore using MySQL.
 type planStore struct {
-	db       *sql.DB
+	db       *rebindDB
 	identity identityInserter
 }
 
diff --git a/pkg/storage/internal/sqlstore/settings.go b/pkg/storage/internal/sqlstore/settings.go
index d04d0f0a3..7f3edd2a3 100644
--- a/pkg/storage/internal/sqlstore/settings.go
+++ b/pkg/storage/internal/sqlstore/settings.go
@@ -15,7 +15,7 @@ const settingColumns = `id, setting_key, setting_value, created_at, updated_at`
 
 // settingsStore implements storage.SettingsStore using MySQL.
 type settingsStore struct {
-	db      *sql.DB
+	db      *rebindDB
 	dialect Dialect
 }
 
diff --git a/pkg/storage/internal/sqlstore/storage.go b/pkg/storage/internal/sqlstore/storage.go
index b9d5b918a..5c2797a1b 100644
--- a/pkg/storage/internal/sqlstore/storage.go
+++ b/pkg/storage/internal/sqlstore/storage.go
@@ -14,7 +14,7 @@ import (
 
 // Storage implements the storage.Storage interface using MySQL.
 type Storage struct {
-	db              *sql.DB
+	db              *rebindDB
 	locks           *lockStore
 	plans           *planStore
 	applies         *applyStore
@@ -33,20 +33,21 @@ var _ storage.Storage = (*Storage)(nil)
 
 // New creates a new MySQL storage instance.
 func New(db *sql.DB) *Storage {
+	rdb := newRebindDB(db, MySQLDialect{})
 	return &Storage{
-		db:              db,
-		locks:           &lockStore{db: db},
-		plans:           &planStore{db: db, identity: MySQLDialect{}},
-		applies:         &applyStore{db: db, dialect: MySQLDialect{}, identity: MySQLDialect{}, locker: namedlock.MySQL{}},
-		tasks:           &taskStore{db: db, identity: MySQLDialect{}},
-		applyLogs:       &applyLogStore{db: db, identity: MySQLDialect{}},
-		controlRequests: &controlRequestStore{db: db, identity: MySQLDialect{}},
-		applyComments:   &applyCommentStore{db: db, dialect: MySQLDialect{}},
-		planComments:    &planCommentStore{db: db, identity: MySQLDialect{}},
-		applyOperations: &applyOperationStore{db: db, dialect: MySQLDialect{}, identity: MySQLDialect{}, locker: namedlock.MySQL{}},
-		checks:          &checkStore{db: db, dialect: MySQLDialect{}},
-		settings:        &settingsStore{db: db, dialect: MySQLDialect{}},
-		webhookEvents:   &webhookEventStore{db: db, dialect: MySQLDialect{}, identity: MySQLDialect{}},
+		db:              rdb,
+		locks:           &lockStore{db: rdb},
+		plans:           &planStore{db: rdb, identity: MySQLDialect{}},
+		applies:         &applyStore{db: rdb, dialect: MySQLDialect{}, identity: MySQLDialect{}, locker: namedlock.MySQL{}},
+		tasks:           &taskStore{db: rdb, identity: MySQLDialect{}},
+		applyLogs:       &applyLogStore{db: rdb, identity: MySQLDialect{}},
+		controlRequests: &controlRequestStore{db: rdb, identity: MySQLDialect{}},
+		applyComments:   &applyCommentStore{db: rdb, dialect: MySQLDialect{}},
+		planComments:    &planCommentStore{db: rdb, identity: MySQLDialect{}},
+		applyOperations: &applyOperationStore{db: rdb, dialect: MySQLDialect{}, identity: MySQLDialect{}, locker: namedlock.MySQL{}},
+		checks:          &checkStore{db: rdb, dialect: MySQLDialect{}},
+		settings:        &settingsStore{db: rdb, dialect: MySQLDialect{}},
+		webhookEvents:   &webhookEventStore{db: rdb, dialect: MySQLDialect{}, identity: MySQLDialect{}},
 	}
 }
 
diff --git a/pkg/storage/internal/sqlstore/tasks.go b/pkg/storage/internal/sqlstore/tasks.go
index d823c8df6..660fae083 100644
--- a/pkg/storage/internal/sqlstore/tasks.go
+++ b/pkg/storage/internal/sqlstore/tasks.go
@@ -41,7 +41,7 @@ var terminalTaskStatesSQL = func() string {
 
 // taskStore implements storage.TaskStore using MySQL.
 type taskStore struct {
-	db       *sql.DB
+	db       *rebindDB
 	identity identityInserter
 }
 
diff --git a/pkg/storage/internal/sqlstore/webhook_events.go b/pkg/storage/internal/sqlstore/webhook_events.go
index 940dc68e6..7e7d06806 100644
--- a/pkg/storage/internal/sqlstore/webhook_events.go
+++ b/pkg/storage/internal/sqlstore/webhook_events.go
@@ -20,7 +20,7 @@ const webhookEventColumns = `id, provider, delivery_id, event, action, repositor
 	received_at, started_at, completed_at, created_at, updated_at`
 
 type webhookEventStore struct {
-	db       *sql.DB
+	db       *rebindDB
 	dialect  Dialect
 	identity identityInserter
 }