feat: add PostgreSQL migration runner

multiplayer-next.md

@@ -41,7 +41,8 @@ product policy are in [`docs/MATCHMAKING.md`](docs/MATCHMAKING.md).
   and serializable store boundaries are implemented, including durable queue
   create/heartbeat/cancel/recovery adapters; an opt-in pgx/Docker harness now
   executes the migrations and real queue create/idempotency/ownership/recovery
-  path, and a TTL-bound Redis candidate index now supports atomic rebuild,
+  path, an executable migration runner now serializes and records forward
+  application, and a TTL-bound Redis candidate index now supports atomic rebuild,
   snapshot and removal with durable-source repair on partial/malformed cache
   state; proposal/result transactions and live Redis restart/failover gates
   remain.

multiplayer-todo.md

@@ -1173,7 +1173,7 @@ the local/CI/community transport, not a silent production fallback.
 | 8.2 `[D:8.1]` | **DONE.** Encode the launch SLOs from `docs/MATCHMAKING.md`: RTT, allocation/connect latency, 99.9% allocation/result success, API latency and tick health | [`docs/MATCHMAKING-SLOs.md`](docs/MATCHMAKING-SLOs.md) defines each metric, denominator, percentile/window, owner, alert threshold and release evidence |
 | 8.3 `[D:8.1]` | **DONE.** Publish versioned OpenAPI + WebSocket contracts for Steam login/session, profile/rating, queue create/heartbeat/cancel/resume, proposal accept/decline, assignment/status, server registration/roster/result/shutdown | [`server/contracts/v1/`](server/contracts/v1/) contains machine-readable REST/events contracts and dependency-free structural tests; REST resync is specified by the contract; `server/api/service.go` also exposes the documented `/api/v1` route names (including server-assigned idempotent queue ticket IDs and DELETE cancellation) alongside the existing client `/v1` routes, covered by `TestDocumentedContractRoutesAdaptToServiceAPI` |
 | 8.4 `[D:8.3]` | **DONE.** Define opaque IDs, legal queue/match state transitions, revisions and idempotency keys | [`server/contracts/v1/state-transitions.json`](server/contracts/v1/state-transitions.json) locks terminal states, legal edges, stale-revision handling and same-key replay/conflict behavior; contract tests cover the invariants |
-| 8.5 `[D:8.4]` | **IN PROGRESS.** Initial PostgreSQL migration now defines durable idempotency keys, queue ownership/active-participation fencing, identities, sessions/revocations, ranked seasons, ratings/events, matches/participants, penalties, results, audits and outbox | `server/migrations/0001_initial.sql` and static checks cover the durable tables, uniqueness/check constraints and Redis-as-cache boundary; opt-in `scripts/run_postgres_integration.sh` now runs the migrations and real queue/assignment ownership, idempotency, revision and expiry checks through pgx; rollback/forward migration, the remaining serializable adapters and cache-loss repair remain |
+| 8.5 `[D:8.4]` | **IN PROGRESS.** Initial PostgreSQL migration now defines durable idempotency keys, queue ownership/active-participation fencing, identities, sessions/revocations, ranked seasons, ratings/events, matches/participants, penalties, results, audits and outbox | `server/migrations/0001_initial.sql`, `migrations/runner.go`, `cmd/migrate` and static checks cover the durable tables, uniqueness/check constraints, Redis-as-cache boundary and serialized forward migration recording; opt-in `scripts/run_postgres_integration.sh` now runs the runner and real queue/assignment ownership, idempotency, revision and expiry checks through pgx; rollback/down migration, the remaining serializable adapters and cache-loss repair remain |
 | 8.6 `[D:8.3,8.4]` | **IN PROGRESS.** Add allocated-mode `ServerConfig` compatibility fields as opt-in defaults | `ServerConfig` now validates allocation mode, match/server IDs, playlist version, client build, future assignment expiry, image digest, transport and EU/NA region; `server_boot.gd` fails closed for the not-yet-wired Steam SDR transport, constrains allocated processes to one match, and emits allocation identity/transport in `server_started`; signed-authorisation admission, dynamic endpoint wiring and full manifest/runtime tests remain |
 
 #### 8B — Authentication and secure control plane
@@ -1239,7 +1239,7 @@ the local/CI/community transport, not a silent production fallback.
 |---|---|---|
 | 8.44 `[D:8.3,8.4,8.28,8.31]` | **IN PROGRESS.** Go observability package encodes queue/proposal/match/server IDs and lifecycle stage in structured events while recursively redacting auth/relay tokens and credentials | `server/observability/` covers correlation fields, nested secret redaction and unnamed-event rejection; production logger/metrics/traces/replay integration and secret-canary coverage remain |
 | 8.45 `[D:8.2,8.44]` | **IN PROGRESS.** Go observability package turns the documented RTT, allocation/connect latency, result-success, API-latency and tick/headroom thresholds into executable window checks | `server/observability/slo.go` covers healthy/violating/empty windows; production metrics export, dashboards, alert routing, wait/MMR/proposal/flood/cost series and runbooks remain |
