From 18888ed5206b1b60c4ac8b891f42b55369599dd7 Mon Sep 17 00:00:00 2001 From: Josh Creek <8179928+jcreek@users.noreply.github.com> Date: Tue, 1 Sep 2026 09:21:26 +0100 Subject: [PATCH] feat: add PostgreSQL migration runner --- multiplayer-next.md | 3 +- multiplayer-todo.md | 4 +- server/cmd/migrate/main.go | 36 +++++++++++ server/migrations/runner.go | 74 +++++++++++++++++++++++ server/migrations/runner_test.go | 15 +++++ server/store/postgres_integration_test.go | 14 ++--- 6 files changed, 133 insertions(+), 13 deletions(-) create mode 100644 server/cmd/migrate/main.go create mode 100644 server/migrations/runner.go create mode 100644 server/migrations/runner_test.go diff --git a/multiplayer-next.md b/multiplayer-next.md index e7f5f043..c24d2e42 100644 --- a/multiplayer-next.md +++ b/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. diff --git a/multiplayer-todo.md b/multiplayer-todo.md index 86166648..03a4869c 100644 --- a/multiplayer-todo.md +++ b/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 | diff --git a/server/cmd/migrate/main.go b/server/cmd/migrate/main.go new file mode 100644 index 00000000..0205bfb8 --- /dev/null +++ b/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") +} diff --git a/server/migrations/runner.go b/server/migrations/runner.go new file mode 100644 index 00000000..cf10b0cb --- /dev/null +++ b/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 +} diff --git a/server/migrations/runner_test.go b/server/migrations/runner_test.go new file mode 100644 index 00000000..48327443 --- /dev/null +++ b/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") + } +} diff --git a/server/store/postgres_integration_test.go b/server/store/postgres_integration_test.go index e13b1bc4..f614be13 100644 --- a/server/store/postgres_integration_test.go +++ b/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) } }