stack/packages/bus/tests/account-migration.test.mjs

import test from 'node:test';
import assert from 'node:assert/strict';
import { chmodSync, copyFileSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { pathToFileURL } from 'node:url';
import { createHash } from 'node:crypto';
import { DatabaseSync } from 'node:sqlite';
import { Store, schemaDigest } from '../src/store.mjs';
// Row 126 (#1620, REQ-LIMIT-4): the 3h→3i step rebuilds usage_records so its account column takes a
// name. It runs on a 3h store built here from the pinned 3h schema (81f26551b's schema.sql) and
// fixture rows. No live store is read or copied. Mos's go (mos-0835) asks for one transaction, a
// schema that differs from 3h only in the account CHECK, and a 3h engine that refuses a 3i store.
const AT = '2026-10-18T00:00:00.000Z';
const SEQUENCE = 50;
const ddl3h = readFileSync(new URL('./fixtures/schema-3h.sql', import.meta.url), 'utf8');
const here = (p) => new URL(p, import.meta.url);
// Row 264a (#1769): what the 3k→3l step rebuilds or adds. sessions.test.mjs tests those objects.
const SESSIONS_3L = ['sessions', 'session_defaults', 'thread_claims', 'message_threads', 'thread_opens'];
// Row 206 (#1704): and the 3l→3m step's thread_shares. thread-shares.test.mjs tests it.
// Row 271 (#1776): and the 3m→3n step's session_ops and message_cancels. session-clear.test.mjs tests them.
const later3l = (r) => [...SESSIONS_3L, 'thread_shares', 'session_ops', 'message_cancels', 'project_archives', 'project_system_types', 'store_tasks', 'request_events', 'request_effects', 'request_snapshots', 'request_snapshot_events', 'request_project_events'].includes(r.tbl_name) || ['delivery_same_thread', 'delivery_same_session', 'threads_id_not_session'].includes(r.name);
function source3h(t) {
  const root = mkdtempSync(join(tmpdir(), 'account-migrate-'));
  t.after(() => rmSync(root, { recursive: true, force: true }));
  mkdirSync(join(root, 'bus'), { mode: 0o700 });
  const path = join(root, 'bus/bus.sqlite'), db = new DatabaseSync(path);
  assert.equal(createHash('sha256').update(ddl3h).digest('hex'), 'bb2ba58175a111bb99681778911507846bef714a6d573f8a8af67580a3636950');
  db.exec(ddl3h);
  const run = (sql, ...args) => db.prepare(sql).run(...args);
  run('INSERT INTO meta VALUES (?,?)', 'schema_version', '3h');
  run('INSERT INTO meta VALUES (?,?)', 'schema_digest', schemaDigest(db));
  run('INSERT INTO meta VALUES (?,?)', 'comment_cursor_epoch', AT);
  run('INSERT INTO events(id,at,business,kind,body) VALUES(?,?,?,?,?)', 'event', AT, 'demo', 'action.allowed', '{}');
  run('INSERT INTO instance_names(at,business,instance,name,by) VALUES(?,?,?,?,?)', AT, 'demo', 'coder', 'Builder', 'jason');
  const usage = 'INSERT INTO usage_records(at,business,run,instance,provider,model,account,turn,source,phase,status,code,cursor,input,output,cache_write,cache_read) VALUES(?,?,?,?,?,?,NULL,?,?,?,?,?,?,?,?,?,?)';
  run(usage, AT, 'demo', 'run1', 'coder', 'anthropic', 'opus', 't1', 'claude', 'begin', 'known', null, '{"input":1}', 0, 0, 0, 0);
  run(usage, AT, 'demo', 'run1', 'coder', 'anthropic', 'opus', 't1', 'claude', 'end', 'known', null, null, 4, 5, 6, 7);
  run(usage, AT, 'demo', 'run2', 'coder', null, null, 't1', 'claude', 'end', 'unknown', 'no-source', null, null, null, null, null);
  // The sequence runs past the last seq, so the step must carry it, not rebuild it from the rows.
  run("UPDATE sqlite_sequence SET seq=? WHERE name='usage_records'", SEQUENCE);
  db.close(); chmodSync(path, 0o600);
  return { root, path };
}
function snapshot(path, columns) {
  const db = new DatabaseSync(path, { readOnly: true });
  try {
    columns ??= Object.fromEntries(db.prepare("SELECT name FROM sqlite_master WHERE type='table'").all()
      .map(({ name }) => [name, db.prepare(`PRAGMA table_info(${name})`).all().map(c => c.name)]));
    const rows = Object.fromEntries(Object.entries(columns).map(([table, keys]) => [table,
      db.prepare(`SELECT ${keys.join(',')} FROM ${table} ORDER BY ${table === 'sqlite_sequence' ? 'name' : 'rowid'}`).all()
        .filter(r => table !== 'meta' || !['schema_version', 'schema_digest'].includes(r.key))]));
    return { columns, rows };
  } finally { db.close(); }
}
// Row 264a: the 3k→3l step fills sessions and session_defaults with INSERT ... SELECT, which gives
// each a sqlite_sequence row even when it copies none (3k-3l.sql:50-52). An after-migration compare
// leaves out the rows SESSIONS_3L names and checks they are exactly those two at seq 0, so any other
// new row still fails.
function migrated3l(rows) {
  const left = rows.sqlite_sequence.filter(r => SESSIONS_3L.includes(r.name));
  assert.deepEqual(left.map(r => ({ ...r })), [{ name: 'session_defaults', seq: 0 }, { name: 'sessions', seq: 0 }]);
  return { ...rows, sqlite_sequence: rows.sqlite_sequence.filter(r => !SESSIONS_3L.includes(r.name)) };
}
function master(path) {
  const db = new DatabaseSync(path, { readOnly: true });
  try {
    return db.prepare('SELECT type,name,tbl_name,sql FROM sqlite_master ORDER BY type,name').all().map(r => ({ ...r }));
  } finally { db.close(); }
}
const version = (path) => {
  const db = new DatabaseSync(path, { readOnly: true });
  try { return db.prepare("SELECT value FROM meta WHERE key='schema_version'").get().value; } finally { db.close(); }
};
// A copy of the bus module whose store.mjs, schema.sql and steps the caller edits.
async function engine(root, name, { store, schema, step = (s) => s }) {
  const moduleRoot = join(root, name);
  mkdirSync(join(moduleRoot, 'src/migrations'), { recursive: true });
  const source = readFileSync(here('../src/store.mjs'), 'utf8'), edited = store ? store(source) : source;
  if (store) assert.notEqual(edited, source);
  writeFileSync(join(moduleRoot, 'src/store.mjs'), edited);
  writeFileSync(join(moduleRoot, 'schema.sql'), schema ?? readFileSync(here('../schema.sql'), 'utf8'));
  for (const file of ['3b.sql', '3b-3c.sql', '3c-3d.sql', '3d-3e.sql', '3e-3f.sql', '3f-3g.sql', '3g-3h.sql'])
    copyFileSync(here('../src/migrations/' + file), join(moduleRoot, 'src/migrations', file));
  writeFileSync(join(moduleRoot, 'src/migrations/3h-3i.sql'), step(readFileSync(here('../src/migrations/3h-3i.sql'), 'utf8')));
  copyFileSync(here('../src/migrations/3i-3j.sql'), join(moduleRoot, 'src/migrations/3i-3j.sql'));
  copyFileSync(here('../src/migrations/3j-3k.sql'), join(moduleRoot, 'src/migrations/3j-3k.sql'));
  copyFileSync(here('../src/migrations/3k-3l.sql'), join(moduleRoot, 'src/migrations/3k-3l.sql'));
  copyFileSync(here('../src/migrations/3l-3m.sql'), join(moduleRoot, 'src/migrations/3l-3m.sql'));
  copyFileSync(here('../src/migrations/3m-3n.sql'), join(moduleRoot, 'src/migrations/3m-3n.sql'));
  copyFileSync(here('../src/migrations/3n-3o.sql'), join(moduleRoot, 'src/migrations/3n-3o.sql'));
  copyFileSync(here('../src/migrations/3o-3p.sql'), join(moduleRoot, 'src/migrations/3o-3p.sql'));
  copyFileSync(here('../src/migrations/3p-3q.sql'), join(moduleRoot, 'src/migrations/3p-3q.sql'));
  copyFileSync(here('../src/migrations/3q-3r.sql'), join(moduleRoot, 'src/migrations/3q-3r.sql'));
  copyFileSync(here('../src/migrations/3r-3s.sql'), join(moduleRoot, 'src/migrations/3r-3s.sql'));
  copyFileSync(here('../src/migrations/3s-3t.sql'), join(moduleRoot, 'src/migrations/3s-3t.sql'));
  copyFileSync(here('../src/migrations/3t-3u.sql'), join(moduleRoot, 'src/migrations/3t-3u.sql'));
  copyFileSync(here('../src/migrations/3u-3v.sql'), join(moduleRoot, 'src/migrations/3u-3v.sql'));
  return (await import(pathToFileURL(join(moduleRoot, 'src/store.mjs')))).Store;
}

test('REQ-LIMIT-4 3h copy migrates to 3i: every row keeps its seq, the sequence carries over, and only the account CHECK changes', t => {
  const { root, path } = source3h(t), before = snapshot(path), was = master(path);
  assert.throws(() => new Store(root), /schema-mismatch/);
  new Store(root, { migrateSchema: true }).close();
  assert.equal(version(path), '3v');
  // Every table's rows as they were, usage_records with its seq values and sqlite_sequence with 50.
  assert.deepEqual(migrated3l(snapshot(path, before.columns).rows), before.rows);
  assert.deepEqual(before.rows.usage_records.map(r => r.seq), [1, 2, 3]);
  assert.deepEqual({ ...before.rows.sqlite_sequence.find(r => r.name === 'usage_records') }, { name: 'usage_records', seq: SEQUENCE });
  // sqlite_master as at 3h, the account line aside: same tables, indexes and triggers, same text.
  const old = '  account TEXT CHECK (account IS NULL),\n';
  assert.equal(was.filter(r => r.sql?.includes(old)).map(r => r.name).join(), 'usage_records');
  // Row 234 (#1740) adds 3i→3j's own objects after this step. Its test is grant-migration.test.mjs.
  // Row 236 (#1742) adds 3j→3k's; its test is gitea-migration.test.mjs.
  const later = ['hosts', 'host_grants', 'events_grant_actor', 'sqlite_autoindex_hosts_1', 'sqlite_autoindex_host_grants_1',
    'gitea_users', 'gitea_grants', 'sqlite_autoindex_gitea_users_1', 'sqlite_autoindex_gitea_grants_1'];
  assert.deepEqual(master(path).filter(r => !later.includes(r.name) && !later3l(r)), was.filter(r => !later3l(r)).map(r => r.name === 'usage_records' ? { ...r, sql: r.sql.replace(old, '  account TEXT,\n') } : r));
  assert.equal(was.filter(r => r.type === 'trigger' && r.tbl_name === 'usage_records').length, 3);
  const store = new Store(root);
  try {
    assert.equal(store.get('PRAGMA quick_check').quick_check, 'ok');
    assert.deepEqual(store.all('PRAGMA foreign_key_check'), []);
    // A new record takes an account, and its seq follows the carried sequence.
    store.run(`INSERT INTO usage_records(at,business,run,instance,provider,model,account,turn,source,phase,status,cursor,input,output,cache_write,cache_read)
      VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, AT, 'demo', 'run3', 'coder', 'anthropic', 'opus', 'jason-max', 't1', 'claude', 'begin', 'known', '{}', 0, 0, 0, 0);
    assert.deepEqual({ ...store.get("SELECT seq,account FROM usage_records WHERE run='run3'") }, { seq: SEQUENCE + 1, account: 'jason-max' });
    // Still append-only on 3i.
    assert.throws(() => store.run("UPDATE usage_records SET account='other' WHERE seq=1"), /usage_records is append-only/);
    assert.throws(() => store.run('DELETE FROM usage_records WHERE seq=1'), /usage_records is append-only/);
    assert.throws(() => store.run(`INSERT OR REPLACE INTO usage_records(at,business,run,instance,turn,source,phase,status)
      VALUES(?,?,?,?,?,?,?,?)`, AT, 'demo', 'run2', 'coder', 't1', 'claude', 'end', 'unknown'), /usage_records is append-only/);
    assert.throws(() => store.run(`INSERT INTO usage_records(seq,at,business,run,instance,turn,source,phase,status)
      VALUES(?,?,?,?,?,?,?,?,?)`, 1, AT, 'demo', 'run9', 'coder', 't1', 'claude', 'end', 'unknown'), /usage_records is append-only/);
    assert.equal(store.get('SELECT count(*) AS n FROM usage_records').n, 4);
    assert.equal(store.get("SELECT account FROM usage_records WHERE seq=1").account, null);
  } finally { store.close(); }
});

test('REQ-LIMIT-4 a 3h→3i step that fails partway leaves the store at 3h with every row, the sequence and the schema intact', async t => {
  const { root, path } = source3h(t), before = snapshot(path), was = master(path);
  // The real step with one statement added after the copy: it copies a row again, which only
  // fails when there are rows, so the empty replay at import passes and the store's run fails
  // after the triggers are dropped, the table renamed, and the new one made and filled.
  const extra = 'INSERT INTO usage_records SELECT * FROM usage_records_3h LIMIT 1;\n';
  const Failing = await engine(root, 'engine-failing', { step: (sql) => {
    assert.equal(sql.split('DROP TABLE usage_records_3h;\n').length, 2);
    return sql.replace('DROP TABLE usage_records_3h;\n', extra + 'DROP TABLE usage_records_3h;\n');
  } });
  assert.throws(() => new Failing(root, { migrateSchema: true }), /UNIQUE constraint failed: usage_records\.seq/);
  assert.equal(version(path), '3h');
  assert.deepEqual(master(path), was);
  assert.deepEqual(snapshot(path, before.columns).rows, before.rows);
  // Still a whole 3h store: the real step migrates it.
  new Store(root, { migrateSchema: true }).close();
  assert.equal(version(path), '3v');
  assert.deepEqual(migrated3l(snapshot(path, before.columns).rows), before.rows);
});

test('REQ-LIMIT-4 the 3h engine refuses a migrated store, migrating or not, and leaves it as it was', async t => {
  const { root, path } = source3h(t);
  new Store(root, { migrateSchema: true }).close();
  const after = snapshot(path), was = master(path);
  const Store3h = await engine(root, 'engine-3h', {
    schema: ddl3h,
    store: (source) => source
      .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3h).digest('hex')}';`)
      .replace("export const SCHEMA_VERSION = '3v';", "export const SCHEMA_VERSION = '3h';")
      .replace(/\}, \{\n {2}\/\/ Row 126 \(#1620\)[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables'),
  });
  const pinned = readFileSync(join(root, 'engine-3h/src/store.mjs'), 'utf8');
  assert.ok(pinned.includes("export const SCHEMA_VERSION = '3h';") && !pinned.includes('3h-3i'), 'the 3h engine has no 3h→3i step');
  assert.throws(() => new Store3h(root), /schema-mismatch/);
  assert.throws(() => new Store3h(root, { migrateSchema: true }), /schema-mismatch/);
  assert.equal(version(path), '3v');
  assert.deepEqual(master(path), was);
  assert.deepEqual(snapshot(path, after.columns).rows, after.rows);
});