feat(db): vendor the pg-boss schema into a committed migration

AGENTS.md

@@ -45,6 +45,7 @@ Crove is the open-source Business OS (AGPL). It must run with `docker compose up
 - Shared dev PostgreSQL: `migrate.db.test.ts` creates cluster-wide `t100_test_runner_*` roles and drops them afterwards; a run killed before that leaves them behind, and every later migrate on the cluster then fails the T17 check with `role t100_test_runner_..._proxy can create temporary objects`. The next run of that test drops leftovers older than 10 minutes; to clear them at once (with no run of it in progress): `psql "$CROVE_TEST_ADMIN_URL" -c "DO \$\$ DECLARE r text; BEGIN FOR r IN SELECT rolname FROM pg_roles WHERE rolname LIKE 't100\_test\_runner\_%' LOOP EXECUTE format('DROP OWNED BY %I', r); EXECUTE format('DROP ROLE %I', r); END LOOP; END \$\$"`
 - `pnpm db:migrate` applies `supabase/migrations` with the built-in runner
 - `pnpm db:types` (needs `CROVE_TEST_ADMIN_URL`) regenerates `packages/db/src/generated/database.ts`, the Kysely types of every table `crove_tenant` may read, with Kanel from a throwaway database the migrations build. Run it after a migration that changes such a table or its grants and commit the file; CI fails when it is out of date. Insert and update types follow `crove_tenant`'s column privileges: a column the role may not write has the write type `NotGranted`, so a forbidden write (an insert into `outbox_event` or the ledger, an update of an append-only table) fails typecheck (through `values()` and `set()`; deletes, raw SQL and insert ... select are refused only at runtime); a refinement cannot lift that. Refine a column (a CHECK-constraint union, a trigger-set column) in `packages/db/src/schema.ts`, never in the generated file. A CHECK-constraint union is a const array there (`TASK_STATUSES` and its type `TaskStatus`); `packages/db/test/check-unions.db.test.ts` compares every fixed-set CHECK constraint with its array, so a migration that adds, widens or narrows one fails until the array follows and is listed in that test.
+- `pnpm db:vendor:pgboss` writes the migration for the pg-boss schema version the pinned pg-boss expects (a unit test fails until it exists after a pg-boss upgrade)
 - `docker compose up --build` runs the full stack on http://localhost:8080
 - `pnpm test:e2e` runs Playwright against a stack started with `pnpm --filter @crove/e2e stack:up` (compose project `crove-e2e`, port 8080; `stack:down` removes it with its volumes). Install the browser once with `pnpm --filter @crove/e2e exec playwright install chromium`
 

apps/kernel/package.json

@@ -10,7 +10,8 @@
     "start": "node src/main.ts all",
     "migrate": "node --env-file-if-exists=../../.env src/main.ts migrate",
     "auth-grants": "node --env-file-if-exists=../../.env src/main.ts auth-grants",
-    "typecheck": "tsc --noEmit -p tsconfig.json"
+    "typecheck": "tsc --noEmit -p tsconfig.json",
+    "vendor:pgboss": "node src/vendor-pgboss.ts"
   },
   "dependencies": {
     "@crove/agent": "workspace:*",

apps/kernel/src/http.ts

@@ -30,6 +30,7 @@ import { publicAuthUrl, type Config } from './config.ts';
 import { exportAad, exportSealer } from './export-seal.ts';
 import { MAX_EMAIL_BYTES, acceptInboundEmail, secretMatches } from './inbound.ts';
 import { createInviteMailSender } from './invite-mail.ts';
+import { pgBossSchemaVersion } from './jobs-schema.ts';
 import type { Mailer } from './mail.ts';
 import { RateLimiter } from './rate-limit.ts';
 import { createRpcHandler, resolveMembership, resolveOrg, type RpcChannels } from './rpc.ts';
@@ -232,6 +233,9 @@ export function createHttpApp(deps: HttpDeps) {
     try {
       const missing = await schemaMissing(client, config.MIGRATIONS_DIR);
       if (missing.length > 0) return c.json({ ready: false, reason: 'schema' }, 503);
+      // The vendored crove_jobs schema must be the version this pg-boss release expects (External-tenant gate T6).
+      const jobs = await client.query<{ version: number | null }>('SELECT crove_kernel.jobs_schema_version() AS version');
+      if (jobs.rows[0]?.version !== pgBossSchemaVersion()) return c.json({ ready: false, reason: 'jobs_schema' }, 503);
       // The session check fails closed until it can read Supabase Auth's tables (ADR-0028).
       if (config.AUTH_SESSION_CHECK === 'db') {
         const { rows } = await client.query<{ ok: boolean }>('SELECT crove_kernel.auth_grants_ready() AS ok');

apps/kernel/src/jobs-schema.ts

@@ -0,0 +1,169 @@
+// The pg-boss schema (`crove_jobs`) is committed SQL like every other schema (plan principle 4, External-tenant
+// gate T6). `pnpm db:vendor:pgboss` renders pg-boss's own plans into a migration; no code path runs pg-boss with
+// `migrate: true`. The worker starts pg-boss with `migrate: false`, which refuses any other schema version, and
+// /readyz compares the database's version with the one the pinned pg-boss expects.
+import { readFileSync } from 'node:fs';
+import { readdir, readFile } from 'node:fs/promises';
+import { createRequire } from 'node:module';
+import { join } from 'node:path';
+import { getConstructionPlans, getMigrationPlans } from 'pg-boss';
+
+export const JOB_SCHEMA = 'crove_jobs';
+
+// Created by the migration that vendors the schema. A new queue, or a new policy, needs a new migration.
+export const QUEUES = {
+  extract: 'finance-file-extract',
+  inbound: 'kernel-inbound-email',
+  signOut: 'kernel-session-signout',
+  orgExport: 'kernel-org-export',
+  retention: 'kernel-retention-purge',
+  agentDispatch: 'kernel-agent-dispatch',
+  agentWake: 'kernel-agent-wake',
+  agentSweep: 'kernel-agent-sweep',
+} as const;
+type QueueName = (typeof QUEUES)[keyof typeof QUEUES];
+
+// The pg-boss policy of each queue that is not 'standard'. kernel-agent-wake is 'stately' with the singleton key
+// <org>:<agent>: at most one queued and one active job per agent, so wake-ups coalesce and an agent runs once at a
+// time (ADR-0029 section 5; the database also allows one running ai_run per agent).
+const QUEUE_POLICIES: Partial<Record<QueueName, 'stately'>> = { [QUEUES.agentWake]: 'stately' };
+
+// What `boss.createQueue(name, { policy })` sends to crove_jobs.create_queue.
+export function queueOptions(queue: string): string {
+  return JSON.stringify({ policy: QUEUE_POLICIES[queue as QueueName] ?? 'standard' });
+}
+
+// First line of every vendored migration.
+const MARKER = /^-- pg-boss schema version: (\d+)$/m;
+const PIN = 'SET search_path = pg_catalog, pg_temp';
+
+let expected: number | null = null;
+
+// The schema version the installed pg-boss release expects (its package.json `pgboss.schema`).
+export function pgBossSchemaVersion(): number {
+  if (expected !== null) return expected;
+  const path = createRequire(import.meta.url).resolve('pg-boss/package.json');
+  const pkg = JSON.parse(readFileSync(path, 'utf8')) as { pgboss?: { schema?: number } };
+  const version = pkg.pgboss?.schema;
+  if (!Number.isInteger(version)) throw new Error('pg-boss package.json has no pgboss.schema');
+  expected = version as number;
+  return expected;
+}
+
+// The highest pg-boss schema version any committed migration vendors, or null if none does.
+export async function vendoredSchemaVersion(migrationsDir: string): Promise<number | null> {
+  let highest: number | null = null;
+  for (const name of (await readdir(migrationsDir)).filter((n) => n.endsWith('.sql'))) {
+    const match = MARKER.exec(await readFile(join(migrationsDir, name), 'utf8'));
+    if (match?.[1]) highest = Math.max(highest ?? 0, Number(match[1]));
+  }
+  return highest;
+}
+
+// pg-boss plans run as their own transaction under an advisory lock; the migration runner already applies each
+// file in one transaction under its lock, so the wrapper goes. Every function pins search_path (AGENTS.md).
+export function adaptPlans(plans: string): string {
+  let body = plans
+    .replace(/^\s*BEGIN;\s*$/m, '')
+    .replace(/^\s*SET LOCAL (?:lock_timeout|idle_in_transaction_session_timeout) = \d+;\s*$/gm, '')
+    .replace(/^\s*SELECT pg_advisory_xact_lock\(.*\);\s*$/m, '')
+    .replace(/^\s*COMMIT;\s*$/m, '');
+  if (/^\s*(?:BEGIN|COMMIT|ROLLBACK);/m.test(body)) throw new Error('pg-boss plans contain more than one transaction');
+  const functions = body.match(/CREATE (?:OR REPLACE )?FUNCTION /g)?.length ?? 0;
+  let pinned = 0;
+  body = body.replace(/(\n\s*LANGUAGE (?:sql|plpgsql)\b[^;\n]*);/g, (_, head: string) => {
+    pinned += 1;
+    return `${head} ${PIN};`;
+  });
+  if (pinned !== functions) throw new Error(`pinned search_path on ${pinned} of ${functions} pg-boss functions`);
+  if (body.includes('$pgboss$')) throw new Error('pg-boss plans contain the $pgboss$ quote tag');
+  return body.trim();
+}
+
+const GRANTS = `-- The worker (pg-boss and the outbox relay) uses crove_jobs; nobody else may.
+REVOKE EXECUTE ON ALL FUNCTIONS IN SCHEMA ${JOB_SCHEMA} FROM PUBLIC;
+GRANT USAGE ON SCHEMA ${JOB_SCHEMA} TO crove_worker;
+GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA ${JOB_SCHEMA} TO crove_worker;
+-- The version row decides whether a vendored upgrade runs, so only migrations write it. pg-boss started with
+-- migrate: false only stamps its maintenance columns.
+REVOKE INSERT, UPDATE, DELETE, TRUNCATE ON ${JOB_SCHEMA}.version FROM crove_worker;
+GRANT UPDATE (cron_on, bam_on, flow_on, reindex_on, monitor_backoff_on) ON ${JOB_SCHEMA}.version TO crove_worker;
+GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA ${JOB_SCHEMA} TO crove_worker;
+GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA ${JOB_SCHEMA} TO crove_worker;
+ALTER DEFAULT PRIVILEGES IN SCHEMA ${JOB_SCHEMA} GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO crove_worker;`;
+
+// Renders the migration that brings crove_jobs to `target`: pg-boss's construction plans on a fresh database
+// (`from` null), else its migration plans from `from`. Re-applying it is a no-op; any other version stops it.
+export function renderJobsMigration(target: number, from: number | null, queues: readonly string[]): string {
+  const plans = from === null ? getConstructionPlans(JOB_SCHEMA) : getMigrationPlans(JOB_SCHEMA, from);
+  // The schema holds exactly one version row; anything else stops the migration.
+  const versions = `(SELECT array_agg(version) FROM ${JOB_SCHEMA}.version)`;
+  const held = `(SELECT string_agg(version::text, ', ') FROM ${JOB_SCHEMA}.version)`;
+  const found =
+    from === null
+      ? `IF ${versions} = ARRAY[${target}] THEN
+      RETURN;
+    END IF;
+    RAISE EXCEPTION '${JOB_SCHEMA} holds pg-boss schema version %, this migration installs ${target}', ${held};`
+      : `IF ${versions} = ARRAY[${target}] THEN
+      RETURN;
+    END IF;
+    IF ${versions} IS DISTINCT FROM ARRAY[${from}] THEN
+      RAISE EXCEPTION '${JOB_SCHEMA} holds pg-boss schema version %, this migration upgrades ${from} to ${target}', ${held};
+    END IF;
+    -- pg-boss's upgrade plans put queue.table_name into SQL as is, and the worker may write that column.
+    IF EXISTS (SELECT 1 FROM ${JOB_SCHEMA}.queue WHERE table_name !~ '^(job_common|j[0-9a-f]{56})$') THEN
+      RAISE EXCEPTION '${JOB_SCHEMA}.queue holds a table name pg-boss does not create';
+    END IF;`;
+  const missing = from === null ? `\n  ELSE\n    EXECUTE $pgboss$\n${adaptPlans(plans)}\n$pgboss$;` : '';
+  const upgrade = from === null ? '' : `\n  EXECUTE $pgboss$\n${adaptPlans(plans)}\n$pgboss$;`;
+  // One create_queue statement per set of options, so each queue gets the policy the worker expects.
+  const byOptions = new Map<string, string[]>();
+  for (const queue of queues) byOptions.set(queueOptions(queue), [...(byOptions.get(queueOptions(queue)) ?? []), `'${queue.replaceAll("'", "''")}'`]);
+  const createQueues = [...byOptions]
+    .map(([options, names]) => `SELECT ${JOB_SCHEMA}.create_queue(q, '${options}'::jsonb) FROM unnest(ARRAY[${names.join(', ')}]) AS q;`)
+    .join('\n');
+  return `-- pg-boss schema version: ${target}
+-- Generated by \`pnpm db:vendor:pgboss\` from pg-boss ${from === null ? 'getConstructionPlans' : `getMigrationPlans (from ${from})`}('${JOB_SCHEMA}').
+-- Do not edit: upgrade pg-boss, run the script again and commit the new migration it writes.
+-- Changes to pg-boss's SQL: no BEGIN, COMMIT or advisory lock (the runner applies this file in one transaction
+-- under its own lock), and every function pins search_path. Before #25 the migrate role let pg-boss create the
+-- schema at runtime, so an existing database at version ${target} keeps its tables and only gets the pins.
+DO $vendor$
+BEGIN
+  IF to_regclass('${JOB_SCHEMA}.version') IS NOT NULL THEN
+    ${found}${missing}
+  END IF;${upgrade}
+END
+$vendor$;
+
+-- A schema pg-boss created at runtime (before #25) has functions without a pinned search_path. The migration role
+-- created them then, so it owns them now.
+DO $pin$
+DECLARE
+  fn regprocedure;
+BEGIN
+  FOR fn IN
+    SELECT p.oid::regprocedure FROM pg_catalog.pg_proc p
+     WHERE p.pronamespace = '${JOB_SCHEMA}'::regnamespace
+       AND p.proconfig IS DISTINCT FROM ARRAY['search_path=pg_catalog, pg_temp']
+  LOOP
+    EXECUTE pg_catalog.format('ALTER FUNCTION %s ${PIN}', fn);
+  END LOOP;
+END
+$pin$;
+${from === null ? `
+-- Queues the worker uses (QUEUES in apps/kernel/src/jobs-schema.ts); create_queue skips existing ones.
+${createQueues}
+
+-- /readyz (the API role) compares this with the version its pg-boss expects.
+CREATE OR REPLACE FUNCTION crove_kernel.jobs_schema_version()
+RETURNS integer LANGUAGE sql STABLE SECURITY DEFINER ${PIN} AS $$
+  SELECT max(v.version) FROM ${JOB_SCHEMA}.version v
+$$;
+REVOKE ALL ON FUNCTION crove_kernel.jobs_schema_version() FROM PUBLIC;
+GRANT EXECUTE ON FUNCTION crove_kernel.jobs_schema_version() TO crove_app;
+` : ''}
+${GRANTS}
+`;
+}

