From f71ac769cc3bc7998ef587adaea835d0312c8bb1 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sun, 25 Oct 2026 01:59:08 -0700 Subject: [PATCH 1/3] feat(db): version PostgreSQL boot migrations and extract read cursor repository Co-authored-by: otto Agent-Model: codex/gpt-6.1-sol/high --- backend/.eslintrc.js | 15 + backend/Dockerfile | 3 +- backend/TESTING.md | 20 +- .../__tests__/service/pgMigrations.test.js | 111 +++++++ .../__tests__/unit/config/init-pg-db.test.js | 45 +-- .../unit/config/run-pg-migrations.test.js | 90 ++++++ .../agentPodReadCursorRepository.test.js | 61 ++++ .../unit/repositories/modelBoundary.test.js | 13 + .../unit/services/agentSendFreshness.test.js | 30 +- backend/__tests__/utils/realPgSchema.js | 37 +++ backend/__tests__/utils/testUtils.js | 11 +- backend/config/init-pg-db.ts | 38 +-- .../config/migrations/0001_legacy_schema.sql | 283 +++++++++++++++++ backend/config/run-pg-migrations.ts | 84 +++++ backend/config/schema.sql | 297 +----------------- backend/package.json | 2 +- .../agentPodReadCursorRepository.ts | 32 ++ backend/scripts/copy-pg-assets.js | 10 + backend/services/agentSendFreshness.ts | 25 +- backend/services/pgBootService.ts | 2 +- docs/database/POSTGRESQL_MIGRATION.md | 3 + docs/database/schema-migrations.md | 92 ++++++ 22 files changed, 888 insertions(+), 416 deletions(-) create mode 100644 backend/__tests__/service/pgMigrations.test.js create mode 100644 backend/__tests__/unit/config/run-pg-migrations.test.js create mode 100644 backend/__tests__/unit/repositories/agentPodReadCursorRepository.test.js create mode 100644 backend/__tests__/unit/repositories/modelBoundary.test.js create mode 100644 backend/__tests__/utils/realPgSchema.js create mode 100644 backend/config/migrations/0001_legacy_schema.sql create mode 100644 backend/config/run-pg-migrations.ts mode change 100644 => 120000 backend/config/schema.sql create mode 100644 backend/repositories/agentPodReadCursorRepository.ts create mode 100644 backend/scripts/copy-pg-assets.js create mode 100644 docs/database/schema-migrations.md diff --git a/backend/.eslintrc.js b/backend/.eslintrc.js index 81302a21bd..7fc5ad652e 100644 --- a/backend/.eslintrc.js +++ b/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 diff --git a/backend/Dockerfile b/backend/Dockerfile index 37f76c0536..47a9a01314 100644 --- a/backend/Dockerfile +++ b/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 diff --git a/backend/TESTING.md b/backend/TESTING.md index fd5ca55bdd..4a1b677579 100644 --- a/backend/TESTING.md +++ b/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. diff --git a/backend/__tests__/service/pgMigrations.test.js b/backend/__tests__/service/pgMigrations.test.js new file mode 100644 index 0000000000..c3132fb43e --- /dev/null +++ b/backend/__tests__/service/pgMigrations.test.js @@ -0,0 +1,111 @@ +/* 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 behind the same session lock 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 = []; + try { + await blocker.query('SELECT pg_advisory_lock($1::integer, $2::integer)', lock); + boots.push(runPgMigrations(pool, directory), runPgMigrations(pool, directory)); + // Observe both waiters in Postgres itself; no sleep-based timing claim. + await waitFor(async () => { + const result = await pool.query(`SELECT count(*)::int AS n 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); + return result.rows[0].n === 2; + }); + } 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); + } + 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' }]); + }); +}); diff --git a/backend/__tests__/unit/config/init-pg-db.test.js b/backend/__tests__/unit/config/init-pg-db.test.js index 22f7fd7cb4..becd44e8c2 100644 --- a/backend/__tests__/unit/config/init-pg-db.test.js +++ b/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); }); diff --git a/backend/__tests__/unit/config/run-pg-migrations.test.js b/backend/__tests__/unit/config/run-pg-migrations.test.js new file mode 100644 index 0000000000..32116f2976 --- /dev/null +++ b/backend/__tests__/unit/config/run-pg-migrations.test.js @@ -0,0 +1,90 @@ +/* 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.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); + }); +}); diff --git a/backend/__tests__/unit/repositories/agentPodReadCursorRepository.test.js b/backend/__tests__/unit/repositories/agentPodReadCursorRepository.test.js new file mode 100644 index 0000000000..9d12863413 --- /dev/null +++ b/backend/__tests__/unit/repositories/agentPodReadCursorRepository.test.js @@ -0,0 +1,61 @@ +/* eslint-disable import/no-unresolved, import/extensions -- TypeScript modules expose CJS exports. */ +const mongoose = require('mongoose'); +const Cursor = require('../../../models/AgentPodReadCursor'); +const { getReadWatermark, advanceReadWatermark } = require('../../../repositories/agentPodReadCursorRepository'); +const { setupMongoDb, closeMongoDb } = require('../../utils/testUtils'); + +const at = (seconds) => new Date(Date.UTC(2026, 9, 25, 0, 0, seconds)); +describe('read cursor repository (Mongo implementation)', () => { + const agent = new mongoose.Types.ObjectId(); + const pod = new mongoose.Types.ObjectId(); + beforeAll(setupMongoDb); + afterAll(closeMongoDb); + beforeEach(async () => { await Cursor.deleteMany({}); }); + + it('returns null for an absent cursor and only a plain watermark after a round trip', async () => { + expect(await getReadWatermark(agent, pod)).toBeNull(); + await advanceReadWatermark(agent, pod, { readAt: at(1) }); + const value = await getReadWatermark(agent, pod); + expect(value).toEqual({ readAt: at(1) }); + expect(Object.getPrototypeOf(value)).toBe(Object.prototype); + expect(value.save).toBeUndefined(); + const row = await Cursor.findById(`${agent}:${pod}`).lean(); + expect(String(row.agentUserId)).toBe(String(agent)); + expect(String(row.podId)).toBe(String(pod)); + expect(row.readRevision).toBeUndefined(); + }); + + it('preserves independent maxima and an existing revision when omitted', async () => { + await advanceReadWatermark(agent, pod, { readAt: at(2), readRevision: 3 }); + await advanceReadWatermark(agent, pod, { readAt: at(1), readRevision: 5 }); + expect(await getReadWatermark(String(agent), String(pod))).toEqual({ readAt: at(2), readRevision: 5 }); + await advanceReadWatermark(agent, pod, { readAt: at(4) }); + expect(await getReadWatermark(agent, pod)).toEqual({ readAt: at(4), readRevision: 5 }); + }); + + it('isolates both the agent and pod parts of the key', async () => { + await advanceReadWatermark(agent, pod, { readAt: at(2) }); + expect(await getReadWatermark(new mongoose.Types.ObjectId(), pod)).toBeNull(); + expect(await getReadWatermark(agent, new mongoose.Types.ObjectId())).toBeNull(); + }); + + it('concurrent first reads converge on one row without rolling either maximum back', async () => { + await Promise.all(Array.from({ length: 20 }, (_, i) => advanceReadWatermark(agent, pod, { + readAt: at(i), readRevision: 20 - i, + }))); + expect(await Cursor.countDocuments({})).toBe(1); + expect(await getReadWatermark(agent, pod)).toEqual({ readAt: at(19), readRevision: 20 }); + }); + + it('retries only duplicate-key upserts and propagates other write/read errors', async () => { + await advanceReadWatermark(agent, pod, { readAt: at(1) }); + const update = jest.spyOn(Cursor, 'updateOne').mockRejectedValueOnce(Object.assign(new Error('duplicate'), { code: 11000 })); + await advanceReadWatermark(agent, pod, { readAt: at(2), readRevision: 0 }); + expect(update.mock.calls[1]).toEqual([{ _id: `${agent}:${pod}` }, { $max: { readAt: at(2), readRevision: 0 } }]); + expect(await getReadWatermark(agent, pod)).toEqual({ readAt: at(2), readRevision: 0 }); + update.mockRejectedValueOnce(new Error('down')); + await expect(advanceReadWatermark(agent, pod, { readAt: at(3) })).rejects.toThrow('down'); + jest.spyOn(Cursor, 'findOne').mockImplementationOnce(() => { throw new Error('read down'); }); + await expect(getReadWatermark(agent, pod)).rejects.toThrow('read down'); + }); +}); diff --git a/backend/__tests__/unit/repositories/modelBoundary.test.js b/backend/__tests__/unit/repositories/modelBoundary.test.js new file mode 100644 index 0000000000..160d3aaa24 --- /dev/null +++ b/backend/__tests__/unit/repositories/modelBoundary.test.js @@ -0,0 +1,13 @@ +const path = require('path'); +const { ESLint } = require('eslint'); + +const eslint = new ESLint({ cwd: path.resolve(__dirname, '../../..') }); +it.each([ + ["import Cursor from '../models/AgentPodReadCursor';", 'no-restricted-imports'], + ["const Cursor = require('../models/AgentPodReadCursor');", 'no-restricted-modules'], +])('blocks production model access: %s', async (source, rule) => { + const [service] = await eslint.lintText(source, { filePath: 'services/boundaryProbe.ts' }); + expect(service.messages.map((message) => message.ruleId)).toContain(rule); + const [repository] = await eslint.lintText(source, { filePath: 'repositories/boundaryProbe.ts' }); + expect(repository.messages.map((message) => message.ruleId)).not.toContain(rule); +}); diff --git a/backend/__tests__/unit/services/agentSendFreshness.test.js b/backend/__tests__/unit/services/agentSendFreshness.test.js index d3bfb110d5..fc70464310 100644 --- a/backend/__tests__/unit/services/agentSendFreshness.test.js +++ b/backend/__tests__/unit/services/agentSendFreshness.test.js @@ -1,6 +1,6 @@ -jest.mock('../../../models/AgentPodReadCursor', () => ({ - findOne: jest.fn(), - updateOne: jest.fn(), +jest.mock('../../../repositories/agentPodReadCursorRepository', () => ({ + getReadWatermark: jest.fn(), + advanceReadWatermark: jest.fn(), })); jest.mock('../../../models/Message', () => ({ countDocuments: jest.fn(), @@ -9,7 +9,7 @@ jest.mock('../../../models/Message', () => ({ })); jest.mock('../../../config/db-pg', () => ({ pool: { query: jest.fn() } })); -const Cursor = require('../../../models/AgentPodReadCursor'); +const Cursor = require('../../../repositories/agentPodReadCursorRepository'); const Message = require('../../../models/Message'); const { pool } = require('../../../config/db-pg'); const Freshness = require('../../../services/agentSendFreshness'); @@ -20,7 +20,7 @@ describe('agent send-time freshness', () => { jest.clearAllMocks(); delete process.env.PG_HOST; Message.exists.mockResolvedValue(null); - Cursor.findOne.mockReturnValue({ select: () => ({ lean: async () => ({ readAt: new Date('2026-10-24T00:00:00Z') }) }) }); + Cursor.getReadWatermark.mockResolvedValue({ readAt: new Date('2026-10-24T00:00:00Z') }); }); afterAll(() => { if (originalPgHost === undefined) delete process.env.PG_HOST; @@ -71,13 +71,9 @@ describe('agent send-time freshness', () => { it('uses a monotonic watermark for a successful read', async () => { const at = new Date('2026-10-24T01:00:00Z'); - Cursor.findOne.mockReturnValueOnce({ select: () => ({ lean: async () => null }) }); + Cursor.getReadWatermark.mockResolvedValueOnce(null); await Freshness.recordRead('agent-id', 'pod-id', { at }, []); - expect(Cursor.updateOne).toHaveBeenCalledWith( - { _id: 'agent-id:pod-id' }, - { $max: { readAt: at }, $setOnInsert: { agentUserId: 'agent-id', podId: 'pod-id' } }, - { upsert: true }, - ); + expect(Cursor.advanceReadWatermark).toHaveBeenCalledWith('agent-id', 'pod-id', { readAt: at }); }); it('does not mistake an unavailable PostgreSQL store for a quiet pod', async () => { @@ -106,9 +102,9 @@ describe('agent send-time freshness', () => { const start = await Freshness.readStart('pod-id'); expect(start).toEqual({ at: expect.any(Date), revision: 7 }); expect(pool.query.mock.calls[0][0]).toContain('pod_message_revisions'); - Cursor.findOne.mockReturnValueOnce({ select: () => ({ lean: async () => null }) }); + Cursor.getReadWatermark.mockResolvedValueOnce(null); await Freshness.recordRead('agent-id', 'pod-id', start, []); - expect(Cursor.updateOne.mock.calls[0][1].$max).toEqual({ readAt: start.at, readRevision: 7 }); + expect(Cursor.advanceReadWatermark.mock.calls[0][2]).toEqual({ readAt: start.at, readRevision: 7 }); }); it('does not stamp a PostgreSQL read that omitted a newly committed message', async () => { @@ -116,17 +112,15 @@ describe('agent send-time freshness', () => { pool.query.mockResolvedValue({ rows: [{ '?column?': 1 }] }); await Freshness.recordRead('agent-id', 'pod-id', { at: new Date(), revision: 12 }, ['19']); expect(pool.query.mock.calls[0][1][4]).toEqual([19]); - expect(Cursor.updateOne).not.toHaveBeenCalled(); + expect(Cursor.advanceReadWatermark).not.toHaveBeenCalled(); }); it('stamps a PostgreSQL read when every new peer message was returned', async () => { process.env.PG_HOST = 'configured'; - Cursor.findOne.mockReturnValueOnce({ - select: () => ({ lean: async () => ({ readAt: new Date('2026-10-24T00:00:00Z'), readRevision: 4 }) }), - }); + Cursor.getReadWatermark.mockResolvedValueOnce({ readAt: new Date('2026-10-24T00:00:00Z'), readRevision: 4 }); pool.query.mockResolvedValue({ rows: [] }); await Freshness.recordRead('agent-id', 'pod-id', { at: new Date(), revision: 6 }, ['5', '6']); expect(pool.query.mock.calls[0][1]).toEqual(['pod-id', 4, 6, 'agent-id', [5, 6]]); - expect(Cursor.updateOne).toHaveBeenCalledTimes(1); + expect(Cursor.advanceReadWatermark).toHaveBeenCalledTimes(1); }); }); diff --git a/backend/__tests__/utils/realPgSchema.js b/backend/__tests__/utils/realPgSchema.js new file mode 100644 index 0000000000..b0f90e9aa4 --- /dev/null +++ b/backend/__tests__/utils/realPgSchema.js @@ -0,0 +1,37 @@ +const { randomUUID } = require('crypto'); +const { Pool } = require('pg'); + +// Isolate DDL tests without dropping a shared database. Each pool connection +// uses the same private schema, including checked-out migration sessions. +async function createRealPgSchema() { + const config = { + host: process.env.PG_HOST, + port: Number(process.env.PG_PORT || 5432), + user: process.env.PG_USER, + password: process.env.PG_PASSWORD, + database: process.env.PG_DATABASE, + ssl: false, + connectionTimeoutMillis: 5000, + }; + const schema = `test_migrations_${randomUUID().replace(/-/g, '')}`; + const admin = new Pool(config); + try { + await admin.query(`CREATE SCHEMA "${schema}"`); + } catch (error) { + await admin.end(); + throw error; + } + const pool = new Pool({ ...config, options: `-c search_path=${schema}`, max: 6 }); + return { + pool, + close: async () => { + await pool.end(); + try { + await admin.query(`DROP SCHEMA "${schema}" CASCADE`); + } finally { + await admin.end(); + } + }, + }; +} +module.exports = { createRealPgSchema }; diff --git a/backend/__tests__/utils/testUtils.js b/backend/__tests__/utils/testUtils.js index b8bd04fe7f..f3a6e5abec 100644 --- a/backend/__tests__/utils/testUtils.js +++ b/backend/__tests__/utils/testUtils.js @@ -1,5 +1,3 @@ -const fs = require('fs'); -const path = require('path'); const mongoose = require('mongoose'); const { MongoMemoryServer } = require('mongodb-memory-server'); const { newDb } = require('pg-mem'); @@ -139,13 +137,14 @@ const setupPgDb = async () => { ssl: false, }); await pgPool.query('CREATE EXTENSION IF NOT EXISTS pgcrypto'); - const schemaPath = path.join(__dirname, '..', '..', 'config', 'schema.sql'); - const schema = fs.readFileSync(schemaPath, 'utf8'); - await pgPool.query(schema); + // Exercise the boot runner so later migrations reach tier-1 fixtures too. + // eslint-disable-next-line global-require, import/no-unresolved, import/extensions + const { runPgMigrations } = require('../../config/run-pg-migrations'); + await runPgMigrations(pgPool); // Same cross-file-contamination concern as Mongo: TRUNCATE on setup so // each suite starts with an empty schema. await pgPool.query('TRUNCATE TABLE messages, pod_members, pods, users RESTART IDENTITY CASCADE'); - console.log('[tier1] Connected to real Postgres, applied schema.sql, truncated tables'); + console.log('[tier1] Connected to real Postgres, applied migrations, truncated tables'); return pgPool; } diff --git a/backend/config/init-pg-db.ts b/backend/config/init-pg-db.ts index 54e7fc495b..50322aed45 100644 --- a/backend/config/init-pg-db.ts +++ b/backend/config/init-pg-db.ts @@ -1,41 +1,17 @@ -// eslint-disable-next-line global-require -const fs = require('fs'); -// eslint-disable-next-line global-require -const path = require('path'); -// eslint-disable-next-line global-require +import { runPgMigrations } from './run-pg-migrations'; + const { pool } = require('./db-pg'); const initializeDatabase = async (): Promise => { try { - console.log('Initializing PostgreSQL database...'); - const p = pool as { - connect: () => Promise<{ - query: (sql: string) => Promise; - release(): void; - }>; - }; - const client = await p.connect(); - - const schemaPath = path.join(__dirname, 'schema.sql') as string; - if (!fs.existsSync(schemaPath)) { - console.error(`Schema file not found at: ${schemaPath}`); - client.release(); - return false; - } - - const schema = fs.readFileSync(schemaPath, 'utf8') as string; - await client.query(schema); - - console.log('PostgreSQL schema created successfully'); - client.release(); + const applied = await runPgMigrations(pool); + console.log(`PostgreSQL migrations ready (${applied.length} applied)`); return true; - } catch (err) { - const e = err as { message?: string }; - console.error('Error initializing PostgreSQL database:', e.message); + } catch (error) { + console.error('Error initializing PostgreSQL database:', (error as Error).message); + // pgBootService retries and keeps PG routes/readiness unavailable. return false; } }; module.exports = initializeDatabase; - -export {}; diff --git a/backend/config/migrations/0001_legacy_schema.sql b/backend/config/migrations/0001_legacy_schema.sql new file mode 100644 index 0000000000..04331c2c11 --- /dev/null +++ b/backend/config/migrations/0001_legacy_schema.sql @@ -0,0 +1,283 @@ +-- Frozen legacy baseline (0001). Do not edit after release. +-- New DDL belongs in the next numbered SQL migration; see docs/database/schema-migrations.md. +-- config/schema.sql is a compatibility alias for older fixtures and operator scripts. +-- +-- This baseline retains the existing additive retrofits: a CREATE TABLE IF +-- NOT EXISTS does not add columns to an existing instance. Later upgrades +-- belong in new numbered files and must be tested on a populated old shape. + +-- Create pods table +CREATE TABLE IF NOT EXISTS pods ( + id VARCHAR(24) PRIMARY KEY, + name VARCHAR(100) NOT NULL, + description TEXT, + type VARCHAR(50) NOT NULL, + created_by VARCHAR(24) NOT NULL, + created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP +); + +-- Create pod_members table for many-to-many relationship +CREATE TABLE IF NOT EXISTS pod_members ( + id SERIAL PRIMARY KEY, + pod_id VARCHAR(24) REFERENCES pods(id) ON DELETE CASCADE, + user_id VARCHAR(24) NOT NULL, + joined_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, + UNIQUE(pod_id, user_id) +); + +-- Create messages table +CREATE TABLE IF NOT EXISTS messages ( + id SERIAL PRIMARY KEY, + pod_id VARCHAR(24) REFERENCES pods(id) ON DELETE CASCADE, + user_id VARCHAR(24) NOT NULL, + content TEXT NOT NULL, + message_type VARCHAR(20) DEFAULT 'text' NOT NULL, + -- Kernel-authored seat posts (install intros and stalled-connect nudges) + -- speak beside an agent without pretending to be that agent. NULL keeps + -- legacy human and agent rows unclassified until a trusted server path opts + -- into the marker. + author_kind VARCHAR(20), + reply_to_message_id INTEGER REFERENCES messages(id) ON DELETE SET NULL, + -- Threading (W-T, TASK-029). DELIBERATELY separate from reply_to_message_id, + -- which is an ADDRESSING edge: it feeds `isRouted` and the implicit-reply + -- chat.mention (agentMentionService :1025 / :1356), so a human replying to an + -- agent addresses that agent. If thread membership rode the same column, + -- joining a thread would ping the parent's author — the opposite of what + -- ambient scoping is for (fable's #1045 ruling). thread_root_id carries no + -- addressing. NULL means "not in a thread"; a root is an ordinary message + -- other messages point at, and a root's own thread_root_id stays NULL. + thread_root_id INTEGER REFERENCES messages(id) ON DELETE SET NULL, + -- Per-pod commit-order fence for agent send freshness. Existing rows stay 0. + freshness_revision BIGINT NOT NULL DEFAULT 0, + created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP +); + +-- Create users table for storing MongoDB user references +CREATE TABLE IF NOT EXISTS users ( + _id VARCHAR(24) PRIMARY KEY, + username VARCHAR(100) NOT NULL, + profile_picture TEXT, + is_bot BOOLEAN DEFAULT false, + created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP +); + +-- Migration: add is_bot column to existing tables (safe to run repeatedly) +ALTER TABLE users ADD COLUMN IF NOT EXISTS is_bot BOOLEAN DEFAULT false; + +-- ADR-020 D3: structured message payload (approval cards and future +-- server-defined message components). Self-applies at boot, same pattern +-- as is_bot above. NULL for ordinary messages. +ALTER TABLE messages ADD COLUMN IF NOT EXISTS payload JSONB; + +-- FUNNEL-5: the CREATE TABLE declaration above reaches fresh databases; this +-- retrofit reaches every instance whose messages table already exists. +ALTER TABLE messages ADD COLUMN IF NOT EXISTS author_kind VARCHAR(20); + +-- Threading (W-T, TASK-029). MUST be retrofitted, not only declared in the +-- CREATE TABLE above — `CREATE TABLE IF NOT EXISTS` is a no-op on an existing +-- table, so a column declared only there NEVER reaches an instance that +-- already has a messages table. Verified on the live dev instance 2026-08-22: +-- 7,001 rows, oldest 2026-07-07, and thread_root_id absent while `payload` +-- (which has this retrofit) is present. +-- +-- Without this line the failure is not subtle: the index below throws at boot +-- and every PGMessage.create fails, because the INSERT names a column that +-- does not exist. Chat posting stops on every existing instance. +-- +-- The tier-1 test suite could not catch it — CI provisions a FRESH postgres:16 +-- per run, where the CREATE TABLE path is the one exercised. Any new column on +-- an existing table needs BOTH declarations, and a test that starts from the +-- old shape (see threadingSchemaRetrofit.test.js). Caught by @sprint-review's +-- caveat on #1106, not by anything I ran. +ALTER TABLE messages + ADD COLUMN IF NOT EXISTS thread_root_id INTEGER REFERENCES messages(id) ON DELETE SET NULL; + +ALTER TABLE messages ADD COLUMN IF NOT EXISTS freshness_revision BIGINT NOT NULL DEFAULT 0; + +-- TASK-313 part 2. The Mongo id of a fallback message this row was reconciled +-- from. Nullable: ordinary rows carry NULL, and several NULLs never collide. +-- The UNIQUE index is NOT declared here: CONCURRENTLY cannot run inside this +-- multi-statement script, which executes as one transaction. The reconciler +-- builds it (services/pgMessageReconcileService.ensureLegacyIdIndex) before +-- its first insert, so it never blocks writes on messages. +ALTER TABLE messages ADD COLUMN IF NOT EXISTS legacy_mongo_id TEXT; + +CREATE TABLE IF NOT EXISTS pod_message_revisions ( + pod_id VARCHAR(24) PRIMARY KEY REFERENCES pods(id) ON DELETE CASCADE, + revision BIGINT NOT NULL DEFAULT 0 +); + +-- The counter update and message insert commit together. Concurrent inserts +-- into one pod serialize on this row, so revision order is commit order even +-- when an INSERT began before a reader and committed after it. +CREATE OR REPLACE FUNCTION assign_message_freshness_revision() +RETURNS TRIGGER AS $$ +BEGIN + INSERT INTO pod_message_revisions (pod_id, revision) + VALUES (NEW.pod_id, 1) + ON CONFLICT (pod_id) DO UPDATE + SET revision = pod_message_revisions.revision + 1 + RETURNING revision INTO NEW.freshness_revision; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS messages_freshness_revision ON messages; +CREATE TRIGGER messages_freshness_revision + BEFORE INSERT ON messages + FOR EACH ROW EXECUTE FUNCTION assign_message_freshness_revision(); + +CREATE INDEX IF NOT EXISTS idx_messages_pod_freshness_revision + ON messages(pod_id, freshness_revision); + +-- Migration records. There was no such table anywhere under backend/ before +-- this (@sprint-review, 56859, checked with a positive control): schema.sql is +-- idempotent boot DDL, which says what the shape IS and never when it changed. +-- That is fine until a RULE needs to reference the change — the threading +-- surface ruling (#1115) says pre-cutoff thread roots default to expanded, +-- "cutoff = the migration timestamp, read from the migration record", and that +-- record did not exist. +-- +-- Deliberately general rather than a threading-specific row. A one-off +-- `threading_migration` table is the thing that has to be generalised the +-- second time someone needs this, and the second time is when it gets done +-- wrong. `details` is JSONB so a migration can record what only IT knew — for +-- the threading backfill, the boundary timestamp it would otherwise discard. +-- +-- NOT a migration RUNNER. These are application/backfill records, not the +-- runner's schema_migrations table. A missing backfill row does not prove +-- that work never ran on an instance predating this application ledger. +CREATE TABLE IF NOT EXISTS migration_records ( + name VARCHAR(255) PRIMARY KEY, + applied_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, + details JSONB +); + +-- Retention execution ledger. This is deliberately NOT `migration_records`: +-- migration records describe one-off changes, while this records every run of +-- a destructive recurring job. A backend restart discards pod logs, so the +-- deletion count and the reasons no deletion occurred must live with the data. +-- A `running` row left behind is evidence of an interrupted run, not success. +CREATE TABLE IF NOT EXISTS pg_retention_runs ( + id BIGSERIAL PRIMARY KEY, + status VARCHAR(16) NOT NULL DEFAULT 'running' + CHECK (status IN ('running', 'completed', 'aborted', 'failed', 'skipped')), + configured_retention_days INTEGER, + final_retention_days INTEGER, + protected_pod_count INTEGER, + deleted_message_count BIGINT NOT NULL DEFAULT 0, + re_rooted_count BIGINT, + target_bytes BIGINT, + initial_size_bytes BIGINT, + final_size_bytes BIGINT, + detail TEXT, + started_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, + finished_at TIMESTAMP WITH TIME ZONE +); +-- `CREATE TABLE IF NOT EXISTS` does not add fields to an already-created +-- ledger, so keep this additive form for instances that booted an earlier DDL. +ALTER TABLE pg_retention_runs ADD COLUMN IF NOT EXISTS re_rooted_count BIGINT; +CREATE INDEX IF NOT EXISTS idx_pg_retention_runs_started_at ON pg_retention_runs(started_at DESC); + +-- Sprint B5: message reactions. One row per (message, user, emoji) — a user +-- can stack different emojis on the same message but each emoji is binary +-- (toggle on/off). PG-only; Mongo fallback path doesn't get reactions in v1. +CREATE TABLE IF NOT EXISTS message_reactions ( + id SERIAL PRIMARY KEY, + message_id INTEGER NOT NULL REFERENCES messages(id) ON DELETE CASCADE, + user_id VARCHAR(24) NOT NULL, + emoji VARCHAR(32) NOT NULL, + created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, + UNIQUE (message_id, user_id, emoji) +); +CREATE INDEX IF NOT EXISTS idx_message_reactions_message_id ON message_reactions(message_id); + +-- Room-grant broker audit and budget ledger. The budget counter is separate +-- from Mongo's grant document so every reservation is a single Postgres +-- conditional write; concurrent broker calls cannot read the same remaining +-- slot and both spend it. `window_started_at` is retained for windowed caps; +-- a NULL window on the grant is represented by a lifetime counter here. +CREATE TABLE IF NOT EXISTS tool_call_budgets ( + grant_id VARCHAR(255) PRIMARY KEY, + calls_used INTEGER NOT NULL DEFAULT 0, + window_started_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP +); + +CREATE TABLE IF NOT EXISTS tool_calls ( + call_id VARCHAR(255) PRIMARY KEY, + grant_id VARCHAR(255) NOT NULL, + pod_id VARCHAR(255), + installation_id VARCHAR(255), + agent_user_id VARCHAR(255) NOT NULL, + tool VARCHAR(255) NOT NULL, + args_digest CHAR(64) NOT NULL, + occurred_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, + outcome VARCHAR(32) NOT NULL CHECK (outcome IN ('ok', 'refused', 'pending_approval', 'failed')), + reason VARCHAR(255), + approval_id VARCHAR(255), + duration_ms INTEGER, + -- Whose credential ran (plan §8, §10 step 5): the Connection's `createdBy` at + -- call time. Copied, not joined — removal ends by deleting the Connection and + -- the trail outlives it. Nullable, and NULL means "no owner was recorded", + -- which is what a call refused before a connection resolved reports. + credential_owner_id VARCHAR(255), + vendor_status INTEGER, + vendor_error VARCHAR(200) +); +CREATE INDEX IF NOT EXISTS idx_tool_calls_grant_at ON tool_calls(grant_id, occurred_at DESC); +-- Per-user, per-thread state (W-T, TASK-029). ONE record carrying both +-- booleans, per ux-lead's ruling in docs/design/threading-surface-ruling.md +-- ("One state record, two booleans"): persisted collapse and follow share the +-- key (user, pod, thread root), so two tables would mean two writes for one +-- gesture. The collapsed column is here at birth rather than in the render PR. +-- +-- NOT a widened User.followedThreads: that field is typed +-- `postId: ObjectId ref 'Post', required` and cannot hold a Postgres +-- messages.id. Widening it would silently drop chat rows from every consumer +-- doing Post.find({_id: {$in: postIds}}) -- postController 457/513/548, +-- activityService 614/712, ActivityFeedPage 584. +-- +-- A row means "this user has non-default state on this thread". Row PRESENCE +-- carries no meaning on its own, which is why `following` is a column and not +-- row-existence: `collapsed` writes create rows for users who are not +-- following, and if presence meant following, expanding a thread would +-- subscribe you to it. +CREATE TABLE IF NOT EXISTS thread_user_state ( + id SERIAL PRIMARY KEY, + thread_root_id INTEGER NOT NULL REFERENCES messages(id) ON DELETE CASCADE, + user_id VARCHAR(255) NOT NULL, + pod_id VARCHAR(255) NOT NULL, + -- TRI-STATE, deliberately nullable. NULL = "no explicit choice, defer to the + -- participation default" (posted in the thread or was @-mentioned => following). + -- TRUE = explicitly followed. FALSE = explicitly unfollowed, which for a + -- participant is a MUTE and must outrank the participation default. + -- A NOT NULL DEFAULT false here would be a live bug: a collapse-only row + -- would silently unfollow a participant who had never touched the toggle. + following BOOLEAN, + -- Defaults TRUE for everyone including followers -- following never implies + -- expanded. Only the expand/collapse gesture writes it. + collapsed BOOLEAN NOT NULL DEFAULT TRUE, + followed_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, + UNIQUE (thread_root_id, user_id) +); +-- The wake path asks "who follows this thread" on every threaded reply, so the +-- root is the hot key. The user index serves the per-pod render read. +CREATE INDEX IF NOT EXISTS idx_thread_user_state_root ON thread_user_state(thread_root_id); +CREATE INDEX IF NOT EXISTS idx_thread_user_state_user_pod ON thread_user_state(user_id, pod_id); + +-- Create indexes for better performance +CREATE INDEX IF NOT EXISTS idx_messages_pod_id ON messages(pod_id); +CREATE INDEX IF NOT EXISTS idx_messages_created_at ON messages(created_at); +-- Threading reads are "give me this thread" and "how many replies has it", both +-- keyed on the root. reply_to_message_id has no index, which is why the root is +-- stored rather than walked: deriving it per read is a seq-scan per level. +CREATE INDEX IF NOT EXISTS idx_messages_thread_root_id ON messages(thread_root_id); +CREATE INDEX IF NOT EXISTS idx_pod_members_pod_id ON pod_members(pod_id); +CREATE INDEX IF NOT EXISTS idx_pod_members_user_id ON pod_members(user_id); +CREATE INDEX IF NOT EXISTS idx_pods_created_by ON pods(created_by); +CREATE INDEX IF NOT EXISTS idx_pods_type ON pods(type); +CREATE INDEX IF NOT EXISTS idx_users_username ON users(username); diff --git a/backend/config/run-pg-migrations.ts b/backend/config/run-pg-migrations.ts new file mode 100644 index 0000000000..687b0a57c6 --- /dev/null +++ b/backend/config/run-pg-migrations.ts @@ -0,0 +1,84 @@ +import { createHash } from 'crypto'; +import { readdirSync, readFileSync } from 'fs'; +import path from 'path'; +import type { Pool } from 'pg'; + +// A session lock: every boot keeps one checked-out connection until unlock. +// Two int32 keys namespace it separately from application advisory locks. +const LOCK = [1129270605, 504]; + +export async function runPgMigrations( + pool: Pool, + directory = path.join(__dirname, 'migrations'), +): Promise { + const names = readdirSync(directory).filter((name) => name.endsWith('.sql')).sort(); + if (!names.length) throw new Error('PostgreSQL migrations are missing'); + const versions = new Set(); + const migrations = names.map((name) => { + if (!/^\d{4}_[a-z0-9_]+\.sql$/.test(name) || versions.has(name.slice(0, 4))) { + throw new Error(`Invalid or duplicate PostgreSQL migration version: ${name}`); + } + versions.add(name.slice(0, 4)); + const sql = readFileSync(path.join(directory, name), 'utf8'); + if (!sql.trim()) throw new Error(`Empty PostgreSQL migration: ${name}`); + return { name, sql, checksum: createHash('sha256').update(sql).digest('hex') }; + }); + + const client = await pool.connect(); + let locked = false; + let discard = false; + const applied: string[] = []; + let failure: { error: unknown } | undefined; + try { + await client.query('SELECT pg_advisory_lock($1::integer, $2::integer)', LOCK); + locked = true; + await client.query(`CREATE TABLE IF NOT EXISTS schema_migrations ( + version TEXT PRIMARY KEY, + checksum TEXT NOT NULL, + applied_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP + )`); + const history = await client.query<{ version: string; checksum: string }>( + 'SELECT version, checksum FROM schema_migrations ORDER BY version', + ); + // Refuse edited, removed, or inserted-before-history files. A migration + // name alone cannot prove this image carries the SQL that actually ran. + history.rows.forEach((row, index) => { + const migration = migrations[index]; + if (migration?.name !== row.version || migration.checksum !== row.checksum) { + throw new Error(`PostgreSQL migration history mismatch: ${row.version}`); + } + }); + for (const migration of migrations.slice(history.rows.length)) { + await client.query('BEGIN'); + try { + await client.query(migration.sql); + await client.query('INSERT INTO schema_migrations (version, checksum) VALUES ($1, $2)', + [migration.name, migration.checksum]); + await client.query('COMMIT'); + } catch (error) { + try { + await client.query('ROLLBACK'); + } catch { + // Never pool a connection in an unknown transaction state. + discard = true; + } + throw error; + } + applied.push(migration.name); + } + } catch (error) { + failure = { error }; + } finally { + try { + if (locked) await client.query('SELECT pg_advisory_unlock($1::integer, $2::integer)', LOCK); + } catch { + // Destroy the session if unlock fails; it may still hold the lock. + discard = true; + failure ??= { error: new Error('Could not release PostgreSQL migration lock') }; + } finally { + client.release(discard); + } + } + if (failure) throw failure.error; + return applied; +} diff --git a/backend/config/schema.sql b/backend/config/schema.sql deleted file mode 100644 index c5a3a8df4c..0000000000 --- a/backend/config/schema.sql +++ /dev/null @@ -1,296 +0,0 @@ --- Boot DDL. Applied verbatim on every backend start, and by testUtils at tier 1. --- --- ADDING A COLUMN TO AN EXISTING TABLE NEEDS TWO DECLARATIONS, NOT ONE. --- `CREATE TABLE IF NOT EXISTS` is a NO-OP wherever the table already exists, --- so a column written only inside a CREATE TABLE below reaches fresh --- databases and NO existing instance. Put it in the CREATE TABLE (for new --- installs) AND in an `ALTER TABLE ADD COLUMN IF NOT EXISTS ...` (for --- every install that already ran). `payload` and `is_bot` are the pattern. --- --- The failure is total, not partial: any index on the missing column throws at --- boot, and any INSERT naming it fails, so the feature does not degrade — the --- table stops accepting writes. Costed live on 2026-08-22, when --- `thread_root_id` shipped with only the CREATE TABLE declaration and would --- have stopped chat on every existing instance. --- --- NEITHER TEST TIER CAN SEE THIS. Tier 0 reads this file as text and finds the --- column declared; tier 1 provisions a FRESH postgres per CI run and only ever --- exercises the CREATE path. A retrofit needs a test that starts from the OLD --- table shape — see `__tests__/unit/models/threadingSchemaRetrofit.test.js`. - --- Create pods table -CREATE TABLE IF NOT EXISTS pods ( - id VARCHAR(24) PRIMARY KEY, - name VARCHAR(100) NOT NULL, - description TEXT, - type VARCHAR(50) NOT NULL, - created_by VARCHAR(24) NOT NULL, - created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, - updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP -); - --- Create pod_members table for many-to-many relationship -CREATE TABLE IF NOT EXISTS pod_members ( - id SERIAL PRIMARY KEY, - pod_id VARCHAR(24) REFERENCES pods(id) ON DELETE CASCADE, - user_id VARCHAR(24) NOT NULL, - joined_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, - UNIQUE(pod_id, user_id) -); - --- Create messages table -CREATE TABLE IF NOT EXISTS messages ( - id SERIAL PRIMARY KEY, - pod_id VARCHAR(24) REFERENCES pods(id) ON DELETE CASCADE, - user_id VARCHAR(24) NOT NULL, - content TEXT NOT NULL, - message_type VARCHAR(20) DEFAULT 'text' NOT NULL, - -- Kernel-authored seat posts (install intros and stalled-connect nudges) - -- speak beside an agent without pretending to be that agent. NULL keeps - -- legacy human and agent rows unclassified until a trusted server path opts - -- into the marker. - author_kind VARCHAR(20), - reply_to_message_id INTEGER REFERENCES messages(id) ON DELETE SET NULL, - -- Threading (W-T, TASK-029). DELIBERATELY separate from reply_to_message_id, - -- which is an ADDRESSING edge: it feeds `isRouted` and the implicit-reply - -- chat.mention (agentMentionService :1025 / :1356), so a human replying to an - -- agent addresses that agent. If thread membership rode the same column, - -- joining a thread would ping the parent's author — the opposite of what - -- ambient scoping is for (fable's #1045 ruling). thread_root_id carries no - -- addressing. NULL means "not in a thread"; a root is an ordinary message - -- other messages point at, and a root's own thread_root_id stays NULL. - thread_root_id INTEGER REFERENCES messages(id) ON DELETE SET NULL, - -- Per-pod commit-order fence for agent send freshness. Existing rows stay 0. - freshness_revision BIGINT NOT NULL DEFAULT 0, - created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, - updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP -); - --- Create users table for storing MongoDB user references -CREATE TABLE IF NOT EXISTS users ( - _id VARCHAR(24) PRIMARY KEY, - username VARCHAR(100) NOT NULL, - profile_picture TEXT, - is_bot BOOLEAN DEFAULT false, - created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, - updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP -); - --- Migration: add is_bot column to existing tables (safe to run repeatedly) -ALTER TABLE users ADD COLUMN IF NOT EXISTS is_bot BOOLEAN DEFAULT false; - --- ADR-020 D3: structured message payload (approval cards and future --- server-defined message components). Self-applies at boot, same pattern --- as is_bot above. NULL for ordinary messages. -ALTER TABLE messages ADD COLUMN IF NOT EXISTS payload JSONB; - --- FUNNEL-5: the CREATE TABLE declaration above reaches fresh databases; this --- retrofit reaches every instance whose messages table already exists. -ALTER TABLE messages ADD COLUMN IF NOT EXISTS author_kind VARCHAR(20); - --- Threading (W-T, TASK-029). MUST be retrofitted, not only declared in the --- CREATE TABLE above — `CREATE TABLE IF NOT EXISTS` is a no-op on an existing --- table, so a column declared only there NEVER reaches an instance that --- already has a messages table. Verified on the live dev instance 2026-08-22: --- 7,001 rows, oldest 2026-07-07, and thread_root_id absent while `payload` --- (which has this retrofit) is present. --- --- Without this line the failure is not subtle: the index below throws at boot --- and every PGMessage.create fails, because the INSERT names a column that --- does not exist. Chat posting stops on every existing instance. --- --- The tier-1 test suite could not catch it — CI provisions a FRESH postgres:16 --- per run, where the CREATE TABLE path is the one exercised. Any new column on --- an existing table needs BOTH declarations, and a test that starts from the --- old shape (see threadingSchemaRetrofit.test.js). Caught by @sprint-review's --- caveat on #1106, not by anything I ran. -ALTER TABLE messages - ADD COLUMN IF NOT EXISTS thread_root_id INTEGER REFERENCES messages(id) ON DELETE SET NULL; - -ALTER TABLE messages ADD COLUMN IF NOT EXISTS freshness_revision BIGINT NOT NULL DEFAULT 0; - --- TASK-313 part 2. The Mongo id of a fallback message this row was reconciled --- from. Nullable: ordinary rows carry NULL, and several NULLs never collide. --- The UNIQUE index is NOT declared here: CONCURRENTLY cannot run inside this --- multi-statement script, which executes as one transaction. The reconciler --- builds it (services/pgMessageReconcileService.ensureLegacyIdIndex) before --- its first insert, so it never blocks writes on messages. -ALTER TABLE messages ADD COLUMN IF NOT EXISTS legacy_mongo_id TEXT; - -CREATE TABLE IF NOT EXISTS pod_message_revisions ( - pod_id VARCHAR(24) PRIMARY KEY REFERENCES pods(id) ON DELETE CASCADE, - revision BIGINT NOT NULL DEFAULT 0 -); - --- The counter update and message insert commit together. Concurrent inserts --- into one pod serialize on this row, so revision order is commit order even --- when an INSERT began before a reader and committed after it. -CREATE OR REPLACE FUNCTION assign_message_freshness_revision() -RETURNS TRIGGER AS $$ -BEGIN - INSERT INTO pod_message_revisions (pod_id, revision) - VALUES (NEW.pod_id, 1) - ON CONFLICT (pod_id) DO UPDATE - SET revision = pod_message_revisions.revision + 1 - RETURNING revision INTO NEW.freshness_revision; - RETURN NEW; -END; -$$ LANGUAGE plpgsql; - -DROP TRIGGER IF EXISTS messages_freshness_revision ON messages; -CREATE TRIGGER messages_freshness_revision - BEFORE INSERT ON messages - FOR EACH ROW EXECUTE FUNCTION assign_message_freshness_revision(); - -CREATE INDEX IF NOT EXISTS idx_messages_pod_freshness_revision - ON messages(pod_id, freshness_revision); - --- Migration records. There was no such table anywhere under backend/ before --- this (@sprint-review, 56859, checked with a positive control): schema.sql is --- idempotent boot DDL, which says what the shape IS and never when it changed. --- That is fine until a RULE needs to reference the change — the threading --- surface ruling (#1115) says pre-cutoff thread roots default to expanded, --- "cutoff = the migration timestamp, read from the migration record", and that --- record did not exist. --- --- Deliberately general rather than a threading-specific row. A one-off --- `threading_migration` table is the thing that has to be generalised the --- second time someone needs this, and the second time is when it gets done --- wrong. `details` is JSONB so a migration can record what only IT knew — for --- the threading backfill, the boundary timestamp it would otherwise discard. --- --- NOT a migration RUNNER. Nothing reads this to decide what to apply; boot DDL --- still owns that. It is a ledger, and the distinction matters: a reader must --- never infer "migration X has not run" from a missing row — it may have run --- on an instance that predates this table. -CREATE TABLE IF NOT EXISTS migration_records ( - name VARCHAR(255) PRIMARY KEY, - applied_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, - details JSONB -); - --- Retention execution ledger. This is deliberately NOT `migration_records`: --- migration records describe one-off changes, while this records every run of --- a destructive recurring job. A backend restart discards pod logs, so the --- deletion count and the reasons no deletion occurred must live with the data. --- A `running` row left behind is evidence of an interrupted run, not success. -CREATE TABLE IF NOT EXISTS pg_retention_runs ( - id BIGSERIAL PRIMARY KEY, - status VARCHAR(16) NOT NULL DEFAULT 'running' - CHECK (status IN ('running', 'completed', 'aborted', 'failed', 'skipped')), - configured_retention_days INTEGER, - final_retention_days INTEGER, - protected_pod_count INTEGER, - deleted_message_count BIGINT NOT NULL DEFAULT 0, - re_rooted_count BIGINT, - target_bytes BIGINT, - initial_size_bytes BIGINT, - final_size_bytes BIGINT, - detail TEXT, - started_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, - finished_at TIMESTAMP WITH TIME ZONE -); --- `CREATE TABLE IF NOT EXISTS` does not add fields to an already-created --- ledger, so keep this additive form for instances that booted an earlier DDL. -ALTER TABLE pg_retention_runs ADD COLUMN IF NOT EXISTS re_rooted_count BIGINT; -CREATE INDEX IF NOT EXISTS idx_pg_retention_runs_started_at ON pg_retention_runs(started_at DESC); - --- Sprint B5: message reactions. One row per (message, user, emoji) — a user --- can stack different emojis on the same message but each emoji is binary --- (toggle on/off). PG-only; Mongo fallback path doesn't get reactions in v1. -CREATE TABLE IF NOT EXISTS message_reactions ( - id SERIAL PRIMARY KEY, - message_id INTEGER NOT NULL REFERENCES messages(id) ON DELETE CASCADE, - user_id VARCHAR(24) NOT NULL, - emoji VARCHAR(32) NOT NULL, - created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, - UNIQUE (message_id, user_id, emoji) -); -CREATE INDEX IF NOT EXISTS idx_message_reactions_message_id ON message_reactions(message_id); - --- Room-grant broker audit and budget ledger. The budget counter is separate --- from Mongo's grant document so every reservation is a single Postgres --- conditional write; concurrent broker calls cannot read the same remaining --- slot and both spend it. `window_started_at` is retained for windowed caps; --- a NULL window on the grant is represented by a lifetime counter here. -CREATE TABLE IF NOT EXISTS tool_call_budgets ( - grant_id VARCHAR(255) PRIMARY KEY, - calls_used INTEGER NOT NULL DEFAULT 0, - window_started_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP -); - -CREATE TABLE IF NOT EXISTS tool_calls ( - call_id VARCHAR(255) PRIMARY KEY, - grant_id VARCHAR(255) NOT NULL, - pod_id VARCHAR(255), - installation_id VARCHAR(255), - agent_user_id VARCHAR(255) NOT NULL, - tool VARCHAR(255) NOT NULL, - args_digest CHAR(64) NOT NULL, - occurred_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, - outcome VARCHAR(32) NOT NULL CHECK (outcome IN ('ok', 'refused', 'pending_approval', 'failed')), - reason VARCHAR(255), - approval_id VARCHAR(255), - duration_ms INTEGER, - -- Whose credential ran (plan §8, §10 step 5): the Connection's `createdBy` at - -- call time. Copied, not joined — removal ends by deleting the Connection and - -- the trail outlives it. Nullable, and NULL means "no owner was recorded", - -- which is what a call refused before a connection resolved reports. - credential_owner_id VARCHAR(255), - vendor_status INTEGER, - vendor_error VARCHAR(200) -); -CREATE INDEX IF NOT EXISTS idx_tool_calls_grant_at ON tool_calls(grant_id, occurred_at DESC); --- Per-user, per-thread state (W-T, TASK-029). ONE record carrying both --- booleans, per ux-lead's ruling in docs/design/threading-surface-ruling.md --- ("One state record, two booleans"): persisted collapse and follow share the --- key (user, pod, thread root), so two tables would mean two writes for one --- gesture. The collapsed column is here at birth rather than in the render PR. --- --- NOT a widened User.followedThreads: that field is typed --- `postId: ObjectId ref 'Post', required` and cannot hold a Postgres --- messages.id. Widening it would silently drop chat rows from every consumer --- doing Post.find({_id: {$in: postIds}}) -- postController 457/513/548, --- activityService 614/712, ActivityFeedPage 584. --- --- A row means "this user has non-default state on this thread". Row PRESENCE --- carries no meaning on its own, which is why `following` is a column and not --- row-existence: `collapsed` writes create rows for users who are not --- following, and if presence meant following, expanding a thread would --- subscribe you to it. -CREATE TABLE IF NOT EXISTS thread_user_state ( - id SERIAL PRIMARY KEY, - thread_root_id INTEGER NOT NULL REFERENCES messages(id) ON DELETE CASCADE, - user_id VARCHAR(255) NOT NULL, - pod_id VARCHAR(255) NOT NULL, - -- TRI-STATE, deliberately nullable. NULL = "no explicit choice, defer to the - -- participation default" (posted in the thread or was @-mentioned => following). - -- TRUE = explicitly followed. FALSE = explicitly unfollowed, which for a - -- participant is a MUTE and must outrank the participation default. - -- A NOT NULL DEFAULT false here would be a live bug: a collapse-only row - -- would silently unfollow a participant who had never touched the toggle. - following BOOLEAN, - -- Defaults TRUE for everyone including followers -- following never implies - -- expanded. Only the expand/collapse gesture writes it. - collapsed BOOLEAN NOT NULL DEFAULT TRUE, - followed_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, - updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, - UNIQUE (thread_root_id, user_id) -); --- The wake path asks "who follows this thread" on every threaded reply, so the --- root is the hot key. The user index serves the per-pod render read. -CREATE INDEX IF NOT EXISTS idx_thread_user_state_root ON thread_user_state(thread_root_id); -CREATE INDEX IF NOT EXISTS idx_thread_user_state_user_pod ON thread_user_state(user_id, pod_id); - --- Create indexes for better performance -CREATE INDEX IF NOT EXISTS idx_messages_pod_id ON messages(pod_id); -CREATE INDEX IF NOT EXISTS idx_messages_created_at ON messages(created_at); --- Threading reads are "give me this thread" and "how many replies has it", both --- keyed on the root. reply_to_message_id has no index, which is why the root is --- stored rather than walked: deriving it per read is a seq-scan per level. -CREATE INDEX IF NOT EXISTS idx_messages_thread_root_id ON messages(thread_root_id); -CREATE INDEX IF NOT EXISTS idx_pod_members_pod_id ON pod_members(pod_id); -CREATE INDEX IF NOT EXISTS idx_pod_members_user_id ON pod_members(user_id); -CREATE INDEX IF NOT EXISTS idx_pods_created_by ON pods(created_by); -CREATE INDEX IF NOT EXISTS idx_pods_type ON pods(type); -CREATE INDEX IF NOT EXISTS idx_users_username ON users(username); diff --git a/backend/config/schema.sql b/backend/config/schema.sql new file mode 120000 index 0000000000..644e5d093f --- /dev/null +++ b/backend/config/schema.sql @@ -0,0 +1 @@ +migrations/0001_legacy_schema.sql \ No newline at end of file diff --git a/backend/package.json b/backend/package.json index 84e0f900cb..3e2c7ef3b0 100644 --- a/backend/package.json +++ b/backend/package.json @@ -4,7 +4,7 @@ "license": "Apache-2.0", "main": "server.js", "scripts": { - "build": "tsc -p tsconfig.build.json", + "build": "tsc -p tsconfig.build.json && node scripts/copy-pg-assets.js", "start": "node dist/server.js", "dev": "nodemon --exec 'ts-node --transpile-only' server.ts", "migrate-files": "node migrations/migrateFilesToDB.js", diff --git a/backend/repositories/agentPodReadCursorRepository.ts b/backend/repositories/agentPodReadCursorRepository.ts new file mode 100644 index 0000000000..90528c900c --- /dev/null +++ b/backend/repositories/agentPodReadCursorRepository.ts @@ -0,0 +1,32 @@ +import Cursor from '../models/AgentPodReadCursor'; + +export type ReadWatermark = { readAt: Date; readRevision?: number }; + +const cursorId = (agentUserId: unknown, podId: unknown): string => `${String(agentUserId)}:${String(podId)}`; + +/** Plain watermark or null. No Mongoose queries/documents escape this module. */ +export async function getReadWatermark(agentUserId: unknown, podId: unknown): Promise { + const row = await Cursor.findOne({ _id: cursorId(agentUserId, podId) }) + .select('readAt readRevision').lean(); + if (!row) return null; + return { readAt: row.readAt, ...(row.readRevision === undefined ? {} : { readRevision: row.readRevision }) }; +} + +/** Independent maxima preserve the existing timestamp/revision fence. */ +export async function advanceReadWatermark( + agentUserId: unknown, podId: unknown, watermark: ReadWatermark, +): Promise { + const filter = { _id: cursorId(agentUserId, podId) }; + const maximum = { + readAt: watermark.readAt, + ...(watermark.readRevision === undefined ? {} : { readRevision: watermark.readRevision }), + }; + try { + await Cursor.updateOne(filter, { $max: maximum, $setOnInsert: { agentUserId, podId } }, { upsert: true }); + } catch (error) { + if ((error as { code?: number }).code !== 11000) throw error; + // Two first reads can race on the deterministic _id. Retry without an + // upsert so the loser advances the winner without rolling it backwards. + await Cursor.updateOne(filter, { $max: maximum }); + } +} diff --git a/backend/scripts/copy-pg-assets.js b/backend/scripts/copy-pg-assets.js new file mode 100644 index 0000000000..e04ede854a --- /dev/null +++ b/backend/scripts/copy-pg-assets.js @@ -0,0 +1,10 @@ +const fs = require('fs'); +const path = require('path'); + +// tsc emits JavaScript only. Keep source and built boot paths equivalent. +const config = path.join(__dirname, '../config'); +const output = path.join(__dirname, '../dist/config'); +fs.mkdirSync(output, { recursive: true }); +fs.rmSync(path.join(output, 'migrations'), { recursive: true, force: true }); +fs.cpSync(path.join(config, 'migrations'), path.join(output, 'migrations'), { recursive: true }); +fs.copyFileSync(path.join(config, 'schema.sql'), path.join(output, 'schema.sql')); diff --git a/backend/services/agentSendFreshness.ts b/backend/services/agentSendFreshness.ts index 5be5a4fee9..541e3ee31b 100644 --- a/backend/services/agentSendFreshness.ts +++ b/backend/services/agentSendFreshness.ts @@ -1,6 +1,6 @@ // A pod-wide read watermark, recorded only after a successful current-tail read. // eslint-disable-next-line @typescript-eslint/no-require-imports -const Cursor = require('../models/AgentPodReadCursor'); +const Cursor = require('../repositories/agentPodReadCursorRepository'); // eslint-disable-next-line @typescript-eslint/no-require-imports const MongoMessage = require('../models/Message'); // eslint-disable-next-line @typescript-eslint/no-require-imports @@ -28,14 +28,12 @@ async function readStart(podId: unknown): Promise { if (!(hello.localTime instanceof Date)) throw new Error('MongoDB did not return a freshness clock'); return { at: hello.localTime }; } -const cursorId = (agentUserId: unknown, podId: unknown): string => `${String(agentUserId)}:${String(podId)}`; async function recordRead( agentUserId: unknown, podId: unknown, start: ReadStart, observedIds: string[], ): Promise { if (!Array.isArray(observedIds)) throw new Error('Read acknowledgement needs returned message IDs'); - const previous = await Cursor.findOne({ _id: cursorId(agentUserId, podId) }) - .select('readAt readRevision').lean(); + const previous = await Cursor.getReadWatermark(agentUserId, podId); // The first read also needs evidence: a truncated tail cannot acknowledge // older history just because no cursor existed yet. if (await omittedMessages( @@ -45,20 +43,10 @@ async function recordRead( } async function advanceCursor(agentUserId: unknown, podId: unknown, start: ReadStart): Promise { - // A slow earlier read must not roll a newer read's watermark backwards. - const filter = { _id: cursorId(agentUserId, podId) }; - const maximum = { + await Cursor.advanceReadWatermark(agentUserId, podId, { readAt: start.at, ...(start.revision === undefined ? {} : { readRevision: start.revision }), - }; - try { - await Cursor.updateOne(filter, { $max: maximum, $setOnInsert: { agentUserId, podId } }, { upsert: true }); - } catch (error) { - if ((error as { code?: number }).code !== 11000) throw error; - // Two first reads can race on the deterministic _id; the loser still - // advances the winner's cursor without ever replacing a later timestamp. - await Cursor.updateOne(filter, { $max: maximum }); - } + }); } async function omittedMessages( @@ -106,8 +94,7 @@ async function omittedMessages( async function recordCatchUp( agentUserId: unknown, podId: unknown, start: ReadStart, observedIds: string[], ): Promise<{ skippedCount: number }> { - const previous = await Cursor.findOne({ _id: cursorId(agentUserId, podId) }) - .select('readAt readRevision').lean(); + const previous = await Cursor.getReadWatermark(agentUserId, podId); const skippedCount = await omittedMessages( agentUserId, podId, previous || { readAt: new Date(0), readRevision: -1 }, start, observedIds, true, ); @@ -116,7 +103,7 @@ async function recordCatchUp( } async function unseenMessages(agentUserId: unknown, podId: unknown): Promise { - const cursor = await Cursor.findOne({ _id: cursorId(agentUserId, podId) }).select('readAt readRevision').lean(); + const cursor = await Cursor.getReadWatermark(agentUserId, podId); const after = cursor?.readAt || new Date(0); const afterRevision = cursor?.readRevision ?? -1; const ownId = String(agentUserId); diff --git a/backend/services/pgBootService.ts b/backend/services/pgBootService.ts index 85d7afc8da..9477a66b11 100644 --- a/backend/services/pgBootService.ts +++ b/backend/services/pgBootService.ts @@ -70,7 +70,7 @@ export interface PgBootDeps { mountRoutes: () => void; /** `connectPG` — resolves the pool, or null when the attempt failed. */ connect: () => Promise; - /** `initializePGDB` — applies schema.sql; false means the schema is not there. */ + /** `initializePGDB` — runs pending SQL migrations; false keeps routes unavailable. */ initialize: () => Promise; /** Started once, after the mount (retention + installation-cleanup crons). */ onMounted?: () => void; diff --git a/docs/database/POSTGRESQL_MIGRATION.md b/docs/database/POSTGRESQL_MIGRATION.md index d1b48a4059..f5fb181823 100644 --- a/docs/database/POSTGRESQL_MIGRATION.md +++ b/docs/database/POSTGRESQL_MIGRATION.md @@ -1,5 +1,8 @@ # PostgreSQL Message Storage Migration +For schema upgrades and the repository extraction template, see +[Versioned PostgreSQL changes](schema-migrations.md). + ## Overview Commonly has been updated to use **PostgreSQL as the default storage for all chat messages**, while maintaining MongoDB for user management and pod metadata. This document describes the current implementation and key changes. diff --git a/docs/database/schema-migrations.md b/docs/database/schema-migrations.md new file mode 100644 index 0000000000..bb6a82f3fb --- /dev/null +++ b/docs/database/schema-migrations.md @@ -0,0 +1,92 @@ +# Versioned PostgreSQL changes and repository extraction + +## Boot and upgrade contract + +`config/init-pg-db.ts` calls `runPgMigrations` before `pgBootService` mounts PG +routes or reports readiness. A failure returns false to the existing retry loop. +Each runner checks out **one connection**, takes the session advisory lock +`(1129270605, 504)`, reads history, and runs pending files in filename order. +Competing backend boots wait for that session to unlock, then reread history. +A process exit releases its session lock through PostgreSQL. + +Files live in `backend/config/migrations/NNNN_name.sql`. Four-digit versions +must be unique. Each file's SQL and its `schema_migrations` row commit in one +transaction. A failed file rolls back and can be retried. Earlier successful +files remain committed. The row stores filename, SHA-256 and application time. +Edited or removed applied files, and files inserted before applied history, +refuse boot. Add a new numbered file to correct an applied migration. + +`0001_legacy_schema.sql` freezes the former boot DDL, including its existing +retrofit ALTERs. On an existing instance it runs once without discarding rows; +on a fresh instance it creates the schema. `config/schema.sql` is a compatibility +symlink to this frozen baseline for older fixtures/scripts, **not the latest +schema**. New schema tests should call the runner. `migration_records` remains +the separate application/backfill ledger (including threading cutoff details); +its rows do not determine which SQL files the runner executes. + +## Adding a migration + +1. Pick the next unused `NNNN_name.sql`. Coordinate numbers across open lanes. +2. Write SQL that upgrades a populated previous schema. A column inside + `CREATE TABLE IF NOT EXISTS` alone never upgrades an existing table. Keep + rolling old/new application versions compatible with additive changes. +3. Do not include BEGIN/COMMIT/ROLLBACK, change the runner's lock, or use + commands that cannot run in a transaction (such as CREATE INDEX CONCURRENTLY). + Large online index builds and data backfills remain separate operator jobs. +4. Test both an empty database and the previous populated shape. Verify failure + rolls back DDL and its ledger row; rerun to prove completed SQL is skipped. +5. Run `npm --prefix backend run build`. The build copies SQL assets into + `dist/config/migrations`; the backend Docker build uses the same command. + +There is no automatic down migration. This image refuses a database whose +recorded migrations it does not carry, so rolling back an image after new SQL +ran requires a compatible image carrying that migration history. Do not delete +history to bypass this check. Take a backup before destructive migrations and +ship destructive changes only after old application versions are drained. + +## Repository template: AgentPodReadCursor + +Production callers import `repositories/agentPodReadCursorRepository`, never +`models/AgentPodReadCursor`. ESLint checks both imports and require calls for +this extracted model. Model tests may still seed and inspect real Mongo rows. +The model/schema is unchanged in this extraction; **Mongo is the only cursor +store** and remains authoritative. No dual writes, shadow reads or PG cursor +table are introduced here. + +- `getReadWatermark(agentUserId, podId)` returns a plain + `{readAt: Date, readRevision?: number}` or null. +- `advanceReadWatermark(agentUserId, podId, watermark)` resolves after writing. + It keeps the deterministic `agent:pod` key, independent `$max` timestamp and + revision updates, omitted-revision behavior, and duplicate-upsert loser retry. + Other database errors propagate to the freshness gate. + +For the next aggregate, expose its domain operations, keep Mongoose queries +inside its repository, and return plain values. Do not expose `find`, `save`, +`populate`, or a generic query builder. Extract without changing storage first, +then compare existing service behavior against a real database. A later storage +switch belongs inside that module and needs its own authority/partial-write +contract before dual writes begin. + +## Verification + +The pg-mem suite executes ordering, retrofit, history and repeat-boot checks. +Its advisory functions are stubs; it **does not establish lock or transaction +semantics**. The service suite uses an isolated schema on real PostgreSQL for +fresh/legacy baseline, DDL rollback/retry, and two competing boot sessions. +The completion-row failure case is essential: PostgreSQL can roll back a +multi-statement SQL query even without an explicit BEGIN. Failing only inside +that query would not prove the SQL and the separate ledger INSERT are atomic. +The Mongo repository suite round-trips rows and races concurrent first reads. + +```sh +cd backend +npx jest --runInBand --runTestsByPath \ + __tests__/unit/config/run-pg-migrations.test.js \ + __tests__/unit/repositories/agentPodReadCursorRepository.test.js +INTEGRATION_TEST=true PG_HOST=localhost PG_PORT=5432 \ + PG_DATABASE=commonly-test PG_USER=postgres PG_PASSWORD=postgres \ + npx jest --runInBand --runTestsByPath __tests__/service/pgMigrations.test.js +``` + +The real test role needs CREATE SCHEMA in a test database. Each case creates +and drops its own schema; it never drops the database or truncates other schemas. From 91c0c540272dc5102335c579451acce5d0b907a8 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sun, 25 Oct 2026 02:15:17 -0700 Subject: [PATCH 2/3] test(db): isolate migration waiters and cover missing history files Co-authored-by: otto Agent-Model: codex/gpt-6.1-sol/high --- .../__tests__/service/pgMigrations.test.js | 29 +++++++++++++++---- .../unit/config/run-pg-migrations.test.js | 20 +++++++++++++ 2 files changed, 43 insertions(+), 6 deletions(-) diff --git a/backend/__tests__/service/pgMigrations.test.js b/backend/__tests__/service/pgMigrations.test.js index c3132fb43e..736142d4eb 100644 --- a/backend/__tests__/service/pgMigrations.test.js +++ b/backend/__tests__/service/pgMigrations.test.js @@ -83,26 +83,43 @@ real('Postgres migrations: real DDL, rollback and competing boots', () => { expect(await runPgMigrations(pool, directory)).toEqual(['0002_second.sql']); }); - it('serializes two boots behind the same session lock and executes SQL once', async () => { + 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; try { await blocker.query('SELECT pg_advisory_lock($1::integer, $2::integer)', lock); - boots.push(runPgMigrations(pool, directory), runPgMigrations(pool, directory)); + const outsider = await pool.connect(); + 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(*)::int AS n FROM pg_locks + const result = await pool.query(`SELECT count(*)::int AS total, + count(*) FILTER (WHERE pid = ANY($3::integer[]))::int AS own 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); - return result.rows[0].n === 2; + AND database = (SELECT oid FROM pg_database WHERE datname = current_database())`, [...lock, bootPids]); + // 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].total >= 3; }); } 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); + await Promise.allSettled([...boots, 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 }]); diff --git a/backend/__tests__/unit/config/run-pg-migrations.test.js b/backend/__tests__/unit/config/run-pg-migrations.test.js index 32116f2976..8af84e6ee3 100644 --- a/backend/__tests__/unit/config/run-pg-migrations.test.js +++ b/backend/__tests__/unit/config/run-pg-migrations.test.js @@ -61,6 +61,26 @@ describe('Postgres migrations (pg-mem)', () => { 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'], From 58dd417de7a93224403bacf5dd12dc179e0b2e32 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sun, 25 Oct 2026 02:56:17 -0700 Subject: [PATCH 3/3] test(db): identify the outsider migration waiter by session Co-authored-by: otto Agent-Model: codex/gpt-6.1-sol/high --- backend/__tests__/service/pgMigrations.test.js | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/backend/__tests__/service/pgMigrations.test.js b/backend/__tests__/service/pgMigrations.test.js index 736142d4eb..42907dae1b 100644 --- a/backend/__tests__/service/pgMigrations.test.js +++ b/backend/__tests__/service/pgMigrations.test.js @@ -98,22 +98,24 @@ real('Postgres migrations: real DDL, rollback and competing boots', () => { }, }; 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(*)::int AS total, - count(*) FILTER (WHERE pid = ANY($3::integer[]))::int AS own FROM pg_locks + 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]); + 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].total >= 3; + return result.rows[0].own === 2 && result.rows[0].outsider === 1; }); } finally { await blocker.query('SELECT pg_advisory_unlock($1::integer, $2::integer)', lock); @@ -121,6 +123,8 @@ real('Postgres migrations: real DDL, rollback and competing boots', () => { // 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' }]);