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 { fileFacts } from "../describe/describe.ts";7import { isTestPath } from "../describe/links.ts";89// Koh 3's schema (§5.1, `src/storage/storage.v`) minus `file_cache`, which only a working tree needs,10// plus `releases`: each version's release.json (§11.2) as read from the bucket, since that is bucket11// data too and pages and feeds need it on every request, `hidden_files`, and what the Atlas reads.12const SCHEMA_VERSION = 4;13// The operator's hidden rows, moved out of `save_files` rather than deleted, so a changed hidden14// list is applied in place, with nothing fetched again (2026-10-02).15const HIDDEN_FILES = `16CREATE TABLE hidden_files (17	save_id    TEXT NOT NULL REFERENCES saves(id),18	path       TEXT NOT NULL,19	hash       TEXT NOT NULL REFERENCES objects(hash),20	executable INTEGER NOT NULL DEFAULT 0,21	PRIMARY KEY (save_id, path)22);23`;24// What pages read, worked out as each save lands rather than on every view (2026-10-02): which25// real saves touched each path (added, changed or removed against the real save before; `n` is the26// touching save's rowid), what each file says about itself, and the objects real saves refer to.27const DERIVED = `28CREATE TABLE known_objects (hash TEXT PRIMARY KEY) WITHOUT ROWID;29CREATE TABLE touches (30	path TEXT NOT NULL,31	n    INTEGER NOT NULL32);33CREATE INDEX touches_n ON touches(n);34CREATE TABLE facts (35	hash        TEXT NOT NULL,36	path        TEXT NOT NULL,37	lines       INTEGER,38	description TEXT,39	outline     TEXT,40	PRIMARY KEY (hash, path)41);42`;43const SCHEMA = `44CREATE TABLE objects (45	hash TEXT PRIMARY KEY,46	size INTEGER NOT NULL,47	data BLOB NOT NULL48);49CREATE TABLE saves (50	id            TEXT PRIMARY KEY,51	parent        TEXT REFERENCES saves(id),52	message       TEXT NOT NULL,53	created_at    INTEGER NOT NULL,54	machine       TEXT NOT NULL,55	is_checkpoint INTEGER NOT NULL DEFAULT 0,56	size_bytes    INTEGER NOT NULL DEFAULT 0,57	agent         TEXT,58	session       TEXT,59	restores      TEXT,60	replayed_from TEXT61);62CREATE TABLE save_files (63	save_id    TEXT NOT NULL REFERENCES saves(id),64	path       TEXT NOT NULL,65	hash       TEXT NOT NULL REFERENCES objects(hash),66	executable INTEGER NOT NULL DEFAULT 0,67	PRIMARY KEY (save_id, path)68);69CREATE TABLE events (70	id         TEXT PRIMARY KEY,71	kind       TEXT NOT NULL,72	save_id    TEXT,73	path       TEXT,74	reason     TEXT,75	created_at INTEGER NOT NULL,76	machine    TEXT NOT NULL77);78CREATE TABLE state (79	key   TEXT PRIMARY KEY,80	value TEXT NOT NULL81);82CREATE TABLE releases (83	version TEXT PRIMARY KEY,84	record  TEXT NOT NULL85);86CREATE INDEX saves_parent ON saves(parent);87CREATE INDEX save_files_hash ON save_files(hash);88${HIDDEN_FILES}89${DERIVED}`;9091/** An object as the bucket stores it: zlib bytes, already checked against its hash. */92export type StoredObject = { compressed: Uint8Array; size: number };9394/** One batch of a sync, landed in one transaction. Saves are oldest first. */95export type Landing = {96	head: string;97	releasesLatest: string;98	saves: SaveRecord[];99	objects: Map<string, StoredObject>;100	events: EventRecord[];101	releases: ReleaseRecord[];102	/** Escaped names of the paths these saves arrived without (§5.5, A3). */103	quarantined: string[];104};105106/** Marks a hidden entry as a name matched at any depth rather than a path from the root. */107export const ANY_DEPTH = "**/";108109// A recount inflates every file at the newest save; a file this large is data, not lines of code.110const MAX_COUNTED = 8 * 1024 * 1024;111112// Display names for the header's language, by extension; data and docs formats never count.113const LANGUAGES: Record<string, string> = {114	ts: "TypeScript",115	tsx: "TypeScript",116	js: "JavaScript",117	mjs: "JavaScript",118	zig: "Zig",119	v: "V",120	rs: "Rust",121	go: "Go",122	py: "Python",123	c: "C",124	h: "C",125	cpp: "C++",126	cc: "C++",127	hpp: "C++",128	swift: "Swift",129	kt: "Kotlin",130	java: "Java",131	rb: "Ruby",132	lua: "Lua",133	sh: "Shell",134	nix: "Nix",135	css: "CSS",136	html: "HTML",137};138139// Vendored code is someone else's: counting Koh 3's bundled SQLite made it read as 290k lines of C.140const VENDORED = /(^|\/)(vendor|third_party|node_modules)\//;141// Bumped when what the recount records changes, so every mirror recounts once when it opens.142const STATS_RULE = "source-only, known objects";143144const open = new Map<string, Database>();145146/** Closes a repo's mirror, so a rename can move its file. */147export function closeMirror(dir: string, repo: string): void {148	const path = join(dir, `${repo}.db`);149	open.get(path)?.close();150	open.delete(path);151}152153/**154 * Opens every mirror in `dir`, bringing each to this Kepr's schema: run at start, before Kepr155 * listens, so no request waits on a migration.156 */157export function openMirrors(dir: string): void {158	if (!existsSync(dir)) return;159	for (const name of readdirSync(dir)) if (name.endsWith(".db")) openMirror(dir, name.slice(0, -3));160}161162/** Closes every mirror this thread holds open; the next use of each opens it again. */163export function closeMirrors(): void {164	for (const db of open.values()) db.close();165	open.clear();166}167168/** The mirror of one repo at `<dir>/<repo>.db`, created on first use and kept open. */169export function openMirror(dir: string, repo: string): Database {170	const path = join(dir, `${repo}.db`);171	const cached = open.get(path);172	if (cached) return cached;173	mkdirSync(dir, { recursive: true });174	const db = new Database(path, { strict: true });175	db.run("PRAGMA journal_mode = WAL");176	db.run("PRAGMA foreign_keys = ON");177	db.run("PRAGMA busy_timeout = 5000");178	// A current mirror needs no lock; only creating or migrating one takes the write lock.179	const found = schemaVersion(db) === SCHEMA_VERSION ? SCHEMA_VERSION : upgrade(db);180	if (found !== 0 && found !== 2 && found !== 3 && found !== SCHEMA_VERSION) {181		db.close();182		throw new Error(183			`${path} has mirror schema version ${found}; this Kepr reads version ${SCHEMA_VERSION}. ` +184				"The mirror is disposable: delete it and the next sync rebuilds it from the bucket.",185		);186	}187	// Mirrors counted under an older rule are recounted now, not at the next save.188	if (mirrorState(db, "head") && mirrorState(db, "stats_rule") !== STATS_RULE) refreshStats(db);189	open.set(path, db);190	return db;191}192193function schemaVersion(db: Database): number {194	return db.query<{ user_version: number }, []>("PRAGMA user_version").get()?.user_version ?? 0;195}196197/**198 * Creates or migrates a mirror and returns the version it found, read again under the write lock:199 * the request thread and the sync's both open mirrors, and must not both create or migrate one.200 */201function upgrade(db: Database): number {202	return db203		.transaction(() => {204			const found = schemaVersion(db);205			if (found === SCHEMA_VERSION || ![0, 2, 3].includes(found)) return found;206			if (found === 0) db.run(SCHEMA);207			if (found === 2) db.run(HIDDEN_FILES);208			if (found === 2 || found === 3) db.run(DERIVED);209			// Version 2 deleted hidden rows, so only the bucket can bring them back on unhide.210			if (found === 2 && mirrorState(db, "hidden")) clearMirror(db);211			else if (found !== 0) {212				recordTouches(db, 0);213				recordFacts(db, null);214			}215			db.run(`PRAGMA user_version = ${SCHEMA_VERSION}`);216			return found;217		})218		.immediate();219}220221export function mirrorState(db: Database, key: string): string | null {222	const row = db223		.query<{ value: string }, [string]>("SELECT value FROM state WHERE key = ?")224		.get(key);225	return row?.value ?? null;226}227228export function knownSave(db: Database, id: string): boolean {229	return db.query("SELECT 1 FROM saves WHERE id = ?").get(id) !== null;230}231232export function knownObject(db: Database, hash: string): boolean {233	return db.query("SELECT 1 FROM objects WHERE hash = ?").get(hash) !== null;234}235236export function knownEventIds(db: Database): Set<string> {237	const rows = db.query<{ id: string }, []>("SELECT id FROM events").all();238	return new Set(rows.map((row) => row.id));239}240241/** Drops a version deleted from the bucket; a sync only ever adds releases. */242export function forgetRelease(db: Database, version: string): void {243	db.query("DELETE FROM releases WHERE version = ?").run(version);244}245246/** Replaces a version's record after Kepr rewrote it in the bucket. */247export function replaceRelease(db: Database, record: ReleaseRecord): void {248	db.query("UPDATE releases SET record = ? WHERE version = ?").run(249		JSON.stringify(record),250		record.version,251	);252}253254export function knownReleases(db: Database): Set<string> {255	const rows = db.query<{ version: string }, []>("SELECT version FROM releases").all();256	return new Set(rows.map((row) => row.version));257}258259/** Stores objects on their own, ahead of the save that refers to them. */260export function storeObjects(db: Database, objects: Map<string, StoredObject>): void {261	const put = db.query("INSERT OR IGNORE INTO objects (hash, size, data) VALUES (?, ?, ?)");262	db.transaction(() => {263		for (const [hash, object] of objects) put.run(hash, object.size, object.compressed);264	})();265}266267/**268 * Lands a sync like Koh's steal: objects, events, saves, then every strip over every save. The new269 * saves are hidden by the list the mirror holds; `applyHidden` changes that list.270 */271export function land(db: Database, landing: Landing): void {272	const putEvent = db.query(273		"INSERT OR IGNORE INTO events (id, kind, save_id, path, reason, created_at, machine) VALUES " +274			"(?, ?, ?, ?, ?, ?, ?)",275	);276	const putSave = db.query(277		`INSERT OR IGNORE INTO saves (id, parent, message, created_at, machine, is_checkpoint, size_bytes,278			agent, session, restores, replayed_from) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,279	);280	const putFile = db.query(281		"INSERT OR IGNORE INTO save_files (save_id, path, hash, executable) VALUES (?, ?, ?, ?)",282	);283	const putRelease = db.query("INSERT OR IGNORE INTO releases (version, record) VALUES (?, ?)");284	const setState = db.query("INSERT OR REPLACE INTO state (key, value) VALUES (?, ?)");285	// Immediate: a landing with nothing new reads before it writes, and the request thread writes286	// mirrors too (a deleted release).287	db.transaction(() => {288		storeObjects(db, landing.objects);289		for (const e of landing.events) {290			putEvent.run(e.id, e.kind, e.save_id, e.path, e.reason, e.created_at, e.machine);291		}292		for (const s of landing.saves) {293			putSave.run(294				s.id,295				s.parent === "" ? null : s.parent,296				s.message,297				s.created_at,298				s.machine,299				s.is_checkpoint ? 1 : 0,300				s.size_bytes,301				s.agent,302				s.session,303				s.restores,304				s.replayed_from,305			);306			for (const f of s.files) putFile.run(s.id, f.path, f.hash, f.executable ? 1 : 0);307		}308		for (const r of landing.releases) putRelease.run(r.version, JSON.stringify(r));309		const stripped = applyStrips(db);310		const landed = landing.saves.map((save) => save.id);311		hideRows(db, hiddenList(db), landed);312		// A strip can reach any save, so every save's touches are worked out again after one.313		recordTouches(db, stripped ? 0 : Number(mirrorState(db, "touches_to") ?? 0));314		recordFacts(db, landed);315		setState.run("head", landing.head);316		setState.run("releases_latest", landing.releasesLatest);317		const quarantined = new Set([...quarantinedPaths(db), ...landing.quarantined]);318		setState.run("quarantined", JSON.stringify([...quarantined].sort()));319		staleStats(db);320	}).immediate();321	// Outside the transaction: the recount can inflate megabytes and needs no lock to be correct.322	refreshStats(db);323}324325/**326 * Makes `hidden` the mirror's hidden list: the operator's hidden paths (June, 2026-09-30) and the327 * names ignored in every repo (June, 2026-10-01), each as `ANY_DEPTH` + name. Rows move between328 * `save_files` and `hidden_files`, so it takes milliseconds and fetches nothing.329 */330export function applyHidden(db: Database, hidden: string[]): void {331	if (hiddenList(db).join("\n") === hidden.join("\n")) return;332	db.transaction(() => {333		db.run(334			`INSERT OR IGNORE INTO save_files (save_id, path, hash, executable)335			SELECT save_id, path, hash, executable FROM hidden_files`,336		);337		db.run("DELETE FROM hidden_files");338		hideRows(db, hidden, null);339		recordTouches(db, 0);340		recordFacts(db, null);341		staleStats(db);342	})();343	refreshStats(db);344}345346// Marks the counts out of date in the change's own transaction: if Kepr stops before the recount347// that follows, the mirror's next open does it.348function staleStats(db: Database): void {349	db.run("DELETE FROM state WHERE key = 'stats_rule'");350}351352/**353 * Records which real saves touched each path, for the saves after rowid `from`; 0 starts again.354 * A save touched a path when it was added, changed or removed against the real save before it;355 * checkpoints are skipped, so a change first recorded by a checkpoint counts at the next real save.356 */357function recordTouches(db: Database, from: number): void {358	if (from === 0) db.run("DELETE FROM touches");359	db.query(360		`WITH line AS (361			SELECT id, rowid AS n, LAG(id) OVER (ORDER BY rowid) AS prev362			FROM saves363			WHERE is_checkpoint = 0364		)365		INSERT INTO touches (path, n)366		SELECT f.path, line.n367		FROM line368		JOIN save_files f ON f.save_id = line.id369		LEFT JOIN save_files p ON p.save_id = line.prev AND p.path = f.path370		WHERE line.n > $from371		AND (p.hash IS NOT f.hash OR p.executable IS NOT f.executable)372		UNION ALL373		SELECT p.path, line.n374		FROM line375		JOIN save_files p ON p.save_id = line.prev376		WHERE line.n > $from377		AND NOT EXISTS (SELECT 1 FROM save_files f WHERE f.save_id = line.id AND f.path = p.path)`,378	).run({ from });379	const last = db.query<{ n: number | null }, []>("SELECT MAX(rowid) AS n FROM saves").get()?.n;380	db.query("INSERT OR REPLACE INTO state (key, value) VALUES ('touches_to', ?)").run(381		String(last ?? 0),382	);383}384385/** Records what each file of the given saves, or (null) every save, says about itself, once per version and path. */386function recordFacts(db: Database, saves: string[] | null): void {387	const wanted = db388		.query<{ hash: string; path: string }, { all: number; saves: string }>(389			`SELECT DISTINCT f.hash, f.path390			FROM save_files f391			WHERE ($all = 1 OR f.save_id IN (SELECT value FROM json_each($saves)))392			AND NOT EXISTS (SELECT 1 FROM facts x WHERE x.hash = f.hash AND x.path = f.path)`,393		)394		.all({ all: saves === null ? 1 : 0, saves: JSON.stringify(saves ?? []) });395	const sizeOf = db.query<{ size: number }, [string]>("SELECT size FROM objects WHERE hash = ?");396	const dataOf = db.query<{ data: Uint8Array }, [string]>(397		"SELECT data FROM objects WHERE hash = ?",398	);399	const put = db.query(400		"INSERT OR IGNORE INTO facts (hash, path, lines, description, outline) VALUES (?, ?, ?, ?, ?)",401	);402	for (const { hash, path } of wanted) {403		const size = sizeOf.get(hash)?.size;404		if (size === undefined) continue;405		// The stored bytes are read only for a file small enough to describe.406		const read = () => inflateSync((dataOf.get(hash) as { data: Uint8Array }).data);407		const facts = fileFacts(path, size, read);408		put.run(409			hash,410			path,411			facts.lines,412			facts.description,413			facts.outline && JSON.stringify(facts.outline),414		);415	}416}417418function hiddenList(db: Database): string[] {419	return mirrorState(db, "hidden")?.split("\n").filter(Boolean) ?? [];420}421422// Moves the rows `hidden` matches, in the `saves` given or (null) every save, into `hidden_files`.423// The list is recorded in the same transaction, so the tarball key never names a list the rows424// don't reflect yet.425function hideRows(db: Database, hidden: string[], saves: string[] | null): void {426	for (const entry of hidden) {427		const { where, params } = pathMatch(entry);428		const scope = { ...params, all: saves === null ? 1 : 0, saves: JSON.stringify(saves ?? []) };429		const chosen = `${where} AND ($all = 1 OR save_id IN (SELECT value FROM json_each($saves)))`;430		db.query(431			`INSERT OR IGNORE INTO hidden_files (save_id, path, hash, executable)432			SELECT save_id, path, hash, executable FROM save_files433			WHERE ${chosen}`,434		).run(scope);435		db.query(`DELETE FROM save_files WHERE ${chosen}`).run(scope);436	}437	db.query("INSERT OR REPLACE INTO state (key, value) VALUES ('hidden', ?)").run(hidden.join("\n"));438}439440/** Escaped names of every path this mirror's saves arrived without. */441export function quarantinedPaths(db: Database): string[] {442	return JSON.parse(mirrorState(db, "quarantined") ?? "[]");443}444445// §10.2: a strip of one save reaches that save; a strip of every save bans its path from all of446// history, whenever a save was made or arrives. Re-applying every strip is harmless, which is how a447// save that arrives after its strip event is covered too.448// Returns whether any save lost a file.449function applyStrips(db: Database): boolean {450	const strips = db451		.query<{ save_id: string; path: string }, []>(452			"SELECT save_id, path FROM events WHERE kind = 'strip'",453		)454		.all();455	if (strips.length === 0) return false;456	let removed = 0;457	for (const strip of strips) {458		const { where, params } = pathMatch(strip.path);459		// Hidden rows too, so unhiding never brings back what a strip removed.460		for (const table of ["save_files", "hidden_files"]) {461			removed += db462				.query(`DELETE FROM ${table} WHERE ${where} AND ($save = '' OR save_id = $save)`)463				.run({ ...params, save: strip.save_id }).changes;464		}465	}466	db.run(467		`DELETE FROM objects468		WHERE hash NOT IN (SELECT hash FROM save_files)469		AND hash NOT IN (SELECT hash FROM hidden_files)`,470	);471	db.run("DELETE FROM facts WHERE hash NOT IN (SELECT hash FROM objects)");472	return removed > 0;473}474475/** The SQL condition matching a path, file or folder, or an `ANY_DEPTH` name at any depth. */476function pathMatch(path: string): { where: string; params: Record<string, string> } {477	if (path.startsWith(ANY_DEPTH)) {478		return {479			where: "instr('/' || path || '/', $segment) > 0",480			params: { segment: `/${path.slice(ANY_DEPTH.length)}/` },481		};482	}483	// Koh compares the prefix exactly, never with LIKE, so `_` and `%` in a path match only themselves.484	return {485		where: "(path = $path OR substr(path, 1, length($dir)) = $dir)",486		params: { path, dir: `${path.replace(/\/+$/, "")}/` },487	};488}489490/**491 * Empties a mirror in place so the next sync rebuilds it from the bucket, as the move to schema492 * version 3 needs when hidden rows were deleted. The file and its open handles stay, so a request holding one reads an empty493 * repo for a moment rather than failing. A new epoch keeps tarballs kept before it from matching.494 */495export function clearMirror(db: Database): void {496	db.transaction(() => {497		// Children before parents, for the foreign keys; every name is a table of SCHEMA's.498		const tables = [499			"save_files",500			"hidden_files",501			"touches",502			"facts",503			"known_objects",504			"events",505			"saves",506			"objects",507			"releases",508			"state",509		];510		for (const table of tables) {511			db.run(`DELETE FROM ${table}`);512		}513		db.query("INSERT INTO state (key, value) VALUES ('epoch', ?)").run(crypto.randomUUID());514	})();515}516517// Lines and languages never change between syncs, so they are counted once here, not per view. They518// come from the newest real save, not `head`: a checkpoint's files are not public history (§21.4).519// Lines are source lines: files in a known language, tests left out; docs and data never count.520// Read and written under one write lock: both threads recount, and a recount must never publish521// what it read before another thread's newer one.522function refreshStats(db: Database): void {523	db.transaction(() => recount(db)).immediate();524}525526function recount(db: Database): void {527	const save =528		db529			.query<{ id: string }, []>(530				"SELECT id FROM saves WHERE is_checkpoint = 0 ORDER BY rowid DESC LIMIT 1",531			)532			.get()?.id ?? "";533	const files = db534		.query<{ path: string; size: number; data: Uint8Array }, [string]>(535			"SELECT f.path, o.size, o.data FROM save_files f JOIN objects o ON o.hash = f.hash WHERE f.save_id = ?",536		)537		.all(save);538	let lines = 0;539	const byLanguage = new Map<string, number>();540	for (const file of files) {541		const language = LANGUAGES[file.path.split(".").pop() ?? ""];542		// Tests are not the program: counted, Koh 3's read as 18k lines of V where it has 11k.543		if (!language || VENDORED.test(file.path) || isTestPath(file.path)) continue;544		if (file.size > MAX_COUNTED) continue;545		const bytes = inflateSync(file.data);546		if (bytes.includes(0)) continue;547		let count = 0;548		for (const byte of bytes) if (byte === 10) count++;549		if (bytes.length > 0 && bytes[bytes.length - 1] !== 10) count++;550		lines += count;551		byLanguage.set(language, (byLanguage.get(language) ?? 0) + count);552	}553	const languages = [...byLanguage].sort((a, b) => b[1] - a[1]);554	const setState = db.query("INSERT OR REPLACE INTO state (key, value) VALUES (?, ?)");555	setState.run("lines", String(lines));556	setState.run("language", languages[0]?.[0] ?? "");557	setState.run("languages", JSON.stringify(languages));558	setState.run("stats_save", save);559	// §21.1's "the library knows N things": the file contents any real save refers to, which the560	// front page unites across public repos. From `objects`, probing save_files_hash: a DISTINCT561	// over save_files reads saves × files rows.562	db.run("DELETE FROM known_objects");563	db.run(564		`INSERT INTO known_objects (hash)565		SELECT o.hash566		FROM objects o567		WHERE EXISTS (568			SELECT 1569			FROM save_files f570			JOIN saves s ON s.id = f.save_id571			WHERE f.hash = o.hash572			AND s.is_checkpoint = 0573		)`,574	);575	// Last, with the recount it names, so one cut short is done again at the next open.576	setState.run("stats_rule", STATS_RULE);577}