apps/kernel/src/main.ts

@@ -1,5 +1,5 @@
 // Crove kernel entry point. One image, several roles:
-//   migrate      apply SQL migrations, set role passwords, create the job schema
+//   migrate      apply SQL migrations (the job schema included) and set role passwords
 //   auth-grants  after Supabase Auth has started: let the session check read auth.sessions, auth.users and
 //                auth.mfa_factors (user_id, status)
 //   api          HTTP API + web UI
@@ -18,7 +18,6 @@ import { runPilotCheck } from './pilot-check.ts';
 import { runProber } from './prober.ts';
 import { describeStorage } from './storage.ts';
 import { runUsageReport } from './usage-report.ts';
-import { migrateJobs } from './worker.ts';
 
 type Role = 'migrate' | 'auth-grants' | 'prober' | 'usage-report' | 'pilot-check' | ServerRole;
 const ROLES: readonly Role[] = ['migrate', 'auth-grants', 'api', 'worker', 'all', 'prober', 'usage-report', 'pilot-check'];
@@ -34,7 +33,6 @@ async function runMigrate(config: Config) {
     crove_reporter: config.CROVE_REPORTER_DB_PASSWORD,
     supabase_auth_admin: config.AUTH_DB_PASSWORD,
   });
-  await migrateJobs(adminUrl);
   console.log(`[migrate] done, ${applied.length} new migration(s)`);
 }
 

