store.ts

1import { Database } from "bun:sqlite";2import { existsSync, mkdirSync, readdirSync } from "node:fs";3import { join } from "node:path";4import { inflateSync } from "node:zlib";5import type { EventRecord, ReleaseRecord, SaveRecord } from "../bucket/repo.ts";6import { isTestPath } from "../describe/links.ts";78// Koh 3's schema (§5.1, `src/storage/storage.v`) minus `file_cache`, which only a working tree needs,9// plus `releases`: each version's release.json (§11.2) as read from the bucket, since that is bucket10// data too and pages and feeds need it on every request, and `hidden_files`.11const SCHEMA_VERSION = 3;12// The operator's hidden rows, moved out of `save_files` rather than deleted, so a changed hidden13// list is applied in place, with nothing fetched again (2026-10-02).14const HIDDEN_FILES = `15CREATE TABLE hidden_files (16	save_id    TEXT NOT NULL REFERENCES saves(id),17	path       TEXT NOT NULL,18	hash       TEXT NOT NULL REFERENCES objects(hash),19	executable INTEGER NOT NULL DEFAULT 0,20	PRIMARY KEY (save_id, path)21);22`;23const SCHEMA = `24CREATE TABLE objects (25	hash TEXT PRIMARY KEY,26	size INTEGER NOT NULL,27	data BLOB NOT NULL28);29CREATE TABLE saves (30	id            TEXT PRIMARY KEY,31	parent        TEXT REFERENCES saves(id),32	message       TEXT NOT NULL,33	created_at    INTEGER NOT NULL,34	machine       TEXT NOT NULL,35	is_checkpoint INTEGER NOT NULL DEFAULT 0,36	size_bytes    INTEGER NOT NULL DEFAULT 0,37	agent         TEXT,38	session       TEXT,39	restores      TEXT,40	replayed_from TEXT41);42CREATE TABLE save_files (43	save_id    TEXT NOT NULL REFERENCES saves(id),44	path       TEXT NOT NULL,45	hash       TEXT NOT NULL REFERENCES objects(hash),46	executable INTEGER NOT NULL DEFAULT 0,47	PRIMARY KEY (save_id, path)48);49CREATE TABLE events (50	id         TEXT PRIMARY KEY,51	kind       TEXT NOT NULL,52	save_id    TEXT,53	path       TEXT,54	reason     TEXT,55	created_at INTEGER NOT NULL,56	machine    TEXT NOT NULL57);58CREATE TABLE state (59	key   TEXT PRIMARY KEY,60	value TEXT NOT NULL61);62CREATE TABLE releases (63	version TEXT PRIMARY KEY,64	record  TEXT NOT NULL65);66CREATE INDEX saves_parent ON saves(parent);67CREATE INDEX save_files_hash ON save_files(hash);68${HIDDEN_FILES}`;6970/** An object as the bucket stores it: zlib bytes, already checked against its hash. */71export type StoredObject = { compressed: Uint8Array; size: number };7273/** One batch of a sync, landed in one transaction. Saves are oldest first. */74export type Landing = {75	head: string;76	releasesLatest: string;77	saves: SaveRecord[];78	objects: Map<string, StoredObject>;79	events: EventRecord[];80	releases: ReleaseRecord[];81	/** Escaped names of the paths these saves arrived without (§5.5, A3). */82	quarantined: string[];83};8485/** Marks a hidden entry as a name matched at any depth rather than a path from the root. */86export const ANY_DEPTH = "**/";8788// A recount inflates every file at the newest save; a file this large is data, not lines of code.89const MAX_COUNTED = 8 * 1024 * 1024;9091// Display names for the header's language, by extension; data and docs formats never count.92const LANGUAGES: Record<string, string> = {93	ts: "TypeScript",94	tsx: "TypeScript",95	js: "JavaScript",96	mjs: "JavaScript",97	zig: "Zig",98	v: "V",99	rs: "Rust",100	go: "Go",101	py: "Python",102	c: "C",103	h: "C",104	cpp: "C++",105	cc: "C++",106	hpp: "C++",107	swift: "Swift",108	kt: "Kotlin",109	java: "Java",110	rb: "Ruby",111	lua: "Lua",112	sh: "Shell",113	nix: "Nix",114	css: "CSS",115	html: "HTML",116};117118// Vendored code is someone else's: counting Koh 3's bundled SQLite made it read as 290k lines of C.119const VENDORED = /(^|\/)(vendor|third_party|node_modules)\//;120// Bumped when what counts as a line changes, so every mirror recounts once when it opens.121const STATS_RULE = "source-only";122123const open = new Map<string, Database>();124125/** Closes a repo's mirror, so a rename can move its file. */126export function closeMirror(dir: string, repo: string): void {127	const path = join(dir, `${repo}.db`);128	open.get(path)?.close();129	open.delete(path);130}131132/**133 * Opens every mirror in `dir`, bringing each to this Kepr's schema: run at start, before Kepr134 * listens, so no request waits on a migration.135 */136export function openMirrors(dir: string): void {137	if (!existsSync(dir)) return;138	for (const name of readdirSync(dir)) if (name.endsWith(".db")) openMirror(dir, name.slice(0, -3));139}140141/** Closes every mirror this thread holds open; the next use of each opens it again. */142export function closeMirrors(): void {143	for (const db of open.values()) db.close();144	open.clear();145}146147/** The mirror of one repo at `<dir>/<repo>.db`, created on first use and kept open. */148export function openMirror(dir: string, repo: string): Database {149	const path = join(dir, `${repo}.db`);150	const cached = open.get(path);151	if (cached) return cached;152	mkdirSync(dir, { recursive: true });153	const db = new Database(path, { strict: true });154	db.run("PRAGMA journal_mode = WAL");155	db.run("PRAGMA foreign_keys = ON");156	db.run("PRAGMA busy_timeout = 5000");157	// A current mirror needs no lock; only creating or migrating one takes the write lock.158	const found = schemaVersion(db) === SCHEMA_VERSION ? SCHEMA_VERSION : upgrade(db);159	if (found !== 0 && found !== 2 && found !== SCHEMA_VERSION) {160		db.close();161		throw new Error(162			`${path} has mirror schema version ${found}; this Kepr reads version ${SCHEMA_VERSION}. ` +163				"The mirror is disposable: delete it and the next sync rebuilds it from the bucket.",164		);165	}166	// Mirrors counted under an older rule are recounted now, not at the next save.167	if (mirrorState(db, "head") && mirrorState(db, "stats_rule") !== STATS_RULE) refreshStats(db);168	open.set(path, db);169	return db;170}171172function schemaVersion(db: Database): number {173	return db.query<{ user_version: number }, []>("PRAGMA user_version").get()?.user_version ?? 0;174}175176/**177 * Creates or migrates a mirror and returns the version it found, read again under the write lock:178 * the request thread and the sync's both open mirrors, and must not both create or migrate one.179 */180function upgrade(db: Database): number {181	return db182		.transaction(() => {183			const found = schemaVersion(db);184			if (found === 0) db.run(SCHEMA);185			if (found === 2) {186				db.run(HIDDEN_FILES);187				// Version 2 deleted hidden rows, so only the bucket can bring them back on unhide.188				if (mirrorState(db, "hidden")) clearMirror(db);189			}190			if (found === 0 || found === 2) db.run(`PRAGMA user_version = ${SCHEMA_VERSION}`);191			return found;192		})193		.immediate();194}195196export function mirrorState(db: Database, key: string): string | null {197	const row = db198		.query<{ value: string }, [string]>("SELECT value FROM state WHERE key = ?")199		.get(key);200	return row?.value ?? null;201}202203export function knownSave(db: Database, id: string): boolean {204	return db.query("SELECT 1 FROM saves WHERE id = ?").get(id) !== null;205}206207export function knownObject(db: Database, hash: string): boolean {208	return db.query("SELECT 1 FROM objects WHERE hash = ?").get(hash) !== null;209}210211export function knownEventIds(db: Database): Set<string> {212	const rows = db.query<{ id: string }, []>("SELECT id FROM events").all();213	return new Set(rows.map((row) => row.id));214}215216/** Drops a version deleted from the bucket; a sync only ever adds releases. */217export function forgetRelease(db: Database, version: string): void {218	db.query("DELETE FROM releases WHERE version = ?").run(version);219}220221/** Replaces a version's record after Kepr rewrote it in the bucket. */222export function replaceRelease(db: Database, record: ReleaseRecord): void {223	db.query("UPDATE releases SET record = ? WHERE version = ?").run(224		JSON.stringify(record),225		record.version,226	);227}228229export function knownReleases(db: Database): Set<string> {230	const rows = db.query<{ version: string }, []>("SELECT version FROM releases").all();231	return new Set(rows.map((row) => row.version));232}233234/** Stores objects on their own, ahead of the save that refers to them. */235export function storeObjects(db: Database, objects: Map<string, StoredObject>): void {236	const put = db.query("INSERT OR IGNORE INTO objects (hash, size, data) VALUES (?, ?, ?)");237	db.transaction(() => {238		for (const [hash, object] of objects) put.run(hash, object.size, object.compressed);239	})();240}241242/**243 * Lands a sync like Koh's steal: objects, events, saves, then every strip over every save. The new244 * saves are hidden by the list the mirror holds; `applyHidden` changes that list.245 */246export function land(db: Database, landing: Landing): void {247	const putEvent = db.query(248		"INSERT OR IGNORE INTO events (id, kind, save_id, path, reason, created_at, machine) VALUES " +249			"(?, ?, ?, ?, ?, ?, ?)",250	);251	const putSave = db.query(252		`INSERT OR IGNORE INTO saves (id, parent, message, created_at, machine, is_checkpoint, size_bytes,253			agent, session, restores, replayed_from) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,254	);255	const putFile = db.query(256		"INSERT OR IGNORE INTO save_files (save_id, path, hash, executable) VALUES (?, ?, ?, ?)",257	);258	const putRelease = db.query("INSERT OR IGNORE INTO releases (version, record) VALUES (?, ?)");259	const setState = db.query("INSERT OR REPLACE INTO state (key, value) VALUES (?, ?)");260	// Immediate: a landing with nothing new reads before it writes, and the request thread writes261	// mirrors too (a deleted release).262	db.transaction(() => {263		storeObjects(db, landing.objects);264		for (const e of landing.events) {265			putEvent.run(e.id, e.kind, e.save_id, e.path, e.reason, e.created_at, e.machine);266		}267		for (const s of landing.saves) {268			putSave.run(269				s.id,270				s.parent === "" ? null : s.parent,271				s.message,272				s.created_at,273				s.machine,274				s.is_checkpoint ? 1 : 0,275				s.size_bytes,276				s.agent,277				s.session,278				s.restores,279				s.replayed_from,280			);281			for (const f of s.files) putFile.run(s.id, f.path, f.hash, f.executable ? 1 : 0);282		}283		for (const r of landing.releases) putRelease.run(r.version, JSON.stringify(r));284		applyStrips(db);285		hideRows(286			db,287			hiddenList(db),288			landing.saves.map((save) => save.id),289		);290		setState.run("head", landing.head);291		setState.run("releases_latest", landing.releasesLatest);292		const quarantined = new Set([...quarantinedPaths(db), ...landing.quarantined]);293		setState.run("quarantined", JSON.stringify([...quarantined].sort()));294	}).immediate();295	// Outside the transaction: the recount can inflate megabytes and needs no lock to be correct.296	refreshStats(db);297}298299/**300 * Makes `hidden` the mirror's hidden list: the operator's hidden paths (June, 2026-09-30) and the301 * names ignored in every repo (June, 2026-10-01), each as `ANY_DEPTH` + name. Rows move between302 * `save_files` and `hidden_files`, so it takes milliseconds and fetches nothing.303 */304export function applyHidden(db: Database, hidden: string[]): void {305	if (hiddenList(db).join("\n") === hidden.join("\n")) return;306	db.transaction(() => {307		db.run(308			`INSERT OR IGNORE INTO save_files (save_id, path, hash, executable)309			SELECT save_id, path, hash, executable FROM hidden_files`,310		);311		db.run("DELETE FROM hidden_files");312		hideRows(db, hidden, null);313	})();314	refreshStats(db);315}316317function hiddenList(db: Database): string[] {318	return mirrorState(db, "hidden")?.split("\n").filter(Boolean) ?? [];319}320321// Moves the rows `hidden` matches, in the `saves` given or (null) every save, into `hidden_files`.322// The list is recorded in the same transaction, so the tarball key never names a list the rows323// don't reflect yet.324function hideRows(db: Database, hidden: string[], saves: string[] | null): void {325	for (const entry of hidden) {326		const { where, params } = pathMatch(entry);327		const scope = { ...params, all: saves === null ? 1 : 0, saves: JSON.stringify(saves ?? []) };328		const chosen = `${where} AND ($all = 1 OR save_id IN (SELECT value FROM json_each($saves)))`;329		db.query(330			`INSERT OR IGNORE INTO hidden_files (save_id, path, hash, executable)331			SELECT save_id, path, hash, executable FROM save_files332			WHERE ${chosen}`,333		).run(scope);334		db.query(`DELETE FROM save_files WHERE ${chosen}`).run(scope);335	}336	db.query("INSERT OR REPLACE INTO state (key, value) VALUES ('hidden', ?)").run(hidden.join("\n"));337}338339/** Escaped names of every path this mirror's saves arrived without. */340export function quarantinedPaths(db: Database): string[] {341	return JSON.parse(mirrorState(db, "quarantined") ?? "[]");342}343344// §10.2: a strip of one save reaches that save; a strip of every save bans its path from all of345// history, whenever a save was made or arrives. Re-applying every strip is harmless, which is how a346// save that arrives after its strip event is covered too.347function applyStrips(db: Database): void {348	const strips = db349		.query<{ save_id: string; path: string }, []>(350			"SELECT save_id, path FROM events WHERE kind = 'strip'",351		)352		.all();353	if (strips.length === 0) return;354	for (const strip of strips) {355		const { where, params } = pathMatch(strip.path);356		// Hidden rows too, so unhiding never brings back what a strip removed.357		for (const table of ["save_files", "hidden_files"]) {358			db.query(`DELETE FROM ${table} WHERE ${where} AND ($save = '' OR save_id = $save)`).run({359				...params,360				save: strip.save_id,361			});362		}363	}364	db.run(365		`DELETE FROM objects366		WHERE hash NOT IN (SELECT hash FROM save_files)367		AND hash NOT IN (SELECT hash FROM hidden_files)`,368	);369}370371/** The SQL condition matching a path, file or folder, or an `ANY_DEPTH` name at any depth. */372function pathMatch(path: string): { where: string; params: Record<string, string> } {373	if (path.startsWith(ANY_DEPTH)) {374		return {375			where: "instr('/' || path || '/', $segment) > 0",376			params: { segment: `/${path.slice(ANY_DEPTH.length)}/` },377		};378	}379	// Koh compares the prefix exactly, never with LIKE, so `_` and `%` in a path match only themselves.380	return {381		where: "(path = $path OR substr(path, 1, length($dir)) = $dir)",382		params: { path, dir: `${path.replace(/\/+$/, "")}/` },383	};384}385386/**387 * Empties a mirror in place so the next sync rebuilds it from the bucket, as the move to schema388 * version 3 needs when hidden rows were deleted. The file and its open handles stay, so a request holding one reads an empty389 * repo for a moment rather than failing. A new epoch keeps tarballs kept before it from matching.390 */391export function clearMirror(db: Database): void {392	db.transaction(() => {393		// Children before parents, for the foreign keys; every name is a table of SCHEMA's.394		const tables = [395			"save_files",396			"hidden_files",397			"events",398			"saves",399			"objects",400			"releases",401			"state",402		];403		for (const table of tables) {404			db.run(`DELETE FROM ${table}`);405		}406		db.query("INSERT INTO state (key, value) VALUES ('epoch', ?)").run(crypto.randomUUID());407	})();408}409410// Lines and languages never change between syncs, so they are counted once here, not per view. They411// come from the newest real save, not `head`: a checkpoint's files are not public history (§21.4).412// Lines are source lines: files in a known language, tests left out; docs and data never count.413function refreshStats(db: Database): void {414	const save =415		db416			.query<{ id: string }, []>(417				"SELECT id FROM saves WHERE is_checkpoint = 0 ORDER BY rowid DESC LIMIT 1",418			)419			.get()?.id ?? "";420	const files = db421		.query<{ path: string; size: number; data: Uint8Array }, [string]>(422			"SELECT f.path, o.size, o.data FROM save_files f JOIN objects o ON o.hash = f.hash WHERE f.save_id = ?",423		)424		.all(save);425	let lines = 0;426	const byLanguage = new Map<string, number>();427	for (const file of files) {428		const language = LANGUAGES[file.path.split(".").pop() ?? ""];429		// Tests are not the program: counted, Koh 3's read as 18k lines of V where it has 11k.430		if (!language || VENDORED.test(file.path) || isTestPath(file.path)) continue;431		if (file.size > MAX_COUNTED) continue;432		const bytes = inflateSync(file.data);433		if (bytes.includes(0)) continue;434		let count = 0;435		for (const byte of bytes) if (byte === 10) count++;436		if (bytes.length > 0 && bytes[bytes.length - 1] !== 10) count++;437		lines += count;438		byLanguage.set(language, (byLanguage.get(language) ?? 0) + count);439	}440	const languages = [...byLanguage].sort((a, b) => b[1] - a[1]);441	const setState = db.query("INSERT OR REPLACE INTO state (key, value) VALUES (?, ?)");442	setState.run("lines", String(lines));443	setState.run("language", languages[0]?.[0] ?? "");444	setState.run("languages", JSON.stringify(languages));445	setState.run("stats_save", save);446	setState.run("stats_rule", STATS_RULE);447}