1/**
2 * The hasher wire protocol (DESIGN §22.7, §22.9) — the part of `Hasher.ts`
3 * a hasher isolate needs and nothing more: no Cloudflare provider imports,
4 * so a dynamic worker bundling it stays small.
5 *
6 * Request: the raw part bytes, coordinates in the query string. Response:
7 * `u32 len | frame` repeated — the scan first (`encodeScanResult`: `u32
8 * jsonLength | json | blob area`, entries referencing content/zdata as
9 * `[offset, length]` into the blob area), then, when a spill was
10 * requested, the uploaded part as JSON.
11 */
12import * as Effect from "effect/Effect";
13import * as Schema from "effect/Schema";
14import type { Oid, ObjectType } from "../Protocol/ObjectCodec.ts";
15import type {
16 DeltaBase,
17 DeltaJob,
18 DeltaResolved,
19 EntryBounds,
20 ScanResult,
21} from "../Protocol/PartialScan.ts";
22
23export class HashError extends Schema.TaggedError<HashError>()("HashError", {
24 reason: Schema.String,
25}) {}
26
27export const encodeBoundsRequest = (
28 payload: Uint8Array,
29 bounds: ReadonlyArray<EntryBounds>,
30): Uint8Array => {
31 const json = new TextEncoder().encode(JSON.stringify(bounds));
32 const out = new Uint8Array(4 + json.length + payload.length);
33 new DataView(out.buffer).setUint32(0, json.length);
34 out.set(json, 4);
35 out.set(payload, 4 + json.length);
36 return out;
37};
38
39export const decodeBoundsRequest = (
40 body: Uint8Array,
41): {
42 readonly bounds: ReadonlyArray<EntryBounds>;
43 readonly payload: Uint8Array;
44} => {
45 const jsonLength = new DataView(body.buffer, body.byteOffset).getUint32(0);
46 const bounds = JSON.parse(
47 new TextDecoder().decode(body.subarray(4, 4 + jsonLength)),
48 ) as ReadonlyArray<EntryBounds>;
49 return { bounds, payload: body.subarray(4 + jsonLength) };
50};
51
52/** The internal route the self-binding layer posts to. */
53export const HASH_ROUTE = "/_alchemy/git/hash";
54
55// ── protocol ────────────────────────────────────────────────────────────────
56
57interface WireEntry {
58 readonly o: string; // oid
59 readonly t: number; // type
60 readonly h: number; // header offset
61 readonly s: number; // size
62 readonly d: number; // dataOffset
63 readonly n: number; // span
64 readonly z?: [number, number]; // zdata [offset, length] in the blob area
65 readonly c?: [number, number]; // content [offset, length]
66 readonly b?: number; // baseOffset (delta-resolved)
67 readonly r?: string; // baseOid (delta-resolved)
68}
69interface WireResult {
70 readonly firstOffset: number;
71 readonly entries: ReadonlyArray<WireEntry>;
72 readonly unresolved: ScanResult["unresolved"];
73 readonly consumedTo: number;
74 readonly count: number;
75}
76
77export const encodeScanResult = (result: ScanResult): Uint8Array => {
78 const blobs: Array<Uint8Array> = [];
79 let at = 0;
80 const put = (bytes: Uint8Array): [number, number] => {
81 blobs.push(bytes);
82 const ref: [number, number] = [at, bytes.length];
83 at += bytes.length;
84 return ref;
85 };
86 const wire: WireResult = {
87 firstOffset: result.firstOffset,
88 entries: result.entries.map((e) => ({
89 o: e.oid,
90 t: e.type,
91 h: e.offset,
92 s: e.size,
93 d: e.dataOffset,
94 n: e.span,
95 ...(e.zdata === undefined ? {} : { z: put(e.zdata) }),
96 ...(e.content === undefined ? {} : { c: put(e.content) }),
97 ...(e.baseOffset === undefined ? {} : { b: e.baseOffset }),
98 ...(e.baseOid === undefined ? {} : { r: e.baseOid }),
99 })),
100 unresolved: result.unresolved,
101 consumedTo: result.consumedTo,
102 count: result.count,
103 };
104 const json = new TextEncoder().encode(JSON.stringify(wire));
105 const out = new Uint8Array(4 + json.length + at);
106 new DataView(out.buffer).setUint32(0, json.length);
107 out.set(json, 4);
108 let cursor = 4 + json.length;
109 for (const b of blobs) {
110 out.set(b, cursor);
111 cursor += b.length;
112 }
113 return out;
114};
115
116export const decodeScanResult = (bytes: Uint8Array): ScanResult => {
117 const jsonLength = new DataView(bytes.buffer, bytes.byteOffset).getUint32(0);
118 const wire = JSON.parse(
119 new TextDecoder().decode(bytes.subarray(4, 4 + jsonLength)),
120 ) as WireResult;
121 const blobs = bytes.subarray(4 + jsonLength);
122 const slice = (ref: [number, number] | undefined) =>
123 ref === undefined ? undefined : blobs.subarray(ref[0], ref[0] + ref[1]);
124 return {
125 firstOffset: wire.firstOffset,
126 entries: wire.entries.map((e) => ({
127 oid: e.o as Oid,
128 type: e.t as ObjectType,
129 offset: e.h,
130 size: e.s,
131 dataOffset: e.d,
132 span: e.n,
133 zdata: slice(e.z),
134 content: slice(e.c),
135 baseOffset: e.b,
136 baseOid: e.r as Oid | undefined,
137 })),
138 unresolved: wire.unresolved,
139 consumedTo: wire.consumedTo,
140 count: wire.count,
141 };
142};
143
144/**
145 * Response framing of the hash route: `u32 len | frame` repeated — the
146 * scan first, then (spill only) the uploaded part as JSON once its upload
147 * has finished. The client acts on the scan without waiting for the part.
148 */
149export const frame = (bytes: Uint8Array): Uint8Array => {
150 const out = new Uint8Array(4 + bytes.length);
151 new DataView(out.buffer).setUint32(0, bytes.length);
152 out.set(bytes, 4);
153 return out;
154};
155
156/** Reads length-prefixed frames off a response body, one at a time. */
157export const makeFrameReader = (body: ReadableStream<Uint8Array>) => {
158 const reader = body.getReader();
159 const pending: Array<Uint8Array> = [];
160 let pendingBytes = 0;
161 const take = (n: number): Uint8Array => {
162 const out = new Uint8Array(n);
163 let written = 0;
164 while (written < n) {
165 const head = pending[0]!;
166 const use = Math.min(head.length, n - written);
167 out.set(head.subarray(0, use), written);
168 written += use;
169 if (use === head.length) pending.shift();
170 else pending[0] = head.subarray(use);
171 }
172 pendingBytes -= n;
173 return out;
174 };
175 const peekLength = (): number | undefined => {
176 if (pendingBytes < 4) return undefined;
177 const head = take(4);
178 const len = new DataView(head.buffer).getUint32(0);
179 pending.unshift(head);
180 pendingBytes += 4;
181 return len;
182 };
183 return (): Effect.Effect<Uint8Array | undefined, HashError> =>
184 Effect.tryPromise({
185 try: async () => {
186 while (true) {
187 const len = peekLength();
188 if (len !== undefined && pendingBytes >= 4 + len) {
189 take(4);
190 return take(len);
191 }
192 const { value, done } = await reader.read();
193 if (done) return undefined;
194 pending.push(value);
195 pendingBytes += value.length;
196 }
197 },
198 catch: (error) =>
199 new HashError({ reason: `hash part body: ${String(error)}` }),
200 });
201};
202
203// ── delta batches (DESIGN §22.13) ──────────────────────────────────────────
204
205/** `u32 jsonLen | json | blobs` with bases and deltas referenced as [offset, length]. */
206export const encodeDeltaBatch = (
207 bases: ReadonlyArray<DeltaBase>,
208 jobs: ReadonlyArray<DeltaJob>,
209): Uint8Array => {
210 const blobs: Array<Uint8Array> = [];
211 let at = 0;
212 const put = (bytes: Uint8Array): [number, number] => {
213 blobs.push(bytes);
214 const ref: [number, number] = [at, bytes.length];
215 at += bytes.length;
216 return ref;
217 };
218 const wire = {
219 bases: bases.map((b) => ({ r: put(b.bytes), c: b.isContent ? 1 : 0 })),
220 jobs: jobs.map((j) => ({ i: j.id, t: j.type, b: j.base, d: put(j.delta) })),
221 };
222 const json = new TextEncoder().encode(JSON.stringify(wire));
223 const out = new Uint8Array(4 + json.length + at);
224 new DataView(out.buffer).setUint32(0, json.length);
225 out.set(json, 4);
226 let cursor = 4 + json.length;
227 for (const b of blobs) {
228 out.set(b, cursor);
229 cursor += b.length;
230 }
231 return out;
232};
233
234export const decodeDeltaBatch = (
235 bytes: Uint8Array,
236): { readonly bases: Array<DeltaBase>; readonly jobs: Array<DeltaJob> } => {
237 const jsonLength = new DataView(bytes.buffer, bytes.byteOffset).getUint32(0);
238 const wire = JSON.parse(
239 new TextDecoder().decode(bytes.subarray(4, 4 + jsonLength)),
240 ) as {
241 bases: Array<{ r: [number, number]; c: number }>;
242 jobs: Array<{ i: number; t: ObjectType; b: number; d: [number, number] }>;
243 };
244 const blobs = bytes.subarray(4 + jsonLength);
245 const slice = (ref: [number, number]) =>
246 blobs.subarray(ref[0], ref[0] + ref[1]);
247 return {
248 bases: wire.bases.map((b) => ({ bytes: slice(b.r), isContent: b.c === 1 })),
249 jobs: wire.jobs.map((j) => ({
250 id: j.i,
251 type: j.t,
252 base: j.b,
253 delta: slice(j.d),
254 })),
255 };
256};
257
258export const encodeDeltaResults = (
259 results: ReadonlyArray<DeltaResolved>,
260): Uint8Array => {
261 const blobs: Array<Uint8Array> = [];
262 let at = 0;
263 const put = (bytes: Uint8Array): [number, number] => {
264 blobs.push(bytes);
265 const ref: [number, number] = [at, bytes.length];
266 at += bytes.length;
267 return ref;
268 };
269 const wire = results.map((r) => ({
270 i: r.id,
271 o: r.oid,
272 s: r.size,
273 z: put(r.zdata),
274 ...(r.content === undefined ? {} : { c: put(r.content) }),
275 }));
276 const json = new TextEncoder().encode(JSON.stringify(wire));
277 const out = new Uint8Array(4 + json.length + at);
278 new DataView(out.buffer).setUint32(0, json.length);
279 out.set(json, 4);
280 let cursor = 4 + json.length;
281 for (const b of blobs) {
282 out.set(b, cursor);
283 cursor += b.length;
284 }
285 return out;
286};
287
288export const decodeDeltaResults = (bytes: Uint8Array): Array<DeltaResolved> => {
289 const jsonLength = new DataView(bytes.buffer, bytes.byteOffset).getUint32(0);
290 const wire = JSON.parse(
291 new TextDecoder().decode(bytes.subarray(4, 4 + jsonLength)),
292 ) as Array<{
293 i: number;
294 o: string;
295 s: number;
296 z: [number, number];
297 c?: [number, number];
298 }>;
299 const blobs = bytes.subarray(4 + jsonLength);
300 const slice = (ref: [number, number]) =>
301 blobs.subarray(ref[0], ref[0] + ref[1]);
302 return wire.map((r) => ({
303 id: r.i,
304 oid: r.o as Oid,
305 size: r.s,
306 zdata: slice(r.z),
307 content: r.c === undefined ? undefined : slice(r.c),
308 }));
309};