apps/kernel/src/vendor-pgboss.ts

@@ -0,0 +1,21 @@
+// `pnpm db:vendor:pgboss`: when the pinned pg-boss expects a schema version no committed migration vendors yet,
+// writes the migration that installs (first time) or upgrades crove_jobs to it. Commit the file it writes.
+import { writeFile } from 'node:fs/promises';
+import { join } from 'node:path';
+import { fileURLToPath } from 'node:url';
+import { QUEUES, pgBossSchemaVersion, renderJobsMigration, vendoredSchemaVersion } from './jobs-schema.ts';
+
+const MIGRATIONS = process.env.MIGRATIONS_DIR ?? fileURLToPath(new URL('../../../supabase/migrations', import.meta.url));
+
+const target = pgBossSchemaVersion();
+const vendored = await vendoredSchemaVersion(MIGRATIONS);
+if (vendored === target) {
+  console.log(`crove_jobs is vendored at pg-boss schema version ${target}; nothing to do`);
+} else if (vendored !== null && vendored > target) {
+  throw new Error(`migrations vendor pg-boss schema version ${vendored}, but the installed pg-boss expects ${target}; upgrade pg-boss`);
+} else {
+  const stamp = new Date().toISOString().replace(/\D/g, '').slice(0, 14);
+  const file = join(MIGRATIONS, `${stamp}_crove_jobs_pgboss_v${target}.sql`);
+  await writeFile(file, renderJobsMigration(target, vendored, Object.values(QUEUES)));
+  console.log(`wrote ${file}${vendored === null ? '' : ` (upgrade from ${vendored})`}`);
+}

