feat(store): OAuth tables migration (forgejo-mcp-broker-cpb)

Adds migrations/0002_oauth_tables.sql per design.md §4.2: clients,
auth_codes, access_tokens, refresh_tokens. Cascading foreign keys
guarantee that revoking a client tears down every dependent row, and
that a refresh token can never outlive its access token.

Storage choices:
- Broker access/refresh tokens stored as hex-encoded SHA-256 hashes;
  plaintext leaves the broker exactly once (when handed to the MCP
  client). Lookups by hash are O(log n) via the PK index.
- Forgejo tokens stored cleartext (subprocess spawning needs them).
  At-rest protection is the volume permissions + optional encrypted
  volume; application-layer encryption is tracked as backlog item -sd4.
- Timestamps are unix epoch INTEGERs, set by the application — keeps
  deadline comparisons trivial and lets phase 5c inject a test clock.
- Tables are not STRICT to stay consistent with the phase-1 broker_meta
  table; converting both is a future cleanup if we want it.

Tests verify column sets via PRAGMA table_info, expected indexes are
present, the FK CASCADE works in both directions (client → tokens, and
access_token → refresh_token), and the oauth_schema_version marker is
written. Existing migration-count assertions parameterised on
embeddedMigrationCount so adding a third migration only needs that
constant bumped.

Closes forgejo-mcp-broker-cpb.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

.beads/issues.jsonl

@@ -9,7 +9,7 @@
 {"id":"forgejo-mcp-broker-b2o","title":"Phase 2d: OAuth discovery endpoints (/.well-known/*)","description":"GET /.well-known/oauth-protected-resource and /.well-known/oauth-authorization-server. Issuer URLs MUST derive from cfg.PublicURL, never inbound headers (host-header attack defense per design.md §8).","acceptance_criteria":"Responses validate against RFC 8414/9728 shapes. Issuer URL sourced from config only. supported_scopes matches cfg.ForgejoOAuthScopes.","status":"open","priority":1,"issue_type":"task","owner":"olemd@glemt.net","created_at":"2026-04-24T15:45:13Z","created_by":"Ole-Morten Duesund","updated_at":"2026-04-24T15:45:13Z","dependencies":[{"issue_id":"forgejo-mcp-broker-b2o","depends_on_id":"forgejo-mcp-broker-pur","type":"blocks","created_at":"2026-04-24T17:45:25Z","created_by":"Ole-Morten Duesund","metadata":"{}"}],"dependency_count":1,"dependent_count":1,"comment_count":0}
 {"id":"forgejo-mcp-broker-b9i","title":"Phase 2b: internal/forgejo OAuth client","description":"Broker-side OAuth client for upstream Forgejo: authorize URL builder, code-to-token exchange, refresh_token grant, userinfo fetch, revoke. Used by AS callback and refresh machinery. Stateless; caller owns persistence.","acceptance_criteria":"Unit tests with httptest.Server fake Forgejo cover each grant plus error paths (wrong code, expired refresh, revoked token). No state persisted in this package.","status":"open","priority":1,"issue_type":"task","owner":"olemd@glemt.net","created_at":"2026-04-24T15:45:12Z","created_by":"Ole-Morten Duesund","updated_at":"2026-04-24T15:45:12Z","dependency_count":0,"dependent_count":1,"comment_count":0}
 {"id":"forgejo-mcp-broker-pur","title":"Phase 2c: internal/oauth AS endpoints (register, authorize, callback, token, revoke)","description":"Five OAuth handlers per design.md §4.1. RFC 7591 DCR with ephemeral client IDs, authorize -\u003e Forgejo delegation, callback minting broker auth codes, token exchange with SHA-256 hashing at rest, revoke. PKCE S256 required.","acceptance_criteria":"End-to-end curl walkthrough from plan.md phase 2 passes. All tokens hashed at rest. Auth codes single-use, 10-min TTL. Rejects missing PKCE, non-S256, wrong verifier, expired codes/tokens. Handler coverage \u003e=80%.","status":"open","priority":1,"issue_type":"task","owner":"olemd@glemt.net","created_at":"2026-04-24T15:45:12Z","created_by":"Ole-Morten Duesund","updated_at":"2026-04-24T15:45:12Z","dependencies":[{"issue_id":"forgejo-mcp-broker-pur","depends_on_id":"forgejo-mcp-broker-b9i","type":"blocks","created_at":"2026-04-24T17:45:24Z","created_by":"Ole-Morten Duesund","metadata":"{}"},{"issue_id":"forgejo-mcp-broker-pur","depends_on_id":"forgejo-mcp-broker-cpb","type":"blocks","created_at":"2026-04-24T17:45:24Z","created_by":"Ole-Morten Duesund","metadata":"{}"}],"dependency_count":2,"dependent_count":5,"comment_count":0}
