1/**
2 * Protocol-v0 fetch/clone choreography (DESIGN §2.3, §4) as pure functions
3 * over a parsed request plus injected {@link ObjectSource} /
4 * {@link ClosureSource}.
5 *
6 * Stateless-RPC reality (HTTP): each POST is independent; the client resends
7 * all wants plus all previously-ACKed haves every round. With
8 * `multi_ack_detailed` + `no-done` and the "lazy-ready" policy the
9 * negotiation converges in ≤ 2 POSTs:
10 *
11 * - POST ends with flush (no `done`): `ACK <oid> common` per recognized
12 * have, `ACK <last> ready` once the want-closure is coverable, else `NAK`.
13 * With `no-done` the server continues straight from `ready` into the pack.
14 * - POST ends with `done`: final `ACK <last common>` (else `NAK`), then the
15 * pack — on side-band band 1 when negotiated, raw bytes otherwise.
16 * - A fresh clone is one POST: wants + flush + `done`, no haves ⇒ `NAK` +
17 * pack.
18 *
19 * Shallow (`deepen <n>`): `shallow <oid>` boundary lines (and `unshallow`
20 * for previously-shallow client tips now complete), then a flush, before the
21 * ACK/NAK section; the pack is depth-truncated.
22 */
23import * as Effect from "effect/Effect";
24import * as Stream from "effect/Stream";
25import { concatBytes, isOid, type Oid } from "./ObjectCodec.ts";
26import {
27 decodePktLines,
28 flushPkt,
29 PktLineError,
30 pktPayloadText,
31 pktText,
32 ProtocolError,
33} from "./Pkt.ts";
34import { packDataBytes, type PackEvent, writePack } from "./PackWriter.ts";
35import { progressMessage, sidebandFrames } from "./Sideband.ts";
36import type { ClosureResult, ClosureSource, ObjectSource } from "./Store.ts";
37import { StoreError } from "./Store.ts";
38
39/**
40 * A parsed `git-upload-pack` request body.
41 */
42export interface UploadPackRequest {
43 /** Requested tips, in request order (first `want` carried the caps). */
44 readonly wants: ReadonlyArray<Oid>;
45 /** The client's `have` lines, in request order. */
46 readonly haves: ReadonlyArray<Oid>;
47 /** Whether the request ended with `done` (vs a bare flush). */
48 readonly done: boolean;
49 /** Capabilities requested on the first `want` line (space-separated). */
50 readonly capabilities: ReadonlySet<string>;
51 /** Depth bound from `deepen <n>`, when present. */
52 readonly depth: number | undefined;
53 /** The client's existing shallow tips (`shallow <oid>` lines). */
54 readonly clientShallow: ReadonlyArray<Oid>;
55}
56
57const parseOidLine = (
58 text: string,
59 keyword: string,
60): Effect.Effect<{ readonly oid: Oid; readonly rest: string }, ProtocolError> =>
61 Effect.suspend(() => {
62 const body = text.slice(keyword.length + 1);
63 const oid = body.slice(0, 40);
64 if (!isOid(oid)) {
65 return Effect.fail(
66 new ProtocolError({ reason: `malformed ${keyword} line: ${text}` }),
67 );
68 }
69 return Effect.succeed({ oid, rest: body.slice(40).trim() });
70 });
71
72/**
73 * Parses a `git-upload-pack` request body (already gunzipped) into a
74 * {@link UploadPackRequest}. Rejects arguments we do not advertise
75 * (`deepen-since`, `deepen-not`, `filter`) with a typed
76 * {@link ProtocolError}.
77 */
78export const parseUploadPackRequest = Effect.fn(function* (body: Uint8Array) {
79 const pkts = yield* decodePktLines(body);
80 const wants: Array<Oid> = [];
81 const haves: Array<Oid> = [];
82 const clientShallow: Array<Oid> = [];
83 const capabilities = new Set<string>();
84 let depth: number | undefined;
85 let done = false;
86 let inHaves = false;
87
88 for (const pkt of pkts) {
89 if (pkt._tag === "flush") {
90 inHaves = true;
91 continue;
92 }
93 if (pkt._tag !== "data") {
94 return yield* new ProtocolError({
95 reason: `unexpected ${pkt._tag} packet in upload-pack request`,
96 });
97 }
98 const text = pktPayloadText(pkt.payload);
99 if (!inHaves) {
100 if (text.startsWith("want ")) {
101 const { oid, rest } = yield* parseOidLine(text, "want");
102 wants.push(oid);
103 if (rest.length > 0) {
104 for (const cap of rest.split(" ")) {
105 if (cap.length > 0) capabilities.add(cap);
106 }
107 }
108 } else if (text.startsWith("shallow ")) {
109 const { oid } = yield* parseOidLine(text, "shallow");
110 clientShallow.push(oid);
111 } else if (text.startsWith("deepen ")) {
112 const n = Number.parseInt(text.slice(7), 10);
113 if (!Number.isInteger(n) || n <= 0) {
114 return yield* new ProtocolError({
115 reason: `invalid deepen depth: ${text}`,
116 });
117 }
118 depth = n;
119 } else if (
120 text.startsWith("deepen-since ") ||
121 text.startsWith("deepen-not ") ||
122 text.startsWith("filter ")
123 ) {
124 return yield* new ProtocolError({
125 reason: `unsupported argument: ${text.split(" ")[0]}`,
126 });
127 } else {
128 return yield* new ProtocolError({
129 reason: `unexpected line in want section: ${text}`,
130 });
131 }
132 } else {
133 if (text.startsWith("have ")) {
134 const { oid } = yield* parseOidLine(text, "have");
135 haves.push(oid);
136 } else if (text === "done") {
137 done = true;
138 } else {
139 return yield* new ProtocolError({
140 reason: `unexpected line in have section: ${text}`,
141 });
142 }
143 }
144 }
145
146 if (wants.length === 0) {
147 return yield* new ProtocolError({
148 reason: "upload-pack request has no want lines",
149 });
150 }
151 const request: UploadPackRequest = {
152 wants,
153 haves,
154 done,
155 capabilities,
156 depth,
157 clientShallow,
158 };
159 return request;
160});
161
162/**
163 * The outcome of one negotiation round.
164 */
165export interface Negotiation {
166 /** Haves that exist in the store, in the client's request order. */
167 readonly common: ReadonlyArray<Oid>;
168 /** Whether the server declared `ready` (lazy-ready: any common found). */
169 readonly ready: boolean;
170 /** Whether this response continues into a pack. */
171 readonly sendPack: boolean;
172 /** The encoded ACK/NAK pkt-line section. */
173 readonly ackSection: Uint8Array;
174}
175
176/**
177 * Runs the v0 `multi_ack_detailed` (+ `no-done`) negotiation for one
178 * stateless round: acks `common` for each recognized have, declares `ready`
179 * as soon as any common exists, and decides whether the pack follows in this
180 * response (`done` received, or `ready` under `no-done`).
181 */
182export const negotiate = Effect.fn(function* (
183 request: UploadPackRequest,
184 objects: ObjectSource,
185) {
186 const existing = new Set(yield* objects.filterExisting(request.haves));
187 const common = request.haves.filter((oid) => existing.has(oid));
188 const noDone = request.capabilities.has("no-done");
189 const multiAckDetailed = request.capabilities.has("multi_ack_detailed");
190 const last = common[common.length - 1];
191
192 const lines: Array<Uint8Array> = [];
193 let ready = false;
194 let sendPack: boolean;
195 if (request.done) {
196 // final round: single ACK of the last common oid, else NAK, then pack
197 lines.push(last !== undefined ? pktText(`ACK ${last}`) : pktText("NAK"));
198 sendPack = true;
199 } else if (last === undefined) {
200 lines.push(pktText("NAK"));
201 sendPack = false;
202 } else if (multiAckDetailed) {
203 for (const oid of common) {
204 lines.push(pktText(`ACK ${oid} common`));
205 }
206 lines.push(pktText(`ACK ${last} ready`));
207 ready = true;
208 if (noDone) {
209 // continue straight into the pack after `ready`
210 sendPack = true;
211 } else {
212 // every non-final round ends with NAK; the client sends `done` next
213 lines.push(pktText("NAK"));
214 sendPack = false;
215 }
216 } else {
217 // legacy single-ack fallback: ack the last common, wait for `done`
218 lines.push(pktText(`ACK ${last}`));
219 ready = true;
220 sendPack = false;
221 }
222 const negotiation: Negotiation = {
223 common,
224 ready,
225 sendPack,
226 ackSection: concatBytes(lines),
227 };
228 return negotiation;
229});
230
231/**
232 * A fully assembled upload-pack response.
233 */
234export interface UploadPackResponse {
235 /** The response body, streamed chunked (`HttpServerResponse.stream`). */
236 readonly body: Stream.Stream<Uint8Array, StoreError>;
237 /** Whether side-band-64k framing was negotiated. */
238 readonly sideband: boolean;
239 /** Whether this response carries a pack (vs a negotiation-only round). */
240 readonly packSent: boolean;
241 /** Number of objects in the pack (0 when `packSent` is false). */
242 readonly objectCount: number;
243}
244
245/** Emit a progress event on band 2 every N pack entries. */
246const PROGRESS_EVERY = 1000;
247
248const shallowSection = (closure: ClosureResult): Uint8Array =>
249 concatBytes([
250 ...closure.shallow.map((oid) => pktText(`shallow ${oid}`)),
251 ...closure.unshallow.map((oid) => pktText(`unshallow ${oid}`)),
252 flushPkt,
253 ]);
254
255const muxPackEvent = (event: PackEvent): Stream.Stream<Uint8Array> =>
256 event._tag === "data"
257 ? Stream.fromArray(sidebandFrames(1, event.bytes))
258 : Stream.succeed(
259 progressMessage(`Counting objects: ${event.written}/${event.total}\r`),
260 );
261
262/**
263 * Runs one full upload-pack round: validates the wants, negotiates, computes
264 * the closure when a pack is due, and assembles the streaming response body
265 * (shallow section, ACK/NAK section, then the pack — sideband-muxed with
266 * band-2 progress when negotiated, raw otherwise).
267 *
268 * Wants that do not exist in the store fail with `ProtocolError`
269 * (`not our ref`); the caller converts pre-stream failures into an `ERR`
270 * pkt / HTTP status. Mid-stream `StoreError`s surface on the body stream for
271 * the caller to translate into a band-3 fatal.
272 */
273export const uploadPack = Effect.fn(function* (
274 request: UploadPackRequest,
275 objects: ObjectSource,
276 closure: ClosureSource,
277) {
278 const sideband = request.capabilities.has("side-band-64k");
279
280 const existingWants = new Set(yield* objects.filterExisting(request.wants));
281 for (const want of request.wants) {
282 if (!existingWants.has(want)) {
283 return yield* new ProtocolError({ reason: `not our ref ${want}` });
284 }
285 }
286
287 const negotiation = yield* negotiate(request, objects);
288 const shallowRequested =
289 request.depth !== undefined || request.clientShallow.length > 0;
290
291 if (!negotiation.sendPack) {
292 // negotiation-only round: shallow section (when requested) + acks
293 const sections: Array<Uint8Array> = [];
294 if (shallowRequested) {
295 const result = yield* closure.commitClosure({
296 wants: request.wants,
297 haves: negotiation.common,
298 depth: request.depth,
299 clientShallow: request.clientShallow,
300 });
301 sections.push(shallowSection(result));
302 }
303 sections.push(negotiation.ackSection);
304 const response: UploadPackResponse = {
305 body: Stream.fromArray(sections),
306 sideband,
307 packSent: false,
308 objectCount: 0,
309 };
310 return response;
311 }
312
313 const result = yield* closure.commitClosure({
314 wants: request.wants,
315 haves: negotiation.common,
316 depth: request.depth,
317 clientShallow: request.clientShallow,
318 });
319
320 const prelude: Array<Uint8Array> = [];
321 if (shallowRequested) prelude.push(shallowSection(result));
322 prelude.push(negotiation.ackSection);
323
324 const events = writePack(result.entries, objects, {
325 progressEvery: sideband ? PROGRESS_EVERY : undefined,
326 });
327
328 const body = sideband
329 ? Stream.concat(
330 Stream.concat(
331 Stream.fromArray([
332 ...prelude,
333 progressMessage(
334 `Enumerating objects: ${result.entries.length}, done.`,
335 ),
336 ]),
337 Stream.flatMap(events, muxPackEvent),
338 ),
339 Stream.succeed(flushPkt),
340 )
341 : Stream.concat(Stream.fromArray(prelude), packDataBytes(events));
342
343 const response: UploadPackResponse = {
344 body,
345 sideband,
346 packSent: true,
347 objectCount: result.entries.length,
348 };
349 return response;
350});
351
352/**
353 * Convenience: parses the request body and runs {@link uploadPack} in one
354 * step.
355 */
356export const handleUploadPack = (
357 body: Uint8Array,
358 objects: ObjectSource,
359 closure: ClosureSource,
360): Effect.Effect<
361 UploadPackResponse,
362 PktLineError | ProtocolError | StoreError
363> =>
364 parseUploadPackRequest(body).pipe(
365 Effect.flatMap((request) => uploadPack(request, objects, closure)),
366 );