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