migration-runner: shared advisory-lock guard for package migrations

docs/package-migrations.md

@@ -41,6 +41,14 @@ tables — how its migrations are written, tracked, and applied.
   ledger row — is left behind half-applied; the next run retries the same
   migration from a clean slate instead of getting stuck on a schema that
   looks partially migrated but has no ledger row to explain why.
+- **A session-level advisory lock guards the whole run.** Two hub replicas
+  can boot at the same time and both call a package's migration runner
+  concurrently; without a lock, both can see a migration as unapplied and
+  race the same ledger `INSERT`, crashing the loser on a duplicate key.
+  `@corbits/migration-runner`'s `applyPackageMigrations` holds
+  `pg_advisory_lock(hashtext(ledgerTable))` for the whole bootstrap-and-apply
+  run and releases it in a `finally`, so the second replica simply waits and
+  then finds every migration already applied.
 
 ## Why not `drizzle-kit` codegen
 
@@ -63,16 +71,18 @@ package onto the transactional pattern — see below):
 
 1. **Self-contained, transactional** — `@corbits/chat`, `@corbits/notify`,
    `@corbits/webhook-triggers`, `@corbits/routines`, `@corbits/insights`,
-   `@corbits/skills`.
-   The package's `src/migrations.ts` owns a literal `{ name, sql }[]` array,
-   opens its own short-lived `postgres` client, creates its ledger table if
-   absent, and applies each not-yet-applied migration inside
-   `sql.begin(async (tx) => { ... })` — the migration's SQL and its ledger
-   insert commit or roll back together. `scripts/db-setup.ts` imports the
+   `@corbits/skills`, `@corbits/bench`, `@corbits/preferences`,
+   `@corbits/inference-catalog`, `@corbits/evals`, `@corbits/access-policy`.
+   The package's `src/migrations.ts` owns only a literal `{ name, sql }[]`
+   array and a thin `applyXMigrations(databaseUrl)` wrapper; the mechanics —
+   schema/ledger bootstrap, the transactional apply loop, and the advisory
+   lock — live once in `@corbits/migration-runner`'s
+   `applyPackageMigrations`, called with the package's schema name, ledger
+   table name, and migration array. `scripts/db-setup.ts` imports the
    package's `applyXMigrations(databaseUrl)` function directly and calls it
    after the platform's own migrations. See
-   `packages/insights/src/migrations.ts` for the reference implementation
-   every other package now matches.
+   `packages/access-policy/src/migrations.ts` for the reference
+   implementation every other package now matches.
 2. **Delegated to an external package** — `@corbits/mailbox`. The package
    ships and owns its entire migration story (its own literal SQL, its own
    ledger, its own `mailbox` schema) behind a single exported runner
@@ -108,11 +118,12 @@ the new schema by running `workbench reset` rather than by an in-place
 ## Which shape to use for a new package
 
 **Transactional, self-contained** (shape 1 above), in a Postgres schema
-named for the package. Copy `packages/insights/src/migrations.ts`'s shape:
-a `pgSchema("<name>")` in `schema.ts`, a literal migration array whose SQL
-qualifies every table/index with that schema, a package-named ledger table
-living in the same schema, and each migration applied inside `sql.begin`.
-Reach for the delegated shape only when the package already ships its own
-migration runner as part of a larger, independently-owned engine (its own
-schema, its own connection handling) — not as a shortcut to skip writing a
-ledger table.
+named for the package. Copy `packages/access-policy/src/migrations.ts`'s
+shape: a `pgSchema("<name>")` in `schema.ts`, a literal migration array
+whose SQL qualifies every table/index with that schema, a package-named
+ledger table living in the same schema, and an `applyXMigrations` that
+calls `@corbits/migration-runner`'s `applyPackageMigrations` with that
+schema, ledger table, and migration array. Reach for the delegated shape
+only when the package already ships its own migration runner as part of a
+larger, independently-owned engine (its own schema, its own connection
+handling) — not as a shortcut to skip writing a ledger table.

packages/access-policy/package.json

