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 { RESERVED_NAMES } from "../server/reserved.ts";14import { dropTarballs } from "../tarball/cache.ts";15import type { Bucket } from "./reader.ts";16import {17	BucketError,18	checkFormat,19	type EventRecord,20	escapePath,21	isSafePath,22	listEventIds,23	listReleaseVersions,24	listRepos,25	OversizedObject,26	type ReleaseRecord,27	readEvent,28	readHead,29	readObject,30	readRelease,31	readReleasesLatest,32	readSave,33	type SaveRecord,34} from "./repo.ts";3536const INTERVAL_MS = 2 * 60 * 1000;3738export type SyncTarget = {39	bucket: Bucket;40	kepr: Database;41	mirrorDir: string;42	tarballDir: 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}5455export function startSyncLoop(target: SyncTarget): void {56	const tick = () => syncAll(target).catch((error) => console.error(`kepr: sync failed: ${error}`));57	tick();58	setInterval(tick, INTERVAL_MS);59}6061async function syncEach(target: SyncTarget): Promise<void> {62	for (const repo of await listRepos(target.bucket)) {63		registerRepo(target.kepr, repo);64		let error: string | null = null;65		if (RESERVED_NAMES.has(repo)) {66			error = `not synced: the name "${repo}" clashes with Kepr's own /${repo} route; rename the repo`;67		} else {68			try {69				const events = await syncRepo(target.bucket, openMirror(target.mirrorDir, repo), repo);70				if (events.some((event) => event.kind === "strip")) dropTarballs(target.tarballDir, repo);71			} catch (cause) {72				error = cause instanceof Error ? cause.message : String(cause);73			}74		}75		recordSync(target.kepr, repo, Math.floor(Date.now() / 1000), error);76	}77}7879/**80 * Koh's steal flow (§9.4) against a mirror: walk back from the remote head to a save the mirror has,81 * fetch what is missing, and land it in one transaction. Events are listed every time, since a82 * flag or a strip arrives without moving `head`; releases only when `head` or `releases/latest`83 * moved (§18), since publishing one moves both. A path §5.5 refuses, or one whose object inflates84 * past Koh's file limit, is quarantined: the save lands without it, as if stripped. Returns the85 * events it landed.86 */87export async function syncRepo(88	bucket: Bucket,89	mirror: Database,90	repo: string,91): Promise<EventRecord[]> {92	await checkFormat(bucket, repo);93	const head = await readHead(bucket, repo);94	if (head === null) return [];95	const saves: SaveRecord[] = [];96	const objects = new Map<string, StoredObject>();97	const oversized = new Set<string>();98	const quarantined: string[] = [];99	const seen = new Set<string>();100	for (let cursor = head; cursor !== "" && !knownSave(mirror, cursor); ) {101		if (seen.has(cursor)) throw new BucketError(`${repo}: save ${cursor} is its own ancestor`);102		seen.add(cursor);103		const save = await readSave(bucket, repo, cursor);104		quarantined.push(...save.quarantined);105		for (const file of save.files) {106			if (objects.has(file.hash) || oversized.has(file.hash) || knownObject(mirror, file.hash)) {107				continue;108			}109			try {110				const { compressed, bytes } = await readObject(bucket, repo, file.hash);111				objects.set(file.hash, { compressed, size: bytes.length });112			} catch (cause) {113				if (!(cause instanceof OversizedObject)) throw cause;114				oversized.add(file.hash);115			}116		}117		quarantined.push(118			...save.files.filter((f) => oversized.has(f.hash)).map((f) => escapePath(f.path)),119		);120		save.files = save.files.filter((file) => !oversized.has(file.hash));121		saves.push(save);122		cursor = save.parent;123	}124	const known = knownEventIds(mirror);125	const events: EventRecord[] = [];126	for (const id of await listEventIds(bucket, repo)) {127		if (known.has(id)) continue;128		const event = await readEvent(bucket, repo, id);129		// Koh ignores it too: no save can hold an illegal path (§5.5).130		if (event.kind !== "strip" || isSafePath(event.path)) events.push(event);131	}132	const releasesLatest = (await readReleasesLatest(bucket, repo)) ?? "";133	const moved =134		mirrorState(mirror, "head") !== head ||135		mirrorState(mirror, "releases_latest") !== releasesLatest;136	const releases = moved ? await newReleases(bucket, mirror, repo) : [];137	if (!moved && events.length === 0) return [];138	land(mirror, {139		head,140		releasesLatest,141		saves: saves.reverse(),142		objects,143		events,144		releases,145		quarantined,146	});147	return events;148}149150async function newReleases(151	bucket: Bucket,152	mirror: Database,153	repo: string,154): Promise<ReleaseRecord[]> {155	const known = knownReleases(mirror);156	const releases: ReleaseRecord[] = [];157	for (const version of await listReleaseVersions(bucket, repo)) {158		if (!known.has(version)) releases.push(await readRelease(bucket, repo, version));159	}160	return releases;161}