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";8import { declaredVersion } from "../describe/version.ts";910// Koh 3's schema (§5.1, `src/storage/storage.v`) minus `file_cache`, which only a working tree needs,11// plus `releases`: each version's release.json (§11.2) as read from the bucket, since that is bucket12// data too and pages and feeds need it on every request, `hidden_files`, and what the Atlas reads.13const SCHEMA_VERSION = 4;14// The operator's hidden rows, moved out of `save_files` rather than deleted, so a changed hidden15// list is applied in place, with nothing fetched again (2026-10-02).16const HIDDEN_FILES = `17CREATE TABLE hidden_files (18	save_id    TEXT NOT NULL REFERENCES saves(id),19	path       TEXT NOT NULL,20	hash       TEXT NOT NULL REFERENCES objects(hash),21	executable INTEGER NOT NULL DEFAULT 0,22	PRIMARY KEY (save_id, path)23);24`;25// What pages read, worked out as each save lands rather than on every view (2026-10-02): which26// real saves touched each path (added, changed or removed against the real save before; `n` is the27// touching save's rowid), what each file says about itself, and the objects real saves refer to.28const DERIVED = `29CREATE TABLE known_objects (hash TEXT PRIMARY KEY) WITHOUT ROWID;30CREATE TABLE touches (31	path TEXT NOT NULL,32	n    INTEGER NOT NULL33);34CREATE INDEX touches_n ON touches(n);35CREATE TABLE facts (36	hash        TEXT NOT NULL,37	path        TEXT NOT NULL,38	lines       INTEGER,39	description TEXT,40	outline     TEXT,41	PRIMARY KEY (hash, path)42);43`;44const SCHEMA = `45CREATE TABLE objects (46	hash TEXT PRIMARY KEY,47	size INTEGER NOT NULL,48	data BLOB NOT NULL49);50CREATE TABLE saves (51	id            TEXT PRIMARY KEY,52	parent        TEXT REFERENCES saves(id),53	message       TEXT NOT NULL,54	created_at    INTEGER NOT NULL,55	machine       TEXT NOT NULL,56	is_checkpoint INTEGER NOT NULL DEFAULT 0,57	size_bytes    INTEGER NOT NULL DEFAULT 0,58	agent         TEXT,59	session       TEXT,60	restores      TEXT,61	replayed_from TEXT62);63CREATE TABLE save_files (64	save_id    TEXT NOT NULL REFERENCES saves(id),65	path       TEXT NOT NULL,66	hash       TEXT NOT NULL REFERENCES objects(hash),67	executable INTEGER NOT NULL DEFAULT 0,68	PRIMARY KEY (save_id, path)69);70CREATE TABLE events (71	id         TEXT PRIMARY KEY,72	kind       TEXT NOT NULL,73	save_id    TEXT,74	path       TEXT,75	reason     TEXT,76	created_at INTEGER NOT NULL,77	machine    TEXT NOT NULL78);79CREATE TABLE state (80	key   TEXT PRIMARY KEY,81	value TEXT NOT NULL82);83CREATE TABLE releases (84	version TEXT PRIMARY KEY,85	record  TEXT NOT NULL86);87CREATE INDEX saves_parent ON saves(parent);88CREATE INDEX save_files_hash ON save_files(hash);89${HIDDEN_FILES}90${DERIVED}`;9192/** An object as the bucket stores it: zlib bytes, already checked against its hash. */93export type StoredObject = { compressed: Uint8Array; size: number };9495/** One batch of a sync, landed in one transaction. Saves are oldest first. */96export type Landing = {97	head: string;98	releasesLatest: string;99	saves: SaveRecord[];100	objects: Map<string, StoredObject>;101	events: EventRecord[];102	releases: ReleaseRecord[];103	/** Escaped names of the paths these saves arrived without (§5.5, A3). */104	quarantined: string[];105};106107/** Marks a hidden entry as a name matched at any depth rather than a path from the root. */108export const ANY_DEPTH = "**/";109110// No project declares its version in a file larger than this.111const MAX_MANIFEST = 1024 * 1024;112113// A recount inflates every file at the newest save; a file this large is data, not lines of code.114const MAX_COUNTED = 8 * 1024 * 1024;115116// Display names for the header's language, by extension; data and docs formats never count.117const LANGUAGES: Record<string, string> = {118	ts: "TypeScript",119	tsx: "TypeScript",120	js: "JavaScript",121	mjs: "JavaScript",122	zig: "Zig",123	v: "V",124	rs: "Rust",125	go: "Go",126	py: "Python",127	c: "C",128	h: "C",129	cpp: "C++",130	cc: "C++",131	hpp: "C++",132	swift: "Swift",133	kt: "Kotlin",134	java: "Java",135	rb: "Ruby",136	lua: "Lua",137	sh: "Shell",138	nix: "Nix",139	css: "CSS",140	html: "HTML",141};142143// Vendored code is someone else's: counting Koh 3's bundled SQLite made it read as 290k lines of C.144const VENDORED = /(^|\/)(vendor|third_party|node_modules)\//;145// Bumped when what the recount records changes, so every mirror recounts once when it opens.146const STATS_RULE = "source-only, known objects, declared version (ASCII whitespace, BOM)";147148const open = new Map<string, Database>();149150/** Closes a repo's mirror, so a rename can move its file. */151export function closeMirror(dir: string, repo: string): void {152	const path = join(dir, `${repo}.db`);153	open.get(path)?.close();154	open.delete(path);155}156157/**158 * Opens every mirror in `dir`, bringing each to this Kepr's schema: run at start, before Kepr159 * listens, so no request waits on a migration.160 */161export function openMirrors(dir: string): void {162	if (!existsSync(dir)) return;163	for (const name of readdirSync(dir)) if (name.endsWith(".db")) openMirror(dir, name.slice(0, -3));164}165166/** Closes every mirror this thread holds open; the next use of each opens it again. */167export function closeMirrors(): void {168	for (const db of open.values()) db.close();169	open.clear();170}171172/** The mirror of one repo at `<dir>/<repo>.db`, created on first use and kept open. */173export function openMirror(dir: string, repo: string): Database {174	const path = join(dir, `${repo}.db`);175	const cached = open.get(path);176	if (cached) return cached;177	mkdirSync(dir, { recursive: true });178	const db = new Database(path, { strict: true });179	db.run("PRAGMA journal_mode = WAL");180	db.run("PRAGMA foreign_keys = ON");181	db.run("PRAGMA busy_timeout = 5000");182	// A current mirror needs no lock; only creating or migrating one takes the write lock.183	const found = schemaVersion(db) === SCHEMA_VERSION ? SCHEMA_VERSION : upgrade(db);184	if (found !== 0 && found !== 2 && found !== 3 && found !== SCHEMA_VERSION) {185		db.close();186		throw new Error(187			`${path} has mirror schema version ${found}; this Kepr reads version ${SCHEMA_VERSION}. ` +188				"The mirror is disposable: delete it and the next sync rebuilds it from the bucket.",189		);190	}191	// Mirrors counted under an older rule are recounted now, not at the next save.192	if (mirrorState(db, "head") && mirrorState(db, "stats_rule") !== STATS_RULE) refreshStats(db);193	open.set(path, db);194	return db;195}196197function schemaVersion(db: Database): number {198	return db.query<{ user_version: number }, []>("PRAGMA user_version").get()?.user_version ?? 0;199}200201/**202 * Creates or migrates a mirror and returns the version it found, read again under the write lock:203 * the request thread and the sync's both open mirrors, and must not both create or migrate one.204 */205function upgrade(db: Database): number {206	return db207		.transaction(() => {208			const found = schemaVersion(db);209			if (found === SCHEMA_VERSION || ![0, 2, 3].includes(found)) return found;210			if (found === 0) db.run(SCHEMA);211			if (found === 2) db.run(HIDDEN_FILES);212			if (found === 2 || found === 3) db.run(DERIVED);213			// Version 2 deleted hidden rows, so only the bucket can bring them back on unhide.214			if (found === 2 && mirrorState(db, "hidden")) clearMirror(db);215			else if (found !== 0) {216				recordTouches(db, 0);217				recordFacts(db, null);218			}219			db.run(`PRAGMA user_version = ${SCHEMA_VERSION}`);220			return found;221		})222		.immediate();223}224225export function mirrorState(db: Database, key: string): string | null {226	const row = db227		.query<{ value: string }, [string]>("SELECT value FROM state WHERE key = ?")228		.get(key);229	return row?.value ?? null;230}231232export function knownSave(db: Database, id: string): boolean {233	return db.query("SELECT 1 FROM saves WHERE id = ?").get(id) !== null;234}235236export function knownObject(db: Database, hash: string): boolean {237	return db.query("SELECT 1 FROM objects WHERE hash = ?").get(hash) !== null;238}239240export function knownEventIds(db: Database): Set<string> {241	const rows = db.query<{ id: string }, []>("SELECT id FROM events").all();242	return new Set(rows.map((row) => row.id));243}244245/** Drops a version deleted from the bucket; a sync only ever adds releases. */246export function forgetRelease(db: Database, version: string): void {247	db.query("DELETE FROM releases WHERE version = ?").run(version);248}249250/** Replaces a version's record after Kepr rewrote it in the bucket. */251export function replaceRelease(db: Database, record: ReleaseRecord): void {252	db.query("UPDATE releases SET record = ? WHERE version = ?").run(253		JSON.stringify(record),254		record.version,255	);256}257258export function knownReleases(db: Database): Set<string> {259	const rows = db.query<{ version: string }, []>("SELECT version FROM releases").all();260	return new Set(rows.map((row) => row.version));261}262263/** Stores objects on their own, ahead of the save that refers to them. */264export function storeObjects(db: Database, objects: Map<string, StoredObject>): void {265	const put = db.query("INSERT OR IGNORE INTO objects (hash, size, data) VALUES (?, ?, ?)");266	db.transaction(() => {267		for (const [hash, object] of objects) put.run(hash, object.size, object.compressed);268	})();269}270271/**272 * Lands a sync like Koh's steal: objects, events, saves, then every strip over every save. The new273 * saves are hidden by the list the mirror holds; `applyHidden` changes that list.274 */275export function land(db: Database, landing: Landing): void {276	const putEvent = db.query(277		"INSERT OR IGNORE INTO events (id, kind, save_id, path, reason, created_at, machine) VALUES " +278			"(?, ?, ?, ?, ?, ?, ?)",279	);280	const putSave = db.query(281		`INSERT OR IGNORE INTO saves (id, parent, message, created_at, machine, is_checkpoint, size_bytes,282			agent, session, restores, replayed_from) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,283	);284	const putFile = db.query(285		"INSERT OR IGNORE INTO save_files (save_id, path, hash, executable) VALUES (?, ?, ?, ?)",286	);287	const putRelease = db.query("INSERT OR IGNORE INTO releases (version, record) VALUES (?, ?)");288	const setState = db.query("INSERT OR REPLACE INTO state (key, value) VALUES (?, ?)");289	// Immediate: a landing with nothing new reads before it writes, and the request thread writes290	// mirrors too (a deleted release).291	db.transaction(() => {292		storeObjects(db, landing.objects);293		for (const e of landing.events) {294			putEvent.run(e.id, e.kind, e.save_id, e.path, e.reason, e.created_at, e.machine);295		}296		for (const s of landing.saves) {297			putSave.run(298				s.id,299				s.parent === "" ? null : s.parent,300				s.message,301				s.created_at,302				s.machine,303				s.is_checkpoint ? 1 : 0,304				s.size_bytes,305				s.agent,306				s.session,307				s.restores,308				s.replayed_from,309			);310			for (const f of s.files) putFile.run(s.id, f.path, f.hash, f.executable ? 1 : 0);311		}312		for (const r of landing.releases) putRelease.run(r.version, JSON.stringify(r));313		const stripped = applyStrips(db);314		const landed = landing.saves.map((save) => save.id);315		hideRows(db, hiddenList(db), landed);316		// A strip can reach any save, so every save's touches are worked out again after one.317		recordTouches(db, stripped ? 0 : Number(mirrorState(db, "touches_to") ?? 0));318		recordFacts(db, landed);319		setState.run("head", landing.head);320		setState.run("releases_latest", landing.releasesLatest);321		const quarantined = new Set([...quarantinedPaths(db), ...landing.quarantined]);322		setState.run("quarantined", JSON.stringify([...quarantined].sort()));323		staleStats(db);324	}).immediate();325	// Outside the transaction: the recount can inflate megabytes and needs no lock to be correct.326	refreshStats(db);327}328329/**330 * Makes `hidden` the mirror's hidden list: the operator's hidden paths (June, 2026-09-30) and the331 * names ignored in every repo (June, 2026-10-01), each as `ANY_DEPTH` + name. Rows move between332 * `save_files` and `hidden_files`, so it takes milliseconds and fetches nothing.333 */334export function applyHidden(db: Database, hidden: string[]): void {335	if (hiddenList(db).join("\n") === hidden.join("\n")) return;336	db.transaction(() => {337		db.run(338			`INSERT OR IGNORE INTO save_files (save_id, path, hash, executable)339			SELECT save_id, path, hash, executable FROM hidden_files`,340		);341		db.run("DELETE FROM hidden_files");342		hideRows(db, hidden, null);343		recordTouches(db, 0);344		recordFacts(db, null);345		staleStats(db);346	})();347	refreshStats(db);348}349350// Marks the counts out of date in the change's own transaction: if Kepr stops before the recount351// that follows, the mirror's next open does it.352function staleStats(db: Database): void {353	db.run("DELETE FROM state WHERE key = 'stats_rule'");354}355356/**357 * Records which real saves touched each path, for the saves after rowid `from`; 0 starts again.358 * A save touched a path when it was added, changed or removed against the real save before it;359 * checkpoints are skipped, so a change first recorded by a checkpoint counts at the next real save.360 */361function recordTouches(db: Database, from: number): void {362	if (from === 0) db.run("DELETE FROM touches");363	db.query(364		`WITH line AS (365			SELECT id, rowid AS n, LAG(id) OVER (ORDER BY rowid) AS prev366			FROM saves367			WHERE is_checkpoint = 0368		)369		INSERT INTO touches (path, n)370		SELECT f.path, line.n371		FROM line372		JOIN save_files f ON f.save_id = line.id373		LEFT JOIN save_files p ON p.save_id = line.prev AND p.path = f.path374		WHERE line.n > $from375		AND (p.hash IS NOT f.hash OR p.executable IS NOT f.executable)376		UNION ALL377		SELECT p.path, line.n378		FROM line379		JOIN save_files p ON p.save_id = line.prev380		WHERE line.n > $from381		AND NOT EXISTS (SELECT 1 FROM save_files f WHERE f.save_id = line.id AND f.path = p.path)`,382	).run({ from });383	const last = db.query<{ n: number | null }, []>("SELECT MAX(rowid) AS n FROM saves").get()?.n;384	db.query("INSERT OR REPLACE INTO state (key, value) VALUES ('touches_to', ?)").run(385		String(last ?? 0),386	);387}388389/** Records what each file of the given saves, or (null) every save, says about itself, once per version and path. */390function recordFacts(db: Database, saves: string[] | null): void {391	const wanted = db392		.query<{ hash: string; path: string }, { all: number; saves: string }>(393			`SELECT DISTINCT f.hash, f.path394			FROM save_files f395			WHERE ($all = 1 OR f.save_id IN (SELECT value FROM json_each($saves)))396			AND NOT EXISTS (SELECT 1 FROM facts x WHERE x.hash = f.hash AND x.path = f.path)`,397		)398		.all({ all: saves === null ? 1 : 0, saves: JSON.stringify(saves ?? []) });399	const sizeOf = db.query<{ size: number }, [string]>("SELECT size FROM objects WHERE hash = ?");400	const dataOf = db.query<{ data: Uint8Array }, [string]>(401		"SELECT data FROM objects WHERE hash = ?",402	);403	const put = db.query(404		"INSERT OR IGNORE INTO facts (hash, path, lines, description, outline) VALUES (?, ?, ?, ?, ?)",405	);406	for (const { hash, path } of wanted) {407		const size = sizeOf.get(hash)?.size;408		if (size === undefined) continue;409		// The stored bytes are read only for a file small enough to describe.410		const read = () => inflateSync((dataOf.get(hash) as { data: Uint8Array }).data);411		const facts = fileFacts(path, size, read);412		put.run(413			hash,414			path,415			facts.lines,416			facts.description,417			facts.outline && JSON.stringify(facts.outline),418		);419	}420}421422function hiddenList(db: Database): string[] {423	return mirrorState(db, "hidden")?.split("\n").filter(Boolean) ?? [];424}425426// Moves the rows `hidden` matches, in the `saves` given or (null) every save, into `hidden_files`.427// The list is recorded in the same transaction, so the tarball key never names a list the rows428// don't reflect yet.429function hideRows(db: Database, hidden: string[], saves: string[] | null): void {430	for (const entry of hidden) {431		const { where, params } = pathMatch(entry);432		const scope = { ...params, all: saves === null ? 1 : 0, saves: JSON.stringify(saves ?? []) };433		const chosen = `${where} AND ($all = 1 OR save_id IN (SELECT value FROM json_each($saves)))`;434		db.query(435			`INSERT OR IGNORE INTO hidden_files (save_id, path, hash, executable)436			SELECT save_id, path, hash, executable FROM save_files437			WHERE ${chosen}`,438		).run(scope);439		db.query(`DELETE FROM save_files WHERE ${chosen}`).run(scope);440	}441	db.query("INSERT OR REPLACE INTO state (key, value) VALUES ('hidden', ?)").run(hidden.join("\n"));442}443444/** Escaped names of every path this mirror's saves arrived without. */445export function quarantinedPaths(db: Database): string[] {446	return JSON.parse(mirrorState(db, "quarantined") ?? "[]");447}448449// §10.2: a strip of one save reaches that save; a strip of every save bans its path from all of450// history, whenever a save was made or arrives. Re-applying every strip is harmless, which is how a451// save that arrives after its strip event is covered too.452// Returns whether any save lost a file.453function applyStrips(db: Database): boolean {454	const strips = db455		.query<{ save_id: string; path: string }, []>(456			"SELECT save_id, path FROM events WHERE kind = 'strip'",457		)458		.all();459	if (strips.length === 0) return false;460	let removed = 0;461	for (const strip of strips) {462		const { where, params } = pathMatch(strip.path);463		// Hidden rows too, so unhiding never brings back what a strip removed.464		for (const table of ["save_files", "hidden_files"]) {465			removed += db466				.query(`DELETE FROM ${table} WHERE ${where} AND ($save = '' OR save_id = $save)`)467				.run({ ...params, save: strip.save_id }).changes;468		}469	}470	db.run(471		`DELETE FROM objects472		WHERE hash NOT IN (SELECT hash FROM save_files)473		AND hash NOT IN (SELECT hash FROM hidden_files)`,474	);475	db.run("DELETE FROM facts WHERE hash NOT IN (SELECT hash FROM objects)");476	return removed > 0;477}478479/** The SQL condition matching a path, file or folder, or an `ANY_DEPTH` name at any depth. */480function pathMatch(path: string): { where: string; params: Record<string, string> } {481	if (path.startsWith(ANY_DEPTH)) {482		return {483			where: "instr('/' || path || '/', $segment) > 0",484			params: { segment: `/${path.slice(ANY_DEPTH.length)}/` },485		};486	}487	// Koh compares the prefix exactly, never with LIKE, so `_` and `%` in a path match only themselves.488	return {489		where: "(path = $path OR substr(path, 1, length($dir)) = $dir)",490		params: { path, dir: `${path.replace(/\/+$/, "")}/` },491	};492}493494/**495 * Empties a mirror in place so the next sync rebuilds it from the bucket, as the move to schema496 * version 3 needs when hidden rows were deleted. The file and its open handles stay, so a request holding one reads an empty497 * repo for a moment rather than failing. A new epoch keeps tarballs kept before it from matching.498 */499export function clearMirror(db: Database): void {500	db.transaction(() => {501		// Children before parents, for the foreign keys; every name is a table of SCHEMA's.502		const tables = [503			"save_files",504			"hidden_files",505			"touches",506			"facts",507			"known_objects",508			"events",509			"saves",510			"objects",511			"releases",512			"state",513		];514		for (const table of tables) {515			db.run(`DELETE FROM ${table}`);516		}517		db.query("INSERT INTO state (key, value) VALUES ('epoch', ?)").run(crypto.randomUUID());518	})();519}520521// A file at a save as text, if it is there and small enough to be a manifest.522function smallFileText(db: Database, save: string, path: string): string | null {523	const row = db524		.query<{ size: number; data: Uint8Array }, [string, string]>(525			`SELECT o.size, o.data526			FROM save_files f527			JOIN objects o ON o.hash = f.hash528			WHERE f.save_id = ?529			AND f.path = ?`,530		)531		.get(save, path);532	return row && row.size <= MAX_MANIFEST533		? new TextDecoder("utf-8", { ignoreBOM: true }).decode(inflateSync(row.data))534		: null;535}536537// Lines and languages never change between syncs, so they are counted once here, not per view. They538// come from the newest real save, not `head`: a checkpoint's files are not public history (§21.4).539// Lines are source lines: files in a known language, tests left out; docs and data never count.540// Read and written under one write lock: both threads recount, and a recount must never publish541// what it read before another thread's newer one.542function refreshStats(db: Database): void {543	db.transaction(() => recount(db)).immediate();544}545546function recount(db: Database): void {547	const save =548		db549			.query<{ id: string }, []>(550				"SELECT id FROM saves WHERE is_checkpoint = 0 ORDER BY rowid DESC LIMIT 1",551			)552			.get()?.id ?? "";553	const files = db554		.query<{ path: string; size: number; data: Uint8Array }, [string]>(555			"SELECT f.path, o.size, o.data FROM save_files f JOIN objects o ON o.hash = f.hash WHERE f.save_id = ?",556		)557		.all(save);558	let lines = 0;559	const byLanguage = new Map<string, number>();560	for (const file of files) {561		const language = LANGUAGES[file.path.split(".").pop() ?? ""];562		// Tests are not the program: counted, Koh 3's read as 18k lines of V where it has 11k.563		if (!language || VENDORED.test(file.path) || isTestPath(file.path)) continue;564		if (file.size > MAX_COUNTED) continue;565		const bytes = inflateSync(file.data);566		if (bytes.includes(0)) continue;567		let count = 0;568		for (const byte of bytes) if (byte === 10) count++;569		if (bytes.length > 0 && bytes[bytes.length - 1] !== 10) count++;570		lines += count;571		byLanguage.set(language, (byLanguage.get(language) ?? 0) + count);572	}573	const languages = [...byLanguage].sort((a, b) => b[1] - a[1]);574	const setState = db.query("INSERT OR REPLACE INTO state (key, value) VALUES (?, ?)");575	setState.run("lines", String(lines));576	setState.run("language", languages[0]?.[0] ?? "");577	setState.run("languages", JSON.stringify(languages));578	setState.run("stats_save", save);579	setState.run("declared_version", declaredVersion((path) => smallFileText(db, save, path)) ?? "");580	// §21.1's "the library knows N things": the file contents any real save refers to, which the581	// front page unites across public repos. From `objects`, probing save_files_hash: a DISTINCT582	// over save_files reads saves × files rows.583	db.run("DELETE FROM known_objects");584	db.run(585		`INSERT INTO known_objects (hash)586		SELECT o.hash587		FROM objects o588		WHERE EXISTS (589			SELECT 1590			FROM save_files f591			JOIN saves s ON s.id = f.save_id592			WHERE f.hash = o.hash593			AND s.is_checkpoint = 0594		)`,595	);596	// Last, with the recount it names, so one cut short is done again at the next open.597	setState.run("stats_rule", STATS_RULE);598}