1/**
2 * Git's HTTP handlers and native Effect route layers.
3 *
4 * ApiLive registers the public REST, smart-HTTP, and GitHub routes on the
5 * application's HttpRouter. InternalApiLive registers internal hashing.
6 * ApiHandlersLive shares the registry, repository clients, and cache across groups.
7 * The application owns its API, authentication, HTTP server, and CORS policy.
8 *
9 * ```typescript
10 * const PublicRoutes = Layer.mergeAll(AppApiLive, Git.ApiLive).pipe(
11 * Layer.provide(Authentication.layer),
12 * );
13 * const Routes = Layer.mergeAll(PublicRoutes, Git.InternalApiLive).pipe(
14 * Layer.provide(Git.ApiHandlersLive),
15 * Layer.provide(Git.ReposDurableObject),
16 * Layer.provide(Git.RegistryDurableObject),
17 * Layer.provide(Git.HasherInline),
18 * Layer.provide(Git.BlobStoreR2(GitObjects)),
19 * Layer.provide(Http.Platform),
20 * );
21 * const fetch = yield* HttpRouter.toHttpEffect(Routes);
22 * return { fetch };
23 * ```
24 */
25
26import * as Context from "effect/Context";
27import * as Effect from "effect/Effect";
28import * as Fiber from "effect/Fiber";
29import * as Layer from "effect/Layer";
30import * as Redacted from "effect/Redacted";
31import * as Result from "effect/Result";
32import * as Stream from "effect/Stream";
33import * as HttpRouter from "effect/http/HttpRouter";
34import * as HttpServerRequest from "effect/http/HttpServerRequest";
35import * as HttpServerResponse from "effect/http/HttpServerResponse";
36import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
37import crypto from "node:crypto";
38import * as Cloudflare from "../Cloudflare/index.ts";
39import { RuntimeContext } from "../RuntimeContext.ts";
40import { GitApi, InternalApi, RepoCreated } from "./Api.ts";
41import { BlobStore, type BlobStoreError } from "./BlobStore.ts";
42import { Engine, EngineLive } from "./Engine.ts";
43import { gitHubCompatRoutes } from "./GitHubCompat.ts";
44import {
45 decodeBoundsRequest,
46 encodeScanResult,
47 frame,
48 Hasher,
49 HASHER_BINDING,
50 InternalSecret,
51} from "./Hasher/Hasher.ts";
52import { decodeDeltaBatch, encodeDeltaResults } from "./Hasher/Protocol.ts";
53import * as ReceivePackHttp from "./Http/ReceivePack.ts";
54import { bundleCovers, type BundleInfo } from "./Jobs/Bundle.ts";
55import { Operations } from "./Operations.ts";
56import { concatBytes, utf8Decode } from "./Protocol/ObjectCodec.ts";
57import { hashBounds, resolveDeltas, scanPart } from "./Protocol/PartialScan.ts";
58import { decodePktLines, flushPkt, pktText } from "./Protocol/Pkt.ts";
59import {
60 progressMessage,
61 pumpPackBody,
62 wrapSideband,
63} from "./Protocol/Sideband.ts";
64import { RegistryStore } from "./RegistryObject.ts";
65import {
66 buildAdvertisement,
67 BUNDLE_COUNT_HEADER,
68 BUNDLE_HASH_HEADER,
69 BUNDLE_KEY_HEADER,
70 BUNDLE_SIDEBAND_HEADER,
71 GitRepo,
72 GitRepoLive,
73 isolatePushGate,
74 parseUploadPackRequest,
75 RepoStore,
76 WWW_AUTHENTICATE,
77} from "./RepoObject.ts";
78import { decodeHeadSnapshot } from "./Store/HeadSnapshot.ts";
79import { headKey } from "./Store/Keys.ts";
80
81/** A `Bearer` credential from the `Authorization` header (the hash route's internal secret). */
82const parseBearer = (
83 headers: Readonly<Record<string, string | undefined>>,
84): string | undefined => {
85 const authorization = headers.authorization;
86 if (authorization === undefined) return undefined;
87 const space = authorization.indexOf(" ");
88 if (space === -1) return undefined;
89 if (authorization.slice(0, space).toLowerCase() !== "bearer")
90 return undefined;
91 const credential = authorization.slice(space + 1).trim();
92 return credential === "" ? undefined : credential;
93};
94
95const sha256 = (input: string): Uint8Array => {
96 const digest = crypto.createHash("sha256").update(input).digest();
97 return new Uint8Array(digest.buffer, digest.byteOffset, digest.byteLength);
98};
99
100/** Timing-safe string equality over sha-256 digests. */
101const timingSafeEqual = (a: string, b: string): Effect.Effect<boolean> =>
102 Effect.sync(() => crypto.timingSafeEqual(sha256(a), sha256(b)));
103
104// ─────────────────────────────────────────────────────────────────────────────
105// Constants
106// ─────────────────────────────────────────────────────────────────────────────
107
108// The Worker
109// ─────────────────────────────────────────────────────────────────────────────
110
111/**
112 * The git-service Worker. Hosts the `GitRepo` and `GitRegistry` Durable
113 * Objects (their Live layers are provided on this Worker's init effect)
114 * plus the shared R2 objects bucket binding, and serves both the REST
115 * management plane and the git smart-HTTP wire protocol.
116 *
117 * Requires `nodejs_compat` (node:zlib / node:crypto in the DO) and a
118 * raised CPU limit (`cpu_ms: 300_000`) — pack inflation cannot run under
119 * the free plan's 10 ms budget (DESIGN.md §1).
120 */
121/**
122 * Worker options every Git host needs: `nodejs_compat` (zlib +
123 * crypto in the codec layer) and a generous CPU ceiling for pack ingest.
124 * Spread into your `Cloudflare.Worker` definition.
125 */
126export const GIT_WORKER_OPTIONS = {
127 compatibility: {
128 flags: ["nodejs_compat"] as Array<"nodejs_compat">,
129 date: "2026-03-17",
130 },
131 limits: { cpuMs: 300_000 },
132 // The push pipeline hashes each spilled part in a fresh isolate of this
133 // same script through a self service binding (DESIGN §22.7, Hasher.ts).
134 env: { [HASHER_BINDING]: Cloudflare.Workers.Self },
135};
136
137const makeCore = Effect.gen(function* () {
138 // ── init: DO namespaces ────────────────────────────────────────────────
139 // RPC-boundary tagged errors are reconstructed by the stubs themselves —
140 // both DO classes declare `errors: [...]` (see DurableObjectProps.errors).
141 const registry = yield* RegistryStore;
142 const repos = yield* RepoStore;
143 // The Worker streams clone bundles straight out of R2 (DESIGN.md §11):
144 // the DO plans the clone, the bytes bypass it entirely.
145 // The Worker-side view of the blob store (clone-bundle splice reads)
146 // — the same BlobStore layer the Repo DO consumes.
147 const workerBlobs = yield* BlobStore;
148 const engine = yield* Engine;
149 const internalSecret = yield* InternalSecret;
150 // The push pipeline's verifier (DESIGN §22.10): pack parts are inflated
151 // and hashed by this service — fanned out across Worker invocations by
152 // `HasherSelf`, inline under `HasherInline`.
153 const hasher = yield* Hasher;
154 const pushGate = yield* isolatePushGate;
155 /** Staging batches in flight to the Repo DO per push (DESIGN §22.10). */
156 const STAGE_CONCURRENCY = 6;
157
158 const operations = yield* Operations;
159 const {
160 resolveCached,
161 refs: refsRoutes,
162 objects: objectsRoutes,
163 pulls: pullsRoutes,
164 } = operations;
165 const remoteResult = <E, R>(effect: Effect.Effect<RepoCreated, E, R>) =>
166 Effect.gen(function* () {
167 const result = yield* effect;
168 const request = yield* HttpServerRequest.HttpServerRequest;
169 const host = request.headers.host;
170 return new RepoCreated({
171 repo: result.repo,
172 remote:
173 host === undefined
174 ? result.remote
175 : `${request.headers["x-forwarded-proto"] ?? "https"}://${host}${result.remote}`,
176 });
177 });
178 const reposRoutes = {
179 ...operations.repos,
180 create: (input: Parameters<typeof operations.repos.create>[0]) =>
181 remoteResult(operations.repos.create(input)),
182 fork: (input: Parameters<typeof operations.repos.fork>[0]) =>
183 remoteResult(operations.repos.fork(input)),
184 import: (input: Parameters<typeof operations.repos.import>[0]) =>
185 remoteResult(operations.repos.import(input)),
186 };
187
188 const wire401 = HttpServerResponse.empty({
189 status: 401,
190 headers: { "www-authenticate": WWW_AUTHENTICATE },
191 });
192 const notFound = HttpServerResponse.text("repository not found", {
193 status: 404,
194 });
195 const internalError = HttpServerResponse.text("internal error", {
196 status: 500,
197 });
198
199 /**
200 * Streams a clone bundle out of the BlobStore as a complete
201 * upload-pack result (NAK + optionally sideband-framed pack).
202 * `undefined` when the bundle bytes are gone (GC raced us).
203 */
204 /**
205 * Wraps pack bytes as a complete upload-pack result (NAK, optional
206 * sideband framing, flush). `shape` distinguishes a bundle streamed
207 * verbatim from one spliced with a delta — surfaced on
208 * `x-git-served-by` so which plane answered is observable, and
209 * assertable in the benchmarks.
210 */
211 const respondWithPack = (
212 packBytes: Stream.Stream<Uint8Array, BlobStoreError>,
213 options: {
214 readonly refsHash: string;
215 readonly objectCount: number;
216 readonly sideband: boolean;
217 readonly via: "do-bundle" | "head-snapshot";
218 /** The store's native stream, when it has one (R2 does). */
219 readonly readable?: ReadableStream<Uint8Array> | undefined;
220 /** `readable` is the pre-framed twin (sideband only). */
221 readonly framed?: boolean | undefined;
222 },
223 shape: "bundle",
224 ) => {
225 const nak = pktText("NAK");
226 const headers = {
227 "cache-control": "no-cache",
228 "x-git-served-by": `${options.via}:${shape}${options.framed === true ? "+framed" : ""}`,
229 [BUNDLE_HASH_HEADER]: options.refsHash,
230 };
231 // Native path (DESIGN §22): R2's own stream pipes into the Response
232 // through web streams only. The bundle bytes never enter an Effect
233 // stream, so there is no fiber hand-off per chunk — that hand-off
234 // was measured as the ceiling on clone throughput. Sideband framing
235 // is a TransformStream emitting 5-byte headers + subarrays (no copy);
236 // the raw case is an IdentityTransformStream, which workerd pipes
237 // without JS touching the bytes at all.
238 if (options.readable !== undefined) {
239 const source = options.readable;
240 const prefix = options.sideband
241 ? concatBytes([
242 nak,
243 progressMessage(
244 `Enumerating objects: ${options.objectCount}, done.`,
245 ),
246 ])
247 : nak;
248 const out = pumpPackBody({
249 prefix,
250 source,
251 sideband: options.sideband,
252 framed: options.framed,
253 });
254 return HttpServerResponse.raw(out, {
255 contentType: "application/x-git-upload-pack-result",
256 headers,
257 });
258 }
259 const body = options.sideband
260 ? Stream.fromArray([
261 nak,
262 progressMessage(`Enumerating objects: ${options.objectCount}, done.`),
263 ]).pipe(
264 Stream.concat(packBytes.pipe(wrapSideband(1))),
265 Stream.concat(Stream.succeed(flushPkt)),
266 )
267 : Stream.fromArray([nak]).pipe(Stream.concat(packBytes));
268 return HttpServerResponse.stream(body, {
269 contentType: "application/x-git-upload-pack-result",
270 headers,
271 });
272 };
273
274 const serveBundle = Effect.fn(function* (options: {
275 readonly key: string;
276 readonly refsHash: string;
277 readonly objectCount: number;
278 readonly sideband: boolean;
279 /** Which plane produced the plan — observability + bench assertions. */
280 readonly via: "do-bundle" | "head-snapshot";
281 }) {
282 if (options.sideband) {
283 // Prefer the pre-framed twin: a pure platform pipe (Keys.ts).
284 const framed = yield* Effect.result(
285 workerBlobs.get(options.key.replace(/\.pack$/, ".sideband")),
286 );
287 if (
288 Result.isSuccess(framed) &&
289 framed.success !== null &&
290 framed.success.readable !== undefined
291 ) {
292 return respondWithPack(
293 framed.success.stream,
294 { ...options, readable: framed.success.readable, framed: true },
295 "bundle",
296 );
297 }
298 }
299 const object = yield* Effect.result(workerBlobs.get(options.key));
300 if (Result.isFailure(object) || object.success === null) {
301 return undefined;
302 }
303 return respondWithPack(
304 object.success.stream,
305 { ...options, readable: object.success.readable },
306 "bundle",
307 );
308 });
309
310 /**
311 * The DO-less read path (DESIGN.md §21): the upload-pack
312 * advertisement and bundle-covered full clones served straight from
313 * the repo's head snapshot in the BlobStore — the Repo DO never
314 * wakes, so read throughput scales with Workers + blob storage
315 * instead of one single-threaded object.
316 *
317 * `undefined` = not eligible; the caller forwards to the DO. Access
318 * was decided before the route ran, by the middleware applied to its route layer — this changes WHERE the bytes come from, never who
319 * gets them.
320 */
321 const headSnapshotFastPath = Effect.fn(function* (
322 request: HttpServerRequest.HttpServerRequest,
323 repoId: string,
324 ) {
325 const target = new URL(request.url, "http://wire");
326 const isAdvertisement =
327 request.method === "GET" &&
328 target.pathname.endsWith("/info/refs") &&
329 target.searchParams.get("service") === "git-upload-pack";
330 const isUploadPack =
331 request.method === "POST" && target.pathname.endsWith("/git-upload-pack");
332 if (!isAdvertisement && !isUploadPack) return undefined;
333 // Compressed bodies carry big negotiation rounds — the DO owns
334 // those (and the gunzip) anyway.
335 if (isUploadPack && request.headers["content-encoding"] !== undefined) {
336 return undefined;
337 }
338
339 const object = yield* Effect.result(workerBlobs.get(headKey(repoId)));
340 if (Result.isFailure(object) || object.success === null) {
341 return undefined;
342 }
343 const raw = yield* Effect.result(object.success.bytes);
344 if (Result.isFailure(raw)) return undefined;
345 const snapshot = decodeHeadSnapshot(utf8Decode(raw.success));
346 if (snapshot === undefined) return undefined;
347
348 if (isAdvertisement) {
349 return HttpServerResponse.uint8Array(
350 buildAdvertisement({
351 service: "git-upload-pack",
352 defaultBranch: snapshot.defaultBranch,
353 refs: snapshot.refs,
354 }),
355 {
356 contentType: "application/x-git-upload-pack-advertisement",
357 headers: {
358 "cache-control": "no-cache",
359 "x-git-served-by": "head-snapshot",
360 },
361 },
362 );
363 }
364
365 const bundle: BundleInfo | undefined = snapshot.bundle;
366 if (bundle === undefined) return undefined;
367 // Read the request body from a CLONE of the platform request so a
368 // fall-through still forwards the original, unconsumed body.
369 const source = request.source;
370 if (!(source instanceof Request)) return undefined;
371 const bodyResult = yield* Effect.result(
372 Effect.tryPromise(() => source.clone().arrayBuffer()),
373 );
374 if (Result.isFailure(bodyResult)) return undefined;
375 const req = yield* decodePktLines(new Uint8Array(bodyResult.success)).pipe(
376 Effect.flatMap(parseUploadPackRequest),
377 Effect.catch(() => Effect.succeed(undefined)),
378 );
379 if (req === undefined) return undefined;
380 if (
381 !req.done ||
382 !bundleCovers(bundle, {
383 wants: req.wants,
384 haves: req.haves,
385 depth: req.depth,
386 clientShallow: req.clientShallow,
387 })
388 ) {
389 return undefined;
390 }
391 return yield* serveBundle({
392 key: bundle.key,
393 refsHash: bundle.refsHash,
394 objectCount: bundle.objectCount,
395 sideband: req.capabilities.has("side-band-64k"),
396 via: "head-snapshot",
397 });
398 });
399
400 const wireProxy = Effect.gen(function* () {
401 const request = yield* HttpServerRequest.HttpServerRequest;
402 const params = yield* HttpRouter.params;
403 const owner = (params.owner ?? "").toLowerCase();
404 let repo = (params.repo ?? "").toLowerCase();
405 if (repo.endsWith(".git")) repo = repo.slice(0, -4);
406
407 const resolved = yield* Effect.result(resolveCached(owner, repo));
408 if (Result.isFailure(resolved)) {
409 return internalError;
410 }
411 const entry = resolved.success;
412 if (entry === undefined) {
413 return notFound;
414 }
415
416 // Reads never wake the DO when the head snapshot covers them: the
417 // middleware in front of the route already decided who may read.
418 const fast = yield* headSnapshotFastPath(request, entry.repoId);
419 if (fast !== undefined) return fast;
420
421 const response = yield* repos.getByName(entry.repoId).fetch(request);
422
423 // Clone-bundle splice (DESIGN.md §11): the DO answered with a marker
424 // naming an immutable R2 object rather than the pack itself. Stream
425 // those bytes to the client from here, so the pack never transits
426 // the Durable Object — this is what makes clone bandwidth scale with
427 // Workers/R2 instead of with one single-threaded object.
428 const bundleKeyHeader = response.headers[BUNDLE_KEY_HEADER];
429 if (bundleKeyHeader === undefined) {
430 return response;
431 }
432 const served = yield* serveBundle({
433 key: bundleKeyHeader,
434 refsHash: response.headers[BUNDLE_HASH_HEADER] ?? "",
435 objectCount: Number(response.headers[BUNDLE_COUNT_HEADER] ?? "0"),
436 sideband: response.headers[BUNDLE_SIDEBAND_HEADER] === "1",
437 via: "do-bundle",
438 });
439 // The bundle vanished (GC raced us): fall back to a plain 500 — git
440 // retries, and the next attempt re-plans against a fresh bundle or
441 // the dynamic path.
442 return served ?? internalError;
443 });
444
445 /**
446 * POST git-receive-pack — the push pipeline (DESIGN §22.10). It runs
447 * HERE, in the stateless Worker: the body streams into the hasher
448 * fan-out and the blob-store spill as it arrives, and the Repo DO
449 * receives only staged ROWS (`stagePush`, promoted rows are coordinates
450 * into the wire pack) and the commit summary (`commitPush`). No pack
451 * byte enters the Durable Object, so its memory, CPU and egress are
452 * untouched by push size.
453 */
454 const receivePackRoute = Effect.scoped(
455 Effect.gen(function* () {
456 const request = yield* HttpServerRequest.HttpServerRequest;
457 const params = yield* HttpRouter.params;
458 const repo = yield* engine.repositories.get({
459 owner: params.owner ?? "",
460 repo: params.repo ?? "",
461 });
462 const push = yield* ReceivePackHttp.decode(request);
463 if (push._tag === "Probe") return ReceivePackHttp.probeResponse();
464 return yield* Effect.gen(function* () {
465 const prepared = yield* engine.preparePush(repo, push.input);
466 return ReceivePackHttp.response(
467 push,
468 yield* engine.commitPush(prepared),
469 );
470 }).pipe(
471 Effect.catchTag("PushDenied", (error) =>
472 Effect.succeed(ReceivePackHttp.reject(push, error.reason)),
473 ),
474 );
475 }),
476 ).pipe(
477 Effect.catchTag("RepoNotFound", () => Effect.succeed(notFound)),
478 Effect.catchTag("StoreError", (error) =>
479 Effect.succeed(ReceivePackHttp.failure(error.reason)),
480 ),
481 Effect.catchTag(["WireProtocolError", "PackIngestError"], (error) =>
482 Effect.succeed(ReceivePackHttp.failure(error.reason)),
483 ),
484 );
485
486 /** Auth + resolve for the raw REST reads; `undefined` = already replied. */
487 const rawRestPrelude = (ownerRaw: string, repoRaw: string) =>
488 Effect.gen(function* () {
489 const request = yield* HttpServerRequest.HttpServerRequest;
490 const resolved = yield* Effect.result(resolveCached(ownerRaw, repoRaw));
491 if (Result.isFailure(resolved)) {
492 return { kind: "halt", response: internalError } as const;
493 }
494 if (resolved.success === undefined) {
495 return { kind: "halt", response: notFound } as const;
496 }
497 return { kind: "ok", entry: resolved.success } as const;
498 });
499
500 /**
501 * `GET /api/v1/repos/:owner/:repo/blobs/:oid/raw` — raw blob bytes,
502 * octet-stream, no size cap (the per-object 64 MiB ingest cap is the
503 * outer bound). Outside HttpApi schema-land by design (DESIGN.md §5).
504 */
505 const blobRawRoute = Effect.gen(function* () {
506 const params = yield* HttpRouter.params;
507 const prelude = yield* rawRestPrelude(
508 params.owner ?? "",
509 params.repo ?? "",
510 );
511 if (prelude.kind === "halt") return prelude.response;
512 return yield* repos
513 .getByName(prelude.entry.repoId)
514 .readObject({ oid: params.oid ?? "", expect: "blob" })
515 .pipe(
516 Effect.map((data) =>
517 HttpServerResponse.uint8Array(data.content, {
518 contentType: "application/octet-stream",
519 }),
520 ),
521 Effect.catchTag(["RepoNotFound", "ObjectNotFound"], () =>
522 Effect.succeed(HttpServerResponse.text("not found", { status: 404 })),
523 ),
524 Effect.catchTag("WrongObjectType", (error) =>
525 Effect.succeed(
526 HttpServerResponse.text(
527 `object ${error.oid} is a ${error.actual}, not a ${error.expected}`,
528 { status: 422 },
529 ),
530 ),
531 ),
532 Effect.catchTag("StoreError", () => Effect.succeed(internalError)),
533 );
534 });
535
536 /**
537 * `GET /api/v1/repos/:owner/:repo/file?ref=<refname|oid>&path=<path>` —
538 * file-at-path bytes via tree walk, octet-stream (DESIGN.md §2.2).
539 */
540 const fileRoute = Effect.gen(function* () {
541 const request = yield* HttpServerRequest.HttpServerRequest;
542 const params = yield* HttpRouter.params;
543 const url = new URL(request.url, "http://worker");
544 const path = url.searchParams.get("path");
545 const ref = url.searchParams.get("ref") ?? undefined;
546 if (path === null || path.length === 0) {
547 return HttpServerResponse.text("missing ?path", { status: 400 });
548 }
549 const prelude = yield* rawRestPrelude(
550 params.owner ?? "",
551 params.repo ?? "",
552 );
553 if (prelude.kind === "halt") return prelude.response;
554 return yield* repos
555 .getByName(prelude.entry.repoId)
556 .readFileAtPath({ ref, path })
557 .pipe(
558 Effect.map((file) =>
559 HttpServerResponse.uint8Array(file.content, {
560 contentType: "application/octet-stream",
561 }),
562 ),
563 Effect.catchTag(["RepoNotFound", "RefNotFound", "ObjectNotFound"], () =>
564 Effect.succeed(HttpServerResponse.text("not found", { status: 404 })),
565 ),
566 Effect.catchTag("StoreError", () => Effect.succeed(internalError)),
567 );
568 });
569
570 // GitHub REST v3 compatibility facade (`gh api`, Octokit): reuses the
571 // same prelude + DO stubs; auth enforcement stays in the DO.
572 const githubRoutes = gitHubCompatRoutes({
573 prelude: rawRestPrelude,
574 stub: (repoId) => repos.getByName(repoId),
575 });
576
577 /**
578 * The push pipeline's hashing endpoint (DESIGN §22.7): a spilled part (plus
579 * carry) in the body, coordinates in the query; the scan result in the
580 * binary form `Hasher.ts` decodes. Internal: admin-authenticated, reached
581 * through the self service binding.
582 */
583 const hashPartRoute = Effect.gen(function* () {
584 const request = yield* HttpServerRequest.HttpServerRequest;
585 const presented = parseBearer(request.headers);
586 const expected = Redacted.value(yield* internalSecret);
587 if (
588 presented === undefined ||
589 !(yield* timingSafeEqual(presented, expected))
590 ) {
591 return HttpServerResponse.text("forbidden", { status: 403 });
592 }
593 const query = new URL(request.url, "http://x").searchParams;
594 const base = Number(query.get("base"));
595 const maxObjectSize = Number(query.get("max"));
596 const boundsMode = query.get("mode") === "bounds";
597 const remaining = boundsMode ? 0 : Number(query.get("remaining"));
598 if (![base, remaining, maxObjectSize].every(Number.isFinite)) {
599 return HttpServerResponse.text("bad coordinates", { status: 400 });
600 }
601 const t0 = Date.now();
602 const body = new Uint8Array(yield* request.arrayBuffer);
603 const tBody = Date.now();
604 if (query.get("mode") === "deltas") {
605 const { bases, jobs } = decodeDeltaBatch(body);
606 const resolved = yield* resolveDeltas(bases, jobs, {
607 maxObjectSize,
608 }).pipe(Effect.result);
609 if (Result.isFailure(resolved)) {
610 return HttpServerResponse.text(
611 `${resolved.failure._tag}: ${"reason" in resolved.failure ? resolved.failure.reason : ""}`,
612 { status: 422 },
613 );
614 }
615 return HttpServerResponse.uint8Array(
616 frame(encodeDeltaResults(resolved.success)),
617 { contentType: "application/octet-stream" },
618 );
619 }
620 // A requested spill part uploads concurrently with the scan (DESIGN
621 // §22.10): this isolate is the writer of the part it verifies.
622 const key = query.get("key");
623 const uploadId = query.get("uploadId");
624 const partNumber = Number(query.get("part"));
625 const upload =
626 key !== null && uploadId !== null && Number.isFinite(partNumber)
627 ? // Detached: it outlives the handler fiber, which returns as soon as
628 // the scan is written; the open response stream keeps the
629 // invocation alive until the part frame follows.
630 yield* Effect.forkDetach(
631 workerBlobs
632 .uploadPart(key, uploadId, partNumber, body)
633 .pipe(Effect.provide(RuntimeContext.phantom), Effect.result),
634 )
635 : undefined;
636 const skip = Number(query.get("skip") ?? "0");
637 const result = yield* (
638 boundsMode
639 ? Effect.suspend(() => {
640 const { bounds, payload } = decodeBoundsRequest(body);
641 return hashBounds(payload, bounds, { base, maxObjectSize });
642 })
643 : scanPart(skip > 0 ? body.subarray(skip) : body, {
644 base,
645 remaining,
646 maxObjectSize,
647 resync: query.get("resync") === "1",
648 })
649 ).pipe(Effect.result);
650 if (Result.isFailure(result)) {
651 return HttpServerResponse.text(
652 `${result.failure._tag}: ${"reason" in result.failure ? result.failure.reason : ""}`,
653 { status: 422 },
654 );
655 }
656 const scanFrame = frame(encodeScanResult(result.success));
657 if (upload === undefined) {
658 return HttpServerResponse.uint8Array(scanFrame, {
659 contentType: "application/octet-stream",
660 });
661 }
662 // Two frames: the scan now, the part once its upload has finished —
663 // the open response stream keeps this invocation alive meanwhile.
664 const { readable, writable } = new IdentityTransformStream();
665 const tScan = Date.now();
666 // A plain async writer, like the clone pump: writes settle only as the
667 // response is read, and the open stream keeps this invocation alive
668 // until the part's upload has finished.
669 const partDone = Effect.runPromise(Fiber.join(upload));
670 yield* Effect.sync(() => {
671 void (async () => {
672 const writer = writable.getWriter();
673 try {
674 await writer.write(scanFrame);
675 const part = await partDone;
676 if (body.length > 1 << 20) {
677 console.log(
678 `[hash] bytes=${body.length} body=${tBody - t0}ms scan=${tScan - tBody}ms upload=${Date.now() - tScan}ms part=${partNumber}`,
679 );
680 }
681 if (Result.isFailure(part)) {
682 await writable
683 .abort(new Error(part.failure.reason))
684 .catch(() => {});
685 return;
686 }
687 await writer.write(
688 frame(new TextEncoder().encode(JSON.stringify(part.success))),
689 );
690 await writer.close();
691 } catch (error) {
692 await writable.abort(error).catch(() => {});
693 }
694 })();
695 });
696 return HttpServerResponse.raw(readable, {
697 contentType: "application/octet-stream",
698 });
699 });
700
701 return {
702 repos: reposRoutes,
703 refs: refsRoutes,
704 objects: {
705 ...objectsRoutes,
706 blobRaw: () => blobRawRoute,
707 file: () => fileRoute,
708 },
709 pulls: pullsRoutes,
710 protocol: {
711 infoRefs: () => wireProxy.pipe(Effect.orDie),
712 uploadPack: () => wireProxy.pipe(Effect.orDie),
713 receivePack: () => receivePackRoute.pipe(Effect.orDie),
714 },
715 github: githubRoutes,
716 internal: {
717 hashPart: () => hashPartRoute.pipe(Effect.orDie),
718 },
719 };
720});
721
722/**
723 * Reusable HTTP handlers for each Git API group. Resolve this service while
724 * building an `HttpApiBuilder.group`, then register its handlers with
725 * `handleAll`. Override individual handlers with ordinary object spread.
726 * The storage clients, resolve cache, and push gate are shared by all groups.
727 *
728 * ```typescript
729 * const ReposLive = HttpApiBuilder.group(AppApi, "repos", (h) =>
730 * Effect.map(Git.Handlers, (git) => h.handleAll(git.repos)),
731 * );
732 * ```
733 */
734export class Handlers extends Context.Service<
735 Handlers,
736 Effect.Success<typeof makeCore>
737>()("alchemy/Git/Handlers") {}
738
739/**
740 * Builds reusable Git handlers from the storage and hasher services.
741 *
742 * ### Implementing a group
743 * **Example:** Repository handlers for an application API
744 * ```typescript
745 * const ReposLive = HttpApiBuilder.group(AppApi, "repos", (h) =>
746 * Effect.map(Git.Handlers, (git) => h.handleAll(git.repos)),
747 * ).pipe(Layer.provide(Git.ApiHandlersLive));
748 * ```
749 */
750export const ApiHandlersLive = Layer.effect(Handlers, makeCore).pipe(
751 Layer.provideMerge(EngineLive),
752);
753
754/**
755 * The internal hash group for {@link InternalApi}. {@link InternalApiLive}
756 * registers it with the application router, separately from public middleware.
757 */
758export const InternalLive = HttpApiBuilder.group(InternalApi, "internal", (h) =>
759 Effect.map(Handlers, (git) => h.handleAll(git.internal)),
760);
761
762/** Native Effect group implementations. Merge an overriding group after these when needed. */
763export const GroupsLive = Layer.mergeAll(
764 HttpApiBuilder.group(GitApi, "repos", (h) =>
765 Effect.map(Handlers, (git) => h.handleAll(git.repos)),
766 ),
767 HttpApiBuilder.group(GitApi, "refs", (h) =>
768 Effect.map(Handlers, (git) => h.handleAll(git.refs)),
769 ),
770 HttpApiBuilder.group(GitApi, "objects", (h) =>
771 Effect.map(Handlers, (git) => h.handleAll(git.objects)),
772 ),
773 HttpApiBuilder.group(GitApi, "pulls", (h) =>
774 Effect.map(Handlers, (git) => h.handleAll(git.pulls)),
775 ),
776 HttpApiBuilder.group(GitApi, "protocol", (h) =>
777 Effect.map(Handlers, (git) => h.handleAll(git.protocol)),
778 ),
779 HttpApiBuilder.group(GitApi, "github", (h) =>
780 Effect.map(Handlers, (git) => h.handleAll(git.github)),
781 ),
782);
783
784/**
785 * Git's public routes as an ordinary HttpRouter layer. Merge this with your
786 * application's routes and provide your route middleware to the result.
787 * The application owns its API, HTTP server, platform, and CORS policy.
788 *
789 * ```typescript
790 * const Routes = Layer.mergeAll(AppApiLive, Git.ApiLive).pipe(
791 * Layer.provide(Authentication.layer),
792 * );
793 * ```
794 *
795 * @layer
796 */
797export const ApiLive = HttpApiBuilder.layer(GitApi).pipe(
798 Layer.provide(GroupsLive),
799);
800
801/**
802 * The authenticated internal hashing route. Mount beside public routes,
803 * outside application authentication; the handler checks InternalSecret.
804 *
805 * @layer
806 */
807export const InternalApiLive = HttpApiBuilder.layer(InternalApi).pipe(
808 Layer.provide(InternalLive),
809);
810
811/**
812 * Hosts the `GitRepo` Durable Object (refs, objects, pulls, the wire
813 * protocol) and provides {@link RepoStore} over its namespace. Requires a
814 * {@link BlobStore} — provide `Git.BlobStoreR2(yourBucket)` in the same
815 * layer graph; one provision serves this DO and the Worker-side reads —
816 * plus the {@link Hasher} and the registry.
817 *
818 * @layer
819 * @provides Git.RepoStore
820 */
821export const ReposDurableObject = Layer.effect(
822 RepoStore,
823 Effect.gen(function* () {
824 const namespace = yield* GitRepo;
825 return { getByName: (repoId) => namespace.getByName(repoId) };
826 }),
827).pipe(Layer.provideMerge(GitRepoLive));