-{"id":"forgejo-mcp-broker-cpb","title":"Phase 2a: OAuth tables migration","description":"Add migrations/0002_oauth_tables.sql creating clients, auth_codes, access_tokens, refresh_tokens per design.md §4.2. Broker tokens stored as SHA-256 hashes; Forgejo tokens cleartext (subprocess spawning requires them). See plan.md phase 2.","acceptance_criteria":"Migration applies on a fresh DB and is idempotent on reopen. Schema matches design.md §4.2. Tests verify every table and key column exists.","status":"open","priority":1,"issue_type":"task","owner":"olemd@glemt.net","created_at":"2026-04-24T15:45:04Z","created_by":"Ole-Morten Duesund","updated_at":"2026-04-24T15:45:04Z","dependency_count":0,"dependent_count":1,"comment_count":0}
+{"id":"forgejo-mcp-broker-cpb","title":"Phase 2a: OAuth tables migration","description":"Add migrations/0002_oauth_tables.sql creating clients, auth_codes, access_tokens, refresh_tokens per design.md §4.2. Broker tokens stored as SHA-256 hashes; Forgejo tokens cleartext (subprocess spawning requires them). See plan.md phase 2.","acceptance_criteria":"Migration applies on a fresh DB and is idempotent on reopen. Schema matches design.md §4.2. Tests verify every table and key column exists.","status":"in_progress","priority":1,"issue_type":"task","assignee":"Ole-Morten Duesund","owner":"olemd@glemt.net","created_at":"2026-04-24T15:45:04Z","created_by":"Ole-Morten Duesund","updated_at":"2026-04-27T11:26:17Z","started_at":"2026-04-27T11:26:17Z","dependency_count":0,"dependent_count":1,"comment_count":0}
 {"id":"forgejo-mcp-broker-8ei","title":"Phase 1: internal/httpserver with /healthz and graceful shutdown","description":"Implement internal/httpserver: constructs a *http.Server bound to cfg.Listen, mounts GET /healthz (returns 200 with JSON build-info from the build-info package, including version, git revision, build date, and current store status), handles SIGTERM/SIGINT by initiating graceful shutdown with a configurable deadline (default 10s). Uses log/slog for structured JSON logs. Exposes a Run(ctx) error method that blocks until shutdown completes.","acceptance_criteria":"go test ./internal/httpserver passes; GET /healthz returns expected JSON; sending SIGTERM causes Run to return nil within 2 seconds after in-flight requests complete; slow in-flight request is allowed to finish within the deadline, then forcibly closed.","status":"closed","priority":1,"issue_type":"task","assignee":"Ole-Morten Duesund","owner":"olemd@glemt.net","created_at":"2026-04-24T14:46:20Z","created_by":"Ole-Morten Duesund","updated_at":"2026-04-24T15:26:43Z","started_at":"2026-04-24T15:24:09Z","closed_at":"2026-04-24T15:26:43Z","close_reason":"httpserver shipped: /healthz with store probe, graceful shutdown with force-close fallback, ExtraHandler extension point. 97.9% coverage. internal/log also implemented in the same commit (100% coverage).","dependencies":[{"issue_id":"forgejo-mcp-broker-8ei","depends_on_id":"forgejo-mcp-broker-n84","type":"blocks","created_at":"2026-04-24T16:46:19Z","created_by":"Ole-Morten Duesund","metadata":"{}"}],"dependency_count":1,"dependent_count":1,"comment_count":0}
 {"id":"forgejo-mcp-broker-t37","title":"Phase 1: wire cmd/broker/main.go and integration test","description":"Final phase 1 task: wire config → log → store → httpserver in cmd/broker/main.go. Parse config, init slog, open store, start httpserver, wait for shutdown signal, close store, exit. Add an integration test under cmd/broker/ that builds the binary, runs it with a valid env + temp store path, curls /healthz, sends SIGTERM, verifies clean exit within 2s. This is the acceptance gate for phase 1.","acceptance_criteria":"make build; make test (incl. integration) pass; running the binary with missing config fails with a clear error; running with valid config serves /healthz; SIGTERM shuts down cleanly within 2s; /healthz JSON includes version, git revision, build date, and store OK status.","status":"closed","priority":1,"issue_type":"task","assignee":"Ole-Morten Duesund","owner":"olemd@glemt.net","created_at":"2026-04-24T14:46:20Z","created_by":"Ole-Morten Duesund","updated_at":"2026-04-24T15:29:44Z","started_at":"2026-04-24T15:27:58Z","closed_at":"2026-04-24T15:29:44Z","close_reason":"Main wired, signal.NotifyContext triggers shutdown cascade, integration tests green. Phase 1 complete: binary starts with valid config, serves /healthz JSON, shuts down cleanly on SIGTERM within 2s.","dependencies":[{"issue_id":"forgejo-mcp-broker-t37","depends_on_id":"forgejo-mcp-broker-8ei","type":"blocks","created_at":"2026-04-24T16:48:29Z","created_by":"Ole-Morten Duesund","metadata":"{}"},{"issue_id":"forgejo-mcp-broker-t37","depends_on_id":"forgejo-mcp-broker-9jh","type":"blocks","created_at":"2026-04-24T16:48:29Z","created_by":"Ole-Morten Duesund","metadata":"{}"},{"issue_id":"forgejo-mcp-broker-t37","depends_on_id":"forgejo-mcp-broker-9nq","type":"blocks","created_at":"2026-04-24T16:48:28Z","created_by":"Ole-Morten Duesund","metadata":"{}"},{"issue_id":"forgejo-mcp-broker-t37","depends_on_id":"forgejo-mcp-broker-n84","type":"blocks","created_at":"2026-04-24T16:48:28Z","created_by":"Ole-Morten Duesund","metadata":"{}"}],"dependency_count":4,"dependent_count":0,"comment_count":0}
 {"id":"forgejo-mcp-broker-9jh","title":"Phase 1: internal/store with SQLite open and embedded schema migrations","description":"Implement internal/store: wraps a modernc.org/sqlite connection, applies embedded schema migrations in order via a schema_migrations table, exposes a *sql.DB and a Close method. Phase 1 schema is just the migrations table itself plus a health_check row — real tables (clients, auth_codes, access_tokens, refresh_tokens) ship in phase 2. Store_path from config; creates parent dirs if missing; fails fast on unwritable path. Migrations embedded via embed.FS under internal/store/migrations/.","acceptance_criteria":"go test ./internal/store passes; opening a fresh db file applies migrations; re-opening is idempotent (no re-application, no errors); corrupt/locked files yield a clear error; Close() leaves no file handles open.","status":"closed","priority":1,"issue_type":"task","assignee":"Ole-Morten Duesund","owner":"olemd@glemt.net","created_at":"2026-04-24T14:46:19Z","created_by":"Ole-Morten Duesund","updated_at":"2026-04-24T15:22:53Z","started_at":"2026-04-24T15:11:36Z","closed_at":"2026-04-24T15:22:53Z","close_reason":"Store package shipped: modernc.org/sqlite, embed.FS migrations, WAL + FK pragmas, idempotent reopen, 90.1% coverage including bad-SQL rollback and record-step PK conflict.","dependencies":[{"issue_id":"forgejo-mcp-broker-9jh","depends_on_id":"forgejo-mcp-broker-n84","type":"blocks","created_at":"2026-04-24T16:46:19Z","created_by":"Ole-Morten Duesund","metadata":"{}"}],"dependency_count":1,"dependent_count":1,"comment_count":0}

