import { DatabaseSync } from 'node:sqlite'; import { readFileSync, mkdirSync, lstatSync, openSync, closeSync, writeFileSync, fsyncSync, unlinkSync, } from 'node:fs'; import { join, isAbsolute } from 'node:path'; import { createHash } from 'node:crypto'; const ddl = readFileSync(new URL('../schema.sql', import.meta.url), 'utf8'); export const SCHEMA_SHA256 = '1c305d842619360c14c8388cd43cb9de81afbe46ff0662fd5de9851be9ff3ea0'; export const SCHEMA_VERSION = '3w'; // Ordered, named offline steps. New schema changes append a step and keep prior DDL pinned. const legacyDDL = readFileSync(new URL('./migrations/3b.sql', import.meta.url), 'utf8'); const LEGACY_SHA256 = '179ffe356d4ff19a49b5ebad39b6c6bfd7771deb8e1b7c55b5e746f039d69e65'; const migrations = [{ from: '3b', to: '3c', sql: readFileSync(new URL('./migrations/3b-3c.sql', import.meta.url), 'utf8'), initialize: (store) => store.run( 'INSERT INTO meta(key,value) VALUES (?,?)', 'comment_cursor_epoch', new Date().toISOString(), ), }, { from: '3c', to: '3d', sql: readFileSync(new URL('./migrations/3c-3d.sql', import.meta.url), 'utf8'), }, { from: '3d', to: '3e', sql: readFileSync(new URL('./migrations/3d-3e.sql', import.meta.url), 'utf8'), // Messages stay as written. Undelivered ones have no thread, so a thread run never receives them. // The store doesn't know which roles are thread instances, so it reports every role's count. report: (store) => store.all( "SELECT business, to_role, count(*) AS n FROM messages m WHERE NOT EXISTS(SELECT 1 FROM deliveries d WHERE d.message=m.id AND d.op IN ('delivered','read')) GROUP BY business, to_role ORDER BY business, to_role", ).map((r) => `kept ${r.n} undelivered message(s) for ${r.business}/${r.to_role} without a thread; if that role is a thread instance, its thread runs never receive them, so the human re-sends them with thread.send`), }, { from: '3e', to: '3f', sql: readFileSync(new URL('./migrations/3e-3f.sql', import.meta.url), 'utf8'), }, { // Row 141 (#1635): instance names, and the name a message was sent to. Messages keep a null to_name. from: '3f', to: '3g', sql: readFileSync(new URL('./migrations/3f-3g.sql', import.meta.url), 'utf8'), }, { from: '3g', to: '3h', sql: readFileSync(new URL('./migrations/3g-3h.sql', import.meta.url), 'utf8'), }, { // Row 126 (#1620): usage records carry the run's account; the table is rebuilt without its CHECK. from: '3h', to: '3i', sql: readFileSync(new URL('./migrations/3h-3i.sql', import.meta.url), 'utf8'), }, { // Row 234 (#1740): enrolled hosts, host grants and the grant event guard. Only new objects. from: '3i', to: '3j', sql: readFileSync(new URL('./migrations/3i-3j.sql', import.meta.url), 'utf8'), }, { // Row 236 (#1742, #436b section 7): the Gitea users the stack made and the Gitea grants. Only new tables. from: '3j', to: '3k', sql: readFileSync(new URL('./migrations/3j-3k.sql', import.meta.url), 'utf8'), }, { // Row 264a (#1769): sessions below a thread. Each thread gets a default session whose id is the // thread id, and thread_claims, message_threads and thread_opens are rebuilt to name it. from: '3k', to: '3l', sql: readFileSync(new URL('./migrations/3k-3l.sql', import.meta.url), 'utf8'), }, { // Row 206 (193b2, #1704): thread shares. Only a new table; no existing row changes. from: '3l', to: '3m', sql: readFileSync(new URL('./migrations/3l-3m.sql', import.meta.url), 'utf8'), }, { // Row 271 (#1776): clear context. The clear journal and the cancelled-message receipts. Only new tables. from: '3m', to: '3n', sql: readFileSync(new URL('./migrations/3m-3n.sql', import.meta.url), 'utf8'), }, { // #797 / #1839 Task 1: append-only request lifecycle evidence. Explicit offline migration only. from: '3n', to: '3o', sql: readFileSync(new URL('./migrations/3n-3o.sql', import.meta.url), 'utf8'), }, { // #797 / #1839 Task 2: the request journal's bootstrap.stale and bootstrap.completed kinds. request_events // is rebuilt with the wider kind list and keeps every row, seq and guard. from: '3o', to: '3p', sql: readFileSync(new URL('./migrations/3o-3p.sql', import.meta.url), 'utf8'), }, { from: '3p', to: '3q', sql: readFileSync(new URL('./migrations/3p-3q.sql', import.meta.url), 'utf8'), }, { // #797 / #1839 Task 3: the request journal's question.asked, question.answered and attach.selected kinds, for the // Console intake controls, added to the kind list 3p-3q left (which already holds the PRD kinds and mode.changed). // request_events is rebuilt with the union and keeps every row, seq and guard. from: '3q', to: '3r', sql: readFileSync(new URL('./migrations/3q-3r.sql', import.meta.url), 'utf8'), }, { // Task 6b: checked advancement after Task 3's complete fifteen-kind union. from: '3r', to: '3s', sql: readFileSync(new URL('./migrations/3r-3s.sql', import.meta.url), 'utf8'), }, { // #797 MVP3: the owner's reversible project archive. Only a new append-only table, its index // and its guards; no existing row changes and nothing is deleted. from: '3s', to: '3t', sql: readFileSync(new URL('./migrations/3s-3t.sql', import.meta.url), 'utf8'), }, { from: '3t', to: '3u', sql: readFileSync(new URL('./migrations/3t-3u.sql', import.meta.url), 'utf8'), }, { // U4a (#1873): the task store built into each organization, the default task-tracker module. // Only a new append-only table, its index and its guards; no existing row changes. from: '3u', to: '3v', sql: readFileSync(new URL('./migrations/3u-3v.sql', import.meta.url), 'utf8'), }, { // SP9 (#1873, Jason's ruling 2026-11-12): drops the table 3t-3u created, its index and its four // guards, with its rows. from: '3v', to: '3w', sql: readFileSync(new URL('./migrations/3v-3w.sql', import.meta.url), 'utf8'), }]; const timedTables = [ 'events', 'role_claims', 'decisions', 'decision_events', 'messages', 'deliveries', 'task_snapshots', 'projects', 'threads', 'sessions', 'session_defaults', 'thread_claims', 'thread_opens', 'thread_shares', 'instance_names', 'usage_records', 'session_ops', 'message_cancels', 'request_events', 'request_effects', 'request_snapshots', 'request_snapshot_events', 'request_project_events', 'project_archives', 'store_tasks', ]; const hex = (value) => createHash('sha256').update(value).digest('hex'); if (hex(ddl) !== SCHEMA_SHA256) throw Error('schema-source-mismatch'); export const schemaDigest = (db) => hex( db .prepare('SELECT type,name,sql FROM sqlite_master WHERE sql IS NOT NULL ORDER BY type,name') .all() .map((r) => `${r.type}|${r.name}|${r.sql}`) .join('\n'), ); const reference = new DatabaseSync(':memory:'); reference.exec(ddl); const EXPECTED_DIGEST = schemaDigest(reference); reference.close(); const legacy = new DatabaseSync(':memory:'); if (hex(legacyDDL) !== LEGACY_SHA256) throw Error('legacy-schema-source-mismatch'); legacy.exec(legacyDDL); for (let i = 0; i < migrations.length; i++) { const step = migrations[i]; if (i && migrations[i - 1].to !== step.from) throw Error('migration-chain-mismatch'); step.fromDigest = schemaDigest(legacy); legacy.exec(step.sql); step.toDigest = schemaDigest(legacy); } if (migrations.at(-1).to !== SCHEMA_VERSION || schemaDigest(legacy) !== EXPECTED_DIGEST) throw Error('migration-target-mismatch'); legacy.close(); function exists(path) { try { return lstatSync(path); } catch (e) { if (e.code === 'ENOENT') return null; throw e; } } function safe(path, directory = false) { const s = lstatSync(path); if ( s.isSymbolicLink() || !(directory ? s.isDirectory() : s.isFile()) || s.uid !== process.getuid() || s.mode & 0o077 ) throw Error('unsafe-path'); return s; } // Internal trusted SQL surface. Only the broker owns a Store; never expose SQL over the socket. export class Store { #db; #lock; #lockStat; #closed = false; #inTransaction = false; #report = []; constructor(dataRoot, { migrateSchema = false } = {}) { if (typeof dataRoot !== 'string' || !isAbsolute(dataRoot)) throw Error('unsafe-path'); const rootStat = lstatSync(dataRoot); if (!rootStat.isDirectory() || rootStat.isSymbolicLink() || rootStat.uid !== process.getuid()) throw Error('unsafe-path'); this.directory = join(dataRoot, 'bus'); if (!exists(this.directory)) mkdirSync(this.directory, { mode: 0o700 }); safe(this.directory, true); this.#lock = join(this.directory, 'writer.lock'); let fd; try { fd = openSync(this.#lock, 'wx', 0o600); } catch (e) { if (e.code === 'EEXIST') throw Error('writer-locked'); throw e; } this.#lockStat = lstatSync(this.#lock); try { writeFileSync(fd, JSON.stringify({ pid: process.pid, at: new Date().toISOString() })); fsyncSync(fd); closeSync(fd); fd = undefined; const dirfd = openSync(this.directory, 'r'); try { fsyncSync(dirfd); } finally { closeSync(dirfd); } this.path = join(this.directory, 'bus.sqlite'); const fresh = !exists(this.path); if (fresh && migrateSchema) throw Error('migration-source-missing'); if (fresh) closeSync(openSync(this.path, 'wx', 0o600)); else safe(this.path); for (const suffix of ['-wal', '-shm']) if (exists(this.path + suffix)) safe(this.path + suffix); this.#db = new DatabaseSync(this.path, { timeout: 5000, defensive: true }); this.#db.exec('PRAGMA foreign_keys=ON; PRAGMA synchronous=FULL'); if (fresh) { // journal_mode must be set outside a transaction. this.#db.exec('PRAGMA journal_mode=WAL'); this.transaction(() => { this.#db.exec(ddl.replace('PRAGMA journal_mode = WAL;', '')); this.run('INSERT INTO meta(key,value) VALUES (?,?)', 'schema_version', SCHEMA_VERSION); this.run('INSERT INTO meta(key,value) VALUES (?,?)', 'schema_digest', EXPECTED_DIGEST); this.run('INSERT INTO meta(key,value) VALUES (?,?)', 'comment_cursor_epoch', new Date().toISOString()); }); } if (!fresh && migrateSchema) this.#migrateSchema(); if ( schemaDigest(this.#db) !== EXPECTED_DIGEST || this.get("SELECT value FROM meta WHERE key='schema_version'")?.value !== SCHEMA_VERSION || this.get("SELECT value FROM meta WHERE key='schema_digest'")?.value !== EXPECTED_DIGEST || this.get('PRAGMA journal_mode').journal_mode !== 'wal' ) throw Error('schema-mismatch'); if (this.get('PRAGMA quick_check').quick_check !== 'ok' || this.all('PRAGMA foreign_key_check').length) throw Error('integrity-failed'); this.#verifyTimes(); const epoch = this.get("SELECT value FROM meta WHERE key='comment_cursor_epoch'")?.value; if (!epoch || !Number.isFinite(Date.parse(epoch)) || new Date(epoch).toISOString() !== epoch) throw Error('invalid-comment-epoch'); } catch (e) { if (fd !== undefined) closeSync(fd); this.close(); throw e; } } // Explicit offline operation only. Normal startup refuses old versions without changing them. #migrateSchema() { let version = this.get("SELECT value FROM meta WHERE key='schema_version'")?.value; while (version !== SCHEMA_VERSION) { const step = migrations.find((m) => m.from === version); if ( !step || schemaDigest(this.#db) !== step.fromDigest || this.get("SELECT value FROM meta WHERE key='schema_digest'")?.value !== step.fromDigest || this.get('PRAGMA journal_mode').journal_mode !== 'wal' ) throw Error('schema-mismatch'); if (this.get('PRAGMA quick_check').quick_check !== 'ok' || this.all('PRAGMA foreign_key_check').length) throw Error('integrity-failed'); this.#verifyTimes(); this.transaction(() => { this.#db.exec(step.sql); // Only version metadata changes. The guard is restored before this step commits. this.#db.exec('DROP TRIGGER meta_no_update'); this.run("UPDATE meta SET value=? WHERE key='schema_version'", step.to); this.run("UPDATE meta SET value=? WHERE key='schema_digest'", step.toDigest); step.initialize?.(this); this.#report.push(...(step.report?.(this) ?? [])); this.#db.exec("CREATE TRIGGER meta_no_update BEFORE UPDATE ON meta BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END"); if (schemaDigest(this.#db) !== step.toDigest) throw Error('schema-mismatch'); }); version = step.to; } } // Lines from the last migrateSchema, for migrate-schema.mjs to print. get migrationReport() { return [...this.#report]; } get(sql, ...args) { return this.#db.prepare(sql).get(...args); } all(sql, ...args) { return this.#db.prepare(sql).all(...args); } run(sql, ...args) { if (!this.#inTransaction) return this.transaction(() => this.run(sql, ...args)); return this.#db.prepare(sql).run(...args); } #verifyTimes() { for (const table of timedTables) { // A migration checks older schemas that lack later tables; the digest checks decide which must exist. if (!this.get("SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", table)) continue; const badAt = "length(at)<>24 OR at NOT GLOB '[0-9][0-9][0-9][0-9]-*' OR strftime('%Y-%m-%dT%H:%M:%fZ',at) IS NOT at"; const badRead = table === 'task_snapshots' ? " OR (read_at IS NOT NULL AND (length(read_at)<>24 OR read_at NOT GLOB '[0-9][0-9][0-9][0-9]-*' OR strftime('%Y-%m-%dT%H:%M:%fZ',read_at) IS NOT read_at))" : ''; if (this.get(`SELECT 1 FROM ${table} WHERE ${badAt}${badRead} LIMIT 1`)) throw Error('invalid-timestamp'); } } transaction(fn) { if (typeof fn !== 'function' || fn.constructor.name === 'AsyncFunction') throw Error('async-transaction-refused'); if (this.#inTransaction) throw Error('nested-transaction'); this.#db.exec('BEGIN IMMEDIATE'); this.#inTransaction = true; try { const result = fn(); if (result?.then) throw Error('async-transaction-refused'); this.#verifyTimes(); this.#db.exec('COMMIT'); return result; } catch (e) { this.#db.exec('ROLLBACK'); throw e; } finally { this.#inTransaction = false; } } close() { if (this.#closed) return; this.#closed = true; this.#db?.close(); const current = exists(this.#lock); if (current?.ino === this.#lockStat?.ino && current?.dev === this.#lockStat?.dev) unlinkSync(this.#lock); } }