@@ -14,6 +14,7 @@
     "test": "bun test"
   },
   "dependencies": {
+    "@corbits/migration-runner": "workspace:*",
     "@intx/hub-api": "workspace:*",
     "@workbench/hub-client": "workspace:*",
     "arktype": "catalog:",

packages/access-policy/src/migrations.ts

@@ -3,13 +3,17 @@
 // disentangling history from the platform drizzle journal. Every table
 // this package owns — including its ledger — lives in its own
 // `access_policy` Postgres schema, never `public`; see
-// docs/package-migrations.md.
-import postgres from "postgres";
+// docs/package-migrations.md. Mechanics (schema/ledger bootstrap,
+// transactional apply, the advisory lock across concurrent hub
+// replicas) live in @corbits/migration-runner — this file owns only
+// the domain SQL.
+import {
+  applyPackageMigrations,
+  type ApplyPackageMigrationsReport,
+  type PackageMigration,
+} from "@corbits/migration-runner";
 
-export interface AccessPolicyMigration {
-  name: string;
-  sql: string;
-}
+export type AccessPolicyMigration = PackageMigration;
 
 const SCHEMA = "access_policy";
 
@@ -56,63 +60,16 @@ export const accessPolicyMigrations: readonly AccessPolicyMigration[] = [
 
 const LEDGER_TABLE = "access_policy_migrations";
 
-function quoteIdentifier(name: string): string {
-  return `"${name.replace(/"/g, '""')}"`;
-}
-
-function quoteQualified(schema: string, name: string): string {
-  return `${quoteIdentifier(schema)}.${quoteIdentifier(name)}`;
-}
-
-export interface ApplyAccessPolicyMigrationsReport {
-  applied: string[];
-  alreadyApplied: string[];
-}
+export type ApplyAccessPolicyMigrationsReport = ApplyPackageMigrationsReport;
 
 export async function applyAccessPolicyMigrations(
   databaseUrl: string,
 ): Promise<ApplyAccessPolicyMigrationsReport> {
-  const sql = postgres(databaseUrl, { max: 1, onnotice: () => undefined });
-  try {
-    await sql.unsafe(`CREATE SCHEMA IF NOT EXISTS ${quoteIdentifier(SCHEMA)}`);
-
-    await sql.unsafe(
-      `CREATE TABLE IF NOT EXISTS ${quoteQualified(SCHEMA, LEDGER_TABLE)} (` +
-        `name text PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())`,
-    );
-
-    const applied: string[] = [];
-    const alreadyApplied: string[] = [];
-
-    for (const migration of accessPolicyMigrations) {
-      const existing = await sql.unsafe(
-        `SELECT 1 FROM ${quoteQualified(SCHEMA, LEDGER_TABLE)} WHERE name = $1`,
-        [migration.name],
-      );
-      if (existing.length > 0) {
-        alreadyApplied.push(migration.name);
-        continue;
-      }
-      try {
-        await sql.begin(async (tx) => {
-          await tx.unsafe(migration.sql);
-          await tx.unsafe(
-            `INSERT INTO ${quoteQualified(SCHEMA, LEDGER_TABLE)} (name) VALUES ($1)`,
-            [migration.name],
-          );
-        });
-        applied.push(migration.name);
-      } catch (err) {
-        const message = err instanceof Error ? err.message : String(err);
-        throw new Error(
-          `access_policy migration ${migration.name} failed: ${message}`,
-          { cause: err },
-        );
-      }
-    }
-
-    return { applied, alreadyApplied };
-  } finally {
-    await sql.end({ timeout: 5 });
-  }
+  return applyPackageMigrations({
+    databaseUrl,
+    schema: SCHEMA,
+    ledgerTable: LEDGER_TABLE,
+    migrations: accessPolicyMigrations,
+    packageLabel: "access_policy",
+  });
 }

packages/bench/package.json

@@ -16,6 +16,7 @@
   },
   "dependencies": {
     "@corbits/api-query": "workspace:*",
+    "@corbits/migration-runner": "workspace:*",
     "@intx/hub-api": "workspace:*",
     "arktype": "catalog:",
     "drizzle-orm": "catalog:",