store.ts

1import { Database } from "bun:sqlite";2import { mkdirSync, rmSync } 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.11const SCHEMA_VERSION = 2;12const SCHEMA = `13CREATE TABLE objects (14	hash TEXT PRIMARY KEY,15	size INTEGER NOT NULL,16	data BLOB NOT NULL17);18CREATE TABLE saves (19	id            TEXT PRIMARY KEY,20	parent        TEXT REFERENCES saves(id),21	message       TEXT NOT NULL,22	created_at    INTEGER NOT NULL,23	machine       TEXT NOT NULL,24	is_checkpoint INTEGER NOT NULL DEFAULT 0,25	size_bytes    INTEGER NOT NULL DEFAULT 0,26	agent         TEXT,27	session       TEXT,28	restores      TEXT,29	replayed_from TEXT30);31CREATE TABLE save_files (32	save_id    TEXT NOT NULL REFERENCES saves(id),33	path       TEXT NOT NULL,34	hash       TEXT NOT NULL REFERENCES objects(hash),35	executable INTEGER NOT NULL DEFAULT 0,36	PRIMARY KEY (save_id, path)37);38CREATE TABLE events (39	id         TEXT PRIMARY KEY,40	kind       TEXT NOT NULL,41	save_id    TEXT,42	path       TEXT,43	reason     TEXT,44	created_at INTEGER NOT NULL,45	machine    TEXT NOT NULL46);47CREATE TABLE state (48	key   TEXT PRIMARY KEY,49	value TEXT NOT NULL50);51CREATE TABLE releases (52	version TEXT PRIMARY KEY,53	record  TEXT NOT NULL54);55CREATE INDEX saves_parent ON saves(parent);56CREATE INDEX save_files_hash ON save_files(hash);57`;5859/** An object as the bucket stores it: zlib bytes, already checked against its hash. */60export type StoredObject = { compressed: Uint8Array; size: number };6162/** What one sync brings in, landed in one transaction. Saves are oldest first. */63export type Landing = {64	head: string;65	releasesLatest: string;66	saves: SaveRecord[];67	objects: Map<string, StoredObject>;68	events: EventRecord[];69	releases: ReleaseRecord[];70	/** Escaped names of the paths these saves arrived without (§5.5, A3). */71	quarantined: string[];72};7374// A recount inflates every file at the newest save; a file this large is data, not lines of code.75const MAX_COUNTED = 8 * 1024 * 1024;7677// Display names for the header's language, by extension; data and docs formats never count.78const LANGUAGES: Record<string, string> = {79	ts: "TypeScript",80	tsx: "TypeScript",81	js: "JavaScript",82	mjs: "JavaScript",83	zig: "Zig",84	v: "V",85	rs: "Rust",86	go: "Go",87	py: "Python",88	c: "C",89	h: "C",90	cpp: "C++",91	cc: "C++",92	hpp: "C++",93	swift: "Swift",94	kt: "Kotlin",95	java: "Java",96	rb: "Ruby",97	lua: "Lua",98	sh: "Shell",99	nix: "Nix",100	css: "CSS",101	html: "HTML",102};103104// Vendored code is someone else's: counting Koh 3's bundled SQLite made it read as 290k lines of C.105const VENDORED = /(^|\/)(vendor|third_party|node_modules)\//;106// Bumped when what counts as a line changes, so every mirror recounts once when it opens.107const STATS_RULE = "source-only";108109const open = new Map<string, Database>();110111/** The mirror of one repo at `<dir>/<repo>.db`, created on first use and kept open. */112export function openMirror(dir: string, repo: string): Database {113	const path = join(dir, `${repo}.db`);114	const cached = open.get(path);115	if (cached) return cached;116	mkdirSync(dir, { recursive: true });117	const db = new Database(path, { strict: true });118	db.run("PRAGMA journal_mode = WAL");119	db.run("PRAGMA foreign_keys = ON");120	db.run("PRAGMA busy_timeout = 5000");121	const { user_version } = db.query<{ user_version: number }, []>("PRAGMA user_version").get() ?? {122		user_version: 0,123	};124	if (user_version !== 0 && user_version !== SCHEMA_VERSION) {125		db.close();126		throw new Error(127			`${path} has mirror schema version ${user_version}; this Kepr reads version ${SCHEMA_VERSION}. ` +128				"The mirror is disposable: delete it and the next sync rebuilds it from the bucket.",129		);130	}131	if (user_version === 0) {132		db.transaction(() => {133			db.run(SCHEMA);134			db.run(`PRAGMA user_version = ${SCHEMA_VERSION}`);135		})();136	}137	// Mirrors counted under an older rule are recounted now, not at the next save.138	if (mirrorState(db, "head") && mirrorState(db, "stats_rule") !== STATS_RULE) refreshStats(db);139	open.set(path, db);140	return db;141}142143export function mirrorState(db: Database, key: string): string | null {144	const row = db145		.query<{ value: string }, [string]>("SELECT value FROM state WHERE key = ?")146		.get(key);147	return row?.value ?? null;148}149150export function knownSave(db: Database, id: string): boolean {151	return db.query("SELECT 1 FROM saves WHERE id = ?").get(id) !== null;152}153154export function knownObject(db: Database, hash: string): boolean {155	return db.query("SELECT 1 FROM objects WHERE hash = ?").get(hash) !== null;156}157158export function knownEventIds(db: Database): Set<string> {159	const rows = db.query<{ id: string }, []>("SELECT id FROM events").all();160	return new Set(rows.map((row) => row.id));161}162163/** Drops a version deleted from the bucket; a sync only ever adds releases. */164export function forgetRelease(db: Database, version: string): void {165	db.query("DELETE FROM releases WHERE version = ?").run(version);166}167168export function knownReleases(db: Database): Set<string> {169	const rows = db.query<{ version: string }, []>("SELECT version FROM releases").all();170	return new Set(rows.map((row) => row.version));171}172173/** Lands a sync like Koh's steal: objects, events, saves, then every strip over every save. */174export function land(db: Database, landing: Landing): void {175	const putObject = db.query("INSERT OR IGNORE INTO objects (hash, size, data) VALUES (?, ?, ?)");176	const putEvent = db.query(177		"INSERT OR IGNORE INTO events (id, kind, save_id, path, reason, created_at, machine) VALUES " +178			"(?, ?, ?, ?, ?, ?, ?)",179	);180	const putSave = db.query(181		`INSERT OR IGNORE INTO saves (id, parent, message, created_at, machine, is_checkpoint, size_bytes,182			agent, session, restores, replayed_from) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,183	);184	const putFile = db.query(185		"INSERT OR IGNORE INTO save_files (save_id, path, hash, executable) VALUES (?, ?, ?, ?)",186	);187	const putRelease = db.query("INSERT OR IGNORE INTO releases (version, record) VALUES (?, ?)");188	const setState = db.query("INSERT OR REPLACE INTO state (key, value) VALUES (?, ?)");189	db.transaction(() => {190		for (const [hash, object] of landing.objects)191			putObject.run(hash, object.size, object.compressed);192		for (const e of landing.events) {193			putEvent.run(e.id, e.kind, e.save_id, e.path, e.reason, e.created_at, e.machine);194		}195		for (const s of landing.saves) {196			putSave.run(197				s.id,198				s.parent === "" ? null : s.parent,199				s.message,200				s.created_at,201				s.machine,202				s.is_checkpoint ? 1 : 0,203				s.size_bytes,204				s.agent,205				s.session,206				s.restores,207				s.replayed_from,208			);209			for (const f of s.files) putFile.run(s.id, f.path, f.hash, f.executable ? 1 : 0);210		}211		for (const r of landing.releases) putRelease.run(r.version, JSON.stringify(r));212		applyStrips(db);213		setState.run("head", landing.head);214		setState.run("releases_latest", landing.releasesLatest);215		const quarantined = new Set([...quarantinedPaths(db), ...landing.quarantined]);216		setState.run("quarantined", JSON.stringify([...quarantined].sort()));217	})();218	// Outside the transaction: the recount can inflate megabytes and needs no lock to be correct.219	refreshStats(db);220}221222/** Escaped names of every path this mirror's saves arrived without. */223export function quarantinedPaths(db: Database): string[] {224	return JSON.parse(mirrorState(db, "quarantined") ?? "[]");225}226227// §10.2: a strip of one save reaches that save; a strip of every save bans its path from all of228// history, whenever a save was made or arrives. Re-applying every strip is harmless, which is how a229// save that arrives after its strip event is covered too.230function applyStrips(db: Database): void {231	const strips = db232		.query<{ save_id: string; path: string }, []>(233			"SELECT save_id, path FROM events WHERE kind = 'strip'",234		)235		.all();236	if (strips.length === 0) return;237	for (const strip of strips) removePath(db, strip.path, strip.save_id);238	db.run("DELETE FROM objects WHERE hash NOT IN (SELECT hash FROM save_files)");239}240241/** Removes a path, file or folder, from one save or ('') every save; returns the rows removed. */242function removePath(db: Database, path: string, save: string): number {243	// Koh compares the prefix exactly, never with LIKE, so `_` and `%` in a path match only themselves.244	const dir = `${path.replace(/\/+$/, "")}/`;245	return db246		.query(247			`DELETE FROM save_files248			WHERE (path = $path OR substr(path, 1, length($dir)) = $dir)249			AND ($save = '' OR save_id = $save)`,250		)251		.run({ path, dir, save }).changes;252}253254/**255 * The operator's hidden paths, removed from every save as a strip would be (June, 2026-09-30).256 * Applied after each sync, so saves arriving later lose them too; unhiding rebuilds the mirror.257 */258export function hidePaths(db: Database, paths: string[]): void {259	let removed = 0;260	db.transaction(() => {261		for (const path of paths) removed += removePath(db, path, "");262		if (removed > 0) db.run("DELETE FROM objects WHERE hash NOT IN (SELECT hash FROM save_files)");263	})();264	if (removed > 0) refreshStats(db);265}266267/** Closes and deletes a repo's mirror, so the next sync rebuilds it from the bucket. */268export function discardMirror(dir: string, repo: string): void {269	const path = join(dir, `${repo}.db`);270	open.get(path)?.close();271	open.delete(path);272	for (const suffix of ["", "-wal", "-shm"]) rmSync(`${path}${suffix}`, { force: true });273}274275// Lines and languages never change between syncs, so they are counted once here, not per view. They276// come from the newest real save, not `head`: a checkpoint's files are not public history (§21.4).277// Lines are source lines: files in a known language, tests left out; docs and data never count.278function refreshStats(db: Database): void {279	const save =280		db281			.query<{ id: string }, []>(282				"SELECT id FROM saves WHERE is_checkpoint = 0 ORDER BY rowid DESC LIMIT 1",283			)284			.get()?.id ?? "";285	const files = db286		.query<{ path: string; size: number; data: Uint8Array }, [string]>(287			"SELECT f.path, o.size, o.data FROM save_files f JOIN objects o ON o.hash = f.hash WHERE f.save_id = ?",288		)289		.all(save);290	let lines = 0;291	const byLanguage = new Map<string, number>();292	for (const file of files) {293		const language = LANGUAGES[file.path.split(".").pop() ?? ""];294		// Tests are not the program: counted, Koh 3's read as 18k lines of V where it has 11k.295		if (!language || VENDORED.test(file.path) || isTestPath(file.path)) continue;296		if (file.size > MAX_COUNTED) continue;297		const bytes = inflateSync(file.data);298		if (bytes.includes(0)) continue;299		let count = 0;300		for (const byte of bytes) if (byte === 10) count++;301		if (bytes.length > 0 && bytes[bytes.length - 1] !== 10) count++;302		lines += count;303		byLanguage.set(language, (byLanguage.get(language) ?? 0) + count);304	}305	const languages = [...byLanguage].sort((a, b) => b[1] - a[1]);306	const setState = db.query("INSERT OR REPLACE INTO state (key, value) VALUES (?, ?)");307	setState.run("lines", String(lines));308	setState.run("language", languages[0]?.[0] ?? "");309	setState.run("languages", JSON.stringify(languages));310	setState.run("stats_save", save);311	setState.run("stats_rule", STATS_RULE);312}