v0.12.0 migrate sqlite3 to psql18, updated auth, deploy docs, account and replay updates

internal/server/storage/storage.go

@@ -3,25 +3,43 @@ package storage
 import (
 	"context"
 	"database/sql"
+	"database/sql/driver"
 	"errors"
 	"fmt"
+	"io"
 	"log/slog"
-	"os"
+	"net"
 	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 
-	_ "github.com/mattn/go-sqlite3"
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/stdlib"
 )
 
 const (
 	writeQueueCapacity = 1000
 	flushTimeout       = 5 * time.Second
-	schemaVersion      = 2
-)
+	queryTimeout       = 5 * time.Second
+	writeTimeout       = 10 * time.Second
+	migrationTimeout   = time.Minute
 
-var memoryStoreCounter atomic.Uint64
+	// A transient failure (PostgreSQL restart, dropped connection) is retried
+	// before storage degrades. Backoff doubles from writeRetryBase, bounding
+	// the stall of the ordered writer to roughly three seconds.
+	writeAttempts  = 6
+	writeRetryBase = 100 * time.Millisecond
+
+	maxOpenConns = 10
+	maxIdleConns = 5
+
+	// schemaLockKey serializes migrations between the server and the CLI.
+	// Advisory locks are scoped to the current database.
+	schemaLockKey int64 = 0x6368657373 // "chess"
+)
 
 var (
 	ErrStorageDegraded = errors.New("storage is degraded")
@@ -32,14 +50,14 @@ var (
 type writeRequest struct {
 	operation string
 	gameID    string
-	run       func(*sql.Tx) error
+	run       func(context.Context, *sql.Tx) error
 	barrier   chan error
 }
 
-// Store handles SQLite database operations with async writes for games and sync writes for auth
+// Store handles PostgreSQL operations with ordered async writes for gameplay
+// and synchronous writes for accounts and sessions.
 type Store struct {
 	db           *sql.DB
-	path         string
 	writeChan    chan writeRequest
 	healthStatus atomic.Bool
 	writeFailed  atomic.Bool
@@ -52,68 +70,55 @@ type Store struct {
 	closeErr     error
 }
 
-// NewStore creates a new storage instance with async writer
-func NewStore(dataSourceName string, devMode bool) (*Store, error) {
-	dsn := sqliteDSN(dataSourceName)
-	db, err := sql.Open("sqlite3", dsn)
+// NewStore connects to PostgreSQL and starts the async writer. dsn is a
+// libpq-style keyword/value string or postgres:// URL; unset fields fall back
+// to the standard PG* environment variables and ~/.pgpass. The schema is
+// resolved through search_path, so the role's default or a search_path DSN
+// parameter selects it.
+func NewStore(dsn string) (*Store, error) {
+	config, err := pgx.ParseConfig(dsn)
 	if err != nil {
-		return nil, fmt.Errorf("failed to open database: %w", err)
+		// Parse errors may echo the DSN, which can contain a password.
+		return nil, errors.New("invalid database connection string")
+	}
+	if _, ok := config.RuntimeParams["application_name"]; !ok {
+		config.RuntimeParams["application_name"] = "chess-server"
 	}
 
-	if err := db.Ping(); err != nil {
+	db := stdlib.OpenDB(*config)
+	db.SetMaxOpenConns(maxOpenConns)
+	db.SetMaxIdleConns(maxIdleConns)
+	db.SetConnMaxIdleTime(5 * time.Minute)
+	db.SetConnMaxLifetime(30 * time.Minute)
+
+	ctx, cancel := context.WithTimeout(context.Background(), queryTimeout)
+	defer cancel()
+	if err := db.PingContext(ctx); err != nil {
 		db.Close()
 		return nil, fmt.Errorf("failed to connect to database: %w", err)
 	}
 
-	// SQLite benefits from a small pool. WAL and busy_timeout are configured in
-	// the DSN for every connection, unlike connection-local PRAGMA calls.
-	db.SetMaxOpenConns(8)
-	db.SetMaxIdleConns(4)
-
-	ctx, cancel := context.WithCancel(context.Background())
-
+	writerCtx, writerCancel := context.WithCancel(context.Background())
 	s := &Store{
 		db:        db,
-		path:      dataSourceName,
 		writeChan: make(chan writeRequest, writeQueueCapacity),
-		ctx:       ctx,
-		cancel:    cancel,
+		ctx:       writerCtx,
+		cancel:    writerCancel,
 	}
-
-	// Initialize health as true
 	s.healthStatus.Store(true)
 
-	// Start async writer
 	s.wg.Add(1)
 	go s.writerLoop()
 	slog.Debug("storage opened",
-		"path", dataSourceName,
-		"dev_mode", devMode,
+		"host", config.Host,
+		"database", config.Database,
+		"user", config.User,
 		"write_queue_capacity", writeQueueCapacity,
-		"max_open_connections", 8,
+		"max_open_connections", maxOpenConns,
 	)
-
 	return s, nil
 }
 
-func sqliteDSN(dataSourceName string) string {
-	if dataSourceName == ":memory:" {
-		// Each Store needs a private shared-cache database: shared cache keeps
-		// that Store's pooled connections on one database, while the unique name
-		// prevents independent in-memory stores from leaking into each other.
-		dataSourceName = fmt.Sprintf(
-			"file:chess-memory-%d?mode=memory&cache=shared",
-			memoryStoreCounter.Add(1),
-		)
-	}
-	separator := "?"
-	if strings.Contains(dataSourceName, "?") {
-		separator = "&"
-	}
-	return dataSourceName + separator +
-		"_foreign_keys=on&_busy_timeout=5000&_journal_mode=WAL&_synchronous=NORMAL"
-}
-
 // IsHealthy returns true if the storage is operational
 func (s *Store) IsHealthy() bool {
 	return s.healthStatus.Load()
@@ -176,49 +181,63 @@ func (s *Store) handleWrite(req writeRequest) {
 	}
 }
 
-// executeWrite runs a transactional write operation
+// executeWrite runs one logical mutation as one transaction. A transient error
+// raised before COMMIT is retried: the transaction was rolled back, so a retry
+// cannot duplicate it, and the single writer preserves ordering while it waits.
+// A COMMIT error has an unknown outcome and is never retried. Any unrecovered
+// failure degrades storage permanently, because skipping one write would leave
+// later writes applied to an incomplete history.
 func (s *Store) executeWrite(req writeRequest) error {
 	started := time.Now()
-	tx, err := s.db.Begin()
-	if err != nil {
-		s.writeFailed.Store(true)
-		s.healthStatus.Store(false)
-		slog.Error("storage degraded: failed to begin transaction",
-			"operation", req.operation, "game_id", req.gameID, "error", err)
-		return err
+	var err error
+	for attempt := 1; ; attempt++ {
+		var committing bool
+		committing, err = s.runWriteTx(req.run)
+		if err == nil {
+			slog.Debug("storage write committed",
+				"operation", req.operation,
+				"game_id", req.gameID,
+				"attempts", attempt,
+				"duration", time.Since(started),
+				"queue_depth", len(s.writeChan),
+			)
+			return nil
+		}
+		if committing || attempt >= writeAttempts || !isTransient(err) {
+			break
+		}
+		backoff := writeRetryBase << (attempt - 1)
+		slog.Warn("storage write failed transiently; retrying",
+			"operation", req.operation, "game_id", req.gameID,
+			"attempt", attempt, "backoff", backoff, "error", err)
+		time.Sleep(backoff)
 	}
 
-	if err := req.run(tx); err != nil {
-		rollbackErr := tx.Rollback()
-		s.writeFailed.Store(true)
-		s.healthStatus.Store(false)
-		slog.Error("storage degraded: write operation failed",
-			"operation", req.operation,
-			"game_id", req.gameID,
-			"error", err,
-			"rollback_error", rollbackErr,
-		)
-		return err
-	}
-
-	if err := tx.Commit(); err != nil {
-		s.writeFailed.Store(true)
-		s.healthStatus.Store(false)
-		slog.Error("storage degraded: failed to commit",
-			"operation", req.operation, "game_id", req.gameID, "error", err)
-		return err
-	}
-
-	slog.Debug("storage write committed",
-		"operation", req.operation,
-		"game_id", req.gameID,
-		"duration", time.Since(started),
-		"queue_depth", len(s.writeChan),
-	)
-	return nil
+	s.writeFailed.Store(true)
+	s.healthStatus.Store(false)
+	slog.Error("storage degraded: write operation failed",
+		"operation", req.operation, "game_id", req.gameID, "error", err)
+	return err
 }
 
-func (s *Store) enqueue(operation, gameID string, fn func(*sql.Tx) error) error {
+func (s *Store) runWriteTx(run func(context.Context, *sql.Tx) error) (committing bool, err error) {
+	ctx, cancel := context.WithTimeout(context.Background(), writeTimeout)
+	defer cancel()
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return false, fmt.Errorf("begin transaction: %w", err)
+	}
+	if err := run(ctx, tx); err != nil {
+		_ = tx.Rollback()
+		return false, err
+	}
+	if err := tx.Commit(); err != nil {
+		return true, fmt.Errorf("commit transaction: %w", err)
+	}
+	return true, nil
+}
+
+func (s *Store) enqueue(operation, gameID string, fn func(context.Context, *sql.Tx) error) error {
 	s.enqueueMu.RLock()
 	defer s.enqueueMu.RUnlock()
 
@@ -285,7 +304,7 @@ func (s *Store) flushBeforeRead() error {
 	return s.Flush(ctx)
 }
 
-// Close gracefully closes the database connection
+// Close drains accepted writes and closes the connection pool.
 func (s *Store) Close() error {
 	s.closeOnce.Do(func() {
 		s.enqueueMu.Lock()
@@ -301,125 +320,189 @@ func (s *Store) Close() error {
 	return s.closeErr
 }
 
-// InitDB creates the database schema
+// InitDB applies pending migrations. It holds a transaction-scoped advisory
+// lock so concurrent migrators serialize, and issues no DDL when the schema is
+// current: a runtime role limited to DML can start the server once an owner
+// role has migrated the schema.
 func (s *Store) InitDB() error {
 	started := time.Now()
-	tx, err := s.db.Begin()
+	ctx, cancel := context.WithTimeout(context.Background(), migrationTimeout)
+	defer cancel()
+
+	tx, err := s.db.BeginTx(ctx, nil)
 	if err != nil {
 		return fmt.Errorf("failed to begin transaction: %w", err)
 	}
 	defer tx.Rollback()
 
-	var currentVersion int
-	if err := tx.QueryRow("PRAGMA user_version").Scan(&currentVersion); err != nil {
-		return fmt.Errorf("failed to read schema version: %w", err)
+	if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock($1)`, schemaLockKey); err != nil {
+		return fmt.Errorf("failed to lock schema: %w", err)
 	}
-	if currentVersion > schemaVersion {
-		return fmt.Errorf(
-			"database schema version %d is newer than supported version %d",
-			currentVersion,
-			schemaVersion,
-		)
+	current, err := readSchemaVersion(ctx, tx)
+	if err != nil {
+		return err
+	}
+	if current > schemaVersion {
+		return fmt.Errorf("database schema version %d is newer than supported version %d",
+			current, schemaVersion)
+	}
+	if current == schemaVersion {
+		slog.Debug("storage schema current", "version", current)
+		return tx.Commit()
 	}
 
-	if _, err := tx.Exec(Schema); err != nil {
-		return fmt.Errorf("failed to create schema: %w", err)
-	}
-
-	columns := []struct {
-		name       string
-		definition string
-	}{
-		{"result", "TEXT CHECK(result IS NULL OR result IN ('white_wins', 'black_wins', 'draw', 'stalemate'))"},
-		{"end_time_utc", "DATETIME"},
-		{"white_claimed_by", "TEXT"},
-		{"black_claimed_by", "TEXT"},
-	}
-	for _, column := range columns {
-		if err := ensureColumn(tx, "games", column.name, column.definition); err != nil {
-			return err
+	for version := current + 1; version <= schemaVersion; version++ {
+		if _, err := tx.ExecContext(ctx, migrations[version-1]); err != nil {
+			return fmt.Errorf("failed to apply schema version %d: %w", version, err)
 		}
 	}
-
-	// These indexes duplicate UNIQUE constraints or are superseded by targeted
-	// partial/composite indexes. Drop them during upgrades as well as omitting
-	// them from new databases.
-	for _, name := range []string{
-		"idx_users_username",
-		"idx_users_email",
-		"idx_users_account_type",
-		"idx_users_expires_at",
-		"idx_sessions_user_id",
-		"idx_moves_game_id",
-		"idx_games_finished_end_time",
-	} {
-		if _, err := tx.Exec("DROP INDEX IF EXISTS " + name); err != nil {
-			return fmt.Errorf("failed to remove redundant index %s: %w", name, err)
-		}
-	}
-
-	if _, err := tx.Exec(Indexes); err != nil {
-		return fmt.Errorf("failed to create indexes: %w", err)
-	}
-	if _, err := tx.Exec(fmt.Sprintf("PRAGMA user_version = %d", schemaVersion)); err != nil {
+	if _, err := tx.ExecContext(ctx, `
+		INSERT INTO schema_version (version) VALUES ($1)
+		ON CONFLICT (singleton) DO UPDATE SET version = excluded.version, updated_at = now()`,
+		schemaVersion,
+	); err != nil {
 		return fmt.Errorf("failed to record schema version: %w", err)
 	}
-
 	if err := tx.Commit(); err != nil {
 		return fmt.Errorf("failed to commit schema: %w", err)
 	}
-	slog.Debug("storage schema ready", "version", schemaVersion, "duration", time.Since(started))
+	slog.Info("storage schema migrated", "from", current, "to", schemaVersion,
+		"duration", time.Since(started))
 	return nil
 }
 
-func ensureColumn(tx *sql.Tx, table, column, definition string) error {
-	rows, err := tx.Query("PRAGMA table_info(" + table + ")")
+// SchemaVersion reports the version recorded in the database; 0 means no
+// schema has been created in the current search_path.
+func (s *Store) SchemaVersion() (int, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), queryTimeout)
+	defer cancel()
+	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
 	if err != nil {
-		return fmt.Errorf("failed to inspect %s schema: %w", table, err)
+		return 0, err
 	}
+	defer tx.Rollback()
+	return readSchemaVersion(ctx, tx)
+}
 
-	found := false
-	for rows.Next() {
-		var cid int
-		var name, columnType string
-		var notNull, primaryKey int
-		var defaultValue any
-		if err := rows.Scan(&cid, &name, &columnType, &notNull, &defaultValue, &primaryKey); err != nil {
-			rows.Close()
-			return fmt.Errorf("failed to inspect %s column: %w", table, err)
-		}
-		if name == column {
-			found = true
+func readSchemaVersion(ctx context.Context, tx *sql.Tx) (int, error) {
+	var exists bool
+	if err := tx.QueryRowContext(ctx,
+		`SELECT to_regclass('schema_version') IS NOT NULL`,
+	).Scan(&exists); err != nil {
+		return 0, fmt.Errorf("failed to inspect schema: %w", err)
+	}
+	if !exists {
+		return 0, nil
+	}
+	var version int
+	err := tx.QueryRowContext(ctx, `SELECT version FROM schema_version`).Scan(&version)
+	if errors.Is(err, sql.ErrNoRows) {
+		return 0, nil
+	}
+	if err != nil {
+		return 0, fmt.Errorf("failed to read schema version: %w", err)
+	}
+	return version, nil
+}
+
+// DropSchema removes every chess table from the current search_path. It is
+// destructive and requires ownership of the tables.
+func (s *Store) DropSchema() error {
+	ctx, cancel := context.WithTimeout(context.Background(), migrationTimeout)
+	defer cancel()
+	// RESTRICT (the default) refuses to drop objects that others depend on,
+	// such as an administrator's reporting view.
+	_, err := s.db.ExecContext(ctx, `DROP TABLE IF EXISTS `+strings.Join(ownedTables, ", "))
+	if err != nil {
+		return fmt.Errorf("failed to drop schema: %w", err)
+	}
+	return nil
+}
+
+// opContext bounds a synchronous operation so a stalled database cannot hold a
+// request handler indefinitely.
+func opContext() (context.Context, context.CancelFunc) {
+	return context.WithTimeout(context.Background(), queryTimeout)
+}
+
+// isTransient reports errors after which the same transaction may succeed on a
+// fresh connection.
+func isTransient(err error) bool {
+	var pgErr *pgconn.PgError
+	if errors.As(err, &pgErr) {
+		switch pgErr.Code {
+		case "40001", // serialization_failure
+			"40P01", // deadlock_detected
+			"57P01", // admin_shutdown
+			"57P02", // crash_shutdown
+			"57P03": // cannot_connect_now
+			return true
 		}
+		return strings.HasPrefix(pgErr.Code, "08") // connection_exception class
 	}
-	if err := rows.Close(); err != nil {
-		return fmt.Errorf("failed to close %s schema rows: %w", table, err)
+	if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
+		return false
 	}
-	if err := rows.Err(); err != nil {
-		return fmt.Errorf("failed to inspect %s schema: %w", table, err)
+	var connectErr *pgconn.ConnectError
+	var netErr net.Error
+	return errors.As(err, &connectErr) ||
+		errors.As(err, &netErr) ||
+		errors.Is(err, driver.ErrBadConn) ||
+		errors.Is(err, io.EOF) ||
+		errors.Is(err, io.ErrUnexpectedEOF) ||
+		pgconn.SafeToRetry(err)
+}
+
+// uniqueViolation returns the violated constraint name for SQLSTATE 23505.
+func uniqueViolation(err error) (string, bool) {
+	var pgErr *pgconn.PgError
+	if errors.As(err, &pgErr) && pgErr.Code == "23505" {
+		return pgErr.ConstraintName, true
 	}
-	if found {
+	return "", false
+}
+
+// validUUID accepts the canonical 36-character form. Lookups treat anything
+// else as not found instead of sending PostgreSQL a value that fails to cast.
+func validUUID(value string) bool {
+	if len(value) != 36 {
+		return false
+	}
+	_, err := uuid.Parse(value)
+	return err == nil
+}
+
+func nullableString(value string) any {
+	if value == "" {
 		return nil
 	}
-
-	if _, err := tx.Exec("ALTER TABLE " + table + " ADD COLUMN " + column + " " + definition); err != nil {
-		return fmt.Errorf("failed to add %s.%s: %w", table, column, err)
-	}
-	slog.Debug("storage schema column added", "table", table, "column", column)
-	return nil
+	return value
 }
 
-// DeleteDB removes the database file
-func (s *Store) DeleteDB() error {
-	// Close connection first
-	if err := s.Close(); err != nil {
-		return fmt.Errorf("failed to close database: %w", err)
+func utcPointer(value sql.NullTime) *time.Time {
+	if !value.Valid {
+		return nil
 	}
+	t := value.Time.UTC()
+	return &t
+}
 
-	// ☣ DESTRUCTIVE: Removes database file
-	if err := os.Remove(s.path); err != nil && !os.IsNotExist(err) {
-		return fmt.Errorf("failed to delete database file: %w", err)
+// ErrSchemaMismatch reports a database whose schema is absent or at a version
+// other than the one this binary supports.
+var ErrSchemaMismatch = errors.New("database schema version mismatch")
+
+// CheckSchema verifies that the schema is current without changing it.
+func (s *Store) CheckSchema() error {
+	version, err := s.SchemaVersion()
+	if err != nil {
+		return err
+	}
+	switch {
+	case version == 0:
+		return fmt.Errorf("%w: schema not initialized (run 'chess-server db init')", ErrSchemaMismatch)
+	case version != schemaVersion:
+		return fmt.Errorf("%w: database has version %d, binary supports %d",
+			ErrSchemaMismatch, version, schemaVersion)
 	}
-
 	return nil
 }