stack/packages/bus/src/store.mjs

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);
  }
}