-| 8.46 `[D:8.5,8.7,8.9,8.10,8.14,8.18,8.21,8.23,8.25]` | **IN PROGRESS.** Go unit/race coverage spans the current domain/store/supervisor policies, and fuzz targets now exercise queue input, result payload hashing and revision events | `server/domain/*_test.go`, `server/store/*_test.go`, `server/supervisor/*_test.go` and `server/domain/fuzz_test.go` pass normal/race suites; `go test -race ./...` passes across API, domain, observability, store, supervisor and testkit; `go vet ./...` passes; each of the three declared domain fuzz targets passes a bounded 4-second run; longer fuzz campaigns, PostgreSQL concurrency/migration execution, fake Steam/allocator and full lost-Redis/transaction fixtures remain |
+| 8.46 `[D:8.5,8.7,8.9,8.10,8.14,8.18,8.21,8.23,8.25]` | **IN PROGRESS.** Go unit/race coverage spans the current domain/store/supervisor policies, and fuzz targets now exercise queue input, result payload hashing and revision events | `server/domain/*_test.go`, `server/store/*_test.go`, `server/supervisor/*_test.go`, `server/migrations/*_test.go` and `server/domain/fuzz_test.go` pass normal/race suites; `go test -race ./...` passes across API, domain, migrations, observability, store, supervisor and testkit; `go vet ./...` passes; each of the three declared domain fuzz targets passes a bounded 4-second run; PostgreSQL live concurrency/migration execution, fake Steam/allocator and full lost-Redis/transaction fixtures remain |
 | 8.47 `[D:8.7,8.30]` | **IN PROGRESS.** Offline testkit provides deterministic fake Steam verification and fake allocation with forced failure injection | `server/testkit/` covers verified identity/replay, unknown identity, wrong App ID, expiry, no capacity, compatibility-key conflict, idempotent allocation replay and cloud-free forced allocation failure in `TestOfflineFakesCoverVerificationAndAllocationFailureMatrix`; API/Compose integration and live exhaustive matrix remain |
 | 8.48 `[D:8.10,8.14,8.17,8.18,8.27,8.31,8.35,8.47]` | **IN PROGRESS.** Offline testkit exercises verified queue projection → ranked six-player proposal → ENet allocation → assignment-ready manifest → certified durable result receipt | `server/testkit/pipeline_test.go` covers the cross-domain success path without Steam/cloud secrets; independent Compose fixture, process shutdown, result ack over HTTP and legacy fixture non-regression remain |
 | 8.49 `[D:8.25,8.26,8.28,8.29,8.30,8.31,8.35,8.36]` | Disposable `kind` + Agones integration gate | CI covers dynamic ports, both readiness stages, roster/no-show, races, multi-match node, result-pending reconciliation, drain and rollback |

server/cmd/migrate/main.go

@@ -0,0 +1,36 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+	"flag"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/cosmic-clash/cosmic-clash/server/migrations"
+	_ "github.com/jackc/pgx/v5/stdlib"
+)
+
+func main() {
+	dsn := flag.String("dsn", os.Getenv("COSMIC_CLASH_POSTGRES_DSN"), "PostgreSQL connection string")
+	directory := flag.String("dir", "migrations", "directory containing numbered SQL migrations")
+	flag.Parse()
+	if *dsn == "" {
+		fmt.Fprintln(os.Stderr, "migrate: --dsn or COSMIC_CLASH_POSTGRES_DSN is required")
+		os.Exit(2)
+	}
+	db, err := sql.Open("pgx", *dsn)
+	if err != nil {
+		fmt.Fprintln(os.Stderr, "migrate:", err)
+		os.Exit(1)
+	}
+	defer db.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	if err := migrations.Apply(ctx, db, *directory); err != nil {
+		fmt.Fprintln(os.Stderr, "migrate:", err)
+		os.Exit(1)
+	}
+	fmt.Println("migrations applied")
+}

server/migrations/runner.go

