fix: address PostgreSQL registry review

internal/postgresql/registry/migrations/001_create_registry.sql

@@ -0,0 +1,25 @@
+CREATE SCHEMA postgresql_tenant_operator;
+
+CREATE TABLE postgresql_tenant_operator.tenant_ownership (
+    instance_uid text NOT NULL,
+    tenant_uid text NOT NULL,
+    tenant_namespace text NOT NULL,
+    tenant_name text NOT NULL,
+    database_name text NOT NULL,
+    role_name text NOT NULL,
+    credential_path text NOT NULL,
+    managed boolean NOT NULL DEFAULT true,
+    created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
+    updated_at timestamptz NOT NULL DEFAULT clock_timestamp(),
+    retained_at timestamptz,
+    PRIMARY KEY (instance_uid, tenant_uid),
+    UNIQUE (instance_uid, tenant_namespace, tenant_name),
+    UNIQUE (instance_uid, database_name),
+    UNIQUE (instance_uid, role_name),
+    UNIQUE (credential_path),
+    CHECK (managed OR retained_at IS NOT NULL)
+);
+
+---- create above / drop below ----
+
+DROP SCHEMA postgresql_tenant_operator CASCADE;

internal/postgresql/registry/registry.go

@@ -21,14 +21,21 @@ package registry
 
 import (
 	"context"
+	"embed"
 	"errors"
 	"fmt"
+	"io/fs"
 	"time"
 
 	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/jackc/tern/v2/migrate"
 )
 
