1/**
2 * The dynamic hasher (DESIGN §22.12): a module Worker loaded by
3 * `HasherWorkerLoader` through a `WorkerLoader` binding, one isolate per
4 * slot. It answers the hash route's own protocol — coordinates in the
5 * query, the part in the body, the scan as one framed response — and
6 * nothing else; it has no bindings and no network.
7 */
8import * as Effect from "effect/Effect";
9import {
10 decodeDeltaBatch,
11 encodeDeltaResults,
12 encodeScanResult,
13 frame,
14} from "./Protocol.ts";
15import { resolveDeltas, scanPart } from "../Protocol/PartialScan.ts";
16
17export default {
18 async fetch(request: Request): Promise<Response> {
19 const query = new URL(request.url).searchParams;
20 if (query.get("mode") === "deltas") {
21 const maxObjectSize = Number(query.get("max"));
22 const { bases, jobs } = decodeDeltaBatch(
23 new Uint8Array(await request.arrayBuffer()),
24 );
25 const resolved = await Effect.runPromise(
26 Effect.result(resolveDeltas(bases, jobs, { maxObjectSize })),
27 );
28 if (resolved._tag === "Failure") {
29 const failure = resolved.failure;
30 return new Response(
31 `${failure._tag}: ${"reason" in failure ? failure.reason : ""}`,
32 { status: 422 },
33 );
34 }
35 return new Response(
36 frame(encodeDeltaResults(resolved.success)) as unknown as BodyInit,
37 { headers: { "content-type": "application/octet-stream" } },
38 );
39 }
40 const base = Number(query.get("base"));
41 const remaining = Number(query.get("remaining"));
42 const maxObjectSize = Number(query.get("max"));
43 const skip = Number(query.get("skip") ?? "0");
44 if (![base, remaining, maxObjectSize, skip].every(Number.isFinite)) {
45 return new Response("bad coordinates", { status: 400 });
46 }
47 const body = new Uint8Array(await request.arrayBuffer());
48 const result = await Effect.runPromise(
49 Effect.result(
50 scanPart(skip > 0 ? body.subarray(skip) : body, {
51 base,
52 remaining,
53 maxObjectSize,
54 resync: query.get("resync") === "1",
55 }),
56 ),
57 );
58 if (result._tag === "Failure") {
59 const failure = result.failure;
60 return new Response(
61 `${failure._tag}: ${"reason" in failure ? failure.reason : ""}`,
62 { status: 422 },
63 );
64 }
65 return new Response(
66 frame(encodeScanResult(result.success)) as unknown as BodyInit,
67 {
68 headers: { "content-type": "application/octet-stream" },
69 },
70 );
71 },
72};