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