-const schemaVersion = 1
+const migrationVersionTable = "public.postgresql_tenant_operator_schema_version"
+
+//go:embed migrations/*.sql
+var migrationFiles embed.FS
 
 var (
 	// ErrNotFound indicates that the registry has no matching ownership record.
@@ -80,35 +87,29 @@ const (
 	ClaimOwned   ClaimResult = "Owned"
 )
 
-// Bootstrap creates the private schema and the current registry table idempotently.
+// Bootstrap applies all pending versioned registry migrations idempotently.
 func (s *Store) Bootstrap(ctx context.Context) error {
 	if s == nil || s.db == nil {
 		return errors.New("bootstrap registry: nil database")
 	}
 
-	tx, err := s.db.Begin(ctx)
-	if err != nil {
-		return fmt.Errorf("begin registry bootstrap: %w", err)
-	}
-	defer func() { _ = tx.Rollback(ctx) }()
-
-	for _, statement := range bootstrapStatements {
-		if _, err := tx.Exec(ctx, statement); err != nil {
-			return fmt.Errorf("bootstrap registry schema version %d: %w", schemaVersion, err)
+	return s.withMigrationConnection(ctx, func(conn *pgx.Conn) error {
+		migrations, err := fs.Sub(migrationFiles, "migrations")
+		if err != nil {
+			return fmt.Errorf("open embedded registry migrations: %w", err)
 		}
-	}
-	var version int
-	if err := tx.QueryRow(ctx, `SELECT version FROM postgresql_tenant_operator.schema_version WHERE singleton`).Scan(&version); err != nil {
-		return fmt.Errorf("read registry schema version: %w", err)
-	}
-	if version != schemaVersion {
-		return fmt.Errorf("registry schema version %d is unsupported; expected %d", version, schemaVersion)
-	}
-
-	if err := tx.Commit(ctx); err != nil {
-		return fmt.Errorf("commit registry bootstrap: %w", err)
-	}
-	return nil
+		migrator, err := migrate.NewMigrator(ctx, conn, migrationVersionTable)
+		if err != nil {
+			return fmt.Errorf("initialize registry migrator: %w", err)
+		}
+		if err := migrator.LoadMigrations(migrations); err != nil {
+			return fmt.Errorf("load registry migrations: %w", err)
+		}
+		if err := migrator.Migrate(ctx); err != nil {
+			return fmt.Errorf("apply registry migrations: %w", err)
+		}
+		return nil
+	})
 }
 
 // Claim reserves all names for an owner. Repeating an identical claim is idempotent.
@@ -126,7 +127,7 @@ func (s *Store) Claim(ctx context.Context, owner Ownership) (ClaimResult, error)
 	}
 	defer func() { _ = tx.Rollback(ctx) }()
 
-	commandTag, err := tx.Exec(ctx, claimStatement,
+	record, err := scanRecord(tx.QueryRow(ctx, claimStatement,
 		owner.InstanceUID,
 		owner.TenantUID,
 		owner.TenantNamespace,
@@ -134,29 +135,47 @@ func (s *Store) Claim(ctx context.Context, owner Ownership) (ClaimResult, error)
 		owner.DatabaseName,
 		owner.RoleName,
 		owner.CredentialPath,
-	)
-	if err != nil {
+	))
+	if err != nil && !errors.Is(err, ErrNotFound) {
 		return "", fmt.Errorf("insert registry claim: %w", err)
 	}
-
-	record, err := getByTenantUID(ctx, tx, owner.InstanceUID, owner.TenantUID)
-	if err != nil {
-		if errors.Is(err, ErrNotFound) {
+	if errors.Is(err, ErrNotFound) {
+		record, err = getByTenantUID(ctx, tx, owner.InstanceUID, owner.TenantUID)
+		if err != nil {
+			if errors.Is(err, ErrNotFound) {
+				return "", fmt.Errorf("%w: database, role, tenant identity, or credential path is already reserved", ErrConflict)
+			}
+			return "", err
+		}
+		if !record.equal(owner) || !record.Managed {
 			return "", fmt.Errorf("%w: database, role, tenant identity, or credential path is already reserved", ErrConflict)
 		}
-		return "", err
-	}
-	if !record.equal(owner) || !record.Managed {
-		return "", fmt.Errorf("%w: existing record does not match the requested managed owner", ErrConflict)
+		if err := tx.Commit(ctx); err != nil {
+			return "", fmt.Errorf("commit registry claim: %w", err)
+		}
+		return ClaimOwned, nil
 	}
 
 	if err := tx.Commit(ctx); err != nil {
 		return "", fmt.Errorf("commit registry claim: %w", err)
 	}
-	if commandTag.RowsAffected() == 1 {
-		return ClaimCreated, nil
+	return ClaimCreated, nil
+}
+
+func (s *Store) withMigrationConnection(ctx context.Context, run func(*pgx.Conn) error) error {
+	switch db := s.db.(type) {
+	case *pgx.Conn:
+		return run(db)
+	case *pgxpool.Pool:
+		conn, err := db.Acquire(ctx)
+		if err != nil {
+			return fmt.Errorf("acquire registry migration connection: %w", err)
+		}
+		defer conn.Release()
+		return run(conn.Conn())
+	default:
+		return fmt.Errorf("bootstrap registry: database type %T cannot provide a migration connection", s.db)
 	}
-	return ClaimOwned, nil
 }
 
 // Get returns the ownership record for an Instance UID and Tenant UID.

internal/postgresql/registry/sql.go

@@ -16,36 +16,6 @@ limitations under the License.
 
 package registry
 
-var bootstrapStatements = []string{
-	`CREATE SCHEMA IF NOT EXISTS postgresql_tenant_operator`,
-	`CREATE TABLE IF NOT EXISTS postgresql_tenant_operator.schema_version (
-    singleton boolean PRIMARY KEY DEFAULT true CHECK (singleton),
-    version integer NOT NULL
-)`,
-	`INSERT INTO postgresql_tenant_operator.schema_version (singleton, version)
-VALUES (true, 1)
-ON CONFLICT (singleton) DO NOTHING`,
-	`CREATE TABLE IF NOT EXISTS postgresql_tenant_operator.tenant_ownership (
-    instance_uid text NOT NULL,
-    tenant_uid text NOT NULL,
-    tenant_namespace text NOT NULL,
-    tenant_name text NOT NULL,
-    database_name text NOT NULL,
-    role_name text NOT NULL,
-    credential_path text NOT NULL,
-    managed boolean NOT NULL DEFAULT true,
-    created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
-    updated_at timestamptz NOT NULL DEFAULT clock_timestamp(),
-    retained_at timestamptz,
-    PRIMARY KEY (instance_uid, tenant_uid),
-    UNIQUE (instance_uid, tenant_namespace, tenant_name),
-    UNIQUE (instance_uid, database_name),
-    UNIQUE (instance_uid, role_name),
-    UNIQUE (credential_path),
-    CHECK (managed OR retained_at IS NOT NULL)
-)`,
-}
-
 const claimStatement = `
 INSERT INTO postgresql_tenant_operator.tenant_ownership (
     instance_uid,
@@ -56,7 +26,19 @@ INSERT INTO postgresql_tenant_operator.tenant_ownership (
     role_name,
     credential_path
 ) VALUES ($1, $2, $3, $4, $5, $6, $7)
-ON CONFLICT DO NOTHING`
+ON CONFLICT DO NOTHING
+RETURNING
+    instance_uid,
+    tenant_uid,
+    tenant_namespace,
+    tenant_name,
+    database_name,
+    role_name,
+    credential_path,
+    managed,
+    created_at,
+    updated_at,
+    retained_at`
 
 const getByTenantUIDStatement = `
 SELECT