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	listEventIds,21	listReleaseVersions,22	listRepos,23	type ReleaseRecord,24	readEvent,25	readHead,26	readObject,27	readRelease,28	readReleasesLatest,29	readSave,30	type SaveRecord,31} from "./repo.ts";3233const INTERVAL_MS = 2 * 60 * 1000;3435export type SyncTarget = {36	bucket: BucketReader;37	kepr: Database;38	mirrorDir: string;39	tarballDir: string;40};4142let running: Promise<void> | null = null;4344/** Syncs every repo in the bucket; the admin console's "sync now" calls this too. Never overlaps itself. */45export function syncAll(target: SyncTarget): Promise<void> {46	running ??= syncEach(target).finally(() => {47		running = null;48	});49	return running;50}5152export function startSyncLoop(target: SyncTarget): void {53	const tick = () => syncAll(target).catch((error) => console.error(`kepr: sync failed: ${error}`));54	tick();55	setInterval(tick, INTERVAL_MS);56}5758async function syncEach(target: SyncTarget): Promise<void> {59	for (const repo of await listRepos(target.bucket)) {60		registerRepo(target.kepr, repo);61		let error: string | null = null;62		if (RESERVED_NAMES.has(repo)) {63			error = `not synced: the name "${repo}" clashes with Kepr's own /${repo} route; rename the repo`;64		} else {65			try {66				const events = await syncRepo(target.bucket, openMirror(target.mirrorDir, repo), repo);67				if (events.some((event) => event.kind === "strip")) dropTarballs(target.tarballDir, repo);68			} catch (cause) {69				error = cause instanceof Error ? cause.message : String(cause);70			}71		}72		recordSync(target.kepr, repo, Math.floor(Date.now() / 1000), error);73	}74}7576/**77 * Koh's steal flow (§9.4) against a mirror: walk back from the remote head to a save the mirror has,78 * fetch what is missing, and land it in one transaction. Events are listed every time, since a79 * flag or a strip arrives without moving `head`; releases only when `head` or `releases/latest`80 * moved (§18), since publishing one moves both. Returns the events it landed.81 */82export async function syncRepo(83	bucket: BucketReader,84	mirror: Database,85	repo: string,86): Promise<EventRecord[]> {87	await checkFormat(bucket, repo);88	const head = await readHead(bucket, repo);89	if (head === null) return [];90	const saves: SaveRecord[] = [];91	const objects = new Map<string, StoredObject>();92	const seen = new Set<string>();93	for (let cursor = head; cursor !== "" && !knownSave(mirror, cursor); ) {94		if (seen.has(cursor)) throw new BucketError(`${repo}: save ${cursor} is its own ancestor`);95		seen.add(cursor);96		const save = await readSave(bucket, repo, cursor);97		for (const file of save.files) {98			if (objects.has(file.hash) || knownObject(mirror, file.hash)) continue;99			const { compressed, bytes } = await readObject(bucket, repo, file.hash);100			objects.set(file.hash, { compressed, size: bytes.length });101		}102		saves.push(save);103		cursor = save.parent;104	}105	const known = knownEventIds(mirror);106	const events: EventRecord[] = [];107	for (const id of await listEventIds(bucket, repo)) {108		if (!known.has(id)) events.push(await readEvent(bucket, repo, id));109	}110	const releasesLatest = (await readReleasesLatest(bucket, repo)) ?? "";111	const moved =112		mirrorState(mirror, "head") !== head ||113		mirrorState(mirror, "releases_latest") !== releasesLatest;114	const releases = moved ? await newReleases(bucket, mirror, repo) : [];115	if (!moved && events.length === 0) return [];116	land(mirror, { head, releasesLatest, saves: saves.reverse(), objects, events, releases });117	return events;118}119120async function newReleases(121	bucket: BucketReader,122	mirror: Database,123	repo: string,124): Promise<ReleaseRecord[]> {125	const known = knownReleases(mirror);126	const releases: ReleaseRecord[] = [];127	for (const version of await listReleaseVersions(bucket, repo)) {128		if (!known.has(version)) releases.push(await readRelease(bucket, repo, version));129	}130	return releases;131}