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 = '179ffe356d4ff19a49b5ebad39b6c6bfd7771deb8e1b7c55b5e746f039d69e65'; const timedTables = [ 'events', 'role_claims', 'decisions', 'decision_events', 'messages', 'deliveries', 'task_snapshots', ]; 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(); 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; constructor(dataRoot) { 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) 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', '3b'); this.run('INSERT INTO meta(key,value) VALUES (?,?)', 'schema_digest', EXPECTED_DIGEST); }); } if ( schemaDigest(this.#db) !== EXPECTED_DIGEST || this.get("SELECT value FROM meta WHERE key='schema_version'")?.value !== '3b' || 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(); } catch (e) { if (fd !== undefined) closeSync(fd); this.close(); throw e; } } 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) { 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); } }