internal/store/migrations/0002_oauth_tables.sql

@@ -0,0 +1,74 @@
+-- 0002_oauth_tables.sql
+-- OAuth 2.1 authorization-server state per design.md §4.2.
+--
+-- Storage rules:
+--   * Broker tokens (access, refresh) are stored as hex-encoded SHA-256 hashes.
+--     Plaintext leaves the broker exactly once, when handed to the MCP client.
+--   * Forgejo tokens are stored cleartext — the broker has to be able to use
+--     them to spawn forgejo-mcp subprocesses. The SQLite file is therefore a
+--     sensitive secret at rest; protect it via volume permissions and (in
+--     production) an encrypted underlying volume.
+--   * Timestamps are unix epoch seconds (INTEGER). Comparing deadlines as
+--     ints is dramatically simpler than parsing ISO-8601 strings, and SQLite's
+--     type affinity makes mixed-type rows easy to misuse.
+--   * No DEFAULT clauses for timestamps — the application supplies them so a
+--     test clock can be injected later (phase 5c).
+
+CREATE TABLE clients (
+    client_id      TEXT    PRIMARY KEY,
+    client_secret  TEXT,                 -- NULL for public clients (PKCE-only)
+    redirect_uris  TEXT    NOT NULL,     -- JSON array of strings
+    metadata_json  TEXT,                 -- raw RFC 7591 client-metadata blob, optional
+    created_at     INTEGER NOT NULL,
+    last_used      INTEGER               -- bumped on successful authorize/token use
+);
+
+CREATE INDEX idx_clients_last_used ON clients(last_used);
+
+CREATE TABLE auth_codes (
+    code                      TEXT    PRIMARY KEY,
+    client_id                 TEXT    NOT NULL REFERENCES clients(client_id) ON DELETE CASCADE,
+    redirect_uri              TEXT    NOT NULL,
+    code_challenge            TEXT    NOT NULL,
+    code_challenge_method     TEXT    NOT NULL,    -- only 'S256' accepted by the AS
+    scopes                    TEXT    NOT NULL,    -- space-separated
+    forgejo_access_token      TEXT    NOT NULL,
+    forgejo_refresh_token     TEXT,
+    forgejo_token_expires_at  INTEGER NOT NULL,
+    forgejo_user_id           INTEGER NOT NULL,
+    forgejo_username          TEXT    NOT NULL,
+    expires_at                INTEGER NOT NULL,    -- short TTL (~600s); reaper sweeps expired rows
+    used_at                   INTEGER              -- non-NULL after exchange; replay defense
+);
+
+CREATE INDEX idx_auth_codes_expires_at ON auth_codes(expires_at);
+
+CREATE TABLE access_tokens (
+    token_hash                TEXT    PRIMARY KEY,    -- hex(sha256(plaintext_token))
+    client_id                 TEXT    NOT NULL REFERENCES clients(client_id) ON DELETE CASCADE,
+    forgejo_user_id           INTEGER NOT NULL,
+    forgejo_username          TEXT    NOT NULL,
+    scopes                    TEXT    NOT NULL,
+    forgejo_access_token      TEXT    NOT NULL,
+    forgejo_refresh_token     TEXT,
+    forgejo_token_expires_at  INTEGER NOT NULL,
+    expires_at                INTEGER NOT NULL,
+    created_at                INTEGER NOT NULL,
+    revoked_at                INTEGER                  -- NULL = active
+);
+
+CREATE INDEX idx_access_tokens_expires_at  ON access_tokens(expires_at);
+CREATE INDEX idx_access_tokens_forgejo_uid ON access_tokens(forgejo_user_id);
+
+CREATE TABLE refresh_tokens (
+    token_hash         TEXT    PRIMARY KEY,            -- hex(sha256(plaintext_token))
+    access_token_hash  TEXT    NOT NULL REFERENCES access_tokens(token_hash) ON DELETE CASCADE,
+    client_id          TEXT    NOT NULL REFERENCES clients(client_id) ON DELETE CASCADE,
+    expires_at         INTEGER NOT NULL,
+    created_at         INTEGER NOT NULL,
+    revoked_at         INTEGER
+);
+
+CREATE INDEX idx_refresh_tokens_expires_at ON refresh_tokens(expires_at);
+
+INSERT INTO broker_meta (key, value) VALUES ('oauth_schema_version', '1');

