sync.ts

1import type { Database } from "bun:sqlite";2import {3	knownEventIds,4	knownObject,5	knownReleases,6	knownSave,7	land,8	mirrorState,9	openMirror,10	type StoredObject,11} from "../mirror/store.ts";12import { recordSync, registerRepo } from "../server/db.ts";13import { dropTarballs } from "../tarball/cache.ts";14import type { BucketReader } from "./reader.ts";15import {16	BucketError,17	checkFormat,18	type EventRecord,19	listEventIds,20	listReleaseVersions,21	listRepos,22	type ReleaseRecord,23	readEvent,24	readHead,25	readObject,26	readRelease,27	readReleasesLatest,28	readSave,29	type SaveRecord,30} from "./repo.ts";3132const INTERVAL_MS = 2 * 60 * 1000;3334export type SyncTarget = {35	bucket: BucketReader;36	kepr: Database;37	mirrorDir: string;38	tarballDir: string;39};4041let running: Promise<void> | null = null;4243/** Syncs every repo in the bucket; the admin console's "sync now" calls this too. Never overlaps itself. */44export function syncAll(target: SyncTarget): Promise<void> {45	running ??= syncEach(target).finally(() => {46		running = null;47	});48	return running;49}5051export function startSyncLoop(target: SyncTarget): void {52	const tick = () => syncAll(target).catch((error) => console.error(`kepr: sync failed: ${error}`));53	tick();54	setInterval(tick, INTERVAL_MS);55}5657async function syncEach(target: SyncTarget): Promise<void> {58	for (const repo of await listRepos(target.bucket)) {59		registerRepo(target.kepr, repo);60		let error: string | null = null;61		try {62			const events = await syncRepo(target.bucket, openMirror(target.mirrorDir, repo), repo);63			if (events.some((event) => event.kind === "strip")) dropTarballs(target.tarballDir, repo);64		} catch (cause) {65			error = cause instanceof Error ? cause.message : String(cause);66		}67		recordSync(target.kepr, repo, Math.floor(Date.now() / 1000), error);68	}69}7071/**72 * Koh's steal flow (§9.4) against a mirror: walk back from the remote head to a save the mirror has,73 * fetch what is missing, and land it in one transaction. Events are listed every time, since a74 * flag or a strip arrives without moving `head`; releases only when `head` or `releases/latest`75 * moved (§18), since publishing one moves both. Returns the events it landed.76 */77export async function syncRepo(78	bucket: BucketReader,79	mirror: Database,80	repo: string,81): Promise<EventRecord[]> {82	await checkFormat(bucket, repo);83	const head = await readHead(bucket, repo);84	if (head === null) return [];85	const saves: SaveRecord[] = [];86	const objects = new Map<string, StoredObject>();87	const seen = new Set<string>();88	for (let cursor = head; cursor !== "" && !knownSave(mirror, cursor); ) {89		if (seen.has(cursor)) throw new BucketError(`${repo}: save ${cursor} is its own ancestor`);90		seen.add(cursor);91		const save = await readSave(bucket, repo, cursor);92		for (const file of save.files) {93			if (objects.has(file.hash) || knownObject(mirror, file.hash)) continue;94			const { compressed, bytes } = await readObject(bucket, repo, file.hash);95			objects.set(file.hash, { compressed, size: bytes.length });96		}97		saves.push(save);98		cursor = save.parent;99	}100	const known = knownEventIds(mirror);101	const events: EventRecord[] = [];102	for (const id of await listEventIds(bucket, repo)) {103		if (!known.has(id)) events.push(await readEvent(bucket, repo, id));104	}105	const releasesLatest = (await readReleasesLatest(bucket, repo)) ?? "";106	const moved =107		mirrorState(mirror, "head") !== head ||108		mirrorState(mirror, "releases_latest") !== releasesLatest;109	const releases = moved ? await newReleases(bucket, mirror, repo) : [];110	if (!moved && events.length === 0) return [];111	land(mirror, { head, releasesLatest, saves: saves.reverse(), objects, events, releases });112	return events;113}114115async function newReleases(116	bucket: BucketReader,117	mirror: Database,118	repo: string,119): Promise<ReleaseRecord[]> {120	const known = knownReleases(mirror);121	const releases: ReleaseRecord[] = [];122	for (const version of await listReleaseVersions(bucket, repo)) {123		if (!known.has(version)) releases.push(await readRelease(bucket, repo, version));124	}125	return releases;126}