store.ts

1import { Database } from "bun:sqlite";2import { mkdirSync } from "node:fs";3import { join } from "node:path";4import { inflateSync } from "node:zlib";5import type { EventRecord, ReleaseRecord, SaveRecord } from "../bucket/repo.ts";67// Koh 3's schema (§5.1, `src/storage/storage.v`) minus `file_cache`, which only a working tree needs,8// plus `releases`: each version's release.json (§11.2) as read from the bucket, since that is bucket9// data too and pages and feeds need it on every request.10const SCHEMA_VERSION = 2;11const SCHEMA = `12CREATE TABLE objects (13	hash TEXT PRIMARY KEY,14	size INTEGER NOT NULL,15	data BLOB NOT NULL16);17CREATE TABLE saves (18	id            TEXT PRIMARY KEY,19	parent        TEXT REFERENCES saves(id),20	message       TEXT NOT NULL,21	created_at    INTEGER NOT NULL,22	machine       TEXT NOT NULL,23	is_checkpoint INTEGER NOT NULL DEFAULT 0,24	size_bytes    INTEGER NOT NULL DEFAULT 0,25	agent         TEXT,26	session       TEXT,27	restores      TEXT,28	replayed_from TEXT29);30CREATE TABLE save_files (31	save_id    TEXT NOT NULL REFERENCES saves(id),32	path       TEXT NOT NULL,33	hash       TEXT NOT NULL REFERENCES objects(hash),34	executable INTEGER NOT NULL DEFAULT 0,35	PRIMARY KEY (save_id, path)36);37CREATE TABLE events (38	id         TEXT PRIMARY KEY,39	kind       TEXT NOT NULL,40	save_id    TEXT,41	path       TEXT,42	reason     TEXT,43	created_at INTEGER NOT NULL,44	machine    TEXT NOT NULL45);46CREATE TABLE state (47	key   TEXT PRIMARY KEY,48	value TEXT NOT NULL49);50CREATE TABLE releases (51	version TEXT PRIMARY KEY,52	record  TEXT NOT NULL53);54CREATE INDEX saves_parent ON saves(parent);55CREATE INDEX save_files_hash ON save_files(hash);56`;5758/** An object as the bucket stores it: zlib bytes, already checked against its hash. */59export type StoredObject = { compressed: Uint8Array; size: number };6061/** What one sync brings in, landed in one transaction. Saves are oldest first. */62export type Landing = {63	head: string;64	releasesLatest: string;65	saves: SaveRecord[];66	objects: Map<string, StoredObject>;67	events: EventRecord[];68	releases: ReleaseRecord[];69};7071// Display names for the header's language, by extension; data and docs formats never count.72const LANGUAGES: Record<string, string> = {73	ts: "TypeScript",74	tsx: "TypeScript",75	js: "JavaScript",76	mjs: "JavaScript",77	zig: "Zig",78	v: "V",79	rs: "Rust",80	go: "Go",81	py: "Python",82	c: "C",83	h: "C",84	cpp: "C++",85	cc: "C++",86	hpp: "C++",87	swift: "Swift",88	kt: "Kotlin",89	java: "Java",90	rb: "Ruby",91	lua: "Lua",92	sh: "Shell",93	nix: "Nix",94	css: "CSS",95	html: "HTML",96};9798// Vendored code is someone else's: counting Koh 3's bundled SQLite made it read as 290k lines of C.99const VENDORED = /(^|\/)(vendor|third_party|node_modules)\//;100101const open = new Map<string, Database>();102103/** The mirror of one repo at `<dir>/<repo>.db`, created on first use and kept open. */104export function openMirror(dir: string, repo: string): Database {105	const path = join(dir, `${repo}.db`);106	const cached = open.get(path);107	if (cached) return cached;108	mkdirSync(dir, { recursive: true });109	const db = new Database(path, { strict: true });110	db.run("PRAGMA journal_mode = WAL");111	db.run("PRAGMA foreign_keys = ON");112	db.run("PRAGMA busy_timeout = 5000");113	const { user_version } = db.query<{ user_version: number }, []>("PRAGMA user_version").get() ?? {114		user_version: 0,115	};116	if (user_version !== 0 && user_version !== SCHEMA_VERSION) {117		db.close();118		throw new Error(119			`${path} has mirror schema version ${user_version}; this Kepr reads version ${SCHEMA_VERSION}. ` +120				"The mirror is disposable: delete it and the next sync rebuilds it from the bucket.",121		);122	}123	if (user_version === 0) {124		db.transaction(() => {125			db.run(SCHEMA);126			db.run(`PRAGMA user_version = ${SCHEMA_VERSION}`);127		})();128	}129	// Mirrors whose stats predate `stats_save` (per-language lines, counted past checkpoints) are130	// recounted now, not at the next save.131	if (mirrorState(db, "head") && mirrorState(db, "stats_save") === null) refreshStats(db);132	open.set(path, db);133	return db;134}135136export function mirrorState(db: Database, key: string): string | null {137	const row = db138		.query<{ value: string }, [string]>("SELECT value FROM state WHERE key = ?")139		.get(key);140	return row?.value ?? null;141}142143export function knownSave(db: Database, id: string): boolean {144	return db.query("SELECT 1 FROM saves WHERE id = ?").get(id) !== null;145}146147export function knownObject(db: Database, hash: string): boolean {148	return db.query("SELECT 1 FROM objects WHERE hash = ?").get(hash) !== null;149}150151export function knownEventIds(db: Database): Set<string> {152	const rows = db.query<{ id: string }, []>("SELECT id FROM events").all();153	return new Set(rows.map((row) => row.id));154}155156export function knownReleases(db: Database): Set<string> {157	const rows = db.query<{ version: string }, []>("SELECT version FROM releases").all();158	return new Set(rows.map((row) => row.version));159}160161/** Lands a sync like Koh's steal: objects, events, existing-save strips, saves, then all strips. */162export function land(db: Database, landing: Landing): void {163	const putObject = db.query("INSERT OR IGNORE INTO objects (hash, size, data) VALUES (?, ?, ?)");164	const putEvent = db.query(165		"INSERT OR IGNORE INTO events (id, kind, save_id, path, reason, created_at, machine) VALUES " +166			"(?, ?, ?, ?, ?, ?, ?)",167	);168	const putSave = db.query(169		`INSERT OR IGNORE INTO saves (id, parent, message, created_at, machine, is_checkpoint, size_bytes,170			agent, session, restores, replayed_from) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,171	);172	const putFile = db.query(173		"INSERT OR IGNORE INTO save_files (save_id, path, hash, executable) VALUES (?, ?, ?, ?)",174	);175	const putRelease = db.query("INSERT OR IGNORE INTO releases (version, record) VALUES (?, ?)");176	const setState = db.query("INSERT OR REPLACE INTO state (key, value) VALUES (?, ?)");177	db.transaction(() => {178		for (const [hash, object] of landing.objects)179			putObject.run(hash, object.size, object.compressed);180		for (const e of landing.events) {181			putEvent.run(e.id, e.kind, e.save_id, e.path, e.reason, e.created_at, e.machine);182		}183		applyNewStrips(db, landing.events);184		for (const s of landing.saves) {185			putSave.run(186				s.id,187				s.parent === "" ? null : s.parent,188				s.message,189				s.created_at,190				s.machine,191				s.is_checkpoint ? 1 : 0,192				s.size_bytes,193				s.agent,194				s.session,195				s.restores,196				s.replayed_from,197			);198			for (const f of s.files) putFile.run(s.id, f.path, f.hash, f.executable ? 1 : 0);199		}200		for (const r of landing.releases) putRelease.run(r.version, JSON.stringify(r));201		applyStrips(db);202		setState.run("head", landing.head);203		setState.run("releases_latest", landing.releasesLatest);204		refreshStats(db);205	})();206}207208function applyNewStrips(db: Database, events: EventRecord[]): void {209	const strips = events.filter((event) => event.kind === "strip");210	if (strips.length === 0) return;211	const remove = db.query(212		`DELETE FROM save_files213		WHERE (path = $path OR substr(path, 1, length($dir)) = $dir)214		AND save_id IN (215			SELECT id FROM saves216			WHERE CASE WHEN $save = '' THEN 1 ELSE id = $save END217		)`,218	);219	for (const strip of strips) {220		const path = strip.path ?? "";221		const dir = `${path.replace(/\/+$/, "")}/`;222		remove.run({ path, dir, save: strip.save_id });223	}224}225226// §10.2: a strip of one save reaches that save; a strip of every save reaches the saves made up to227// and including its own time, so a file saved again after the strip survives. Re-applying every228// strip is harmless, which lets a save that arrives after its strip event be covered too.229function applyStrips(db: Database): void {230	const strips = db231		.query<{ save_id: string; path: string; created_at: number }, []>(232			"SELECT save_id, path, created_at FROM events WHERE kind = 'strip'",233		)234		.all();235	if (strips.length === 0) return;236	const remove = db.query(237		`DELETE FROM save_files238		WHERE (path = $path OR substr(path, 1, length($dir)) = $dir)239		AND save_id IN (SELECT id FROM saves WHERE240			CASE WHEN $save = '' THEN created_at <= $at ELSE id = $save END)`,241	);242	for (const strip of strips) {243		// Koh compares the prefix exactly, never with LIKE, so `_` and `%` in a path match only themselves.244		const dir = `${strip.path.replace(/\/+$/, "")}/`;245		remove.run({ path: strip.path, dir, save: strip.save_id, at: strip.created_at });246	}247	db.run("DELETE FROM objects WHERE hash NOT IN (SELECT hash FROM save_files)");248}249250// Lines and languages never change between syncs, so they are counted once here, not per view. They251// come from the newest real save, not `head`: a checkpoint's files are not public history (§21.4).252function refreshStats(db: Database): void {253	const save =254		db255			.query<{ id: string }, []>(256				"SELECT id FROM saves WHERE is_checkpoint = 0 ORDER BY rowid DESC LIMIT 1",257			)258			.get()?.id ?? "";259	const files = db260		.query<{ path: string; data: Uint8Array }, [string]>(261			"SELECT f.path, o.data FROM save_files f JOIN objects o ON o.hash = f.hash WHERE f.save_id = ?",262		)263		.all(save);264	let lines = 0;265	const byLanguage = new Map<string, number>();266	for (const file of files) {267		if (VENDORED.test(file.path)) continue;268		const bytes = inflateSync(file.data);269		if (bytes.includes(0)) continue;270		let count = 0;271		for (const byte of bytes) if (byte === 10) count++;272		if (bytes.length > 0 && bytes[bytes.length - 1] !== 10) count++;273		lines += count;274		const language = LANGUAGES[file.path.split(".").pop() ?? ""];275		if (language) byLanguage.set(language, (byLanguage.get(language) ?? 0) + count);276	}277	const languages = [...byLanguage].sort((a, b) => b[1] - a[1]);278	const setState = db.query("INSERT OR REPLACE INTO state (key, value) VALUES (?, ?)");279	setState.run("lines", String(lines));280	setState.run("language", languages[0]?.[0] ?? "");281	setState.run("languages", JSON.stringify(languages));282	setState.run("stats_save", save);283}