zz_

app/engine/store.ts

zzmq.dev@gmail.com/zz / app / engine / store.ts @ 4a679371 Download
Ladder Summary
  • file This module implements a SQLite‑backed persistence layer (one `zz.db` per repository) for **Lane**, **AgentRoles**, and decision records, lazily opening and caching a `DatabaseSync` connection, creating the necessary tables, and migrating any existing flat‑JSON files on first use. It exports the CRUD helpers `readLane`, `writeLane`, `listLanes`, `withLane` (an atomic read‑modify‑write transaction for a single lane), `readRoles`, `writeRoles`, and `writeDecision`. Internal helpers (`zzRoot`, `getDb`, `migrateFlatFiles`) support path handling, DB initialization, and migration.
 // .zz/ persistence: a SQLite db (one per repo, `.zz/zz.db`) via Node's
 // built-in `node:sqlite` -- no new dependency, no native-module build
 // risk. Replaces the earlier flat-JSON-file version specifically to fix
 // a real race: two `zz work` processes concurrently acting on the same
 // Lane (a plain read-mutate-write across two file I/O calls has a
 // window where one process's write can clobber the other's). SQLite's
 // own locking (BEGIN IMMEDIATE below) makes the read-modify-write in
 // `withLane` atomic across processes, not just within one. This also
 // doubles as a closer rehearsal for the eventual Durable Object (which
 // is SQLite-backed too, per site/content/remote-backend/index.md) than
 // flat files ever were.
 //
 // Migration: a repo's existing `.zz/lanes/*.json` (and `.zz/roles/*.json`)
 // files, if present, are imported into the db once, the first time it's
 // created -- the old files are left in place afterward (not read again,
 // not deleted) rather than risking data loss on an already-committed
 // demo repo's state.
 import { DatabaseSync } from "node:sqlite";
 import { existsSync, mkdirSync, readFileSync, readdirSync } from "node:fs";
 import { join } from "node:path";
 import type { AgentRoles, Lane } from "./types.js";
 
 function zzRoot(repoStateDir: string): string {
  return join(repoStateDir, ".zz");
 }
 
 const dbCache = new Map<string, DatabaseSync>();
 
 function migrateFlatFiles(db: DatabaseSync, root: string): void {
  const lanesDir = join(root, "lanes");
  if (existsSync(lanesDir)) {
  for (const f of readdirSync(lanesDir)) {
  if (!f.endsWith(".json")) continue;
  const data = readFileSync(join(lanesDir, f), "utf8");
  const lane = JSON.parse(data) as Lane;
  db.prepare("INSERT OR REPLACE INTO lanes (id, data) VALUES (?, ?)").run(lane.id, data);
  }
  }
  const rolesDir = join(root, "roles");
  if (existsSync(rolesDir)) {
  for (const f of readdirSync(rolesDir)) {
  if (!f.endsWith(".json")) continue;
  const data = readFileSync(join(rolesDir, f), "utf8");
  const roles = JSON.parse(data) as AgentRoles;
  db.prepare("INSERT OR REPLACE INTO roles (agent, data) VALUES (?, ?)").run(roles.agent, data);
  }
  }
 }
 
 function getDb(repoStateDir: string): DatabaseSync {
  const root = zzRoot(repoStateDir);
  const dbPath = join(root, "zz.db");
  const cached = dbCache.get(dbPath);
  if (cached) return cached;
 
  mkdirSync(root, { recursive: true });
  const isNew = !existsSync(dbPath);
  const db = new DatabaseSync(dbPath);
  db.exec("PRAGMA busy_timeout = 5000"); // wait briefly on a lock held by another process instead of failing immediately -- set before anything else that could itself contend for the lock
  db.exec("PRAGMA journal_mode = WAL");
  db.exec("CREATE TABLE IF NOT EXISTS lanes (id TEXT PRIMARY KEY, data TEXT NOT NULL)");
  db.exec("CREATE TABLE IF NOT EXISTS roles (agent TEXT PRIMARY KEY, data TEXT NOT NULL)");
  db.exec("CREATE TABLE IF NOT EXISTS decisions (commit_sha TEXT PRIMARY KEY, data TEXT NOT NULL)");
  if (isNew) migrateFlatFiles(db, root);
  dbCache.set(dbPath, db);
  return db;
 }
 
 export function readLane(repoStateDir: string, laneId: string): Lane | null {
  const row = getDb(repoStateDir).prepare("SELECT data FROM lanes WHERE id = ?").get(laneId) as { data: string } | undefined;
  return row ? (JSON.parse(row.data) as Lane) : null;
 }
 
 export function writeLane(repoStateDir: string, lane: Lane): void {
  getDb(repoStateDir).prepare("INSERT OR REPLACE INTO lanes (id, data) VALUES (?, ?)").run(lane.id, JSON.stringify(lane, null, 2));
 }
 
 export function listLanes(repoStateDir: string): Lane[] {
  const rows = getDb(repoStateDir).prepare("SELECT data FROM lanes").all() as { data: string }[];
  return rows.map((r) => JSON.parse(r.data) as Lane);
 }
 
 /**
  * Atomic read-modify-write for a single Lane: the whole thing runs
  * inside one SQLite transaction (BEGIN IMMEDIATE takes the write lock
  * up front), so two processes racing to act on the same Lane serialize
  * instead of one silently clobbering the other's change. `mutate` edits
  * the Lane in place; its return value is what gets persisted and
  * returned.
  */
 export function withLane(repoStateDir: string, laneId: string, mutate: (lane: Lane) => void): Lane {
  const db = getDb(repoStateDir);
  db.exec("BEGIN IMMEDIATE");
  try {
  const row = db.prepare("SELECT data FROM lanes WHERE id = ?").get(laneId) as { data: string } | undefined;
  if (!row) throw new Error(`no such lane: ${laneId}`);
  const lane = JSON.parse(row.data) as Lane;
  mutate(lane);
  db.prepare("INSERT OR REPLACE INTO lanes (id, data) VALUES (?, ?)").run(lane.id, JSON.stringify(lane, null, 2));
  db.exec("COMMIT");
  return lane;
  } catch (err) {
  db.exec("ROLLBACK");
  throw err;
  }
 }
 
 export function readRoles(repoStateDir: string, agent: string): AgentRoles | null {
  const row = getDb(repoStateDir).prepare("SELECT data FROM roles WHERE agent = ?").get(agent) as { data: string } | undefined;
  return row ? (JSON.parse(row.data) as AgentRoles) : null;
 }
 
 export function writeRoles(repoStateDir: string, agentRoles: AgentRoles): void {
  getDb(repoStateDir).prepare("INSERT OR REPLACE INTO roles (agent, data) VALUES (?, ?)").run(agentRoles.agent, JSON.stringify(agentRoles, null, 2));
 }
 
 export function writeDecision(repoStateDir: string, commitSha: string, decision: unknown): void {
  getDb(repoStateDir).prepare("INSERT OR REPLACE INTO decisions (commit_sha, data) VALUES (?, ?)").run(commitSha, JSON.stringify(decision, null, 2));
 }
 
 // Ladder storage lives in git notes (see engine/ladderNotes.ts), not here:
 // a rung is an annotation on a git object, not a standalone coordination
 // record the way a Lane or Role grant is.