feat: add migration runner with structured failure diagnostic (E00-S03-T05)

MigrationRunner applies pending migrations through the migration ledger
exactly once; when a migration fails it throws a MigrationFailedError
whose diagnostic is a structured object identifying the failing migration
(version), the failure phase (apply/record), the underlying cause, and the
applied/pending ledger state, serializable via toJSON. Re-exported from
the driver boundary so no other package needs the pg driver to run
migrations. Advisory lock (T04) and ready gate (T06) remain out of scope.

packages/database-postgres/src/runner.ts

@@ -0,0 +1,206 @@
+/**
+ * Migration runner with failure diagnostic — [E00-S03-T05].
+ *
+ * The migration runner applies pending migrations to the database exactly
+ * once and, when a migration fails, produces a structured diagnostic that
+ * identifies the failing migration. It is built on the boundary that came
+ * before it: the migration ledger (E00-S03-T03) records applied migrations,
+ * so a rerun never double-applies. The advisory lock (E00-S03-T04) serializes
+ * concurrent runners, but wiring the lock into the runner is out of scope for
+ * this story — the runner performs no locking itself; a runner that wants to
+ * serialize takes the lock (E00-S03-T04) around `run()`.
+ *
+ * Failure model:
+ *   - migrations run in the order given, oldest first; a migration whose
+ *     version is already recorded in the ledger is skipped;
+ *   - when a migration's `up` throws, the runner wraps the failure into a
+ *     `MigrationFailedError` whose `diagnostic` is a structured object that
+ *     identifies the failing migration (`migration` — its version), where the
+ *     run failed (`phase`: 'apply' when the migration's `up` threw, 'record'
+ *     when the ledger insert threw after a successful `up`), the underlying
+ *     cause, and the ledger state at failure time (`applied`/`pending` —
+ *     `applied` + `pending` cover the runner's migrations exactly);
+ *   - the error is serializable: `toJSON()` returns a plain structured object
+ *     (including a structured cause — for pg errors the `code`, e.g. `42P01`),
+ *     so operators can log/parse the diagnostic without string-matching.
+ *
+ * The runner lives in `database-postgres` — the single workspace package
+ * allowed to import the PostgreSQL driver (E00-S03-T02) — and talks to the
+ * database exclusively through the package-owned `pg` Pool and the
+ * `MigrationLedger`, so no other package needs the driver to run migrations.
+ *
+ * Rollback note from the issue: revert the diagnostic/error handling changes.
+ */
+
+import type { Pool } from 'pg';
+import { MigrationLedger } from './ledger.js';
+
+/**
+ * A single migration step: an identifier (recorded in the ledger once the
+ * step has been applied) and the apply function. `up` receives the
+ * package-owned pool, so a migration can run any SQL (and multi-statement
+ * work) through the same driver boundary the runner itself uses.
+ */
+export interface Migration {
+  version: string;
+  up(pool: Pool): Promise<void> | void;
+}
+
+/**
+ * Where a migration run failed: 'apply' when the migration's `up` threw, or
+ * 'record' when the ledger insert threw after a successful `up`.
+ */
+export type MigrationFailurePhase = 'apply' | 'record';
+
+/**
+ * The structured diagnostic produced when a migration fails. `migration`
+ * identifies the failing migration; `applied` and `pending` are disjoint and
+ * together cover the runner's migration list in run order.
+ */
+export interface MigrationDiagnostic {
+  /** Version of the migration that failed — identifies the failing migration. */
+  migration: string;
+  /** Where the run failed: 'apply' (the migration's `up` threw) or 'record' (the ledger insert threw). */
+  phase: MigrationFailurePhase;
+  /** The underlying failure (e.g. the pg error), preserved for inspection. */
+  cause: unknown;
+  /** Versions recorded in the ledger when the failure happened, in run order. */
+  applied: string[];
+  /** Versions not yet recorded when the failure happened, in run order — includes the failing migration. */
+  pending: string[];
+}
+
+/** Result of a successful migration run. */
+export interface MigrationRunResult {
+  /** Versions applied by this run, in run order (oldest first). */
+  applied: string[];
+  /** Versions skipped because they were already recorded in the ledger. */
+  skipped: string[];
+}
+
+/**
+ * The error thrown when a migration fails. Carries the structured diagnostic
+ * (`.diagnostic`) and is serializable (`toJSON()`), so callers and operators
+ * can inspect and parse the failure without string-matching the message.
+ */
+export class MigrationFailedError extends Error {
+  readonly diagnostic: MigrationDiagnostic;
+
+  constructor(diagnostic: MigrationDiagnostic) {
+    super(`migration "${diagnostic.migration}" failed during ${diagnostic.phase}`);
+    this.name = 'MigrationFailedError';
+    this.diagnostic = diagnostic;
+  }
+
+  /** Serializable form of the error and its structured diagnostic. */
+  toJSON(): Record<string, unknown> {
+    return {
+      name: this.name,
+      message: this.message,
+      diagnostic: {
+        migration: this.diagnostic.migration,
+        phase: this.diagnostic.phase,
+        applied: [...this.diagnostic.applied],
+        pending: [...this.diagnostic.pending],
+        cause: structuredCause(this.diagnostic.cause),
+      },
+    };
+  }
+}
+
+/**
+ * Reduces the underlying cause to a structured, serializable shape: for an
+ * `Error` the name/message (plus the pg error `code` when present, e.g.
+ * `42P01`); for anything else a `{ value }` wrapper, so `toJSON()` never
+ * stringifies to an empty object.
+ */
+function structuredCause(cause: unknown): Record<string, unknown> {
+  if (cause instanceof Error) {
+    const structured: Record<string, unknown> = { name: cause.name, message: cause.message };
+    const code = (cause as Error & { code?: unknown }).code;
+    if (code !== undefined) structured.code = code;
+    return structured;
+  }
+  return { value: cause };
+}
+
+/**
+ * The migration runner: applies pending migrations in order, exactly once,
+ * through the migration ledger. Instances are cheap and share the caller's
+ * pool and ledger; the runner performs no locking (advisory lock is
+ * E00-S03-T04) and no ready gating (E00-S03-T06).
+ */
+export class MigrationRunner {
+  private readonly pool: Pool;
+  private readonly ledger: MigrationLedger;
+  private readonly migrations: readonly Migration[];
+
+  /**
+   * @param pool The package-owned PostgreSQL pool (`pg.Pool`).
+   * @param migrations The migrations to run, in apply order (oldest first).
+   * @param ledger The migration ledger; defaults to one sharing `pool`.
+   */
+  constructor(pool: Pool, migrations: readonly Migration[], ledger?: MigrationLedger) {
+    this.pool = pool;
+    this.migrations = migrations;
+    this.ledger = ledger ?? new MigrationLedger(pool);
+  }
+
+  /**
+   * Runs the pending migrations: creates the ledger if missing, then applies
+   * every migration whose version is not yet recorded, recording each one as
+   * it completes. Idempotent: a rerun skips everything already recorded, so a
+   * second run never double-applies. When a migration fails — its `up` throws
+   * ('apply') or the ledger insert throws ('record') — the runner throws a
+   * `MigrationFailedError` whose `diagnostic` identifies the failing migration
+   * and the ledger state at failure time.
+   */
+  async run(): Promise<MigrationRunResult> {
+    await this.ledger.ensure();
+    const recorded = new Set(await this.ledger.applied());
+    const applied: string[] = [];
+    const skipped: string[] = [];
+
+    for (const migration of this.migrations) {
+      if (recorded.has(migration.version)) {
+        skipped.push(migration.version);
+        continue;
+      }
+      try {
+        await migration.up(this.pool);
+      } catch (error) {
+        throw this.failure(migration.version, 'apply', error, recorded);
+      }
+      try {
+        await this.ledger.record(migration.version);
+      } catch (error) {
+        throw this.failure(migration.version, 'record', error, recorded);
+      }
+      recorded.add(migration.version);
+      applied.push(migration.version);
+    }
+
+    return { applied, skipped };
+  }
+
+  /**
+   * Builds the structured diagnostic for a failing migration: `migration` is
+   * the failing version, `applied`/`pending` are the ledger state at failure
+   * time scoped to this runner's migrations (disjoint, in run order — the
+   * failing migration is still pending, since it was never recorded).
+   */
+  private failure(
+    version: string,
+    phase: MigrationFailurePhase,
+    cause: unknown,
+    recorded: ReadonlySet<string>,
+  ): MigrationFailedError {
+    const applied: string[] = [];
+    const pending: string[] = [];
+    for (const migration of this.migrations) {
+      if (recorded.has(migration.version)) applied.push(migration.version);
+      else pending.push(migration.version);
+    }
+    return new MigrationFailedError({ migration: version, phase, cause, applied, pending });
+  }
+}