sync.ts

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