1/**
2 * The hash event a {@link HasherFunction} answers (DESIGN §22.11): one pack
3 * chunk, base64 in a JSON invoke payload (Lambda invokes are JSON; a 4 MiB
4 * chunk encodes to ~5.6 MB, under the 6 MB payload limit), answered with
5 * the encoded scan — bounded to the same limit by demoting the largest
6 * delta-resolved entries back to `unresolved` (the receiver resolves those
7 * itself from its retained bytes) and by omitting non-blob content (the
8 * receiver inflates it).
9 */
10import * as Effect from "effect/Effect";
11import {
12 decodeScanResult,
13 encodeScanResult,
14 HashError,
15 type HashPartOptions,
16} from "./Hasher.ts";
17import {
18 scanPart,
19 type ScanResult,
20 type UnresolvedDelta,
21} from "../Protocol/PartialScan.ts";
22
23/** Chunk size the Lambda hasher asks the pump for (4 MiB → ~5.6 MB base64). */
24export const LAMBDA_CHUNK_BYTES = 4 * 1024 * 1024;
25/** Raw bytes the encoded scan may carry before demotion (base64 adds a third). */
26export const RESPONSE_BUDGET_BYTES = 4 * 1024 * 1024;
27
28export interface HashEvent {
29 readonly alchemyGitHash: 1;
30 readonly base: number;
31 readonly remaining: number;
32 readonly max: number;
33 readonly resync: boolean;
34 readonly skip: number;
35 /** The chunk, base64. */
36 readonly payload: string;
37}
38
39export type HashResponse =
40 | { readonly scan: string }
41 | { readonly error: string };
42
43export const isHashEvent = (event: unknown): event is HashEvent =>
44 typeof event === "object" &&
45 event !== null &&
46 (event as { alchemyGitHash?: unknown }).alchemyGitHash === 1 &&
47 typeof (event as { payload?: unknown }).payload === "string";
48
49export const encodeHashEvent = (
50 payload: Uint8Array,
51 options: HashPartOptions,
52): HashEvent => ({
53 alchemyGitHash: 1,
54 base: options.base,
55 remaining: options.remaining,
56 max: options.maxObjectSize,
57 resync: options.resync === true,
58 skip: options.skip ?? 0,
59 payload: toBase64(payload),
60});
61
62const toBase64 = (bytes: Uint8Array): string =>
63 typeof Buffer !== "undefined"
64 ? Buffer.from(bytes.buffer, bytes.byteOffset, bytes.byteLength).toString(
65 "base64",
66 )
67 : btoa(String.fromCharCode(...bytes));
68
69export const fromBase64 = (text: string): Uint8Array =>
70 typeof Buffer !== "undefined"
71 ? new Uint8Array(Buffer.from(text, "base64"))
72 : Uint8Array.from(atob(text), (c) => c.charCodeAt(0));
73
74/**
75 * Bounds a scan to `budget` raw bytes: strips non-blob content, then
76 * demotes delta-resolved entries — largest first — to `unresolved` until
77 * the fresh zdata fits. Every demoted entry keeps its base reference.
78 */
79export const boundScan = (scan: ScanResult, budget: number): ScanResult => {
80 const entries = scan.entries.map((e) =>
81 e.content === undefined ? e : { ...e, content: undefined },
82 );
83 let bytes = entries.length * 120;
84 for (const e of entries) bytes += e.zdata?.byteLength ?? 0;
85 if (bytes <= budget) {
86 return { ...scan, entries };
87 }
88 const order = entries
89 .map((e, i) => ({ i, z: e.zdata?.byteLength ?? 0 }))
90 .filter((x) => x.z > 0)
91 .sort((a, b) => b.z - a.z);
92 const demoted = new Set<number>();
93 const unresolved: Array<UnresolvedDelta> = [...scan.unresolved];
94 for (const { i, z } of order) {
95 if (bytes <= budget) break;
96 const e = entries[i]!;
97 demoted.add(i);
98 bytes -= z;
99 unresolved.push({
100 offset: e.offset,
101 dataOffset: e.dataOffset,
102 span: e.span,
103 baseOffset: e.baseOffset,
104 baseOid: e.baseOid,
105 size: e.size,
106 });
107 }
108 return {
109 ...scan,
110 entries: entries.filter((_, i) => !demoted.has(i)),
111 unresolved,
112 };
113};
114
115/** Answers one hash event: scan, bound, encode. Pure — the Lambda's body. */
116export const handleHashEvent = (
117 event: HashEvent,
118): Effect.Effect<HashResponse> =>
119 Effect.gen(function* () {
120 const payload = fromBase64(event.payload);
121 const result = yield* scanPart(
122 event.skip > 0 ? payload.subarray(event.skip) : payload,
123 {
124 base: event.base,
125 remaining: event.remaining,
126 maxObjectSize: event.max,
127 resync: event.resync,
128 },
129 ).pipe(Effect.result);
130 if (result._tag === "Failure") {
131 const failure = result.failure;
132 return {
133 error: `${failure._tag}${"reason" in failure ? `: ${failure.reason}` : ""}`,
134 };
135 }
136 return {
137 scan: toBase64(
138 encodeScanResult(boundScan(result.success, RESPONSE_BUDGET_BYTES)),
139 ),
140 };
141 });
142
143/** Decodes a response into the pump's scan result (or a typed error). */
144export const decodeHashResponse = (
145 response: HashResponse,
146): Effect.Effect<ScanResult, HashError> =>
147 "error" in response
148 ? Effect.fail(new HashError({ reason: `lambda hasher: ${response.error}` }))
149 : Effect.succeed(decodeScanResult(fromBase64(response.scan)));