feat(cli): add storage resync-identity-sequences admin subcommand

docs/configuration.md

@@ -18,6 +18,7 @@
 - [Pending Drops](#pending-drops)
 - [Direct Execution](#direct-execution)
 - [Storage Dialect](#storage-dialect)
+  - [Resyncing PostgreSQL identity sequences](#resyncing-postgresql-identity-sequences)
 - [Storage Connection Pool](#storage-connection-pool)
 - [Spirit Run Settings](#spirit-run-settings)
 - [PlanetScale mTLS](#planetscale-mtls)
@@ -503,6 +504,29 @@ supports both `mysql` and `postgres`.) See
 [Storage Schema Changes](#storage-schema-changes) for how schema
 bootstrapping differs between the two dialects.
 
+### Resyncing PostgreSQL identity sequences
+
+After an explicit-ID bulk load into PostgreSQL storage, advance the identity
+sequences before SchemaBot resumes default inserts. Run the command only after
+the load has fully committed. It advances sequences when needed, never rewinds
+them, and is idempotent and safe to rerun.
+
+Pass the storage DSN directly:
+
+```shell
+schemabot storage resync-identity-sequences --dsn "$STORAGE_DSN"
+```
+
+Or resolve it from the server configuration:
+
+```shell
+schemabot storage resync-identity-sequences --config /etc/schemabot/config.yaml
+```
+
+The command refuses to run when none of SchemaBot's storage tables exist in
+the target database. Confirm the target and complete the resync before
+restarting the server or otherwise allowing default inserts.
+
 ## Storage Connection Pool
 
 SchemaBot tunes the `database/sql` pool for its internal storage database with

pkg/api/resync_postgres_identity.go

@@ -36,33 +36,48 @@ import (
 func ResyncPostgresIdentitySequences(ctx context.Context, db *sql.DB, logger *slog.Logger) error {
 	tables, _, err := readEmbeddedPostgresSchemaFiles()
 	if err != nil {
-		return err
+		return fmt.Errorf("read embedded PostgreSQL storage schema: %w", err)
+	}
+	missing, err := missingPostgresTables(ctx, db, tables)
+	if err != nil {
+		return fmt.Errorf("check target for storage tables: %w", err)
+	}
+	if len(missing) == len(tables) {
+		return fmt.Errorf("none of the %d storage tables exist in the target database; it does not look like SchemaBot's storage database", len(tables))
 	}
 	columns, err := postgresIdentityColumns(ctx, db, tables)
 	if err != nil {
-		return err
+		return fmt.Errorf("find identity columns on storage tables: %w", err)
 	}
 
+	advanced := 0
+	skipped := 0
 	for _, col := range columns {
 		newValue, outcome, err := advancePostgresIdentitySequence(ctx, db, col)
 		if err != nil {
-			return err
+			return fmt.Errorf("resync identity column %s.%s: %w", col.table, col.column, err)
 		}
 		switch outcome {
 		case sequenceAdvanced:
+			advanced++
 			logger.Info("advanced identity sequence past stored maximum",
 				"table", col.table, "column", col.column, "sequence_value", newValue)
 		case sequenceSkippedEmptyTable:
+			skipped++
 			logger.Debug("identity column has no stored rows; sequence left untouched",
 				"table", col.table, "column", col.column)
 		case sequenceSkippedAlreadyAhead:
+			skipped++
 			logger.Debug("identity sequence already at or past stored maximum; left untouched",
 				"table", col.table, "column", col.column)
 		case sequenceSkippedDescending:
+			skipped++
 			logger.Warn("identity sequence is descending; resync skipped it — a stored-maximum resync only applies to ascending sequences",
 				"table", col.table, "column", col.column)
 		}
 	}
+	logger.Info("identity sequence resync summary",
+		"tables", len(tables), "examined", len(columns), "advanced", advanced, "skipped", skipped)
 	return nil
 }
 

pkg/api/resync_postgres_identity_integration_test.go

@@ -102,6 +102,16 @@ func resyncLogger(t *testing.T) (*slog.Logger, *recordingLogHandler) {
 	return slog.New(handler), handler
 }
 
+// A target without any SchemaBot storage tables is rejected before sequence
+// discovery, preventing an unrelated PostgreSQL database from appearing to
+// resync successfully.
+func TestResyncPostgresIdentitySequences_RejectsTargetWithoutStorageTables(t *testing.T) {
+	_, db := startPostgresStorage(t)
+	err := ResyncPostgresIdentitySequences(t.Context(), db, slog.New(slog.DiscardHandler))
+	require.ErrorContains(t, err, "none of the")
+	require.ErrorContains(t, err, "does not look like SchemaBot's storage database")
+}
+
 // After an explicit-id bulk load, the identity sequences still point below
 // the loaded rows — GENERATED BY DEFAULT AS IDENTITY accepts explicit values
 // without advancing the sequence — so the next default insert collides with

pkg/cmd/commands/storage.go

@@ -0,0 +1,142 @@
+package commands
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/block/spirit/pkg/utils"
+
+	"github.com/block/schemabot/pkg/api"
+	"github.com/block/schemabot/pkg/postgresconn"
+	"github.com/block/schemabot/pkg/schema"
+)
+
+// StorageCmd groups operator commands that act directly on SchemaBot's own
+// storage database. Unlike the API-client commands, these connect to storage
+// themselves and work while the server is down — they exist for maintenance
+// windows such as a cross-dialect data move or a restore from a dump.
+type StorageCmd struct {
+	ResyncIdentitySequences ResyncIdentitySequencesCmd `cmd:"" name:"resync-identity-sequences" help:"Advance PostgreSQL identity sequences on storage tables past their columns' stored maxima after an explicit-id bulk load; run after the load has fully committed and before default inserts resume — advance-only and safe to rerun."`
+}
+
+// ResyncIdentitySequencesCmd resyncs the identity sequences of SchemaBot's
+// PostgreSQL storage tables after an explicit-id bulk load — a data move
+// that preserves ids, or a restore from a dump without sequence state —
+// so default inserts resume above the loaded ids instead of colliding with
+// them. Run it after the load has fully committed and before the server
+// resumes default inserts. The resync is advance-only and idempotent, so
+// rerunning it is safe.
+//
+// The storage DSN comes from --dsn directly, or from the server config
+// (--config, falling back to $SCHEMABOT_CONFIG_FILE) whose storage dialect
+// must be postgres.
+type ResyncIdentitySequencesCmd struct {
+	DSN    string `help:"PostgreSQL DSN of the storage database to resync; bypasses the server config"`
+	Config string `help:"Server config file to resolve the storage DSN from; defaults to $SCHEMABOT_CONFIG_FILE when neither flag is set"`
+}
+
+// storagePingTimeout bounds the connection check so an unreachable storage
+// database fails the command promptly instead of hanging it.
+const storagePingTimeout = 10 * time.Second
+
+func (cmd *ResyncIdentitySequencesCmd) Run(ctx context.Context, g *Globals) error {
+	// A text handler, deliberately: this is a one-shot operator command read
+	// at a terminal during a maintenance window, not a long-running server
+	// whose stdout feeds a JSON log collector. Diagnostics go to stderr so
+	// stdout stays free for machine-readable output.
+	logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{
+		Level: logLevel(),
+	})).With("schemabot_version", g.Version)
+
+	dsn, source, err := cmd.resolveStorageDSN()
+	if err != nil {
+		return err
+	}
+	logger.Info("resolved storage DSN", "source", source)
+
+	db, err := postgresconn.Open(dsn)
+	if err != nil {
+		return fmt.Errorf("open storage database: %w", err)
+	}
+	defer utils.CloseAndLog(db)
+	pingCtx, cancel := context.WithTimeout(ctx, storagePingTimeout)
+	defer cancel()
+	if err := db.PingContext(pingCtx); err != nil {
+		return fmt.Errorf("ping storage database: %w", err)
+	}
+
+	if err := api.ResyncPostgresIdentitySequences(ctx, db, logger); err != nil {
+		return fmt.Errorf("resync identity sequences on storage tables: %w", err)
+	}
+	logger.Info("identity sequence resync complete")
+	return nil
+}
+
+// resolveStorageDSN returns the storage DSN and a loggable description of
+// where it came from. A direct --dsn is used as-is; otherwise the server
+// config (--config, then $SCHEMABOT_CONFIG_FILE) is loaded and its resolved
+// storage DSN is used, failing closed when the configured storage dialect is
+// not postgres. The source never contains the DSN itself, which may embed
+// credentials.
+func (cmd *ResyncIdentitySequencesCmd) resolveStorageDSN() (string, string, error) {
+	directDSN := strings.TrimSpace(cmd.DSN)
+	if directDSN != "" && cmd.Config != "" {
+		return "", "", fmt.Errorf("--dsn and --config are mutually exclusive; pass the storage DSN directly or resolve it from a server config, not both")
+	}
+	if cmd.DSN != "" {
+		if directDSN == "" {
+			return "", "", fmt.Errorf("storage DSN not configured: --dsn contains only whitespace")
+		}
+		return directDSN, "--dsn flag", nil
+	}
+
+	configPath := cmd.Config
+	source := fmt.Sprintf("server config %s", configPath)
+	if configPath == "" {
+		configPath = os.Getenv("SCHEMABOT_CONFIG_FILE")
+		if configPath == "" {
+			return "", "", fmt.Errorf("no storage DSN source: set --dsn, --config, or the SCHEMABOT_CONFIG_FILE environment variable")
+		}
+		source = fmt.Sprintf("server config %s ($SCHEMABOT_CONFIG_FILE)", configPath)
+	}
+
+	var cfg *api.ServerConfig
+	var err error
+	if cmd.Config == "" {
+		cfg, err = api.LoadServerConfig()
+	} else {
+		cfg, err = api.LoadServerConfigFromFile(configPath)
+	}
+	if err != nil {
+		return "", "", fmt.Errorf("load %s: %w", source, err)
+	}
+
+	dialect, err := cfg.Storage.ResolveDialect()
+	if err != nil {
+		return "", "", fmt.Errorf("resolve storage dialect from %s: %w", source, err)
+	}
+	if dialect != schema.DialectPostgres {
+		return "", "", fmt.Errorf("storage dialect in %s is %q; the identity sequence resync only applies to %q storage", source, dialect, schema.DialectPostgres)
+	}
+
+	dsn, err := cfg.StorageDSN()
+	if err != nil {
+		return "", "", fmt.Errorf("resolve storage DSN from %s: %w", source, err)
+	}
+	dsn = strings.TrimSpace(dsn)
+	if dsn == "" {
+		return "", "", fmt.Errorf("storage DSN not configured (set --dsn, config storage.dsn or storage.dsn_from, STORAGE_DSN, or MYSQL_DSN)")
+	}
+	if cfg.Storage.DSN == "" && cfg.Storage.DSNFrom == nil {
+		if strings.TrimSpace(os.Getenv("STORAGE_DSN")) != "" {
+			source = "STORAGE_DSN environment variable"
+		} else if strings.TrimSpace(os.Getenv("MYSQL_DSN")) != "" {
+			source = "MYSQL_DSN environment variable"
+		}
+	}
+	return dsn, source, nil
+}

pkg/cmd/commands/storage_integration_test.go

@@ -0,0 +1,87 @@
+//go:build integration
+
+package commands
+
+import (
+	"database/sql"
+	"errors"
+	"log/slog"
+	"testing"
+
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/stretchr/testify/require"
+
+	"github.com/block/schemabot/pkg/api"
+	"github.com/block/schemabot/pkg/schema"
+	"github.com/block/schemabot/pkg/testutil"
+)
+
+// startResyncStorage boots a PostgreSQL storage database with the full
+// storage schema and seeds the settings table with explicit-id rows, the way
+// an id-preserving bulk load writes them. The seeded rows leave the identity
+// sequence behind the stored maximum, so a default insert collides until the
+// resync runs.
+func startResyncStorage(t *testing.T) (string, *sql.DB) {
+	t.Helper()
+	dsn, db := testutil.StartPostgres(t, "schemabot")
+	require.NoError(t, api.EnsureSchema(dsn, slog.New(slog.DiscardHandler), api.WithDialect(schema.DialectPostgres)))
+	for id, key := range map[int64]string{1: "loaded-1", 2: "loaded-2", 3: "loaded-3"} {
+		_, err := db.ExecContext(t.Context(),
+			`INSERT INTO settings (id, setting_key, setting_value) VALUES ($1, $2, '')`, id, key)
+		require.NoError(t, err)
+	}
+	return dsn, db
+}
+
+// requireDefaultInsertResumes asserts that a default insert into settings
+// draws the id above the loaded maximum, proving the resync advanced the
+// sequence past the explicit-id rows.
+func requireDefaultInsertResumes(t *testing.T, db *sql.DB) {
+	t.Helper()
+	var id int64
+	err := db.QueryRowContext(t.Context(),
+		`INSERT INTO settings (setting_key, setting_value) VALUES ('after-resync', '') RETURNING id`).Scan(&id)
+	require.NoError(t, err, "default insert must succeed after the resync")
+	require.Equal(t, int64(4), id, "the first default insert after the resync draws max+1")
+}
+
+func requireDefaultInsertCollides(t *testing.T, db *sql.DB) {
+	t.Helper()
+	_, err := db.ExecContext(t.Context(),
+		`INSERT INTO settings (setting_key, setting_value) VALUES ('before-resync', '')`)
+	require.Error(t, err)
+	var pgErr *pgconn.PgError
+	require.True(t, errors.As(err, &pgErr), "default insert must return a PostgreSQL error")
+	require.Equal(t, "23505", pgErr.Code, "default insert must collide before the resync")
+}
+
+// After an explicit-id bulk load, the operator runs the resync subcommand
+// with the storage DSN passed directly; afterwards default inserts resume
+// above the loaded ids instead of colliding with them.
+func TestResyncIdentitySequencesCmd_DSNFlag(t *testing.T) {
+	dsn, db := startResyncStorage(t)
+	requireDefaultInsertCollides(t, db)
+
+	cmd := &ResyncIdentitySequencesCmd{DSN: dsn}
+	require.NoError(t, cmd.Run(t.Context(), &Globals{Version: "test"}))
+
+	requireDefaultInsertResumes(t, db)
+}
+
+// In a deployed pod the operator points the resync subcommand at the server
+// config instead of hand-building a DSN; the command resolves the storage
+// DSN from the config's storage section and resyncs the same way.
+func TestResyncIdentitySequencesCmd_ConfigFile(t *testing.T) {
+	dsn, db := startResyncStorage(t)
+	requireDefaultInsertCollides(t, db)
+
+	path := writeStorageTestConfig(t, `
+storage:
+  dialect: postgres
+  dsn: `+dsn+`
+`)
+	cmd := &ResyncIdentitySequencesCmd{Config: path}
+	require.NoError(t, cmd.Run(t.Context(), &Globals{Version: "test"}))
+
+	requireDefaultInsertResumes(t, db)
+}