internal/store/oauth_schema_test.go

@@ -0,0 +1,248 @@
+package store_test
+
+import (
+	"context"
+	"sort"
+	"testing"
+
+	"kode.naiv.no/olemd/forgejo-mcp-broker/internal/store"
+)
+
+// expectedSchema lists every column we expect each OAuth table to have. The
+// test sorts both sides before comparison, so column order in 0002 is free
+// to change without breaking tests as long as the set is stable.
+var expectedSchema = map[string][]string{
+	"clients": {
+		"client_id", "client_secret", "redirect_uris",
+		"metadata_json", "created_at", "last_used",
+	},
+	"auth_codes": {
+		"code", "client_id", "redirect_uri",
+		"code_challenge", "code_challenge_method", "scopes",
+		"forgejo_access_token", "forgejo_refresh_token", "forgejo_token_expires_at",
+		"forgejo_user_id", "forgejo_username",
+		"expires_at", "used_at",
+	},
+	"access_tokens": {
+		"token_hash", "client_id",
+		"forgejo_user_id", "forgejo_username", "scopes",
+		"forgejo_access_token", "forgejo_refresh_token", "forgejo_token_expires_at",
+		"expires_at", "created_at", "revoked_at",
+	},
+	"refresh_tokens": {
+		"token_hash", "access_token_hash", "client_id",
+		"expires_at", "created_at", "revoked_at",
+	},
+}
+
+func TestOAuthSchema_TablesAndColumns(t *testing.T) {
+	ctx := t.Context()
+	s, err := store.Open(ctx, tempStorePath(t))
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	defer s.Close()
+
+	for table, want := range expectedSchema {
+		got := tableColumns(t, ctx, s, table)
+		sort.Strings(got)
+		w := append([]string(nil), want...)
+		sort.Strings(w)
+		if !equalStrings(got, w) {
+			t.Errorf("table %q columns = %v, want %v", table, got, w)
+		}
+	}
+}
+
+func TestOAuthSchema_OAuthSchemaVersionRecorded(t *testing.T) {
+	ctx := t.Context()
+	s, err := store.Open(ctx, tempStorePath(t))
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	defer s.Close()
+
+	var v string
+	if err := s.DB().QueryRowContext(ctx,
+		`SELECT value FROM broker_meta WHERE key = 'oauth_schema_version'`,
+	).Scan(&v); err != nil {
+		t.Fatalf("read oauth_schema_version: %v", err)
+	}
+	if v != "1" {
+		t.Errorf("oauth_schema_version = %q, want %q", v, "1")
+	}
+}
+
+func TestOAuthSchema_ForeignKeyCascade(t *testing.T) {
+	// Foreign keys are enforced via the DSN pragma set in store.buildDSN.
+	// Verify the cascade works end-to-end: deleting a client tears down its
+	// dependent auth_codes, access_tokens, and refresh_tokens rows.
+	ctx := t.Context()
+	s, err := store.Open(ctx, tempStorePath(t))
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	defer s.Close()
+
+	mustExec(t, ctx, s,
+		`INSERT INTO clients (client_id, redirect_uris, created_at)
+		 VALUES ('c1', '["https://example.com/cb"]', 1000)`)
+
+	mustExec(t, ctx, s,
+		`INSERT INTO auth_codes
+		   (code, client_id, redirect_uri, code_challenge, code_challenge_method,
+		    scopes, forgejo_access_token, forgejo_token_expires_at,
+		    forgejo_user_id, forgejo_username, expires_at)
+		 VALUES
+		   ('code-1', 'c1', 'https://example.com/cb', 'chal', 'S256',
+		    'read:user', 'forgejo-tok', 9999, 42, 'alice', 1600)`)
+
+	mustExec(t, ctx, s,
+		`INSERT INTO access_tokens
+		   (token_hash, client_id, forgejo_user_id, forgejo_username, scopes,
+		    forgejo_access_token, forgejo_token_expires_at,
+		    expires_at, created_at)
+		 VALUES
+		   ('hash-a', 'c1', 42, 'alice', 'read:user',
+		    'forgejo-tok', 9999, 9999, 1000)`)
+
+	mustExec(t, ctx, s,
+		`INSERT INTO refresh_tokens
+		   (token_hash, access_token_hash, client_id, expires_at, created_at)
+		 VALUES
+		   ('hash-r', 'hash-a', 'c1', 99999, 1000)`)
+
+	// Deleting the client must cascade to everything else.
+	mustExec(t, ctx, s, `DELETE FROM clients WHERE client_id = 'c1'`)
+
+	for _, table := range []string{"auth_codes", "access_tokens", "refresh_tokens"} {
+		var n int
+		row := s.DB().QueryRowContext(ctx, "SELECT COUNT(*) FROM "+table)
+		if err := row.Scan(&n); err != nil {
+			t.Fatalf("count %s: %v", table, err)
+		}
+		if n != 0 {
+			t.Errorf("%s row count after cascade delete = %d, want 0", table, n)
+		}
+	}
+}
+
+func TestOAuthSchema_RefreshTokenCascadeFromAccessToken(t *testing.T) {
+	// access_tokens -> refresh_tokens cascade is independent of the
+	// client-level cascade. A refresh token outliving its access token is a
+	// bug class to prevent at the schema level.
+	ctx := t.Context()
+	s, err := store.Open(ctx, tempStorePath(t))
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	defer s.Close()
+
+	mustExec(t, ctx, s,
+		`INSERT INTO clients (client_id, redirect_uris, created_at)
+		 VALUES ('c1', '[]', 1000)`)
+	mustExec(t, ctx, s,
+		`INSERT INTO access_tokens
+		   (token_hash, client_id, forgejo_user_id, forgejo_username, scopes,
+		    forgejo_access_token, forgejo_token_expires_at, expires_at, created_at)
+		 VALUES
+		   ('hash-a', 'c1', 42, 'alice', '', 'forgejo-tok', 9999, 9999, 1000)`)
+	mustExec(t, ctx, s,
+		`INSERT INTO refresh_tokens
+		   (token_hash, access_token_hash, client_id, expires_at, created_at)
+		 VALUES
+		   ('hash-r', 'hash-a', 'c1', 99999, 1000)`)
+
+	mustExec(t, ctx, s, `DELETE FROM access_tokens WHERE token_hash = 'hash-a'`)
+
+	var n int
+	if err := s.DB().QueryRowContext(ctx,
+		`SELECT COUNT(*) FROM refresh_tokens`).Scan(&n); err != nil {
+		t.Fatalf("count refresh_tokens: %v", err)
+	}
+	if n != 0 {
+		t.Errorf("refresh_tokens count after access_token delete = %d, want 0", n)
+	}
+}
+
+func TestOAuthSchema_IndexesPresent(t *testing.T) {
+	wantIndexes := []string{
+		"idx_clients_last_used",
+		"idx_auth_codes_expires_at",
+		"idx_access_tokens_expires_at",
+		"idx_access_tokens_forgejo_uid",
+		"idx_refresh_tokens_expires_at",
+	}
+
+	ctx := t.Context()
+	s, err := store.Open(ctx, tempStorePath(t))
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	defer s.Close()
+
+	rows, err := s.DB().QueryContext(ctx,
+		`SELECT name FROM sqlite_master WHERE type = 'index' AND name LIKE 'idx_%'`)
+	if err != nil {
+		t.Fatalf("query indexes: %v", err)
+	}
+	defer rows.Close()
+
+	got := map[string]bool{}
+	for rows.Next() {
+		var name string
+		if err := rows.Scan(&name); err != nil {
+			t.Fatalf("scan: %v", err)
+		}
+		got[name] = true
+	}
+	for _, idx := range wantIndexes {
+		if !got[idx] {
+			t.Errorf("missing index %q", idx)
+		}
+	}
+}
+
+func tableColumns(t *testing.T, ctx context.Context, s *store.Store, table string) []string {
+	t.Helper()
+	rows, err := s.DB().QueryContext(ctx, "PRAGMA table_info("+table+")")
+	if err != nil {
+		t.Fatalf("PRAGMA table_info(%s): %v", table, err)
+	}
+	defer rows.Close()
+	var cols []string
+	for rows.Next() {
+		// PRAGMA table_info returns: cid, name, type, notnull, dflt_value, pk
+		var cid int
+		var name, ctype string
+		var notnull, pk int
+		var dflt any
+		if err := rows.Scan(&cid, &name, &ctype, &notnull, &dflt, &pk); err != nil {
+			t.Fatalf("scan column: %v", err)
+		}
+		cols = append(cols, name)
+	}
+	if len(cols) == 0 {
+		t.Fatalf("table %q has no columns (does it exist?)", table)
+	}
+	return cols
+}
+
+func mustExec(t *testing.T, ctx context.Context, s *store.Store, query string, args ...any) {
+	t.Helper()
+	if _, err := s.DB().ExecContext(ctx, query, args...); err != nil {
+		t.Fatalf("exec %q: %v", query, err)
+	}
+}
+
+func equalStrings(a, b []string) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := range a {
+		if a[i] != b[i] {
+			return false
+		}
+	}
+	return true
+}

