feat(db): version PostgreSQL boot migrations and extract read cursor repository

backend/.eslintrc.js

@@ -33,6 +33,21 @@ module.exports = {
     'prefer-regex-literals': 'off',
   },
   overrides: [
+    {
+      // Extracted aggregates are private to their repositories. Fixtures may
+      // still seed/inspect the real model to test the persistence contract.
+      files: ['**/*.ts', '**/*.js'],
+      excludedFiles: ['repositories/**', '__tests__/**', 'models/**', 'dist/**'],
+      rules: {
+        'no-restricted-imports': ['error', { patterns: [{
+          group: ['**/models/AgentPodReadCursor', '**/models/AgentPodReadCursor.*'],
+          message: 'Use repositories/agentPodReadCursorRepository.',
+        }] }],
+        'no-restricted-modules': ['error', { patterns: [
+          '**/models/AgentPodReadCursor', '**/models/AgentPodReadCursor.*',
+        ] }],
+      },
+    },
     {
       // TypeScript. Added 2026-08-28 for TASK-024, which measured backend
       // eslint reaching 0 of 286 `.ts` files: `npm run lint` was scoped

backend/Dockerfile

@@ -57,10 +57,9 @@ RUN npm install --include=dev
 COPY . .
 
 # Compile TypeScript to dist/
-RUN npx tsc -p tsconfig.build.json
+RUN npm run build
 
 # Copy non-JS assets that tsc doesn't copy
-RUN cp config/schema.sql dist/config/schema.sql
 RUN cp -r external dist/external 2>/dev/null || true
 
 EXPOSE 5000

backend/TESTING.md

@@ -43,21 +43,21 @@ cd backend && INTEGRATION_TEST=true \
   npx jest --testPathPattern="__tests__/service" --forceExit --runInBand
 ```
 
-Schema source: `backend/config/schema.sql`. `testUtils.setupPgDb()` applies it verbatim after `CREATE EXTENSION IF NOT EXISTS pgcrypto`.
-
-> **Blind spot both tiers share: neither exercises an EXISTING database.**
-> Tier 1 provisions a fresh `postgres:16` per CI run, so `CREATE TABLE IF NOT
-> EXISTS` always takes the CREATE branch. Tier 0 reads `schema.sql` as text.
-> A column added to an existing table therefore needs a *third* arrangement to
-> be tested at all: start from the pre-threading table shape, apply the
-> retrofit `ALTER`s, assert the column appears. See
+Schema source: ordered files in `backend/config/migrations/`. `testUtils.setupPgDb()` runs the boot migration runner after `CREATE EXTENSION IF NOT EXISTS pgcrypto`. `config/schema.sql` aliases the frozen first migration for older fixtures. See [the migration contract](../docs/database/schema-migrations.md).
+
+> **A fresh database alone does not prove an upgrade.**
+> Tier 1 normally provisions fresh PostgreSQL per CI run, and many Tier 0
+> fixtures inspect SQL as text. An upgrade needs an additional arrangement:
+> start from the previous populated table shape, apply pending migrations,
+> and assert the new column appears without losing data. See
 > `__tests__/unit/models/threadingSchemaRetrofit.test.js`.
 >
 > This is not hypothetical. On 2026-08-22 `thread_root_id` shipped declared
 > only inside its `CREATE TABLE`; both tiers were green and the column would
 > never have existed on the live instance, throwing at boot on the index and
-> failing every INSERT that named it. Any PR adding a column to an existing
-> table should be read for both declarations.
+> failing every INSERT that named it. New migrations need an existing-shape
+> test. `__tests__/service/pgMigrations.test.js` now exercises that path with
+> real PostgreSQL, plus rollback and concurrent boots.
 
 Tier-1 setup logs `[tier1] Connected to real MongoDB …` and `[tier1] Connected to real Postgres …` so the CI run log makes the mode obvious.
 

backend/__tests__/service/pgMigrations.test.js

@@ -0,0 +1,132 @@
+/* eslint-disable import/no-unresolved, import/extensions -- TypeScript modules expose CJS exports. */
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const { runPgMigrations } = require('../../config/run-pg-migrations');
+const { createRealPgSchema } = require('../utils/realPgSchema');
+const { waitFor } = require('../utils/testUtils');
+
+const real = process.env.INTEGRATION_TEST === 'true' ? describe : describe.skip;
+real('Postgres migrations: real DDL, rollback and competing boots', () => {
+  let database;
+  let pool;
+  let directory;
+  const write = (name, sql) => fs.writeFileSync(path.join(directory, name), sql);
+  beforeEach(async () => {
+    database = await createRealPgSchema();
+    pool = database.pool;
+    directory = fs.mkdtempSync(path.join(os.tmpdir(), 'pg-migration-real-'));
+  });
+  afterEach(async () => {
+    if (database) await database.close();
+    if (directory) fs.rmSync(directory, { recursive: true, force: true });
+  });
+
+  it('boots the shipped baseline on a fresh database and does not reapply it', async () => {
+    expect(await runPgMigrations(pool)).toEqual(['0001_legacy_schema.sql']);
+    const history = (await pool.query('SELECT * FROM schema_migrations')).rows;
+    expect(history[0].applied_at).toBeInstanceOf(Date);
+    expect(await runPgMigrations(pool)).toEqual([]);
+    expect((await pool.query('SELECT * FROM schema_migrations')).rows).toEqual(history);
+    // The application/backfill ledger has distinct semantics and is preserved.
+    expect((await pool.query('SELECT * FROM migration_records')).rows).toEqual([]);
+  });
+
+  it('retrofits a populated pre-runner schema without dropping messages or ledger details', async () => {
+    await pool.query(`
+      CREATE TABLE pods (id VARCHAR(24) PRIMARY KEY, name VARCHAR(100) NOT NULL,
+        type VARCHAR(50) NOT NULL, created_by VARCHAR(24) NOT NULL);
+      CREATE TABLE users (_id VARCHAR(24) PRIMARY KEY, username VARCHAR(100) NOT NULL);
+      CREATE TABLE messages (id SERIAL PRIMARY KEY, pod_id VARCHAR(24) REFERENCES pods(id),
+        user_id VARCHAR(24) NOT NULL, content TEXT NOT NULL, created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP);
+      CREATE TABLE migration_records (name VARCHAR(255) PRIMARY KEY,
+        applied_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, details JSONB);
+      INSERT INTO pods VALUES ('pod', 'kept', 'chat', 'user');
+      INSERT INTO users VALUES ('user', 'peer');
+      INSERT INTO messages (pod_id, user_id, content) VALUES ('pod', 'user', 'old message');
+      INSERT INTO migration_records (name, details) VALUES ('thread-cutoff', '{"cutoff":"kept"}');
+    `);
+    expect(await runPgMigrations(pool)).toEqual(['0001_legacy_schema.sql']);
+    expect((await pool.query('SELECT id, content, thread_root_id, freshness_revision FROM messages')).rows)
+      .toEqual([{
+        id: 1, content: 'old message', thread_root_id: null, freshness_revision: '0',
+      }]);
+    expect((await pool.query('SELECT is_bot FROM users')).rows).toEqual([{ is_bot: false }]);
+    expect((await pool.query('SELECT details FROM migration_records')).rows).toEqual([{ details: { cutoff: 'kept' } }]);
+    const inserted = await pool.query("INSERT INTO messages (pod_id,user_id,content) VALUES ('pod','user','new') RETURNING freshness_revision");
+    expect(inserted.rows[0].freshness_revision).toBe('1');
+  });
+
+  it('rolls back both DDL and completion on failure, then retries only the failed migration', async () => {
+    write('0001_first.sql', 'CREATE TABLE example (id INTEGER PRIMARY KEY); INSERT INTO example VALUES (7);');
+    write('0002_upgrade.sql', 'ALTER TABLE example ADD COLUMN label TEXT; INSERT INTO missing_table VALUES (1);');
+    await expect(runPgMigrations(pool, directory)).rejects.toThrow('missing_table');
+    expect((await pool.query('SELECT * FROM example')).rows).toEqual([{ id: 7 }]);
+    const first = (await pool.query('SELECT * FROM schema_migrations')).rows;
+    expect(first.map((row) => row.version)).toEqual(['0001_first.sql']);
+    write('0002_upgrade.sql', "ALTER TABLE example ADD COLUMN label TEXT; UPDATE example SET label = 'upgraded';");
+    expect(await runPgMigrations(pool, directory)).toEqual(['0002_upgrade.sql']);
+    expect((await pool.query('SELECT * FROM example')).rows).toEqual([{ id: 7, label: 'upgraded' }]);
+    expect((await pool.query('SELECT * FROM schema_migrations ORDER BY version')).rows[0]).toEqual(first[0]);
+    expect(await runPgMigrations(pool, directory)).toEqual([]);
+  });
+
+  it('rolls back schema changes if writing the completion row fails', async () => {
+    write('0001_first.sql', 'CREATE TABLE example (id INTEGER);');
+    await runPgMigrations(pool, directory);
+    await pool.query("ALTER TABLE schema_migrations ADD CONSTRAINT refuse_second CHECK (version <> '0002_second.sql')");
+    write('0002_second.sql', 'ALTER TABLE example ADD COLUMN label TEXT;');
+    await expect(runPgMigrations(pool, directory)).rejects.toThrow('refuse_second');
+    await expect(pool.query('SELECT label FROM example')).rejects.toThrow('label');
+    expect((await pool.query('SELECT version FROM schema_migrations')).rows).toEqual([{ version: '0001_first.sql' }]);
+    await pool.query('ALTER TABLE schema_migrations DROP CONSTRAINT refuse_second');
+    expect(await runPgMigrations(pool, directory)).toEqual(['0002_second.sql']);
+  });
+
+  it('serializes two boots despite an unrelated waiter and executes SQL once', async () => {
+    write('0001_once.sql', 'CREATE TABLE once_only (id INTEGER); INSERT INTO once_only VALUES (1);');
+    const blocker = await pool.connect();
+    const lock = [1129270605, 504];
+    const boots = [];
+    const bootPids = [];
+    // Observe each real checked-out connection without replacing its queries.
+    const bootPool = {
+      connect: async () => {
+        const client = await pool.connect();
+        bootPids.push(client.processID);
+        return client;
+      },
+    };
+    let unrelated;
+    let outsiderPid;
+    try {
+      await blocker.query('SELECT pg_advisory_lock($1::integer, $2::integer)', lock);
+      const outsider = await pool.connect();
+      outsiderPid = outsider.processID;
+      unrelated = outsider.query('SELECT pg_advisory_lock($1::integer, $2::integer)', lock)
+        .then(() => outsider.query('SELECT pg_advisory_unlock($1::integer, $2::integer)', lock))
+        .finally(() => outsider.release());
+      boots.push(runPgMigrations(bootPool, directory), runPgMigrations(bootPool, directory));
+      // Observe both waiters in Postgres itself; no sleep-based timing claim.
+      await waitFor(async () => {
+        const result = await pool.query(`SELECT count(*) FILTER (WHERE pid = ANY($3::integer[]))::int AS own,
+            count(*) FILTER (WHERE pid = $4::integer)::int AS outsider FROM pg_locks
+          WHERE locktype = 'advisory' AND classid = $1 AND objid = $2 AND NOT granted
+            AND database = (SELECT oid FROM pg_database WHERE datname = current_database())`, [...lock, bootPids, outsiderPid]);
+        // Other Tier-1 workers share the lock even in separate schemas. Count
+        // our two PIDs, and prove the extra waiter really is present too.
+        return result.rows[0].own === 2 && result.rows[0].outsider === 1;
+      });
+    } finally {
+      await blocker.query('SELECT pg_advisory_unlock($1::integer, $2::integer)', lock);
+      blocker.release();
+      // Drain both even if the lock assertion fails, so cleanup never hangs.
+      await Promise.allSettled([...boots, unrelated]);
+    }
+    // allSettled above is cleanup, not permission to hide the control failing.
+    await unrelated;
+    expect((await Promise.all(boots)).sort((a, b) => a.length - b.length)).toEqual([[], ['0001_once.sql']]);
+    expect((await pool.query('SELECT * FROM once_only')).rows).toEqual([{ id: 1 }]);
+    expect((await pool.query('SELECT version FROM schema_migrations')).rows).toEqual([{ version: '0001_once.sql' }]);
+  });
+});

backend/__tests__/unit/config/init-pg-db.test.js

@@ -1,37 +1,18 @@
-/* eslint-disable import/no-unresolved, import/extensions */
-jest.mock('fs');
-// The unit under test is initializeDatabase; its pool dependency is stubbed. This
-// suite previously reached for a REAL Pool object, which existed only because
-// setup.js left PG_HOST as the truthy string 'undefined' and db-pg.ts:75 built a
-// Pool for that host — so it broke the moment the harness stopped lying about the
-// host, not because initializeDatabase changed (TASK-128).
+/* eslint-disable import/no-unresolved, import/extensions -- TypeScript modules expose CJS exports. */
 jest.mock('../../../config/db-pg', () => ({ pool: { connect: jest.fn() } }));
-const fs = require('fs');
+jest.mock('../../../config/run-pg-migrations', () => ({ runPgMigrations: jest.fn() }));
 const { pool } = require('../../../config/db-pg');
+const { runPgMigrations } = require('../../../config/run-pg-migrations');
+const initialize = require('../../../config/init-pg-db');
 
-delete require.cache[require.resolve('../../../config/init-pg-db')];
-const initDb = require('../../../config/init-pg-db');
-
-describe('initializeDatabase', () => {
-  beforeEach(() => {
-    jest.clearAllMocks();
-  });
-
-  it('executes schema when file exists', async () => {
-    fs.existsSync.mockReturnValue(true);
-    fs.readFileSync.mockReturnValue('SQL');
-    const client = { query: jest.fn(), release: jest.fn() };
-    pool.connect = jest.fn().mockResolvedValue(client);
-    const result = await initDb();
-    expect(client.query).toHaveBeenCalledWith('SQL');
-    expect(result).toBe(true);
-  });
+it('reports ready only after all pending migrations complete', async () => {
+  runPgMigrations.mockResolvedValue(['0001_legacy_schema.sql']);
+  expect(await initialize()).toBe(true);
+  expect(runPgMigrations).toHaveBeenCalledWith(pool);
+});
 
-  it('returns false when schema file missing', async () => {
-    fs.existsSync.mockReturnValue(false);
-    const client = { release: jest.fn() };
-    pool.connect = jest.fn().mockResolvedValue(client);
-    const result = await initDb();
-    expect(result).toBe(false);
-  });
+it('keeps boot unready on migration failure and can retry', async () => {
+  runPgMigrations.mockRejectedValueOnce(new Error('bad migration')).mockResolvedValueOnce([]);
+  expect(await initialize()).toBe(false);
+  expect(await initialize()).toBe(true);
 });

backend/__tests__/unit/config/run-pg-migrations.test.js

@@ -0,0 +1,110 @@
+/* eslint-disable import/no-unresolved, import/extensions -- TypeScript modules expose CJS exports. */
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const { newDb, DataType } = require('pg-mem');
+const { runPgMigrations } = require('../../../config/run-pg-migrations');
+
+// pg-mem executes the DDL/ledger contract. It cannot prove locks or rollback;
+// those assertions live in service/pgMigrations.test.js against real Postgres.
+describe('Postgres migrations (pg-mem)', () => {
+  let pool;
+  let directory;
+  const write = (name, sql) => fs.writeFileSync(path.join(directory, name), sql);
+  beforeEach(() => {
+    // pg-mem misreports coverage on repeated CREATE TABLE IF NOT EXISTS.
+    const db = newDb({ noAstCoverageCheck: true });
+    ['pg_advisory_lock', 'pg_advisory_unlock'].forEach((name) => db.public.registerFunction({
+      name,
+      args: [DataType.integer, DataType.integer],
+      returns: DataType.integer,
+      implementation: () => 1,
+    }));
+    pool = new (db.adapters.createPg().Pool)();
+    directory = fs.mkdtempSync(path.join(os.tmpdir(), 'pg-migration-unit-'));
+    write('0001_initial.sql', 'CREATE TABLE example (id INTEGER PRIMARY KEY);');
+  });
+  afterEach(async () => {
+    await pool.end();
+    fs.rmSync(directory, { recursive: true, force: true });
+  });
+
+  it('orders files, upgrades an existing table, preserves data and skips completed work', async () => {
+    expect(await runPgMigrations(pool, directory)).toEqual(['0001_initial.sql']);
+    await pool.query('INSERT INTO example (id) VALUES (7)');
+    write('0003_third.sql', "UPDATE example SET label = 'kept' WHERE id = 7;");
+    write('0002_second.sql', 'ALTER TABLE example ADD COLUMN label TEXT;');
+    expect(await runPgMigrations(pool, directory)).toEqual(['0002_second.sql', '0003_third.sql']);
+    expect((await pool.query('SELECT * FROM example')).rows).toEqual([{ id: 7, label: 'kept' }]);
+    const history = (await pool.query('SELECT * FROM schema_migrations ORDER BY version')).rows;
+    expect(history).toHaveLength(3);
+    expect(history[0]).toMatchObject({ checksum: expect.stringMatching(/^[a-f0-9]{64}$/), applied_at: expect.any(Date) });
+    expect(await runPgMigrations(pool, directory)).toEqual([]);
+    expect((await pool.query('SELECT * FROM schema_migrations ORDER BY version')).rows).toEqual(history);
+  });
+
+  it('refuses changed applied SQL before executing a later migration', async () => {
+    await runPgMigrations(pool, directory);
+    write('0001_initial.sql', 'DROP TABLE example;');
+    write('0002_later.sql', 'CREATE TABLE should_not_exist (id INTEGER);');
+    await expect(runPgMigrations(pool, directory)).rejects.toThrow('history mismatch');
+    expect((await pool.query('SELECT * FROM example')).rows).toEqual([]);
+    expect((await pool.query("SELECT table_name FROM information_schema.tables WHERE table_name = 'should_not_exist'")).rows).toEqual([]);
+  });
+
+  it('refuses missing files and migrations inserted before already-applied history', async () => {
+    await runPgMigrations(pool, directory);
+    write('0000_earlier.sql', 'DROP TABLE example;');
+    await expect(runPgMigrations(pool, directory)).rejects.toThrow('history mismatch');
+    fs.rmSync(path.join(directory, '0000_earlier.sql'));
+    fs.rmSync(path.join(directory, '0001_initial.sql'));
+    await expect(runPgMigrations(pool, directory)).rejects.toThrow('missing');
+  });
+
+  it('refuses a removed applied file while a later file remains, even with identical SQL', async () => {
+    const initial = fs.readFileSync(path.join(directory, '0001_initial.sql'), 'utf8');
+    await runPgMigrations(pool, directory);
+    fs.rmSync(path.join(directory, '0001_initial.sql'));
+    // Same checksum, different version: checksum alone cannot identify history.
+    write('0002_later.sql', initial);
+    await expect(runPgMigrations(pool, directory)).rejects.toThrow('history mismatch: 0001_initial.sql');
+    expect((await pool.query('SELECT version FROM schema_migrations')).rows).toEqual([{ version: '0001_initial.sql' }]);
+  });
+
+  it('refuses an older image whose file list is shorter than applied history', async () => {
+    await runPgMigrations(pool, directory);
+    write('0002_later.sql', 'ALTER TABLE example ADD COLUMN label TEXT;');
+    await runPgMigrations(pool, directory);
+    fs.rmSync(path.join(directory, '0002_later.sql'));
+    await expect(runPgMigrations(pool, directory)).rejects.toThrow('history mismatch: 0002_later.sql');
+    expect((await pool.query('SELECT version FROM schema_migrations ORDER BY version')).rows)
+      .toEqual([{ version: '0001_initial.sql' }, { version: '0002_later.sql' }]);
+  });
+
+  it.each([
+    ['0001_duplicate.sql', 'SELECT 1;', 'duplicate'],
+    ['oops.sql', 'SELECT 1;', 'Invalid'],
+    ['0002_empty.sql', '', 'Empty'],
+  ])('refuses invalid migration %s before connecting', async (name, sql, error) => {
+    write(name, sql);
+    const connect = jest.spyOn(pool, 'connect');
+    await expect(runPgMigrations(pool, directory)).rejects.toThrow(error);
+    expect(connect).not.toHaveBeenCalled();
+  });
+
+  it('releases its client when the advisory lock query fails', async () => {
+    const client = { query: jest.fn().mockRejectedValue(new Error('offline')), release: jest.fn() };
+    await expect(runPgMigrations({ connect: async () => client }, directory)).rejects.toThrow('offline');
+    expect(client.release).toHaveBeenCalledWith(false);
+  });
+
+  it('destroys a client whose unlock fails instead of pooling a held lock', async () => {
+    const query = jest.fn(async (sql) => {
+      if (sql.includes('pg_advisory_unlock')) throw new Error('offline');
+      return { rows: [] };
+    });
+    const client = { query, release: jest.fn() };
+    await expect(runPgMigrations({ connect: async () => client }, directory)).rejects.toThrow('release');
+    expect(client.release).toHaveBeenCalledWith(true);
+  });
+});