@@ -0,0 +1,74 @@
+package migrations
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+)
+
+const migrationTableSQL = `CREATE TABLE IF NOT EXISTS schema_migrations (
+    version TEXT PRIMARY KEY,
+    applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
+)`
+
+// Apply executes numbered SQL files in lexical order. A transaction-level
+// advisory lock serializes concurrent API/worker starts, while each migration
+// is committed together with its schema_migrations marker so a failed
+// migration can be retried safely.
+func Apply(ctx context.Context, db *sql.DB, directory string) error {
+	if db == nil || strings.TrimSpace(directory) == "" {
+		return fmt.Errorf("database and migration directory are required")
+	}
+	paths, err := filepath.Glob(filepath.Join(directory, "*.sql"))
+	if err != nil {
+		return fmt.Errorf("find migrations: %w", err)
+	}
+	sort.Slice(paths, func(i, j int) bool { return filepath.Base(paths[i]) < filepath.Base(paths[j]) })
+	if len(paths) == 0 {
+		return fmt.Errorf("no migrations found in %s", directory)
+	}
+	if _, err := db.ExecContext(ctx, migrationTableSQL); err != nil {
+		return fmt.Errorf("create migration table: %w", err)
+	}
+	for _, path := range paths {
+		version := filepath.Base(path)
+		sqlBytes, err := os.ReadFile(path)
+		if err != nil {
+			return fmt.Errorf("read migration %s: %w", version, err)
+		}
+		tx, err := db.BeginTx(ctx, nil)
+		if err != nil {
+			return fmt.Errorf("begin migration %s: %w", version, err)
+		}
+		committed := false
+		defer func() {
+			if !committed {
+				_ = tx.Rollback()
+			}
+		}()
+		if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock(hashtext('cosmic-clash:migrations'))`); err != nil {
+			return fmt.Errorf("lock migration %s: %w", version, err)
+		}
+		var applied bool
+		if err := tx.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM schema_migrations WHERE version = $1)`, version).Scan(&applied); err != nil {
+			return fmt.Errorf("check migration %s: %w", version, err)
+		}
+		if !applied {
+			if _, err := tx.ExecContext(ctx, string(sqlBytes)); err != nil {
+				return fmt.Errorf("apply migration %s: %w", version, err)
+			}
+			if _, err := tx.ExecContext(ctx, `INSERT INTO schema_migrations (version) VALUES ($1)`, version); err != nil {
+				return fmt.Errorf("record migration %s: %w", version, err)
+			}
+		}
+		if err := tx.Commit(); err != nil {
+			return fmt.Errorf("commit migration %s: %w", version, err)
+		}
+		committed = true
+	}
+	return nil
+}

server/migrations/runner_test.go

@@ -0,0 +1,15 @@
+package migrations
+
+import (
+	"context"
+	"testing"
+)
+
+func TestApplyRejectsMissingDatabaseOrDirectory(t *testing.T) {
+	if err := Apply(context.Background(), nil, "."); err == nil {
+		t.Fatal("nil database accepted")
+	}
+	if err := Apply(context.Background(), nil, ""); err == nil {
+		t.Fatal("empty directory accepted")
+	}
+}

server/store/postgres_integration_test.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"github.com/cosmic-clash/cosmic-clash/server/domain"
+	"github.com/cosmic-clash/cosmic-clash/server/migrations"
 	_ "github.com/jackc/pgx/v5/stdlib"
 )
 
@@ -40,18 +41,11 @@ func openIntegrationPostgres(t *testing.T) *sql.DB {
 
 func applyIntegrationMigrations(t *testing.T, db *sql.DB) {
 	t.Helper()
-	if _, err := db.ExecContext(context.Background(), `DROP TABLE IF EXISTS assignments, audit_events, outbox, result_receipts, ranked_season_rollovers, penalties, seasons, ratings, match_participants, matches, proposal_participants, proposals, queue_tickets, idempotency_keys, sessions, identities CASCADE`); err != nil {
+	if _, err := db.ExecContext(context.Background(), `DROP TABLE IF EXISTS schema_migrations, assignments, audit_events, outbox, result_receipts, ranked_season_rollovers, penalties, seasons, ratings, match_participants, matches, proposal_participants, proposals, queue_tickets, idempotency_keys, sessions, identities CASCADE`); err != nil {
 		t.Fatalf("reset PostgreSQL schema: %v", err)
 	}
-	for _, name := range []string{"0001_initial.sql", "0002_assignments.sql"} {
-		path := filepath.Join("..", "migrations", name)
-		sqlBytes, err := os.ReadFile(path)
-		if err != nil {
-			t.Fatal(err)
-		}
-		if _, err := db.ExecContext(context.Background(), string(sqlBytes)); err != nil {
-			t.Fatalf("apply %s: %v", name, err)
-		}
+	if err := migrations.Apply(context.Background(), db, filepath.Join("..", "migrations")); err != nil {
+		t.Fatalf("apply migrations: %v", err)
 	}
 }