import { Database } from "bun:sqlite"; import { mkdirSync } from "node:fs"; import { join } from "node:path"; import { inflateSync } from "node:zlib"; import type { EventRecord, ReleaseRecord, SaveRecord } from "../bucket/repo.ts"; import { isTestPath } from "../describe/links.ts"; // Koh 3's schema (§5.1, `src/storage/storage.v`) minus `file_cache`, which only a working tree needs, // plus `releases`: each version's release.json (§11.2) as read from the bucket, since that is bucket // data too and pages and feeds need it on every request. const SCHEMA_VERSION = 2; const SCHEMA = ` CREATE TABLE objects ( hash TEXT PRIMARY KEY, size INTEGER NOT NULL, data BLOB NOT NULL ); CREATE TABLE saves ( id TEXT PRIMARY KEY, parent TEXT REFERENCES saves(id), message TEXT NOT NULL, created_at INTEGER NOT NULL, machine TEXT NOT NULL, is_checkpoint INTEGER NOT NULL DEFAULT 0, size_bytes INTEGER NOT NULL DEFAULT 0, agent TEXT, session TEXT, restores TEXT, replayed_from TEXT ); CREATE TABLE save_files ( save_id TEXT NOT NULL REFERENCES saves(id), path TEXT NOT NULL, hash TEXT NOT NULL REFERENCES objects(hash), executable INTEGER NOT NULL DEFAULT 0, PRIMARY KEY (save_id, path) ); CREATE TABLE events ( id TEXT PRIMARY KEY, kind TEXT NOT NULL, save_id TEXT, path TEXT, reason TEXT, created_at INTEGER NOT NULL, machine TEXT NOT NULL ); CREATE TABLE state ( key TEXT PRIMARY KEY, value TEXT NOT NULL ); CREATE TABLE releases ( version TEXT PRIMARY KEY, record TEXT NOT NULL ); CREATE INDEX saves_parent ON saves(parent); CREATE INDEX save_files_hash ON save_files(hash); `; /** An object as the bucket stores it: zlib bytes, already checked against its hash. */ export type StoredObject = { compressed: Uint8Array; size: number }; /** One batch of a sync, landed in one transaction. Saves are oldest first. */ export type Landing = { head: string; releasesLatest: string; saves: SaveRecord[]; objects: Map; events: EventRecord[]; releases: ReleaseRecord[]; /** Escaped names of the paths these saves arrived without (§5.5, A3). */ quarantined: string[]; /** * The operator's hidden paths (June, 2026-09-30), removed from every save like a strip, and the * names ignored in every repo (June, 2026-10-01), each as `ANY_DEPTH` + name. */ hidden: string[]; }; /** Marks a hidden entry as a name matched at any depth rather than a path from the root. */ export const ANY_DEPTH = "**/"; // A recount inflates every file at the newest save; a file this large is data, not lines of code. const MAX_COUNTED = 8 * 1024 * 1024; // Display names for the header's language, by extension; data and docs formats never count. const LANGUAGES: Record = { ts: "TypeScript", tsx: "TypeScript", js: "JavaScript", mjs: "JavaScript", zig: "Zig", v: "V", rs: "Rust", go: "Go", py: "Python", c: "C", h: "C", cpp: "C++", cc: "C++", hpp: "C++", swift: "Swift", kt: "Kotlin", java: "Java", rb: "Ruby", lua: "Lua", sh: "Shell", nix: "Nix", css: "CSS", html: "HTML", }; // Vendored code is someone else's: counting Koh 3's bundled SQLite made it read as 290k lines of C. const VENDORED = /(^|\/)(vendor|third_party|node_modules)\//; // Bumped when what counts as a line changes, so every mirror recounts once when it opens. const STATS_RULE = "source-only"; const open = new Map(); /** The mirror of one repo at `/.db`, created on first use and kept open. */ export function openMirror(dir: string, repo: string): Database { const path = join(dir, `${repo}.db`); const cached = open.get(path); if (cached) return cached; mkdirSync(dir, { recursive: true }); const db = new Database(path, { strict: true }); db.run("PRAGMA journal_mode = WAL"); db.run("PRAGMA foreign_keys = ON"); db.run("PRAGMA busy_timeout = 5000"); const { user_version } = db.query<{ user_version: number }, []>("PRAGMA user_version").get() ?? { user_version: 0, }; if (user_version !== 0 && user_version !== SCHEMA_VERSION) { db.close(); throw new Error( `${path} has mirror schema version ${user_version}; this Kepr reads version ${SCHEMA_VERSION}. ` + "The mirror is disposable: delete it and the next sync rebuilds it from the bucket.", ); } if (user_version === 0) { db.transaction(() => { db.run(SCHEMA); db.run(`PRAGMA user_version = ${SCHEMA_VERSION}`); })(); } // Mirrors counted under an older rule are recounted now, not at the next save. if (mirrorState(db, "head") && mirrorState(db, "stats_rule") !== STATS_RULE) refreshStats(db); open.set(path, db); return db; } export function mirrorState(db: Database, key: string): string | null { const row = db .query<{ value: string }, [string]>("SELECT value FROM state WHERE key = ?") .get(key); return row?.value ?? null; } export function knownSave(db: Database, id: string): boolean { return db.query("SELECT 1 FROM saves WHERE id = ?").get(id) !== null; } export function knownObject(db: Database, hash: string): boolean { return db.query("SELECT 1 FROM objects WHERE hash = ?").get(hash) !== null; } export function knownEventIds(db: Database): Set { const rows = db.query<{ id: string }, []>("SELECT id FROM events").all(); return new Set(rows.map((row) => row.id)); } /** Drops a version deleted from the bucket; a sync only ever adds releases. */ export function forgetRelease(db: Database, version: string): void { db.query("DELETE FROM releases WHERE version = ?").run(version); } /** Replaces a version's record after Kepr rewrote it in the bucket. */ export function replaceRelease(db: Database, record: ReleaseRecord): void { db.query("UPDATE releases SET record = ? WHERE version = ?").run( JSON.stringify(record), record.version, ); } export function knownReleases(db: Database): Set { const rows = db.query<{ version: string }, []>("SELECT version FROM releases").all(); return new Set(rows.map((row) => row.version)); } /** Lands a sync like Koh's steal: objects, events, saves, then every strip over every save. */ /** Stores objects on their own, ahead of the save that refers to them. */ export function storeObjects(db: Database, objects: Map): void { const put = db.query("INSERT OR IGNORE INTO objects (hash, size, data) VALUES (?, ?, ?)"); db.transaction(() => { for (const [hash, object] of objects) put.run(hash, object.size, object.compressed); })(); } export function land(db: Database, landing: Landing): void { const putEvent = db.query( "INSERT OR IGNORE INTO events (id, kind, save_id, path, reason, created_at, machine) VALUES " + "(?, ?, ?, ?, ?, ?, ?)", ); const putSave = db.query( `INSERT OR IGNORE INTO saves (id, parent, message, created_at, machine, is_checkpoint, size_bytes, agent, session, restores, replayed_from) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, ); const putFile = db.query( "INSERT OR IGNORE INTO save_files (save_id, path, hash, executable) VALUES (?, ?, ?, ?)", ); const putRelease = db.query("INSERT OR IGNORE INTO releases (version, record) VALUES (?, ?)"); const setState = db.query("INSERT OR REPLACE INTO state (key, value) VALUES (?, ?)"); db.transaction(() => { storeObjects(db, landing.objects); for (const e of landing.events) { putEvent.run(e.id, e.kind, e.save_id, e.path, e.reason, e.created_at, e.machine); } for (const s of landing.saves) { putSave.run( s.id, s.parent === "" ? null : s.parent, s.message, s.created_at, s.machine, s.is_checkpoint ? 1 : 0, s.size_bytes, s.agent, s.session, s.restores, s.replayed_from, ); for (const f of s.files) putFile.run(s.id, f.path, f.hash, f.executable ? 1 : 0); } for (const r of landing.releases) putRelease.run(r.version, JSON.stringify(r)); applyStrips(db); let hidden = 0; for (const path of landing.hidden) hidden += removePath(db, path, ""); if (hidden > 0) db.run("DELETE FROM objects WHERE hash NOT IN (SELECT hash FROM save_files)"); // In the same transaction as the removal, so the tarball key never names a list the rows // don't reflect yet. setState.run("hidden", landing.hidden.join("\n")); setState.run("head", landing.head); setState.run("releases_latest", landing.releasesLatest); const quarantined = new Set([...quarantinedPaths(db), ...landing.quarantined]); setState.run("quarantined", JSON.stringify([...quarantined].sort())); })(); // Outside the transaction: the recount can inflate megabytes and needs no lock to be correct. refreshStats(db); } /** Escaped names of every path this mirror's saves arrived without. */ export function quarantinedPaths(db: Database): string[] { return JSON.parse(mirrorState(db, "quarantined") ?? "[]"); } // §10.2: a strip of one save reaches that save; a strip of every save bans its path from all of // history, whenever a save was made or arrives. Re-applying every strip is harmless, which is how a // save that arrives after its strip event is covered too. function applyStrips(db: Database): void { const strips = db .query<{ save_id: string; path: string }, []>( "SELECT save_id, path FROM events WHERE kind = 'strip'", ) .all(); if (strips.length === 0) return; for (const strip of strips) removePath(db, strip.path, strip.save_id); db.run("DELETE FROM objects WHERE hash NOT IN (SELECT hash FROM save_files)"); } /** Removes a path, file or folder, from one save or ('') every save; returns the rows removed. */ function removePath(db: Database, path: string, save: string): number { if (path.startsWith(ANY_DEPTH)) { // A name ignored in every repo: any path with a segment of exactly that name. return db .query( `DELETE FROM save_files WHERE instr('/' || path || '/', $segment) > 0 AND ($save = '' OR save_id = $save)`, ) .run({ segment: `/${path.slice(ANY_DEPTH.length)}/`, save }).changes; } // Koh compares the prefix exactly, never with LIKE, so `_` and `%` in a path match only themselves. const dir = `${path.replace(/\/+$/, "")}/`; return db .query( `DELETE FROM save_files WHERE (path = $path OR substr(path, 1, length($dir)) = $dir) AND ($save = '' OR save_id = $save)`, ) .run({ path, dir, save }).changes; } /** * Empties a mirror in place so the next sync rebuilds it from the bucket, as a change to the * hidden paths needs. The file and its open handles stay, so a request holding one reads an empty * repo for a moment rather than failing. A new epoch keeps tarballs kept before it from matching. */ export function clearMirror(db: Database): void { db.transaction(() => { // Children before parents, for the foreign keys; every name is a table of SCHEMA's. for (const table of ["save_files", "events", "saves", "objects", "releases", "state"]) { db.run(`DELETE FROM ${table}`); } db.query("INSERT INTO state (key, value) VALUES ('epoch', ?)").run(crypto.randomUUID()); })(); } // Lines and languages never change between syncs, so they are counted once here, not per view. They // come from the newest real save, not `head`: a checkpoint's files are not public history (§21.4). // Lines are source lines: files in a known language, tests left out; docs and data never count. function refreshStats(db: Database): void { const save = db .query<{ id: string }, []>( "SELECT id FROM saves WHERE is_checkpoint = 0 ORDER BY rowid DESC LIMIT 1", ) .get()?.id ?? ""; const files = db .query<{ path: string; size: number; data: Uint8Array }, [string]>( "SELECT f.path, o.size, o.data FROM save_files f JOIN objects o ON o.hash = f.hash WHERE f.save_id = ?", ) .all(save); let lines = 0; const byLanguage = new Map(); for (const file of files) { const language = LANGUAGES[file.path.split(".").pop() ?? ""]; // Tests are not the program: counted, Koh 3's read as 18k lines of V where it has 11k. if (!language || VENDORED.test(file.path) || isTestPath(file.path)) continue; if (file.size > MAX_COUNTED) continue; const bytes = inflateSync(file.data); if (bytes.includes(0)) continue; let count = 0; for (const byte of bytes) if (byte === 10) count++; if (bytes.length > 0 && bytes[bytes.length - 1] !== 10) count++; lines += count; byLanguage.set(language, (byLanguage.get(language) ?? 0) + count); } const languages = [...byLanguage].sort((a, b) => b[1] - a[1]); const setState = db.query("INSERT OR REPLACE INTO state (key, value) VALUES (?, ?)"); setState.run("lines", String(lines)); setState.run("language", languages[0]?.[0] ?? ""); setState.run("languages", JSON.stringify(languages)); setState.run("stats_save", save); setState.run("stats_rule", STATS_RULE); }