sync.ts

1import type { Database } from "bun:sqlite";2import {3	knownEventIds,4	knownObject,5	knownSave,6	land,7	mirrorState,8	openMirror,9	type StoredObject,10} from "../mirror/store.ts";11import { recordSync, registerRepo } from "../server/db.ts";12import type { BucketReader } from "./reader.ts";13import {14	BucketError,15	checkFormat,16	type EventRecord,17	listEventIds,18	listRepos,19	readEvent,20	readHead,21	readObject,22	readSave,23	type SaveRecord,24} from "./repo.ts";2526const INTERVAL_MS = 2 * 60 * 1000;2728export type SyncTarget = { bucket: BucketReader; kepr: Database; mirrorDir: string };2930let running: Promise<void> | null = null;3132/** Syncs every repo in the bucket; the admin console's "sync now" calls this too. Never overlaps itself. */33export function syncAll(target: SyncTarget): Promise<void> {34	running ??= syncEach(target).finally(() => {35		running = null;36	});37	return running;38}3940export function startSyncLoop(target: SyncTarget): void {41	const tick = () => syncAll(target).catch((error) => console.error(`kepr: sync failed: ${error}`));42	tick();43	setInterval(tick, INTERVAL_MS);44}4546async function syncEach(target: SyncTarget): Promise<void> {47	for (const repo of await listRepos(target.bucket)) {48		registerRepo(target.kepr, repo);49		let error: string | null = null;50		try {51			await syncRepo(target.bucket, openMirror(target.mirrorDir, repo), repo);52		} catch (cause) {53			error = cause instanceof Error ? cause.message : String(cause);54		}55		recordSync(target.kepr, repo, Math.floor(Date.now() / 1000), error);56	}57}5859/**60 * Koh's steal flow (§9.4) against a mirror: walk back from the remote head to a save the mirror has,61 * fetch what is missing, and land it in one transaction. Events are listed every time, since a62 * flag or a strip arrives without moving `head`.63 */64export async function syncRepo(65	bucket: BucketReader,66	mirror: Database,67	repo: string,68): Promise<void> {69	await checkFormat(bucket, repo);70	const head = await readHead(bucket, repo);71	if (head === null) return;72	const saves: SaveRecord[] = [];73	const objects = new Map<string, StoredObject>();74	const seen = new Set<string>();75	for (let cursor = head; cursor !== "" && !knownSave(mirror, cursor); ) {76		if (seen.has(cursor)) throw new BucketError(`${repo}: save ${cursor} is its own ancestor`);77		seen.add(cursor);78		const save = await readSave(bucket, repo, cursor);79		for (const file of save.files) {80			if (objects.has(file.hash) || knownObject(mirror, file.hash)) continue;81			const { compressed, bytes } = await readObject(bucket, repo, file.hash);82			objects.set(file.hash, { compressed, size: bytes.length });83		}84		saves.push(save);85		cursor = save.parent;86	}87	const known = knownEventIds(mirror);88	const events: EventRecord[] = [];89	for (const id of await listEventIds(bucket, repo)) {90		if (!known.has(id)) events.push(await readEvent(bucket, repo, id));91	}92	if (saves.length === 0 && events.length === 0 && mirrorState(mirror, "head") === head) return;93	land(mirror, { head, saves: saves.reverse(), objects, events });94}