1/**
2 * The Worker ↔ Repo DO push protocol (DESIGN §22.10): the pack never
3 * enters the Durable Object. The Worker receives the body, spills it,
4 * verifies it through the hasher fan-out, and ships the DO only *rows* —
5 * staged-object batches (promoted rows are coordinates into the wire
6 * pack, inline rows carry their zlib bytes) — plus the commit summary.
7 *
8 * Batches cross the RPC boundary as one binary buffer:
9 * `u32 jsonLen | json | zdata…` where the JSON array lists each row's
10 * metadata and the byte length of its zdata run, in order.
11 */
12import type { StagedObject } from "./Store/ObjectStore.ts";
13import type { ObjectType } from "./Protocol/ObjectCodec.ts";
14
15interface RowMeta {
16 /** oid */
17 readonly o: string;
18 /** type */
19 readonly t: ObjectType;
20 /** size (inflated) */
21 readonly s: number;
22 /** zdata length in the blob section */
23 readonly n: number;
24 /** zsize (promoted rows: the span in the pack) */
25 readonly z?: number | undefined;
26 /** pack coordinates (promoted rows) */
27 readonly p?: { readonly i: string; readonly f: number } | undefined;
28}
29
30/** Encodes staged rows for one `stagePush` call. */
31export const encodeStagedBatch = (
32 objects: ReadonlyArray<StagedObject>,
33): Uint8Array => {
34 const meta: Array<RowMeta> = [];
35 let blobBytes = 0;
36 for (const object of objects) {
37 meta.push({
38 o: object.oid,
39 t: object.type,
40 s: object.size,
41 n: object.zdata.byteLength,
42 z: object.zsize,
43 p:
44 object.pack === undefined
45 ? undefined
46 : { i: object.pack.packId, f: object.pack.offset },
47 });
48 blobBytes += object.zdata.byteLength;
49 }
50 const json = new TextEncoder().encode(JSON.stringify(meta));
51 const out = new Uint8Array(4 + json.length + blobBytes);
52 new DataView(out.buffer).setUint32(0, json.length);
53 out.set(json, 4);
54 let at = 4 + json.length;
55 for (const object of objects) {
56 out.set(object.zdata, at);
57 at += object.zdata.byteLength;
58 }
59 return out;
60};
61
62/** Decodes a `stagePush` batch back into staged rows (views into `bytes`). */
63export const decodeStagedBatch = (
64 bytes: Uint8Array,
65): ReadonlyArray<StagedObject> => {
66 const jsonLen = new DataView(
67 bytes.buffer,
68 bytes.byteOffset,
69 bytes.byteLength,
70 ).getUint32(0);
71 const meta = JSON.parse(
72 new TextDecoder().decode(bytes.subarray(4, 4 + jsonLen)),
73 ) as ReadonlyArray<RowMeta>;
74 const out: Array<StagedObject> = [];
75 let at = 4 + jsonLen;
76 for (const row of meta) {
77 const zdata = bytes.subarray(at, at + row.n);
78 at += row.n;
79 out.push({
80 oid: row.o,
81 type: row.t,
82 size: row.s,
83 zdata,
84 zsize: row.z,
85 pack:
86 row.p === undefined ? undefined : { packId: row.p.i, offset: row.p.f },
87 });
88 }
89 return out;
90};