worker.ts

1import { join } from "node:path";2import { closeMirrors } from "../mirror/store.ts";3import { openKeprDb } from "../server/db.ts";4import { openBucket } from "./reader.ts";5import {6	localSyncer,7	type Syncer,8	type SyncReply,9	type SyncRequest,10	startSyncLoop,11} from "./sync.ts";1213// The sync's own thread: started by `startSyncWorker`, it runs the loop and answers its requests.1415declare const self: Worker;1617const reply = (message: SyncReply) => self.postMessage(message);18const settle = (id: number, work: Promise<void>) =>19	work.then(20		() => reply({ kind: "done", id }),21		(error) => reply({ kind: "done", id, error: String(error) }),22	);23const released = new Map<number, () => void>();24// Settles when the hold in force, if any, is released: a rename moves mirror files under it.25let unheld = Promise.resolve();26let syncer: Syncer;2728self.addEventListener("message", ({ data }: MessageEvent<SyncRequest>) => {29	switch (data.kind) {30		case "start": {31			const target = {32				bucket: openBucket(data.source),33				kepr: openKeprDb(data.dataDir),34				mirrorDir: join(data.dataDir, "mirror"),35			};36			syncer = localSyncer(target);37			startSyncLoop(target);38			return;39		}40		case "sync":41			return void settle(data.id, syncer.syncAll());42		case "hold":43			// Answered once no sync runs; mirrors are closed so a rename can move their files.44			return void syncer.hold(() => {45				const held = new Promise<void>((release) => released.set(data.id, release));46				unheld = held;47				closeMirrors();48				reply({ kind: "done", id: data.id });49				return held;50			});51		case "release":52			released.get(data.id)?.();53			released.delete(data.id);54			return;55		case "hidden":56			// Not behind a sync under way, which can take minutes; only behind a rename's hold.57			return void settle(58				data.id,59				unheld.then(() => syncer.applyHidden(data.repos)),60			);61		case "bucket":62			return syncer.setBucket(data.source);63	}64});