stack/packages/bus/tests/store.test.mjs

import test from 'node:test';
import assert from 'node:assert/strict';
import { mkdtempSync, rmSync, statSync, symlinkSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { DatabaseSync } from 'node:sqlite';
const { Store } = await import('../src/store.mjs').catch((e) => {
  if (e.code === 'ERR_MODULE_NOT_FOUND') return {};
  throw e;
});
const scratch = (t) => {
  const root = mkdtempSync(join(tmpdir(), 'bus-store-'));
  t.after(() => rmSync(root, { recursive: true, force: true }));
  return root;
};
test('creates private WAL store and excludes a second writer until explicit close', (t) => {
  assert.equal(typeof Store, 'function', 'Store implementation exists');
  const root = scratch(t);
  const s = new Store(root);
  t.after(() => s.close());
  assert.equal(statSync(join(root, 'bus')).mode & 0o777, 0o700);
  assert.equal(statSync(join(root, 'bus/bus.sqlite')).mode & 0o777, 0o600);
  assert.equal(s.get('PRAGMA journal_mode').journal_mode, 'wal');
  assert.throws(() => new Store(root), /writer-locked/);
  s.close();
  const next = new Store(root);
  next.close();
});
test('rollback is atomic and schema metadata is checked against trusted DDL, not just itself', (t) => {
  assert.equal(typeof Store, 'function');
  const root = scratch(t);
  const s = new Store(root);
  assert.throws(
    () =>
      s.transaction(() => {
        s.run("INSERT INTO events(id,at,business,kind,body) VALUES('e','x','b','action.allowed','{}')");
        throw Error('rollback');
      }),
    /rollback/,
  );
  assert.equal(s.get('SELECT count(*) n FROM events').n, 0);
  s.close();
  const db = new DatabaseSync(join(root, 'bus/bus.sqlite'));
  db.exec('DROP TRIGGER events_no_update');
  db.close();
  assert.throws(() => new Store(root), /schema-mismatch/);
});
test('existing empty database and symlink runtime directory refuse, never initialize over damage', (t) => {
  assert.equal(typeof Store, 'function');
  const root = scratch(t);
  const s = new Store(root);
  s.close();
  const db = new DatabaseSync(join(root, 'bus/bus.sqlite'));
  db.exec('DROP TRIGGER meta_no_delete; DELETE FROM meta');
  db.close();
  assert.throws(() => new Store(root), /schema-mismatch/);
  const other = scratch(t);
  symlinkSync(join(root, 'bus'), join(other, 'bus'));
  assert.throws(() => new Store(other), /unsafe-path/);
});
test('crash during a transaction recovers no partial event after explicit fixture-only lock removal', async (t) => {
  const { spawn } = await import('node:child_process');
  const { once } = await import('node:events');
  const { unlinkSync } = await import('node:fs');
  const root = scratch(t);
  const child = spawn(
    process.execPath,
    [new URL('./fixtures/crash-writer.mjs', import.meta.url).pathname, root],
    { stdio: ['ignore', 'pipe', 'pipe'] },
  );
  t.after(() => {
    if (child.exitCode === null) child.kill('SIGKILL');
  });
  const ready = await Promise.race([
    once(child.stdout, 'data'),
    once(child, 'exit').then(() => {
      throw Error('crash fixture exited early');
    }),
  ]);
  assert.match(String(ready[0]), /inserted/);
  const exit = once(child, 'exit');
  child.kill('SIGKILL');
  await exit;
  assert.throws(() => new Store(root), /writer-locked/);
  unlinkSync(join(root, 'bus/writer.lock')); // Known dead test child, never production recovery.
  const recovered = new Store(root);
  try {
    assert.equal(recovered.get("SELECT count(*) n FROM events WHERE id='uncommitted'").n, 0);
  } finally {
    recovered.close();
  }
});
test('writer refuses mixed at/read_at forms atomically, even through trusted SQL helpers', (t) => {
  const root = scratch(t),
    store = new Store(root);
  t.after(() => store.close());
  const insert = (at) =>
    store.run(
      'INSERT INTO events(id,at,business,kind,body) VALUES(?,?,?,?,?)',
      'time-' + at,
      at,
      'demo',
      'action.allowed',
      '{}',
    );
  for (const at of [
    '2026-10-04T00:00:00Z',
    '2026-10-04T01:00:00.000+01:00',
    '2026-10-04T00:00:00.000000Z',
    '2026-02-30T00:00:00.000Z',
  ])
    assert.throws(() => insert(at), /invalid-timestamp/);
  assert.equal(store.get('SELECT count(*) n FROM events').n, 0);
  insert('2026-10-04T00:00:00.000Z');
  assert.throws(
    () =>
      store.run(
        "INSERT INTO events(seq,id,at,business,kind,body) VALUES(-1,'low','2026-10-04T00:00:00Z','demo','action.allowed','{}')",
      ),
    /invalid-timestamp/,
  );
  assert.throws(
    () =>
      store.transaction(() => {
        insert('2026-10-04T00:00:01.000Z');
        store.run(
          'INSERT INTO task_snapshots(at,business,task_ref,updated,digest,fields,source,via,read_at) VALUES(?,?,?,?,?,?,?,?,?)',
          '2026-10-04T00:00:02.000Z',
          'demo',
          'vikunja:1/1',
          'external',
          'a'.repeat(64),
          '{"bucket":1}',
          'poll',
          'board',
          '2026-10-04T00:00:01Z',
        );
      }),
    /invalid-timestamp/,
  );
  assert.equal(store.get('SELECT count(*) n FROM events').n, 1, 'whole transaction rolls back');
  assert.equal(store.get('SELECT count(*) n FROM task_snapshots').n, 0);
});
test('async transactions refuse before invoking their function', async (t) => {
  const root = scratch(t),
    store = new Store(root);
  t.after(() => store.close());
  let ran = false;
  assert.throws(
    () =>
      store.transaction(async () => {
        ran = true;
        await Promise.resolve();
      }),
    /async-transaction-refused/,
  );
  await Promise.resolve();
  assert.equal(ran, false);
});