apps/kernel/src/worker.ts

@@ -11,25 +11,11 @@ import type { ExtractClient } from '@crove/extract-contract';
 import { extractAndPropose, recordExtractionFailure } from '@crove/mod-finance';
 import { processInboundEmail, type InboundJob } from './inbound.ts';
 import { ExportSealer } from './export-seal.ts';
+import { JOB_SCHEMA, QUEUES } from './jobs-schema.ts';
 import { DEFAULT_EXPORT_RETENTION_DAYS, runOrgExport } from './org-export.ts';
 import { RETENTION_CRON, RETENTION_TZ, expireExports, runRetention, sweepInboundBlobs, sweepStaging } from './retention.ts';
 import type { SessionRevoker, SignOutTarget } from './sessions.ts';
 
-export const JOB_SCHEMA = 'crove_jobs';
-export const QUEUES = {
-  extract: 'finance-file-extract',
-  inbound: 'kernel-inbound-email',
-  signOut: 'kernel-session-signout',
-  orgExport: 'kernel-org-export',
-  retention: 'kernel-retention-purge',
-  agentDispatch: 'kernel-agent-dispatch',
-  agentWake: 'kernel-agent-wake',
-  agentSweep: 'kernel-agent-sweep',
-} as const;
-// Queue options by queue. kernel-agent-wake is 'stately' with the singleton key <org>:<agent>: at most one queued and
-// one active job per agent, so wake-ups coalesce and an agent runs once at a time (ADR-0029 section 5; the database
-// also allows one running ai_run per agent).
-const QUEUE_OPTIONS: Partial<Record<(typeof QUEUES)[keyof typeof QUEUES], { policy: 'stately' }>> = { [QUEUES.agentWake]: { policy: 'stately' } };
 const AGENT_RETRY_LIMIT = 3;
 // Every 10 minutes the worker wakes agents whose wake-up is still pending and that can run now
 // (crove_kernel.agent_wakeups_due): a run refused earlier by the daily cap, a kill switch or a budget that has since