internal/store/store_test.go

@@ -14,6 +14,11 @@ func tempStorePath(t *testing.T) string {
 	return filepath.Join(t.TempDir(), "broker.db")
 }
 
+// embeddedMigrationCount is the number of *.sql files shipped under
+// internal/store/migrations/. Bump it when adding a migration so the
+// idempotency tests stay accurate.
+const embeddedMigrationCount = 2
+
 func TestOpen_FreshDB_AppliesMigrations(t *testing.T) {
 	ctx := t.Context()
 	s, err := store.Open(ctx, tempStorePath(t))
@@ -33,14 +38,14 @@ func TestOpen_FreshDB_AppliesMigrations(t *testing.T) {
 		t.Errorf("schema_version = %q, want %q", version, "1")
 	}
 
-	// schema_migrations should have recorded the initial migration.
+	// schema_migrations should have recorded every embedded migration.
 	var count int
 	row = s.DB().QueryRowContext(ctx, `SELECT COUNT(*) FROM schema_migrations`)
 	if err := row.Scan(&count); err != nil {
 		t.Fatalf("counting schema_migrations: %v", err)
 	}
-	if count != 1 {
+	if count != embeddedMigrationCount {
-		t.Errorf("schema_migrations row count = %d, want 1", count)
+		t.Errorf("schema_migrations row count = %d, want %d", count, embeddedMigrationCount)
 	}
 }
 
@@ -68,8 +73,8 @@ func TestOpen_ReopenIsIdempotent(t *testing.T) {
 		`SELECT COUNT(*) FROM schema_migrations`).Scan(&count); err != nil {
 		t.Fatalf("counting schema_migrations: %v", err)
 	}
-	if count != 1 {
+	if count != embeddedMigrationCount {
-		t.Errorf("after reopen, schema_migrations count = %d, want 1", count)
+		t.Errorf("after reopen, schema_migrations count = %d, want %d", count, embeddedMigrationCount)
 	}
 }