1/**
2 * No-delta pack emission (DESIGN §3.5) — streaming, constant memory:
3 *
4 * ```
5 * PACK + version(2) + count(manifest.length) — count known upfront
6 * for each entry: varint(type, size) + zdata — stored zlib bytes, verbatim
7 * sha1 trailer — incremental over everything
8 * ```
9 *
10 * No compression CPU at all: objects are stored pre-deflated (`zdata` is
11 * `zlib(content)` without the loose header), so pack generation is literally
12 * concatenation + SHA-1. Serving no deltas / no thin packs is legal
13 * (thin-pack/ofs-delta are protocol MAYs).
14 *
15 * The writer emits {@link PackEvent}s — raw pack bytes plus optional progress
16 * marks — so the choreography layer can mux data onto side-band 1 and
17 * progress onto band 2 without this module knowing about sideband framing.
18 */
19import * as Effect from "effect/Effect";
20import * as Stream from "effect/Stream";
21import { encodeTypeSize, makeSha1, utf8Encode } from "./ObjectCodec.ts";
22import type { ManifestEntry, ObjectSource, StoreError } from "./Store.ts";
23
24/**
25 * One item of the pack emission stream: raw pack bytes, or a progress mark
26 * after an entry finished (`written` of `total` objects emitted).
27 */
28export type PackEvent =
29 | { readonly _tag: "data"; readonly bytes: Uint8Array }
30 | {
31 readonly _tag: "progress";
32 readonly written: number;
33 readonly total: number;
34 };
35
36/**
37 * Builds the 12-byte pack header: `"PACK"`, version 2, entry count (both
38 * 32-bit big-endian).
39 */
40export const packHeader = (count: number): Uint8Array => {
41 const out = new Uint8Array(12);
42 out.set(utf8Encode("PACK"), 0);
43 const view = new DataView(out.buffer);
44 view.setUint32(4, 2);
45 view.setUint32(8, count);
46 return out;
47};
48
49/**
50 * Options for {@link writePack}.
51 */
52export interface WritePackOptions {
53 /**
54 * Emit a `progress` event after every N entries (and after the last). When
55 * omitted, no progress events are emitted.
56 */
57 readonly progressEvery?: number | undefined;
58}
59
60/**
61 * Streams a complete no-delta pack for the given manifest: header + count,
62 * per-object `varint(type, size)` + stored zlib bytes (read from the
63 * {@link ObjectSource}), and the incremental SHA-1 trailer.
64 *
65 * The manifest is materialized by the closure walk before streaming so the
66 * header count is exact. Emission is sequential and constant-memory: one
67 * object's zdata chunks are in flight at a time.
68 */
69export const writePack = (
70 entries: ReadonlyArray<ManifestEntry>,
71 objects: ObjectSource,
72 options?: WritePackOptions,
73): Stream.Stream<PackEvent, StoreError> =>
74 Stream.unwrap(
75 Effect.sync(() => {
76 const sha = makeSha1();
77 const total = entries.length;
78 const progressEvery = options?.progressEvery;
79 let written = 0;
80
81 const data = (bytes: Uint8Array): PackEvent => {
82 sha.update(bytes);
83 return { _tag: "data", bytes };
84 };
85
86 const header: Stream.Stream<PackEvent> = Stream.fromEffect(
87 Effect.sync(() => data(packHeader(total))),
88 );
89
90 const body: Stream.Stream<PackEvent, StoreError> = Stream.flatMap(
91 Stream.fromArray(entries),
92 (entry) => {
93 const head = Stream.fromEffect(
94 Effect.sync(() => data(encodeTypeSize(entry.type, entry.size))),
95 );
96 const zdata = Stream.map(objects.readZData(entry.oid), data);
97 const progress: Stream.Stream<PackEvent> = Stream.unwrap(
98 Effect.sync(() => {
99 written += 1;
100 return progressEvery !== undefined &&
101 (written % progressEvery === 0 || written === total)
102 ? Stream.succeed<PackEvent>({
103 _tag: "progress",
104 written,
105 total,
106 })
107 : Stream.empty;
108 }),
109 );
110 return Stream.concat(Stream.concat(head, zdata), progress);
111 },
112 );
113
114 const trailer: Stream.Stream<PackEvent> = Stream.fromEffect(
115 Effect.sync((): PackEvent => ({ _tag: "data", bytes: sha.digest() })),
116 );
117
118 return Stream.concat(Stream.concat(header, body), trailer);
119 }),
120 );
121
122/**
123 * Drops progress events, keeping only the raw pack bytes (the non-sideband
124 * response shape).
125 */
126export const packDataBytes = <E, R>(
127 events: Stream.Stream<PackEvent, E, R>,
128): Stream.Stream<Uint8Array, E, R> =>
129 Stream.flatMap(events, (event) =>
130 event._tag === "data" ? Stream.succeed(event.bytes) : Stream.empty,
131 );
132
133/**
134 * Convenience: the raw pack byte stream for a manifest, with no progress
135 * events.
136 */
137export const writePackBytes = (
138 entries: ReadonlyArray<ManifestEntry>,
139 objects: ObjectSource,
140): Stream.Stream<Uint8Array, StoreError> =>
141 packDataBytes(writePack(entries, objects));