@@ -397,24 +383,7 @@ export async function startWorker(deps: WorkerDeps): Promise<() => Promise<void>
   };
 }
 
-// Creates the pg-boss schema and queues with an admin connection, then grants the worker role.
-export async function migrateJobs(adminUrl: string): Promise<void> {
-  const boss = new PgBoss({ connectionString: adminUrl, schema: JOB_SCHEMA, migrate: true, supervise: false, schedule: false });
-  boss.on('error', (error) => console.error('[pg-boss migrate]', error));
-  await boss.start();
-  for (const queue of Object.values(QUEUES)) await boss.createQueue(queue, QUEUE_OPTIONS[queue] ?? {});
-  await boss.stop({ graceful: false });
-  const client = new pg.Client({ connectionString: adminUrl, application_name: 'crove-migrate' });
-  await client.connect();
-  try {
-    await client.query(`
-      GRANT USAGE ON SCHEMA ${JOB_SCHEMA} TO crove_worker;
-      GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA ${JOB_SCHEMA} TO crove_worker;
-      GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA ${JOB_SCHEMA} TO crove_worker;
-      GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA ${JOB_SCHEMA} TO crove_worker;
-      ALTER DEFAULT PRIVILEGES IN SCHEMA ${JOB_SCHEMA} GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO crove_worker;
-    `);
-  } finally {
-    await client.end();
-  }
-}
+// Kept only because the invariant suite on main still calls it, and protected.yml runs main's copy of that suite
+// against this code. The schema now comes from the vendored migration (#25), so this does nothing; delete it
+// together with those calls once they are gone from main.
+export async function migrateJobs(_adminUrl: string): Promise<void> {}

apps/kernel/test/blob-sweep.db.test.ts

@@ -12,7 +12,8 @@ import { createDb, createPool, type Db } from '@crove/db';
 import { migrate, setRolePasswords } from '@crove/db/migrate';
 import { BlobNotFoundError, LocalDiskStore, inboundKey, newInboundBlobId, sha256Hex } from '@crove/kernel-runtime';
 import { INBOUND_BLOB_MAX_AGE_DAYS, sweepInboundBlobs } from '../src/retention.ts';
-import { QUEUES, migrateJobs } from '../src/worker.ts';
+import { QUEUES } from '../src/jobs-schema.ts';
+import { migrateJobs } from '../src/worker.ts';
 
 const ADMIN = process.env.CROVE_TEST_ADMIN_URL;
 const MIGRATIONS = fileURLToPath(new URL('../../../supabase/migrations', import.meta.url));