sync.ts

1import type { Database } from "bun:sqlite";2import {3	clearMirror,4	knownEventIds,5	knownObject,6	knownReleases,7	knownSave,8	land,9	mirrorState,10	openMirror,11	type StoredObject,12} from "../mirror/store.ts";13import { hiddenPaths, recordSync, registerRepo } from "../server/db.ts";14import { RESERVED_NAMES } from "../server/reserved.ts";15import { dropTarballs } from "../tarball/cache.ts";16import type { Bucket } from "./reader.ts";17import {18	BucketError,19	checkFormat,20	type EventRecord,21	escapePath,22	isSafePath,23	listEventIds,24	listReleaseVersions,25	listRepos,26	OversizedObject,27	type ReleaseRecord,28	readEvent,29	readHead,30	readObject,31	readRelease,32	readReleasesLatest,33	readSave,34	type SaveRecord,35} from "./repo.ts";3637const INTERVAL_MS = 2 * 60 * 1000;3839export type SyncTarget = {40	bucket: Bucket;41	kepr: Database;42	mirrorDir: string;43};4445let running: Promise<void> | null = null;4647/** Syncs every repo in the bucket; the admin console's "sync now" calls this too. Never overlaps itself. */48export function syncAll(target: SyncTarget): Promise<void> {49	running ??= syncEach(target).finally(() => {50		running = null;51	});52	return running;53}5455/** Rebuilds one repo's mirror from the bucket, once any sync under way has finished. */56export async function rebuildRepo(target: SyncTarget, repo: string): Promise<void> {57	await syncAll(target);58	clearMirror(openMirror(target.mirrorDir, repo));59	await syncAll(target);60}6162export function startSyncLoop(target: SyncTarget): void {63	const tick = () => syncAll(target).catch((error) => console.error(`kepr: sync failed: ${error}`));64	tick();65	setInterval(tick, INTERVAL_MS);66}6768async function syncEach(target: SyncTarget): Promise<void> {69	for (const repo of await listRepos(target.bucket)) {70		registerRepo(target.kepr, repo);71		let error: string | null = null;72		if (RESERVED_NAMES.has(repo)) {73			error = `not synced: the name "${repo}" clashes with Kepr's own /${repo} route; rename the repo`;74		} else {75			try {76				const mirror = openMirror(target.mirrorDir, repo);77				const events = await syncRepo(target.bucket, mirror, repo, hiddenPaths(target.kepr, repo));78				if (events.some((event) => event.kind === "strip")) await dropTarballs(target.bucket, repo);79			} catch (cause) {80				error = cause instanceof Error ? cause.message : String(cause);81			}82		}83		recordSync(target.kepr, repo, Math.floor(Date.now() / 1000), error);84	}85}8687/**88 * Koh's steal flow (§9.4) against a mirror: walk back from the remote head to a save the mirror has,89 * fetch what is missing, and land it in one transaction. Events are listed every time, since a90 * flag or a strip arrives without moving `head`; releases only when `head` or `releases/latest`91 * moved (§18), since publishing one moves both. A path §5.5 refuses, or one whose object inflates92 * past Koh's file limit, is quarantined: the save lands without it, as if stripped. Returns the93 * events it landed.94 */95export async function syncRepo(96	bucket: Bucket,97	mirror: Database,98	repo: string,99	hidden: string[] = [],100): Promise<EventRecord[]> {101	await checkFormat(bucket, repo);102	const head = await readHead(bucket, repo);103	if (head === null) return [];104	const saves: SaveRecord[] = [];105	const objects = new Map<string, StoredObject>();106	const oversized = new Set<string>();107	const quarantined: string[] = [];108	const seen = new Set<string>();109	for (let cursor = head; cursor !== "" && !knownSave(mirror, cursor); ) {110		if (seen.has(cursor)) throw new BucketError(`${repo}: save ${cursor} is its own ancestor`);111		seen.add(cursor);112		const save = await readSave(bucket, repo, cursor);113		quarantined.push(...save.quarantined);114		for (const file of save.files) {115			if (objects.has(file.hash) || oversized.has(file.hash) || knownObject(mirror, file.hash)) {116				continue;117			}118			try {119				const { compressed, bytes } = await readObject(bucket, repo, file.hash);120				objects.set(file.hash, { compressed, size: bytes.length });121			} catch (cause) {122				if (!(cause instanceof OversizedObject)) throw cause;123				oversized.add(file.hash);124			}125		}126		quarantined.push(127			...save.files.filter((f) => oversized.has(f.hash)).map((f) => escapePath(f.path)),128		);129		save.files = save.files.filter((file) => !oversized.has(file.hash));130		saves.push(save);131		cursor = save.parent;132	}133	const known = knownEventIds(mirror);134	const events: EventRecord[] = [];135	for (const id of await listEventIds(bucket, repo)) {136		if (known.has(id)) continue;137		const event = await readEvent(bucket, repo, id);138		// Koh ignores it too: no save can hold an illegal path (§5.5).139		if (event.kind !== "strip" || isSafePath(event.path)) events.push(event);140	}141	const releasesLatest = (await readReleasesLatest(bucket, repo)) ?? "";142	const moved =143		mirrorState(mirror, "head") !== head ||144		mirrorState(mirror, "releases_latest") !== releasesLatest;145	const releases = moved ? await newReleases(bucket, mirror, repo) : [];146	if (!moved && events.length === 0) return [];147	land(mirror, {148		head,149		releasesLatest,150		saves: saves.reverse(),151		objects,152		events,153		releases,154		quarantined,155		hidden,156	});157	return events;158}159160async function newReleases(161	bucket: Bucket,162	mirror: Database,163	repo: string,164): Promise<ReleaseRecord[]> {165	const known = knownReleases(mirror);166	const releases: ReleaseRecord[] = [];167	for (const version of await listReleaseVersions(bucket, repo)) {168		if (!known.has(version)) releases.push(await readRelease(bucket, repo, version));169	}170	return releases;171}