clover's git

the-knife

1	/**
2	* The per-repo Durable Object (DESIGN.md §2, §3.4, §3.6, §8).
3	*
4	* One DO per repo, addressed by `getByName(repoId)` (a ULID — never
5	* `owner/name`, so renames never move data). The DO owns refs, the object
6	* index and loose bytes, tokens, the commit graph, push staging, and the
7	* alarm jobs (import / fork / delete-purge / staging-GC). It runs the git
8	* v0 wire protocol in its `fetch` handler — advertisement, upload-pack,
9	* receive-pack — because every byte those need lives in this DO's SQLite.
10	*
11	* NOTE: the protocol *choreography* (advertisement builder, request
12	* parsers, pack writer/parser) is implemented in this file against the
13	* pure codec primitives in `src/Git/Protocol/` (`Pkt`, `Sideband`, `ObjectCodec`,
14	* `Delta`, `Zlib`, `Store`). DESIGN.md places these in
15	* `src/git/{Advertise,UploadPack,ReceivePack,PackParser,PackWriter}.ts`;
16	* they live here (exported) until the git-layer split lands — moving them
17	* is a mechanical extraction.
18	*
19	* ### Wire endpoints
20	* The Worker proxies these untouched to `stub.fetch`:
21	* **Example:** Routes served by this DO's fetch handler
22	* ```
23	* GET /:owner/:repo[.git]/info/refs?service=git-(upload|receive)-pack
24	* POST /:owner/:repo[.git]/git-upload-pack
25	* POST /:owner/:repo[.git]/git-receive-pack
26	* ```
27	*/
28	import * as Cloudflare from "../Cloudflare/index.ts";
29	import type { HttpEffect } from "../Http.ts";
30	import { RuntimeContext } from "../RuntimeContext.ts";
31	import * as Context from "effect/Context";
32	import * as Effect from "effect/Effect";
33	import * as Fiber from "effect/Fiber";
34	import * as Option from "effect/Option";
35	import * as Redacted from "effect/Redacted";
36	import * as Result from "effect/Result";
37	import * as Schema from "effect/Schema";
38	import * as Semaphore from "effect/Semaphore";
39	import * as Stream from "effect/Stream";
40	import * as FetchHttpClient from "effect/http/FetchHttpClient";
41	import * as HttpServerRequest from "effect/http/HttpServerRequest";
42	import * as HttpServerResponse from "effect/http/HttpServerResponse";
43	import {
44	BranchMissing,
45	MergeConflict,
46	NoMergeBase,
47	NothingToMerge,
48	ObjectNotFound,
49	PullExists,
50	PullNotFound,
51	PullStateConflict,
52	ReadOnlyRepo,
53	RefConflict,
54	RefNotFound,
55	RepoNotFound,
56	ValidationError,
57	WrongObjectType,
58	type Oid as ApiOid,
59	type RepoStatus,
60	} from "./Api.ts";
61	import { applyDelta } from "./Protocol/Delta.ts";
62	import * as PackParser from "./Protocol/PackParser.ts";
63	import type { RandomAccess } from "./Protocol/PackParser.ts";
64	import {
65	bytesToHex,
66	concatBytes,
67	decodeOfsDeltaOffset,
68	decodeTypeSize,
69	encodeCommit,
70	encodeTypeSize,
71	hashObject,
72	isDeltaType,
73	isOid,
74	makeSha1,
75	ObjectType,
76	objectTypeName,
77	parseCommit,
78	parseTag,
79	parseTree,
80	treeEntryKind,
81	utf8Encode,
82	ZERO_OID,
83	type ObjectTypeName,
84	type Oid,
85	} from "./Protocol/ObjectCodec.ts";
86	import {
87	decodePktLines,
88	errPkt,
89	flushPkt,
90	pktPayloadText,
91	pktText,
92	readPktLineAt,
93	type PktLine,
94	} from "./Protocol/Pkt.ts";
95	import {
96	progressMessage,
97	pumpPackBody,
98	sidebandFrames,
99	} from "./Protocol/Sideband.ts";
100	import {
101	StoreError,
102	type ManifestEntry,
103	type ObjectSource,
104	} from "./Protocol/Store.ts";
105	import {
106	applyTreeChanges,
107	conflictingPaths,
108	diffTrees,
109	type DiffEntryData,
110	} from "./Protocol/TreeDiff.ts";
111	import * as Zlib from "./Protocol/Zlib.ts";
112	import { runImport, type ImportSource } from "./Jobs/Import.ts";
113	import { runForkJob, snapshotStream, type SnapshotChunk } from "./Jobs/Fork.ts";
114	import { bundleCovers, runBundleJob, type BundleInfo } from "./Jobs/Bundle.ts";
115	import {
116	runCompactJob,
117	runGeometricMergeJob,
118	shouldCompact,
119	} from "./Jobs/Compact.ts";
120	import { headKey, incomingKey, packKeyOf, wirePackId } from "./Store/Keys.ts";
121	import { encodeHeadSnapshot, type HeadSnapshot } from "./Store/HeadSnapshot.ts";
122	import { blobRandomAccess, sliceRandomAccess } from "./Store/PackSource.ts";
123	import { SPILL_PART_BYTES } from "./Store/IncomingBody.ts";
124	import type { BlobMultipart, UploadedPart } from "./BlobStore.ts";
125	import type { StreamingFeeder } from "./Store/StreamingSource.ts";
126	import {
127	BACKPRESSURE_BYTES,
128	makeStreamingSource,
129	RETAIN_BYTES,
130	} from "./Store/StreamingSource.ts";
131	import {
132	Hasher,
133	type HasherShape,
134	type HashPartResult,
135	} from "./Hasher/Hasher.ts";
136	import * as PartialScan from "./Protocol/PartialScan.ts";
137	import type { ScanResult } from "./Protocol/PartialScan.ts";
138	import type {
139	DeltaBase,
140	DeltaJob,
141	UnresolvedDelta,
142	} from "./Protocol/PartialScan.ts";
143	import { BlobStore, type BlobStoreShape } from "./BlobStore.ts";
144	import { runPurgeJob } from "./Jobs/Purge.ts";
145	import { RegistryStore, ulid } from "./RegistryObject.ts";
146	import { decodeStagedBatch } from "./PushWire.ts";
147	import { computeClosure } from "./Store/Closure.ts";
148	import {
149	makeObjectStore,
150	MAX_OBJECT_SIZE,
151	type ObjectStore,
152	type StagedObject,
153	LIVE_OBJECTS,
154	} from "./Store/ObjectStore.ts";
155	import type * as cf from "@cloudflare/workers-types";
156	import {
157	initRepoSchema,
158	makeSqlClient,
159	type JobRow,
160	type PullRow,
161	type RefRow,
162	} from "./Store/Sql.ts";
163
164	// ─────────────────────────────────────────────────────────────────────────────
165	// Constants
166	// ─────────────────────────────────────────────────────────────────────────────
167
168	/**
169	* Pushes up to this size are ingested straight from memory; larger ones are
170	* parked in R2 and parsed from there (DESIGN.md §3.6). A **buffering
171	* threshold, not a limit** — a push larger than this is no longer rejected.
172	*
173	* The spill threshold of the streaming receive (`receiveWireBody`): bodies
174	* up to this size are buffered and parsed from memory (the fast path, at
175	* 50 MiB the bound v1 validated for a 128 MB isolate); larger bodies are
176	* never materialized — they stream into an R2 multipart upload as they
177	* arrive and are parsed back through bounded windows.
178	*
179	* Measured on the 38 MiB alchemy-repo push: ~20 s in memory, ~37 s through
180	* R2. The R2 path costs about 2x, and is unbounded — the full 67 MiB
181	* alchemy history (44k objects) pushed in 92 s.
182	*/
183	export const MAX_PACK_BYTES = 4 * 1024 * 1024;
184	// 4 MiB (was 50, then 24): with promoted wire packs (DESIGN §22.5) the
185	// spilled path is the FAST one — it never writes blob bytes to SQLite, so
186	// staging and the staged→live flip cost milliseconds instead of seconds
187	// (a 5.7 MiB in-memory push spent 6 s in finalize on production; a 40 MiB
188	// spilled one 0.7 s). Only genuinely small pushes stay in memory.
189	// 24 MiB, down from the 50 MiB v1 validated with ONE active repo: instances
190	// of this class share an isolate, so the in-memory pack, the ingest's
191	// resolved-content LRU (20 MiB) and staging batch (8 MiB) sit beside every
192	// other active repo's working set. DESIGN §22.4 records the OOM that
193	// motivated it; the R2 path costs ~2x on the bytes above the threshold.
194
195	/** Objects staged per SQL transaction during ingest (DESIGN.md §16.6). */
196	/** Bytes per hasher part — the spill part size, so a part is one R2 part (DESIGN §22.7). */
197	export const HASH_PART_BYTES = 4 * 1024 * 1024;
198	/** Hasher calls in flight per push (DESIGN §22.8). */
199	export const HASH_CONCURRENCY = 8;
200	// 4 MiB (was 8): the hasher chain is sequential (each part needs the
201	// previous part's boundary), so a smaller part shortens both the tail after
202	// the upload ends and the granularity at which the upload can run ahead.
203
204	export const STAGE_BATCH_OBJECTS = 2048;
205	// 2048 (was 256): measured on production, staging cost tracked the number
206	// of transaction COMMITs (~100 ms each), not rows — 61 commits for a
207	// 15.6k-object push. Objects average a few KiB, so 2048 stays well under
208	// STAGE_BATCH_BYTES; the byte cap still bounds memory for large objects.
209
210	/** Byte budget per staging batch, so a batch of large blobs stays bounded. */
211	export const STAGE_BATCH_BYTES = 8 * 1024 * 1024;
212
213	// The tree-diff caps (`MAX_DIFF_FILES`, `MAX_DIFF_TREE_READS`) live with
214	// the pure walk in `Protocol/TreeDiff.ts`; re-exported here alongside the other
215	// serving-plane bounds.
216	export { MAX_DIFF_FILES, MAX_DIFF_TREE_READS } from "./Protocol/TreeDiff.ts";
217
218	/** Max commits popped by the merge-base paint walk. */
219	export const MAX_COMPARE_WALK = 10_000;
220
221	/** Max commits returned in a Comparison's `commits` array. */
222	export const MAX_COMPARE_COMMITS = 250;
223
224	/**
225	* Max commits popped by the PR live-field walk (`getPull`). Saturation
226	* makes the numeric detail fields `null` (reason `"unknown"`); merge still
227	* works because {@link MAX_PULL_MERGE_WALK} gives the merge path a deeper
228	* budget.
229	*/
230	export const MAX_PULL_WALK = 5_000;
231
232	/** Deeper walk budget for `mergePull` (needs only reachability + one base). */
233	export const MAX_PULL_MERGE_WALK = 50_000;
234
235	/** Conflicting paths reported on a `MergeConflict` (the set may be larger). */
236	export const MAX_CONFLICT_PATHS = 20;
237
238	/** Agent string advertised on the wire. */
239	export const GIT_AGENT = "git-service/1";
240
241	/** How long a push waits for ingest memory before answering 503. */
242	export const PUSH_WAIT_TIMEOUT = "30 seconds";
243
244	/**
245	* Memory the Repo DO lends to concurrent push ingests, in MiB — one
246	* permit per MiB (DESIGN.md §2.1).
247	*
248	* The semaphore exists for MEMORY, not correctness: a DO is
249	* single-threaded, but its input gate is only closed across *storage*
250	* awaits, and receive-pack awaits the network and blob storage — so two
251	* pushes genuinely interleave. Correctness across that interleave is
252	* already handled by data scoping (`staged_push = pushId`, and every
253	* read filtering `staged_push IS NULL`) plus a synchronous
254	* `transactionSync` finalize. What is left is that two concurrent
255	* 50 MiB buffers would exhaust the 128 MB isolate.
256	*
257	* Bounding the resource rather than the count is what lets pushes to
258	* different branches proceed together: a normal push is kilobytes and
259	* takes one permit, so it no longer queues behind a large one.
260	*/
261	export const PUSH_MEMORY_BUDGET_MB = 64;
262
263	let sharedPushGate: Semaphore.Semaphore | undefined;
264	/**
265	* The push-admission semaphore, ONE per isolate (DESIGN §21.2, §22.4). DO
266	* instances of a class share an isolate, so a per-instance gate would let
267	* N repos each admit a full budget's worth of concurrent pushes.
268	*/
269	export const isolatePushGate: Effect.Effect<Semaphore.Semaphore> =
270	Effect.suspend(() =>
271	sharedPushGate === undefined
272	? Semaphore.make(PUSH_MEMORY_BUDGET_MB).pipe(
273	Effect.tap((gate) =>
274	Effect.sync(() => {
275	sharedPushGate = gate;
276	}),
277	),
278	)
279	: Effect.succeed(sharedPushGate),
280	);
281
282	/**
283	* Permits a push must reserve: its buffered ceiling in MiB. A body
284	* larger than {@link MAX_PACK_BYTES} spills to blob storage instead of
285	* growing in memory, so that is the cap; an absent `content-length`
286	* (chunked upload) is charged the same worst case.
287	*/
288	export const pushPermitsFor = (contentLength: number | undefined): number => {
289	const mib = (bytes: number) => Math.ceil(bytes / (1024 * 1024));
290	// What a push can hold at once, in MiB — never its body. An unknown
291	// length is treated as spilled. Memory safety beats concurrency: on a
292	// 64 MiB budget that admits one large push per isolate alongside small ones.
293	// Streaming (DESIGN §22.6–7): the feeder retains RETAIN_BYTES behind the
294	// pump and buffers BACKPRESSURE_BYTES ahead of it, plus one staging batch.
295	const spilled = mib(RETAIN_BYTES + BACKPRESSURE_BYTES + STAGE_BATCH_BYTES);
296	// In memory the caches cannot outgrow what the (small) body can inflate
297	// to, so charge the body twice (compressed + inflated) plus one for the
298	// batch, rather than the caches' ceilings.
299	const permits =
300	contentLength === undefined ||
301	!Number.isFinite(contentLength) ||
302	contentLength > MAX_PACK_BYTES
303	? spilled
304	: mib(Math.max(contentLength, 0)) * 2 + 1;
305	return Math.max(1, Math.min(PUSH_MEMORY_BUDGET_MB, permits));
306	};
307
308	/** How long crashed push staging survives before the GC alarm reaps it. */
309	export const STAGING_TTL_MS = 24 * 60 * 60 * 1000;
310
311	/** The realm sent on wire 401s. */
312	export const WWW_AUTHENTICATE = 'Basic realm="git-service"' as const;
313
314	/**
315	* Worker↔DO headers for the clone-bundle splice (DESIGN.md §11): the DO
316	* answers an eligible clone with these *instead of* the pack, and the
317	* Worker streams the named R2 object to the client. Only
318	* {@link BUNDLE_HASH_HEADER} is kept on the client-visible response, as
319	* observable proof the fast path served the clone.
320	*/
321	export const BUNDLE_KEY_HEADER = "x-git-bundle-key" as const;
322	export const BUNDLE_HASH_HEADER = "x-git-bundle" as const;
323	export const BUNDLE_COUNT_HEADER = "x-git-bundle-count" as const;
324	export const BUNDLE_SIDEBAND_HEADER = "x-git-bundle-sideband" as const;
325
326	// ─────────────────────────────────────────────────────────────────────────────
327	// Errors
328	// ─────────────────────────────────────────────────────────────────────────────
329
330	/**
331	* A push pack failed to ingest (malformed, oversize, checksum mismatch,
332	* missing thin base, delta error). Reported in-band as `unpack <reason>` +
333	* per-ref `ng` — never as an HTTP error.
334	*/
335	export class PackIngestError extends Schema.TaggedError<PackIngestError>()(
336	"PackIngestError",
337	{ reason: Schema.String },
338	) {}
339
340	/**
341	* A syntactically valid request that the v0 grammar does not allow
342	* (missing wants, malformed command lines, unsupported deepen flavors).
343	* Answered with an `ERR` pkt.
344	*/
345	export class WireProtocolError extends Schema.TaggedError<WireProtocolError>()(
346	"WireProtocolError",
347	{ reason: Schema.String },
348	) {}
349
350	// ─────────────────────────────────────────────────────────────────────────────
351	// RPC data shapes (plain serializable data — no Schema classes over RPC)
352	// ─────────────────────────────────────────────────────────────────────────────
353
354	/** Repo metadata as stored in the `config` table. */
355	export interface RepoMetaData {
356	readonly repoId: string;
357	readonly owner: string;
358	readonly name: string;
359	readonly defaultBranch: string;
360	readonly description: string | null;
361	readonly readOnly: boolean;
362	/** Anyone can read/clone without a token (GitHub public-repo model). */
363	readonly public: boolean;
364	readonly forkOf: string | null;
365	readonly status: RepoStatus;
366	readonly createdAt: number;
367	/** Object storage breakdown (DESIGN.md §12.1). */
368	readonly objects: ObjectStatsData;
369	/** Server-side timing of the last push (DESIGN.md §19 phase 0). */
370	readonly lastPush: PushStatsData | null;
371	}
372
373	/** `beginPush` input: the parsed ref commands (DESIGN §22.10). */
374	export interface BeginPushInput {
375	readonly commands: ReadonlyArray<RefCommand>;
376	}
377	/** `beginPush` outcome. `denied` is reported per ref (`ng <ref> <reason>`). */
378	export type BeginPushResult =
379	| { readonly _tag: "ok"; readonly pushId: string }
380	| { readonly _tag: "denied"; readonly reason: string };
381	/** One ref's outcome in a push report. */
382	export interface PushRefResult {
383	readonly ref: string;
384	readonly ok: boolean;
385	readonly reason?: string | undefined;
386	}
387	/** `commitPush` input: what the Worker-side ingest learned (DESIGN §22.10). */
388	export interface CommitPushInput {
389	readonly pushId: string;
390	readonly commands: ReadonlyArray<RefCommand>;
391	readonly atomic: boolean;
392	readonly commits: ReadonlyArray<StagedCommitInfo>;
393	readonly referenced: ReadonlyArray<Oid>;
394	readonly referencedParents: ReadonlyArray<Oid>;
395	/** Rows staged as references into the wire pack (it must then outlive the push). */
396	readonly promoted: number;
397	readonly stats: {
398	readonly objects: number;
399	readonly bytes: number;
400	readonly ingestMs: number;
401	readonly stageMs: number;
402	readonly phases: Record<string, number> | undefined;
403	};
404	}
405	/** `commitPush` outcome: the `unpack` line and per-ref results. */
406	export interface CommitPushResult {
407	readonly unpack: string;
408	readonly results: ReadonlyArray<PushRefResult>;
409	}
410
411	/** Server-side push timing — see the REST `PushStats` schema. */
412	export interface PushStatsData {
413	readonly objects: number;
414	readonly bytes: number;
415	readonly ingestMs: number;
416	readonly stageMs: number;
417	readonly connectivityMs: number;
418	readonly finalizeMs: number;
419	readonly totalMs: number;
420	/** Ingest split by phase (ms) — see the REST `PushStats.phases` caveat. */
421	readonly phases?: Record<string, number> | undefined;
422	}
423
424	/** Where a repo's objects live — see the REST `ObjectStats` schema. */
425	export interface ObjectStatsData {
426	/** Blobs still in SQLite rows, awaiting compaction into an R2 pack. */
427	readonly loose: number;
428	/** Commits, trees and tags: always SQLite rows (never packed). */
429	readonly resident: number;
430	/** Blobs in R2 packs. */
431	readonly packed: number;
432	readonly r2: number;
433	readonly bytes: number;
434	}
435
436	/** Patch accepted by {@link GitRepoShape.updateRepoMeta}. */
437	export interface RepoMetaPatch {
438	readonly description?: string | null | undefined;
439	/** Must resolve to an existing branch. */
440	readonly defaultBranch?: string | undefined;
441	readonly readOnly?: boolean | undefined;
442	/** Anyone can read/clone without a token when `true`. */
443	readonly public?: boolean | undefined;
444	}
445
446	/** One ref (peeled target present for annotated tags). */
447	export interface RefData {
448	readonly name: string;
449	readonly oid: string;
450	readonly peeled?: string | undefined;
451	}
452
453	/** Result of {@link GitRepoShape.listRefs}. */
454	export interface RefsPage {
455	/** The default branch's full ref name, `null` on an unborn repo. */
456	readonly head: string | null;
457	readonly refs: ReadonlyArray<RefData>;
458	}
459
460	/** Input of {@link GitRepoShape.initRepo} (also fork/import bootstrap). */
461	export interface InitRepoInput {
462	readonly repoId: string;
463	readonly owner: string;
464	readonly name: string;
465	readonly defaultBranch: string;
466	readonly description: string | null;
467	readonly readOnly: boolean;
468	/** Anyone can read/clone without a token when `true`. */
469	readonly public: boolean;
470	readonly forkOf: string | null;
471	}
472
473	/** Result of init/fork/import bootstrap: the meta + a bootstrap token. */
474	export interface InitRepoResult {
475	readonly meta: RepoMetaData;
476	}
477
478	/** Input of {@link GitRepoShape.updateRef}. */
479	export interface UpdateRefInput {
480	readonly name: string;
481	readonly newOid: string;
482	/**
483	* CAS token: `string` = must currently equal; `null` = must not exist;
484	* absent = unconditional.
485	*/
486	readonly expectedOid?: string | null | undefined;
487	}
488
489	/** Input of {@link GitRepoShape.removeRef}. */
490	export interface RemoveRefInput {
491	readonly name: string;
492	readonly expectedOid?: string | undefined;
493	}
494
495	/** Input of {@link GitRepoShape.readObject}. */
496	export interface ReadObjectInput {
497	readonly oid: string;
498	/** When set, a different actual type fails with `WrongObjectType`. */
499	readonly expect?: ObjectTypeName | undefined;
500	}
501
502	/** A fully-read object (content inflated, ≤ 64 MiB by the ingest cap). */
503	export interface ObjectData {
504	readonly oid: string;
505	readonly type: ObjectTypeName;
506	readonly size: number;
507	readonly content: Uint8Array;
508	}
509
510	/** One commit for the REST read surface. */
511	export interface CommitData {
512	readonly oid: string;
513	readonly tree: string;
514	readonly parents: ReadonlyArray<string>;
515	readonly author: SignatureData;
516	readonly committer: SignatureData;
517	readonly message: string;
518	}
519
520	/** An author/committer signature. */
521	export interface SignatureData {
522	readonly name: string;
523	readonly email: string;
524	readonly date: number;
525	readonly tz: string;
526	}
527
528	/** Input of {@link GitRepoShape.readCommitLog}. */
529	export interface CommitLogInput {
530	/** Refname or 40-hex oid; default = HEAD (the default branch). */
531	readonly ref?: string | undefined;
532	readonly cursor?: string | undefined;
533	readonly limit?: number | undefined;
534	}
535
536	/** One page of {@link GitRepoShape.readCommitLog}. */
537	export interface CommitLogPage {
538	readonly items: ReadonlyArray<CommitData>;
539	readonly nextCursor: string | null;
540	readonly hasMore: boolean;
541	}
542
543	// `DiffEntryData` (one changed file) is defined next to the pure walk in
544	// `Protocol/TreeDiff.ts`; re-exported here with the other RPC data shapes.
545	export type { DiffEntryData } from "./Protocol/TreeDiff.ts";
546
547	/** Result of {@link GitRepoShape.readCommitDiff}. */
548	export interface CommitDiffData {
549	readonly oid: string;
550	readonly parent: string | null;
551	readonly files: ReadonlyArray<DiffEntryData>;
552	readonly truncated: boolean;
553	}
554
555	/** Input of {@link GitRepoShape.compareCommits}. */
556	export interface CompareInput {
557	/** Refname or 40-hex oid. */
558	readonly base: string;
559	readonly head: string;
560	}
561
562	/** Result of {@link GitRepoShape.compareCommits}. */
563	export interface CompareData {
564	readonly base: string;
565	readonly head: string;
566	readonly mergeBase: string;
567	readonly aheadBy: number;
568	readonly behindBy: number;
569	readonly commits: ReadonlyArray<CommitData>;
570	readonly commitsTruncated: boolean;
571	readonly files: ReadonlyArray<DiffEntryData>;
572	readonly filesTruncated: boolean;
573	}
574
575	/** Input of {@link GitRepoShape.readFileAtPath}. */
576	export interface ReadFileInput {
577	/** Refname or 40-hex oid; default = HEAD. */
578	readonly ref?: string | undefined;
579	/** Slash-separated path inside the tree. */
580	readonly path: string;
581	}
582
583	/** A file read via tree walk. */
584	export interface FileData {
585	readonly oid: string;
586	/** Octal mode string, e.g. `100644`. */
587	readonly mode: string;
588	readonly size: number;
589	readonly content: Uint8Array;
590	}
591
592	/** One pull-request row (see the REST `Pull` schema). */
593	export interface PullData {
594	readonly number: number;
595	readonly title: string;
596	readonly body: string | null;
597	/** Full base ref name, e.g. `refs/heads/main`. */
598	readonly baseRef: string;
599	/** Full head ref name, e.g. `refs/heads/feature`. */
600	readonly headRef: string;
601	readonly state: "open" | "closed" | "merged";
602	readonly createdAt: number;
603	readonly updatedAt: number;
604	readonly mergedAt: number | null;
605	/** FF: the head tip; the merge commit otherwise. Set iff state=merged. */
606	readonly mergeCommit: string | null;
607	}
608
609	/** Detail = row + live compare fields (`null` = uncomputable, see REST docs). */
610	export interface PullDetailData extends PullData {
611	readonly baseOid: string | null;
612	readonly headOid: string | null;
613	readonly mergeBase: string | null;
614	readonly aheadBy: number | null;
615	readonly behindBy: number | null;
616	readonly mergeable: boolean | null;
617	readonly mergeableReason:
618	| "ff"
619	| "merge-commit"
620	| "conflict"
621	| "up-to-date"
622	| "unknown"
623	| null;
624	}
625
626	/** Input of {@link GitRepoShape.createPull}. */
627	export interface CreatePullInput {
628	readonly title: string;
629	readonly body?: string | undefined;
630	/** Short or full branch name; normalized to `refs/heads/…` here. */
631	readonly base: string;
632	readonly head: string;
633	}
634
635	/** Input of {@link GitRepoShape.listPulls}. */
636	export interface ListPullsInput {
637	readonly state?: "open" | "closed" | "merged" | "all" | undefined;
638	/** Last-seen PR number as a string (keyset pagination, newest first). */
639	readonly cursor?: string | undefined;
640	/** Default 20, max 100. */
641	readonly limit?: number | undefined;
642	}
643
644	/** One page of {@link GitRepoShape.listPulls}. */
645	export interface PullsPage {
646	readonly items: ReadonlyArray<PullData>;
647	readonly nextCursor: string | null;
648	readonly hasMore: boolean;
649	}
650
651	/** Input of {@link GitRepoShape.updatePull}. */
652	export interface UpdatePullInput {
653	readonly number: number;
654	readonly title?: string | undefined;
655	/** `null` clears the body. */
656	readonly body?: string | null | undefined;
657	readonly state?: "open" | "closed" | undefined;
658	}
659
660	/** Input of {@link GitRepoShape.mergePull}. */
661	export interface MergePullInput {
662	readonly number: number;
663	/** Merge-commit message; ignored on fast-forward. */
664	readonly message?: string | undefined;
665	/** Race guard: fail RefConflict when the head tip moved since inspection. */
666	readonly expectedHeadOid?: string | undefined;
667	/** Race guard for application inspection of the base branch. */
668	readonly expectedBaseOid?: string | undefined;
669	}
670
671	/** Result of {@link GitRepoShape.mergePull}. */
672	export interface MergePullResult {
673	readonly method: "ff" | "merge-commit";
674	/** The oid the base ref now points at. */
675	readonly oid: string;
676	readonly pull: PullData;
677	}
678
679	/** Input of {@link GitRepoShape.startImport}. */
680	export interface StartImportInput extends InitRepoInput {
681	readonly source: ImportSource;
682	}
683
684	/** Input of {@link GitRepoShape.startFork}. */
685	export interface StartForkInput extends InitRepoInput {
686	readonly parentRepoId: string;
687	}
688
689	/** The common auth+repo error union of most RPC methods. */
690	export type RepoAuthError = RepoNotFound | StoreError;
691
692	/**
693	* The Repo DO's typed RPC surface. All methods are `RuntimeContext`-colored
694	* (callable only through a stub / inside the deployed Worker). Note: no
695	* method may be named `delete` (Cloudflare stub proxy reserves it).
696	*/
697	export interface GitRepoShape {
698	/** Seeds config (repo create). */
699	readonly initRepo: (
700	input: InitRepoInput,
701	) => Effect.Effect<InitRepoResult, StoreError, RuntimeContext>;
702	/** Seeds config with `status: importing`, arms the import alarm job. */
703	readonly startImport: (
704	input: StartImportInput,
705	) => Effect.Effect<InitRepoResult, StoreError, RuntimeContext>;
706	/** Seeds config with `status: forking`, arms the fork alarm job. */
707	readonly startFork: (
708	input: StartForkInput,
709	) => Effect.Effect<InitRepoResult, StoreError, RuntimeContext>;
710	/** Flips `status: deleting` and arms the purge alarm job. */
711	/**
712	* Arms a compaction run now (admin): loose object bytes are moved into an
713	* immutable R2 pack by the alarm (DESIGN.md §12.1). Normally armed
714	* automatically by the post-push size check.
715	*/
716	readonly startCompact: () => Effect.Effect<
717	void,
718	RepoAuthError,
719	RuntimeContext
720	>;
721	readonly startPurge: () => Effect.Effect<void, RepoAuthError, RuntimeContext>;
722	readonly getRepoMeta: () => Effect.Effect<
723	RepoMetaData,
724	RepoAuthError,
725	RuntimeContext
726	>;
727	/** Repo metadata for the Worker's own bookkeeping. */
728	readonly readMeta: () => Effect.Effect<
729	RepoMetaData,
730	RepoNotFound | StoreError,
731	RuntimeContext
732	>;
733	readonly updateRepoMeta: (
734	patch: RepoMetaPatch,
735	) => Effect.Effect<RepoMetaData, RepoAuthError | RefNotFound, RuntimeContext>;
736	readonly listRefs: (
737	prefix?: string | undefined,
738	) => Effect.Effect<RefsPage, RepoAuthError, RuntimeContext>;
739	readonly getRef: (
740	name: string,
741	) => Effect.Effect<RefData, RepoAuthError | RefNotFound, RuntimeContext>;
742	readonly updateRef: (
743	input: UpdateRefInput,
744	) => Effect.Effect<
745	RefData,
746	RepoAuthError | RefConflict | ObjectNotFound | ReadOnlyRepo,
747	RuntimeContext
748	>;
749	readonly removeRef: (
750	input: RemoveRefInput,
751	) => Effect.Effect<
752	void,
753	RepoAuthError | RefNotFound | RefConflict | ReadOnlyRepo,
754	RuntimeContext
755	>;
756	readonly readObject: (
757	input: ReadObjectInput,
758	) => Effect.Effect<
759	ObjectData,
760	RepoAuthError | ObjectNotFound | WrongObjectType,
761	RuntimeContext
762	>;
763	readonly readCommitLog: (
764	input: CommitLogInput,
765	) => Effect.Effect<
766	CommitLogPage,
767	RepoAuthError | RefNotFound,
768	RuntimeContext
769	>;
770	/**
771	* The changed-file list of a commit vs its FIRST parent (empty tree for
772	* a root commit) — see the REST `diff` endpoint.
773	*/
774	readonly readCommitDiff: (input: {
775	readonly oid: string;
776	}) => Effect.Effect<
777	CommitDiffData,
778	RepoAuthError | ObjectNotFound | WrongObjectType,
779	RuntimeContext
780	>;
781	/**
782	* Three-dot comparison of two revisions: merge base, ahead/behind
783	* counts, head-side commits, and the mergeBase..head file diff — see
784	* the REST `compare` endpoint.
785	*/
786	readonly compareCommits: (
787	input: CompareInput,
788	) => Effect.Effect<
789	CompareData,
790	| RepoAuthError
791	| RefNotFound
792	| ObjectNotFound
793	| WrongObjectType
794	| NoMergeBase,
795	RuntimeContext
796	>;
797	readonly readFileAtPath: (
798	input: ReadFileInput,
799	) => Effect.Effect<
800	FileData,
801	RepoAuthError | RefNotFound | ObjectNotFound,
802	RuntimeContext
803	>;
804	/** Opens a pull request (same-repo branches, `write` scope). */
805	readonly createPull: (
806	input: CreatePullInput,
807	) => Effect.Effect<
808	PullData,
809	RepoAuthError | BranchMissing | PullExists | ValidationError,
810	RuntimeContext
811	>;
812	/** Lists PRs newest-first with keyset pagination (`read` scope). */
813	readonly listPulls: (
814	input: ListPullsInput,
815	) => Effect.Effect<PullsPage, RepoAuthError, RuntimeContext>;
816	/** Reads one PR with live compare fields recomputed from current tips. */
817	readonly getPull: (
818	number: number,
819	) => Effect.Effect<
820	PullDetailData,
821	RepoAuthError | PullNotFound,
822	RuntimeContext
823	>;
824	/** Patches title/body, or closes/reopens via `state` (`write` scope). */
825	readonly updatePull: (
826	input: UpdatePullInput,
827	) => Effect.Effect<
828	PullData,
829	RepoAuthError | PullNotFound | PullStateConflict,
830	RuntimeContext
831	>;
832	/**
833	* Merges an open PR: fast-forward when possible, else a two-parent merge
834	* commit iff the three-way tree merge is trivial (no path changed on both
835	* sides vs the merge base). Conflicts are a typed 409 — the server never
836	* writes conflict markers.
837	*/
838	readonly mergePull: (
839	input: MergePullInput,
840	) => Effect.Effect<
841	MergePullResult,
842	| RepoAuthError
843	| PullNotFound
844	| PullStateConflict
845	| BranchMissing
846	| NothingToMerge
847	| MergeConflict
848	| RefConflict
849	| ReadOnlyRepo,
850	RuntimeContext
851	>;
852	/**
853	* Streams a row snapshot for a fork (DESIGN.md §2.3 Fork). Internal —
854	* called only by a sibling Repo DO's fork alarm job; never routed from
855	* the Worker.
856	*/
857	readonly snapshotRows: () => Stream.Stream<SnapshotChunk, StoreError>;
858	/**
859	* Push, phase 1 (DESIGN §22.10): authorize the parsed commands and open
860	* a staging push. The Worker then streams the pack through the hasher
861	* fan-out and the spill, staging rows with `stagePush`.
862	*/
863	readonly beginPush: (
864	input: BeginPushInput,
865	) => Effect.Effect<
866	BeginPushResult,
867	RepoNotFound | StoreError,
868	RuntimeContext
869	>;
870	/** Push, phase 2: stage one batch of rows (see `PushWire.ts`). */
871	readonly stagePush: (
872	pushId: string,
873	batch: Uint8Array,
874	) => Effect.Effect<void, RepoNotFound | StoreError, RuntimeContext>;
875	/** Abort uncommitted staging; false means committed data must be retained. */
876	readonly abortPush: (
877	pushId: string,
878	packIds?: ReadonlyArray<string>,
879	) => Effect.Effect<boolean, RepoNotFound | StoreError, RuntimeContext>;
880	/** Validate connectivity without moving refs. Commit repeats this check. */
881	readonly validatePush: (
882	input: CommitPushInput,
883	) => Effect.Effect<void, RepoNotFound | StoreError, RuntimeContext>;
884	/** Read from this push's staging area and the live store only. */
885	readonly readPreparedObject: (
886	pushId: string,
887	oid: Oid,
888	maxBytes: number,
889	) => Effect.Effect<
890	{ readonly type: ObjectType; readonly content: Uint8Array } | undefined,
891	RepoNotFound | StoreError,
892	RuntimeContext
893	>;
894	/** A live object for thin-delta base resolution, or `undefined`. */
895	readonly readPushBase: (
896	oid: Oid,
897	) => Effect.Effect<
898	{ readonly type: ObjectType; readonly content: Uint8Array } | undefined,
899	RepoNotFound | StoreError,
900	RuntimeContext
901	>;
902	/**
903	* Push, phase 3: connectivity over the staged rows, the commit-graph
904	* rows, the transactional ref CAS, and post-commit bookkeeping.
905	*/
906	readonly commitPush: (
907	input: CommitPushInput,
908	) => Effect.Effect<
909	CommitPushResult,
910	RepoNotFound | StoreError,
911	RuntimeContext
912	>;
913	/** The git wire protocol (info/refs, upload-pack). */
914	readonly fetch: HttpEffect<Cloudflare.DurableObjectState | RuntimeContext>;
915	/** Alarm job dispatcher (import / fork / purge / gc). */
916	readonly alarm: (
917	info?: Cloudflare.AlarmInvocationInfo,
918	) => Effect.Effect<void, never, RuntimeContext>;
919	}
920
921	// ─────────────────────────────────────────────────────────────────────────────
922	// Protocol choreography — pure helpers (exported for tests / future split)
923	// ─────────────────────────────────────────────────────────────────────────────
924
925	/** Upload-pack capability list (DESIGN.md §4). */
926	export const uploadCapabilities = (defaultBranch: string): string =>
927	`multi_ack_detailed no-done side-band-64k shallow ofs-delta agent=${GIT_AGENT} symref=HEAD:refs/heads/${defaultBranch} object-format=sha1`;
928
929	/** Receive-pack capability list (DESIGN.md §4). */
930	export const receiveCapabilities = (): string =>
931	`report-status report-status-v2 delete-refs side-band-64k atomic ofs-delta object-format=sha1 agent=${GIT_AGENT}`;
932
933	/** The two smart-HTTP services. */
934	export type GitService = "git-upload-pack" | "git-receive-pack";
935
936	/**
937	* Builds the v0 `info/refs` advertisement (with `# service=` prelude):
938	* HEAD first with the capability line, refs sorted, annotated tags peeled
939	* (`^{}` lines), zero-id `capabilities^{}` on an empty repo.
940	*/
941	export const buildAdvertisement = (options: {
942	readonly service: GitService;
943	readonly defaultBranch: string;
944	/** Sorted by name; peeled targets included for annotated tags. */
945	readonly refs: ReadonlyArray<RefData>;
946	}): Uint8Array => {
947	const caps =
948	options.service === "git-upload-pack"
949	? uploadCapabilities(options.defaultBranch)
950	: receiveCapabilities();
951	const parts: Array<Uint8Array> = [
952	pktText(`# service=${options.service}`),
953	flushPkt,
954	];
955	const lines: Array<string> = [];
956	const headRef = options.refs.find(
957	(ref) => ref.name === `refs/heads/${options.defaultBranch}`,
958	);
959	if (options.service === "git-upload-pack" && headRef !== undefined) {
960	lines.push(`${headRef.oid} HEAD`);
961	}
962	for (const ref of options.refs) {
963	lines.push(`${ref.oid} ${ref.name}`);
964	if (ref.peeled !== undefined) {
965	lines.push(`${ref.peeled} ${ref.name}^{}`);
966	}
967	}
968	if (lines.length === 0) {
969	parts.push(pktText(`${ZERO_OID} capabilities^{}\0${caps}`));
970	} else {
971	lines.forEach((line, index) => {
972	parts.push(index === 0 ? pktText(`${line}\0${caps}`) : pktText(line));
973	});
974	}
975	parts.push(flushPkt);
976	return concatBytes(parts);
977	};
978
979	/** A parsed v0 upload-pack request. */
980	export interface UploadPackRequest {
981	readonly wants: ReadonlyArray<Oid>;
982	readonly haves: ReadonlyArray<Oid>;
983	readonly done: boolean;
984	readonly depth: number | undefined;
985	readonly clientShallow: ReadonlyArray<Oid>;
986	readonly capabilities: ReadonlySet<string>;
987	}
988
989	/**
990	* Parses a v0 upload-pack POST body (already gunzipped) into wants /
991	* shallow / deepen / haves / done. Rejects `deepen-since` / `deepen-not` /
992	* `filter` (not advertised) with {@link WireProtocolError}.
993	*/
994	export const parseUploadPackRequest = (
995	pkts: ReadonlyArray<PktLine>,
996	): Effect.Effect<UploadPackRequest, WireProtocolError> =>
997	Effect.suspend(() => {
998	const wants: Array<Oid> = [];
999	const haves: Array<Oid> = [];
1000	const clientShallow: Array<Oid> = [];
1001	let done = false;
1002	let depth: number | undefined;
1003	let capabilities: ReadonlySet<string> = new Set();
1004
1005	for (const pkt of pkts) {
1006	if (pkt._tag !== "data") continue;
1007	const text = pktPayloadText(pkt.payload);
1008	if (text === "done") {
1009	done = true;
1010	continue;
1011	}
1012	if (text.startsWith("want ")) {
1013	let rest = text.slice(5);
1014	if (wants.length === 0) {
1015	const space = rest.indexOf(" ");
1016	if (space !== -1) {
1017	capabilities = new Set(
1018	rest
1019	.slice(space + 1)
1020	.split(" ")
1021	.filter(Boolean),
1022	);
1023	rest = rest.slice(0, space);
1024	}
1025	}
1026	if (!isOid(rest)) {
1027	return Effect.fail(
1028	new WireProtocolError({ reason: `malformed want line: ${text}` }),
1029	);
1030	}
1031	wants.push(rest);
1032	continue;
1033	}
1034	if (text.startsWith("have ")) {
1035	const oid = text.slice(5, 45);
1036	if (isOid(oid)) haves.push(oid);
1037	continue;
1038	}
1039	if (text.startsWith("shallow ")) {
1040	const oid = text.slice(8);
1041	if (!isOid(oid)) {
1042	return Effect.fail(
1043	new WireProtocolError({
1044	reason: `malformed shallow line: ${text}`,
1045	}),
1046	);
1047	}
1048	clientShallow.push(oid);
1049	continue;
1050	}
1051	if (text.startsWith("deepen ")) {
1052	const n = Number.parseInt(text.slice(7), 10);
1053	if (!Number.isInteger(n) || n <= 0) {
1054	return Effect.fail(
1055	new WireProtocolError({ reason: `malformed deepen line: ${text}` }),
1056	);
1057	}
1058	depth = n;
1059	continue;
1060	}
1061	if (text.startsWith("deepen-since ") || text.startsWith("deepen-not ")) {
1062	return Effect.fail(
1063	new WireProtocolError({
1064	reason: `${text.split(" ")[0]} is not supported`,
1065	}),
1066	);
1067	}
1068	if (text.startsWith("filter ")) {
1069	return Effect.fail(
1070	new WireProtocolError({ reason: "filter is not supported" }),
1071	);
1072	}
1073	// ignore unknown lines defensively
1074	}
1075	if (wants.length === 0) {
1076	return Effect.fail(
1077	new WireProtocolError({ reason: "upload-pack request has no wants" }),
1078	);
1079	}
1080	return Effect.succeed({
1081	wants,
1082	haves,
1083	done,
1084	depth,
1085	clientShallow,
1086	capabilities,
1087	});
1088	});
1089
1090	/** One receive-pack command (`<old> <new> <refname>`). */
1091	export interface RefCommand {
1092	readonly oldOid: Oid;
1093	readonly newOid: Oid;
1094	readonly ref: string;
1095	}
1096
1097	/** A parsed v0 receive-pack request head. */
1098	export interface ReceivePackRequest {
1099	readonly commands: ReadonlyArray<RefCommand>;
1100	readonly capabilities: ReadonlySet<string>;
1101	/** Byte offset of the pack within the body (== length when no pack). */
1102	readonly packStart: number;
1103	/** True for git's bare-flush `http.postBuffer` probe. */
1104	readonly probe: boolean;
1105	}
1106
1107	const REF_NAME_REGEX = /^refs\/[^\s~^:?*\[\\]+$/;
1108
1109	/**
1110	* Parses a v0 receive-pack POST body (already gunzipped): command
1111	* pkt-lines (first carries `\0caps`) up to the flush, leaving the raw pack
1112	* bytes at `packStart`. Detects the empty-flush probe (DESIGN.md §2.3
1113	* Push step 2).
1114	*/
1115	export const parseReceivePackRequest = (
1116	body: Uint8Array,
1117	): Effect.Effect<ReceivePackRequest, WireProtocolError> =>
1118	Effect.suspend(() => {
1119	const commands: Array<RefCommand> = [];
1120	let capabilities: ReadonlySet<string> = new Set();
1121	let pos = 0;
1122	let first = true;
1123
1124	for (;;) {
1125	if (pos >= body.length) {
1126	// Body ended without a flush: only legal as the bare probe of an
1127	// entirely empty body.
1128	if (commands.length === 0 && pos === 0) {
1129	return Effect.succeed({
1130	commands: [],
1131	capabilities,
1132	packStart: body.length,
1133	probe: true,
1134	});
1135	}
1136	return Effect.fail(
1137	new WireProtocolError({
1138	reason: "receive-pack request ended before flush",
1139	}),
1140	);
1141	}
1142	const r = readPktLineAt(body, pos);
1143	if (r._tag === "incomplete" || r._tag === "invalid") {
1144	return Effect.fail(
1145	new WireProtocolError({
1146	reason:
1147	r._tag === "invalid"
1148	? r.reason
1149	: "truncated receive-pack request",
1150	}),
1151	);
1152	}
1153	pos = r.next;
1154	if (r.pkt._tag === "flush") {
1155	return Effect.succeed({
1156	commands,
1157	capabilities,
1158	packStart: pos,
1159	probe: commands.length === 0 && pos >= body.length,
1160	});
1161	}
1162	if (r.pkt._tag !== "data") continue;
1163	let text = pktPayloadText(r.pkt.payload);
1164	if (first) {
1165	const nul = text.indexOf("\0");
1166	if (nul !== -1) {
1167	capabilities = new Set(
1168	text
1169	.slice(nul + 1)
1170	.split(" ")
1171	.filter(Boolean),
1172	);
1173	text = text.slice(0, nul);
1174	}
1175	first = false;
1176	}
1177	const oldOid = text.slice(0, 40);
1178	const newOid = text.slice(41, 81);
1179	const ref = text.slice(82);
1180	if (
1181	!isOid(oldOid) ||
1182	!isOid(newOid) ||
1183	text[40] !== " " ||
1184	text[81] !== " " ||
1185	!REF_NAME_REGEX.test(ref)
1186	) {
1187	return Effect.fail(
1188	new WireProtocolError({ reason: `malformed command line: ${text}` }),
1189	);
1190	}
1191	commands.push({ oldOid, newOid, ref });
1192	}
1193	});
1194
1195	/** The 12-byte pack header: `PACK`, version 2, entry count. */
1196	export const packHeader = (count: number): Uint8Array => {
1197	const out = new Uint8Array(12);
1198	out.set(utf8Encode("PACK"), 0);
1199	const view = new DataView(out.buffer);
1200	view.setUint32(4, 2);
1201	view.setUint32(8, count);
1202	return out;
1203	};
1204
1205	/**
1206	* The §3.5 no-delta pack emitter: header + per-entry
1207	* `varint(type,size) + stored zdata` (verbatim — zero compression CPU) +
1208	* incremental SHA-1 trailer. Constant memory; the manifest is already
1209	* materialized so the count is known upfront.
1210	*/
1211	export const packStream = (
1212	entries: ReadonlyArray<ManifestEntry>,
1213	objects: ObjectSource,
1214	): Stream.Stream<Uint8Array, StoreError> =>
1215	Stream.suspend(() => {
1216	const sha = makeSha1();
1217	const tap = (bytes: Uint8Array): Uint8Array => {
1218	sha.update(bytes);
1219	return bytes;
1220	};
1221	const header = Stream.sync(() => tap(packHeader(entries.length)));
1222	const body = packEntryStream(entries, objects).pipe(Stream.map(tap));
1223	const trailer = Stream.sync(() => sha.digest());
1224	return header.pipe(Stream.concat(body), Stream.concat(trailer));
1225	});
1226
1227	/**
1228	* The entry stream alone — no header, no trailer. Because entries are
1229	* stored non-delta (`varint(type,size) + zdata`), entry streams from
1230	* different sources concatenate into one valid pack under a rewritten
1231	* header and a fresh trailer. That is what lets a clone be served as
1232	* "bundle bytes + a small delta" (DESIGN.md §21.2), and it is the same
1233	* property geometric pack merging relies on.
1234	*/
1235	export const packEntryStream = (
1236	entries: ReadonlyArray<ManifestEntry>,
1237	objects: ObjectSource,
1238	): Stream.Stream<Uint8Array, StoreError> =>
1239	objects.packEntries !== undefined
1240	? objects.packEntries(entries)
1241	: Stream.fromIterable(entries).pipe(
1242	Stream.flatMap((entry) =>
1243	Stream.sync(() => encodeTypeSize(entry.type, entry.size)).pipe(
1244	Stream.concat(objects.readZData(entry.oid)),
1245	),
1246	),
1247	);
1248
1249	// ─────────────────────────────────────────────────────────────────────────────
1250	// Pack ingest (DESIGN.md §3.6) — buffered v1, RandomAccess seam is the buffer
1251	// ─────────────────────────────────────────────────────────────────────────────
1252
1253	/** A commit staged by an ingest, with what the commit graph needs. */
1254	export interface StagedCommitInfo {
1255	readonly oid: Oid;
1256	readonly tree: Oid;
1257	readonly parents: ReadonlyArray<Oid>;
1258	readonly commitTime: number;
1259	}
1260
1261	/** The result of a successful pack ingest. */
1262	export interface IngestResult {
1263	/** Objects in the pack (including already-existing re-pushed ones). */
1264	readonly objectCount: number;
1265	/** SQL staging time, when measured by the streaming ingest path. */
1266	readonly stageMs?: number | undefined;
1267	/** Per-phase CPU split of ingest (ms) — see `PackParser` `phases`. */
1268	readonly phases?: Record<string, number> | undefined;
1269	/** Objects staged as references into a promoted wire pack (DESIGN §22.5). */
1270	readonly promoted?: number | undefined;
1271	/** The completed spilled object's key, when the body spilled (DESIGN §22.10). */
1272	readonly parkedKey?: string | undefined;
1273	/** The resolved pack's key — delta-resolved blobs, promoted (DESIGN §22.13). */
1274	readonly resolvedKey?: string | undefined;
1275	/** The pack's trailer checksum (hex). */
1276	readonly packSha: string;
1277	/** Staged commits (commit-graph rows are inserted at finalize). */
1278	readonly commits: ReadonlyArray<StagedCommitInfo>;
1279	/** Oids referenced by staged trees/tags + commit trees (hard edges). */
1280	readonly referenced: ReadonlyArray<Oid>;
1281	/** Oids referenced as commit parents (soft under shallow import). */
1282	readonly referencedParents: ReadonlyArray<Oid>;
1283	}
1284
1285	/** Bounded LRU of inflated contents shared across delta resolution. */
1286	const makeContentCache = (budget: number) => {
1287	const map = new Map<string, Uint8Array>();
1288	let total = 0;
1289	return {
1290	get: (key: string): Uint8Array | undefined => {
1291	const value = map.get(key);
1292	if (value !== undefined) {
1293	map.delete(key);
1294	map.set(key, value);
1295	}
1296	return value;
1297	},
1298	set: (key: string, value: Uint8Array): void => {
1299	if (value.byteLength > budget || map.has(key)) return;
1300	map.set(key, value);
1301	total += value.byteLength;
1302	for (const [k, v] of map) {
1303	if (total <= budget) break;
1304	map.delete(k);
1305	total -= v.byteLength;
1306	}
1307	},
1308	};
1309	};
1310
1311	/** Delta-resolution content cache budget (DESIGN.md §3.6): 20 MiB. */
1312	const CACHE_BUDGET = 20 * 1024 * 1024;
1313	/** Maximum delta-chain depth tolerated (git's own default is 50). */
1314	const MAX_DELTA_DEPTH = 64;
1315
1316	interface RawEntry {
1317	readonly offset: number;
1318	readonly kind: "object" | "ofs" | "ref";
1319	/** Real object type for `kind: "object"`. */
1320	readonly type: ObjectType | undefined;
1321	/** Header size (object size, or delta payload size for deltas). */
1322	readonly declaredSize: number;
1323	/** Where this entry's zlib stream starts. */
1324	readonly zstart: number;
1325	/** Exact compressed span length. */
1326	readonly zlen: number;
1327	readonly baseOffset: number | undefined;
1328	readonly baseOid: Oid | undefined;
1329	/** Filled once resolved. */
1330	oid: Oid | undefined;
1331	resolvedType: ObjectType | undefined;
1332	}
1333
1334	/**
1335	* The store surface ingest needs (DESIGN §22.10): staging rows and thin
1336	* delta bases. The Repo DO satisfies it directly; the Worker satisfies it
1337	* over RPC so the pack never enters the Durable Object.
1338	*/
1339	export interface IngestStore {
1340	readonly insertStagedBatch: ObjectStore["insertStagedBatch"];
1341	/** Awaits staging that `insertStagedBatch` may have left in flight. */
1342	readonly settle?: Effect.Effect<void, StoreError> | undefined;
1343	/** A live base object for a thin delta, or `undefined` when absent. */
1344	readonly readBase: (
1345	oid: Oid,
1346	) => Effect.Effect<
1347	{ readonly type: ObjectType; readonly content: Uint8Array } | undefined,
1348	StoreError
1349	>;
1350	}
1351
1352	/** An {@link IngestStore} over a full {@link ObjectStore}. */
1353	export const ingestStoreOf = (store: ObjectStore): IngestStore => ({
1354	insertStagedBatch: store.insertStagedBatch,
1355	readBase: (oid) =>
1356	store
1357	.getMeta(oid)
1358	.pipe(
1359	Effect.flatMap((meta) =>
1360	meta === undefined
1361	? Effect.succeed(undefined)
1362	: store
1363	.readContent(oid)
1364	.pipe(Effect.map((content) => ({ type: meta.type, content }))),
1365	),
1366	),
1367	});
1368
1369	/** Options of {@link ingestPack}. */
1370	export interface IngestOptions {
1371	readonly store: IngestStore;
1372	readonly pushId: string;
1373	/**
1374	* When the wire body was spilled to blob storage, stage non-delta blobs
1375	* as references into it (DESIGN §22.5): `packId` is the wire pack's id,
1376	* `base` the pack's byte offset within the spilled object (the request's
1377	* pkt-line commands precede it).
1378	*/
1379	readonly promote?:
1380	| (() => { readonly packId: string; readonly base: number } | undefined)
1381	| undefined;
1382	/**
1383	* When set, entries are inflated and hashed by the hasher service one
1384	* spilled part at a time (DESIGN §22.7) instead of by the in-process
1385	* parser; the DO only pumps bytes and stages rows.
1386	*/
1387	readonly hasher?: HasherShape | undefined;
1388	/** Bytes per hasher part (tests shrink it). @default HASH_PART_BYTES */
1389	readonly partBytes?: number | undefined;
1390	/**
1391	* The streaming spill (DESIGN §22.10): the pump reads the whole request
1392	* BODY in uniform parts, each hasher writes its part to blob storage,
1393	* and the completed object is the promotion target (`packId`) and the
1394	* source's fallback reader. Bodies that end within `threshold` bytes
1395	* stay in memory and nothing is written.
1396	*/
1397	readonly spill?:
1398	| {
1399	readonly body: RandomAccess;
1400	readonly feeder: StreamingFeeder;
1401	readonly packStart: number;
1402	readonly blobs: BlobStoreShape;
1403	readonly key: string;
1404	readonly packId: string;
1405	readonly repoId: string;
1406	readonly threshold: number;
1407	}
1408	| undefined;
1409	/**
1410	* Shallow-import mode: commit parents may be absent (they become walk
1411	* boundaries); trees/blobs/tags must still be fully connected.
1412	*/
1413	readonly allowMissingParents?: boolean | undefined;
1414	}
1415
1416	/**
1417	* Single-pass buffered pack ingest (DESIGN.md §3.6): parse entries,
1418	* store non-delta compressed spans **verbatim**, resolve ofs/ref deltas
1419	* (thin bases from the live store), enforce the 64 MiB object cap, verify
1420	* the trailer SHA-1, and stage everything under `pushId`.
1421	*
1422	* The v1.x streaming upgrade seam: this function's only view of the pack
1423	* is `pack: Uint8Array` random access — swapping in an R2-backed
1424	* `RandomAccess` touches this implementation, not the protocol drivers.
1425	*/
1426	/**
1427	* Ingests a pack from a {@link RandomAccess} source — the path that has **no
1428	* size cap** (DESIGN.md §3.6). Memory is bounded by the source's read
1429	* window and the delta LRU rather than by the pack, so the source can be an
1430	* R2 object of any size (see `Store/PackSource.ts`).
1431	*
1432	* Delegates the wire-format work to the pure `PackParser` (which the pack
1433	* fixtures in `test/pack.test.ts` cover directly) and does the repo-side
1434	* work in its sink: stage the object, and record commit-graph edges.
1435	* Blobs are staged without inflating — only commits, trees and tags need
1436	* their content parsed, and those are small.
1437	*/
1438	export const ingestPackFrom = (
1439	source: RandomAccess,
1440	options: IngestOptions,
1441	): Effect.Effect<IngestResult, PackIngestError | StoreError> =>
1442	Effect.gen(function* () {
1443	const { store, pushId } = options;
1444	const commits: Array<StagedCommitInfo> = [];
1445	const referenced = new Set<Oid>();
1446	const referencedParents = new Set<Oid>();
1447
1448	const record = Effect.fn(function* (
1449	oid: Oid,
1450	type: ObjectType,
1451	content: Uint8Array,
1452	) {
1453	switch (type) {
1454	case ObjectType.commit: {
1455	const parsed = yield* parseCommit(content).pipe(
1456	Effect.mapError(
1457	(error) =>
1458	new PackIngestError({
1459	reason: `bad commit ${oid}: ${error.reason}`,
1460	}),
1461	),
1462	);
1463	commits.push({
1464	oid,
1465	tree: parsed.tree,
1466	parents: parsed.parents,
1467	commitTime: parsed.committer.when,
1468	});
1469	referenced.add(parsed.tree);
1470	for (const parent of parsed.parents) referencedParents.add(parent);
1471	return;
1472	}
1473	case ObjectType.tree: {
1474	const parsed = yield* parseTree(content).pipe(
1475	Effect.mapError(
1476	(error) =>
1477	new PackIngestError({
1478	reason: `bad tree ${oid}: ${error.reason}`,
1479	}),
1480	),
1481	);
1482	for (const entry of parsed) {
1483	if (treeEntryKind(entry.mode) !== "commit") {
1484	referenced.add(entry.oid);
1485	}
1486	}
1487	return;
1488	}
1489	case ObjectType.tag: {
1490	const parsed = yield* parseTag(content).pipe(
1491	Effect.mapError(
1492	(error) =>
1493	new PackIngestError({
1494	reason: `bad tag ${oid}: ${error.reason}`,
1495	}),
1496	),
1497	);
1498	referenced.add(parsed.object);
1499	return;
1500	}
1501	case ObjectType.blob:
1502	return;
1503	}
1504	});
1505
1506	// Stage in batches: one transaction per batch instead of one (plus an
1507	// existence probe) per object. That per-object cost is what made a
1508	// 13.7k-object push take ~20 s inside a single-threaded DO — the work
1509	// is inherently serial, so the only lever is fewer operations
1510	// (DESIGN.md §16.6).
1511	// Buffered entries keep their pack coordinates so the promote/inline
1512	// decision is made at FLUSH time (DESIGN §22.6): a streaming body only
1513	// becomes a wire pack once it crosses the spill threshold, but the wire
1514	// object holds the body from byte 0, so entries buffered before that
1515	// moment can still be promoted when the flush happens after it.
1516	/** What the parser or the hasher hands over per entry. */
1517	interface Incoming {
1518	readonly oid: Oid;
1519	readonly type: ObjectType;
1520	readonly size: number;
1521	/** Absent for hasher-scanned non-delta entries: read from the source if the row stays inline. */
1522	readonly zdata?: Uint8Array | undefined;
1523	readonly span?: number | undefined;
1524	readonly dataOffset: number;
1525	readonly fromDelta: boolean;
1526	readonly content?: Uint8Array | undefined;
1527	/** Already in a pack in blob storage (the resolved pack, DESIGN §22.13). */
1528	readonly pack?:
1529	| { readonly packId: string; readonly offset: number }
1530	| undefined;
1531	}
1532	let batch: Array<Incoming> = [];
1533	let batchBytes = 0;
1534	let promoted = 0;
1535	/** Time spent in SQL staging; the rest of ingest is CPU. */
1536	let stageMs = 0;
1537	const flush = Effect.fn(function* () {
1538	if (batch.length === 0) return;
1539	const pending = batch;
1540	batch = [];
1541	batchBytes = 0;
1542	const target =
1543	options.promote?.() ??
1544	(spilling && options.spill !== undefined
1545	? { packId: options.spill.packId, base: options.spill.packStart }
1546	: undefined);
1547	const staged: Array<StagedObject> = [];
1548	for (const entry of pending) {
1549	const pack =
1550	entry.pack ??
1551	(target !== undefined &&
1552	!entry.fromDelta &&
1553	entry.type === ObjectType.blob &&
1554	entry.dataOffset >= 0
1555	? { packId: target.packId, offset: target.base + entry.dataOffset }
1556	: undefined);
1557	if (pack !== undefined) promoted += 1;
1558	// A promoted row needs no bytes; an inline row needs its zdata —
1559	// the hasher path did not ship it, so read the span back.
1560	const zdata =
1561	entry.zdata ??
1562	(pack === undefined
1563	? Uint8Array.from(
1564	readSpan?.(entry.dataOffset, entry.span ?? 0) ??
1565	(yield* source.read(entry.dataOffset, entry.span ?? 0)),
1566	)
1567	: new Uint8Array(0));
1568	staged.push({
1569	oid: entry.oid,
1570	type: entry.type,
1571	size: entry.size,
1572	zdata,
1573	zsize: entry.zdata?.byteLength ?? entry.span,
1574	pack,
1575	});
1576	}
1577	const at = yield* Effect.sync(() => performance.now());
1578	yield* store.insertStagedBatch(pushId, staged);
1579	stageMs += yield* Effect.sync(() => performance.now() - at);
1580	});
1581
1582	const phases: Record<string, number> = {};
1583	/** The spill's multipart upload, once the pump decided to spill. */
1584	let upload: BlobMultipart | undefined;
1585	let spilling = false;
1586	/** Set once the spilled object is complete: it is the wire pack. */
1587	let parkedKey: string | undefined;
1588	/** Set once the resolved pack(s) are written (DESIGN §22.13). */
1589	let resolvedKey: string | undefined;
1590	const resolvedKeys: Array<string> = [];
1591	/**
1592	* Delta-resolved blobs held back from staging while the body spills:
1593	* their fresh zlib goes into one resolved pack in blob storage and the
1594	* rows become coordinates, so the DO never writes those bytes.
1595	*/
1596	const resolvedBlobs: Array<Incoming> = [];
1597	/** In-memory span reads over the retained parts (set by the pump). */
1598	let readSpan:
1599	| ((at: number, length: number) => Uint8Array | undefined)
1600	| undefined;
1601	/**
1602	* Stages a run of resolved entries (DESIGN §22.5): the parser's
1603	* synchronous fast path hands over up to SINK_BATCH non-delta entries
1604	* per fiber hop; delta-resolved entries arrive one at a time.
1605	*/
1606	/** Every oid this push stages: connectivity only asks the store about the rest. */
1607	const stagedOids = new Set<Oid>();
1608	const stage = (entries: ReadonlyArray<Incoming>) =>
1609	Effect.gen(function* () {
1610	for (const entry of entries) {
1611	stagedOids.add(entry.oid);
1612	if (
1613	spilling &&
1614	entry.fromDelta &&
1615	entry.type === ObjectType.blob &&
1616	entry.pack === undefined &&
1617	entry.zdata !== undefined
1618	) {
1619	resolvedBlobs.push(entry);
1620	continue;
1621	}
1622	batch.push(entry);
1623	// Counted as inline until the flush decides; promoted rows carry
1624	// no BLOB, so this over-counts — safe for a memory cap.
1625	batchBytes += entry.zdata?.byteLength ?? entry.span ?? 0;
1626	if (
1627	batch.length >= STAGE_BATCH_OBJECTS ||
1628	batchBytes >= STAGE_BATCH_BYTES
1629	) {
1630	yield* flush();
1631	}
1632	if (entry.type !== ObjectType.blob) {
1633	if (entry.content === undefined) {
1634	return yield* new PackIngestError({
1635	reason: `no content for ${entry.oid} (type ${entry.type})`,
1636	});
1637	}
1638	yield* record(entry.oid, entry.type, entry.content);
1639	}
1640	}
1641	});
1642	/**
1643	* The hasher pipeline (DESIGN §22.7): pump the pack in parts through
1644	* the hasher; every part carries the previous part's incomplete tail.
1645	* The DO's own CPU here is a memcpy, the trailer SHA-1 and staging.
1646	*/
1647	const pumpThroughHasher = (hasher: HasherShape) =>
1648	Effect.gen(function* () {
1649	const spill = options.spill;
1650	// Spill parts (uniform, ≥ 5 MiB for R2) and hash chunks: a hasher
1651	// may want smaller chunks than a part (a Lambda's 6 MB payload);
1652	// chunks then divide parts and the pump uploads the parts itself.
1653	const partBytes =
1654	options.partBytes ??
1655	(spill === undefined ? HASH_PART_BYTES : SPILL_PART_BYTES);
1656	const wanted = hasher.chunkBytes ?? partBytes;
1657	const hashBytes =
1658	wanted < partBytes && partBytes % wanted === 0 ? wanted : partBytes;
1659	const asIngest = (error: {
1660	readonly _tag: string;
1661	readonly reason?: string;
1662	}) =>
1663	new PackIngestError({
1664	reason: `${error._tag}${error.reason === undefined ? "" : `: ${error.reason}`}`,
1665	});
1666	const header = yield* source.read(0, 12);
1667	if (
1668	header.length < 12 ||
1669	header[0] !== 0x50 ||
1670	header[1] !== 0x41 ||
1671	header[2] !== 0x43 ||
1672	header[3] !== 0x4b
1673	) {
1674	return yield* new PackIngestError({ reason: "bad pack magic" });
1675	}
1676	const count = new DataView(header.buffer, header.byteOffset).getUint32(
1677	8,
1678	);
1679	// Trailer SHA-1 over [0, total − 20): hash with a 20-byte lag, so
1680	// the last 20 bytes seen are the trailer itself.
1681	const trailerSha = makeSha1();
1682	let lag = new Uint8Array(0);
1683	const feed = (bytes: Uint8Array) => {
1684	const joined = lag.length === 0 ? bytes : concatBytes([lag, bytes]);
1685	if (joined.length > 20) {
1686	trailerSha.update(joined.subarray(0, joined.length - 20));
1687	lag = Uint8Array.from(joined.subarray(joined.length - 20));
1688	} else {
1689	lag = Uint8Array.from(joined);
1690	}
1691	};
1692	interface Known {
1693	readonly oid: Oid;
1694	readonly type: ObjectType;
1695	readonly dataOffset: number;
1696	readonly span: number;
1697	readonly zdata?: Uint8Array | undefined;
1698	readonly content?: Uint8Array | undefined;
1699	}
1700	const known = new Map<number, Known>(); // by header offset
1701	const knownByOid = new Map<string, Known>();
1702	const unresolved: Array<UnresolvedDelta> = [];
1703	// Raw-chunk dispatch (DESIGN §22.9–10): NO scanning here. Every
1704	// chunk goes to a hasher as it arrives (bounded concurrency), which
1705	// resyncs to the first entry boundary inside the chunk and scans and
1706	// hashes from there — and, when the body spills, writes the chunk
1707	// as its part of the multipart upload. Results are consumed in
1708	// chunk order, concurrently with the receive; the one entry
1709	// straddling each chunk edge is stitched from the neighbouring
1710	// payloads and hashed by a small extra call. A boundary that does
1711	// not line up (a resync false positive) falls back to a sequential
1712	// rescan of that chunk from the known boundary.
1713	const gate = yield* Semaphore.make(
1714	hasher.concurrency ?? HASH_CONCURRENCY,
1715	);
1716	interface Chunk {
1717	readonly index: number;
1718	/** Pack-relative offset of payload[0] (negative for a body-aligned first part). */
1719	readonly base: number;
1720	readonly payload: Uint8Array;
1721	readonly fiber: Fiber.Fiber<HashPartResult, PackIngestError>;
1722	}
1723	const chunks: Array<Chunk> = [];
1724	const bytesFrom = (from: number, to: number): Uint8Array => {
1725	const pieces: Array<Uint8Array> = [];
1726	for (const c of chunks) {
1727	const start = Math.max(from, c.base);
1728	const end = Math.min(to, c.base + c.payload.length);
1729	if (end > start)
1730	pieces.push(c.payload.subarray(start - c.base, end - c.base));
1731	}
1732	if (pieces.length === 1) return pieces[0]!;
1733	// Neighbouring chunks are views of one slab: hand out one view.
1734	const first = pieces[0]!;
1735	let contiguous = true;
1736	let end = first.byteOffset + first.byteLength;
1737	for (let i = 1; i < pieces.length; i++) {
1738	const piece = pieces[i]!;
1739	if (piece.buffer !== first.buffer || piece.byteOffset !== end) {
1740	contiguous = false;
1741	break;
1742	}
1743	end += piece.byteLength;
1744	}
1745	return contiguous
1746	? new Uint8Array(
1747	first.buffer,
1748	first.byteOffset,
1749	end - first.byteOffset,
1750	)
1751	: concatBytes(pieces);
1752	};
1753	// Retained payloads serve every span read (inline rows, delta
1754	// bases and deltas) so nothing goes back to the source or the
1755	// spilled object during ingest.
1756	readSpan = (at, length) => {
1757	const bytes = bytesFrom(at, at + length);
1758	return bytes.length === length ? bytes : undefined;
1759	};
1760	const stageResult = (
1761	payload: Uint8Array,
1762	base: number,
1763	result: ScanResult,
1764	) =>
1765	Effect.gen(function* () {
1766	const incoming: Array<Incoming> = [];
1767	for (const e of result.entries) {
1768	// A hasher may omit non-blob content to bound its response
1769	// (a Lambda): inflate it here from the fresh zdata or the
1770	// retained span — trees, commits and tags are small.
1771	const content =
1772	e.content ??
1773	(e.type === ObjectType.blob
1774	? undefined
1775	: yield* Zlib.inflate(
1776	e.zdata ??
1777	payload.subarray(
1778	e.dataOffset - base,
1779	e.dataOffset - base + e.span,
1780	),
1781	).pipe(Effect.mapError(asIngest)));
1782	const item: Known = {
1783	oid: e.oid,
1784	type: e.type,
1785	dataOffset: e.dataOffset,
1786	span: e.span,
1787	zdata: e.zdata,
1788	content,
1789	};
1790	known.set(e.offset, item);
1791	knownByOid.set(e.oid, item);
1792	incoming.push({
1793	oid: e.oid,
1794	type: e.type,
1795	size: e.size,
1796	zdata:
1797	e.zdata ??
1798	payload.subarray(
1799	e.dataOffset - base,
1800	e.dataOffset - base + e.span,
1801	),
1802	span: e.span,
1803	dataOffset: e.dataOffset,
1804	fromDelta: e.zdata !== undefined,
1805	content,
1806	});
1807	}
1808	yield* stage(incoming);
1809	// Batches flush by size (STAGE_BATCH_OBJECTS / _BYTES), not per
1810	// chunk: the retained parts hold every span until ingest ends.
1811	for (const u of result.unresolved) unresolved.push(u);
1812	return result.count;
1813	});
1814
1815	// ── phase 1 (producer): read parts as they arrive, dispatch each ──
1816	// Wall-clock phase split (Date.now advances across awaits; the
1817	// platform's cpuTime is the CPU attribution): `PushStats.phases`.
1818	const pumpStarted = Date.now();
1819	let regionCalls = 0;
1820	// Each part's identity arrives after its upload; joined before
1821	// the multipart upload is completed, never on the scan path.
1822	const parts: Array<Effect.Effect<UploadedPart, PackIngestError>> = [];
1823	let producerDone = false;
1824	const wakers: Array<() => void> = [];
1825	const notifyConsumer = () => {
1826	for (const w of wakers.splice(0)) w();
1827	};
1828	const awaitChunk = (k: number) =>
1829	Effect.suspend(() =>
1830	k < chunks.length || producerDone
1831	? Effect.void
1832	: Effect.callback<void>((resume) => {
1833	wakers.push(() => resume(Effect.void));
1834	}),
1835	);
1836	const availableEnd = () => {
1837	const last = chunks[chunks.length - 1];
1838	return last === undefined ? 12 : last.base + last.payload.length;
1839	};
1840	const dispatch = (
1841	payload: Uint8Array,
1842	opts: {
1843	readonly base: number;
1844	readonly skip: number;
1845	readonly resync: boolean;
1846	readonly partNumber: number;
1847	},
1848	) =>
1849	Effect.forkChild(
1850	Semaphore.withPermits(
1851	gate,
1852	1,
1853	)(
1854	hasher
1855	.hashPart(payload, {
1856	base: opts.base,
1857	skip: opts.skip,
1858	remaining: count,
1859	maxObjectSize: MAX_OBJECT_SIZE,
1860	resync: opts.resync,
1861	spill:
1862	upload === undefined ||
1863	spill === undefined ||
1864	!hasher.writesSpill
1865	? undefined
1866	: {
1867	key: spill.key,
1868	uploadId: upload.uploadId,
1869	partNumber: opts.partNumber,
1870	},
1871	})
1872	.pipe(Effect.mapError(asIngest)),
1873	),
1874	);
1875	const produce = Effect.gen(function* () {
1876	if (spill === undefined) {
1877	// Pack-relative parts from the (in-memory or spilled) source.
1878	feed(header);
1879	let offset = 12;
1880	let index = 0;
1881	while (true) {
1882	const chunk = yield* source.read(offset, partBytes);
1883	if (chunk.length === 0) break;
1884	feed(chunk);
1885	const fiber = yield* dispatch(chunk, {
1886	base: offset,
1887	skip: 0,
1888	resync: index > 0,
1889	partNumber: index + 1,
1890	});
1891	chunks.push({ index, base: offset, payload: chunk, fiber });
1892	offset += chunk.length;
1893	index += 1;
1894	notifyConsumer();
1895	source.release?.(offset);
1896	if (chunk.length < partBytes) break;
1897	}
1898	return;
1899	}
1900	// Body-aligned parts (DESIGN §22.10): part k is body
1901	// [k·P, (k+1)·P); the first carries the command section and
1902	// the pack header before its first entry (`skip`). Chunks of
1903	// `hashBytes` divide the parts.
1904	const { body, packStart } = spill;
1905	const uploadGate = yield* Semaphore.make(3);
1906	let partStart = 0;
1907	/** Upload body [partStart, end) as the next part (pump-owned spill). */
1908	const uploadPart = (end: number) =>
1909	Effect.gen(function* () {
1910	if (upload === undefined || end <= partStart) return;
1911	const settled = upload;
1912	const partNumber = Math.floor(partStart / partBytes) + 1;
1913	const bytes = bytesFrom(partStart - packStart, end - packStart);
1914	partStart = end;
1915	const fiber = yield* Effect.forkDetach(
1916	Semaphore.withPermits(
1917	uploadGate,
1918	1,
1919	)(
1920	spill.blobs
1921	.uploadPart(spill.key, settled.uploadId, partNumber, bytes)
1922	.pipe(
1923	Effect.mapError(
1924	(error) =>
1925	new PackIngestError({
1926	reason: `spill part ${partNumber}: ${error.reason}`,
1927	}),
1928	),
1929	Effect.provide(RuntimeContext.phantom),
1930	),
1931	),
1932	);
1933	parts.push(Fiber.join(fiber));
1934	});
1935	let index = 0;
1936	let received = 0;
1937	while (true) {
1938	const at = index * hashBytes;
1939	const chunk = yield* body.read(at, hashBytes);
1940	if (chunk.length === 0) break;
1941	if (index === 0) {
1942	// Decide the spill once the first chunk is in: a body that
1943	// ended within the in-memory threshold stays in memory.
1944	const ended = spill.feeder.source.ended();
1945	if (!(ended && spill.feeder.source.size <= spill.threshold)) {
1946	upload = yield* spill.blobs.multipart(spill.key).pipe(
1947	Effect.mapError(
1948	(error) =>
1949	new PackIngestError({
1950	reason: `spill: ${error.reason}`,
1951	}),
1952	),
1953	Effect.provide(RuntimeContext.phantom),
1954	);
1955	spill.feeder.expectFallback();
1956	spilling = true;
1957	}
1958	}
1959	const skip = index === 0 ? packStart + 12 : 0;
1960	feed(index === 0 ? chunk.subarray(packStart) : chunk);
1961	const base = at - packStart;
1962	const fiber = yield* dispatch(chunk, {
1963	base: base + skip,
1964	skip,
1965	resync: index > 0,
1966	partNumber: Math.floor(at / partBytes) + 1,
1967	});
1968	console.log(
1969	`[pump] chunk ${index + 1} read at +${Date.now() - pumpStarted}ms (${chunk.length} bytes)`,
1970	);
1971	chunks.push({ index, base, payload: chunk, fiber });
1972	index += 1;
1973	received = at + chunk.length;
1974	notifyConsumer();
1975	body.release?.(received);
1976	if (!hasher.writesSpill && received % partBytes === 0) {
1977	yield* uploadPart(received);
1978	}
1979	if (chunk.length < hashBytes) break;
1980	}
1981	if (!hasher.writesSpill) yield* uploadPart(received);
1982	}).pipe(
1983	Effect.ensuring(
1984	Effect.sync(() => {
1985	producerDone = true;
1986	notifyConsumer();
1987	}),
1988	),
1989	);
1990
1991	// ── phase 2 (consumer): settle results in chunk order ──────────
1992	// `prevEnd` is the last settled boundary. A chunk's resync result
1993	// covers [firstOffset, …); whatever lies in [prevEnd, firstOffset)
1994	// — usually one straddling entry, but any number when chunks are
1995	// tiny — is scanned sequentially as one region from the retained
1996	// payloads. Entries the region scan reaches at or past
1997	// `firstOffset` are duplicates of the chunk's own result and are
1998	// dropped; a region that does not end exactly at `firstOffset`
1999	// means the resync was a false positive, and the chunk is rescanned
2000	// sequentially from `prevEnd` instead.
2001	const keepBelow = (result: ScanResult, limit: number): ScanResult => {
2002	const entries = result.entries.filter((e) => e.offset < limit);
2003	const unresolved = result.unresolved.filter((u) => u.offset < limit);
2004	// The settled boundary is the END of the last kept entry — which
2005	// may lie past `limit` when that entry straddles a chunk edge.
2006	let end = result.count === 0 ? result.consumedTo : -1;
2007	for (const e of entries) end = Math.max(end, e.dataOffset + e.span);
2008	for (const u of unresolved)
2009	end = Math.max(end, u.dataOffset + u.span);
2010	return {
2011	firstOffset: result.firstOffset,
2012	entries,
2013	unresolved,
2014	consumedTo: end < 0 ? result.consumedTo : end,
2015	count: entries.length + unresolved.length,
2016	};
2017	};
2018	const trace =
2019	process.env.GIT_PUMP_DEBUG === "1"
2020	? (...args: Array<unknown>) => console.log("[pump]", ...args)
2021	: () => {};
2022	// The consumer is the forked side: hash fibers are children of THIS
2023	// fiber (a child's children die with it), so the producer runs here
2024	// and the consumer settles results as they land.
2025	// The entry straddling a part edge is hashed as soon as its part
2026	// settles — sized from its own header, from the retained bytes —
2027	// so a large object crossing several parts (a 10 MB blob spans
2028	// three) is verified alongside the parts, not after the last one.
2029	interface Speculative {
2030	readonly start: number;
2031	readonly fiber: Fiber.Fiber<
2032	{ readonly bytes: Uint8Array; readonly r: ScanResult },
2033	PackIngestError
2034	>;
2035	}
2036	const speculate = (start: number) =>
2037	Effect.gen(function* () {
2038	const head = bytesFrom(start, Math.min(start + 32, availableEnd()));
2039	let need = start + 64;
2040	try {
2041	const h = decodeTypeSize(head, 0);
2042	// zlib's worst case is ~0.1% + a few bytes; deltas carry an
2043	// offset varint before their stream.
2044	need = start + h.next + Math.ceil(h.size * 1.001) + 80;
2045	} catch {
2046	// Header cut by the received edge: wait for more below.
2047	}
2048	while (true) {
2049	const end = Math.min(need, availableEnd());
2050	const bytes = bytesFrom(start, end);
2051	const r = yield* hasher
2052	.hashPart(bytes, {
2053	base: start,
2054	remaining: 1,
2055	maxObjectSize: MAX_OBJECT_SIZE,
2056	})
2057	.pipe(Effect.mapError(asIngest));
2058	if (r.count === 1) return { bytes, r };
2059	if (producerDone && end >= availableEnd()) return { bytes, r };
2060	// Not complete in what has arrived: wait for another part.
2061	if (end >= availableEnd()) yield* awaitChunk(chunks.length);
2062	if (end >= need) need = availableEnd();
2063	}
2064	});
2065	const consume = Effect.gen(function* () {
2066	let staged = 0;
2067	let prevEnd = 12;
2068	let speculative: Speculative | undefined;
2069	for (let k = 0; ; k++) {
2070	yield* awaitChunk(k);
2071	if (k >= chunks.length) break;
2072	const current = chunks[k]!;
2073	const result = yield* Fiber.join(current.fiber);
2074	console.log(
2075	`[pump] chunk ${k + 1} settled at +${Date.now() - pumpStarted}ms`,
2076	);
2077	if (result.part !== undefined) {
2078	parts.push(result.part.pipe(Effect.mapError(asIngest)));
2079	}
2080	const chunkEnd = current.base + current.payload.length;
2081	trace(
2082	`chunk ${k} base=${current.base} len=${current.payload.length} first=${result.firstOffset} consumedTo=${result.consumedTo} count=${result.count} prevEnd=${prevEnd} staged=${staged}`,
2083	);
2084	const sequential = () =>
2085	Effect.gen(function* () {
2086	// Everything from the settled boundary through this chunk,
2087	// scanned in order (plus a little of the next chunk so the last
2088	// entry can complete, if there is one).
2089	yield* awaitChunk(k + 1);
2090	const next = chunks[k + 1];
2091	const to =
2092	next === undefined
2093	? chunkEnd
2094	: Math.min(chunkEnd + 64, next.base + next.payload.length);
2095	const bytes = bytesFrom(prevEnd, to);
2096	const r = yield* hasher
2097	.hashPart(bytes, {
2098	base: prevEnd,
2099	remaining: count - staged,
2100	maxObjectSize: MAX_OBJECT_SIZE,
2101	})
2102	.pipe(Effect.mapError(asIngest));
2103	return { bytes, base: prevEnd, r: keepBelow(r, chunkEnd) };
2104	});
2105	if (result.firstOffset < 0) {
2106	// No boundary in this chunk (tiny, or inside one huge entry):
2107	// its bytes are settled by the pending straddler or a later
2108	// chunk's region scan.
2109	continue;
2110	}
2111	if (speculative !== undefined) {
2112	const sp = yield* Fiber.join(speculative.fiber);
2113	const start = speculative.start;
2114	speculative = undefined;
2115	if (
2116	sp.r.count === 1 &&
2117	start === prevEnd &&
2118	sp.r.consumedTo <= result.firstOffset
2119	) {
2120	staged += yield* stageResult(sp.bytes, start, sp.r);
2121	prevEnd = sp.r.consumedTo;
2122	}
2123	// Otherwise the boundary is suspect: the paths below rescan.
2124	}
2125	if (k === 0 && result.firstOffset !== 12) {
2126	const again = yield* sequential();
2127	staged += yield* stageResult(again.bytes, again.base, again.r);
2128	prevEnd = again.r.consumedTo;
2129	continue;
2130	}
2131	if (result.firstOffset < prevEnd) {
2132	// Resync landed inside an already-settled entry: rescan.
2133	const again = yield* sequential();
2134	staged += yield* stageResult(again.bytes, again.base, again.r);
2135	prevEnd = again.r.consumedTo;
2136	continue;
2137	}
2138	if (result.firstOffset > prevEnd) {
2139	trace(` region [${prevEnd}, ${result.firstOffset})`);
2140	regionCalls += 1;
2141	const region = bytesFrom(
2142	prevEnd,
2143	Math.min(result.firstOffset + 64, chunkEnd),
2144	);
2145	const scanned = yield* hasher
2146	.hashPart(region, {
2147	base: prevEnd,
2148	remaining: count - staged,
2149	maxObjectSize: MAX_OBJECT_SIZE,
2150	})
2151	.pipe(Effect.mapError(asIngest));
2152	const ends =
2153	scanned.entries.some(
2154	(e) => e.dataOffset + e.span === result.firstOffset,
2155	) ||
2156	scanned.unresolved.some(
2157	(u) => u.dataOffset + u.span === result.firstOffset,
2158	);
2159	trace(
2160	` region scan count=${scanned.count} consumedTo=${scanned.consumedTo} ends=${ends}`,
2161	);
2162	if (!ends) {
2163	trace(` → sequential fallback`);
2164	const again = yield* sequential();
2165	staged += yield* stageResult(again.bytes, again.base, again.r);
2166	prevEnd = again.r.consumedTo;
2167	continue;
2168	}
2169	staged += yield* stageResult(
2170	region,
2171	prevEnd,
2172	keepBelow(scanned, result.firstOffset),
2173	);
2174	}
2175	staged += yield* stageResult(current.payload, current.base, result);
2176	prevEnd = result.consumedTo;
2177	if (prevEnd < chunkEnd && staged < count) {
2178	const start = prevEnd;
2179	speculative = {
2180	start,
2181	fiber: yield* Effect.forkChild(speculate(start)),
2182	};
2183	}
2184	}
2185	// A straddler still pending at the end (the last parts held no
2186	// boundary): settle it before the tail scan.
2187	if (speculative !== undefined) {
2188	const sp = yield* Fiber.join(speculative.fiber);
2189	const start = speculative.start;
2190	speculative = undefined;
2191	if (sp.r.count === 1 && start === prevEnd) {
2192	staged += yield* stageResult(sp.bytes, start, sp.r);
2193	prevEnd = sp.r.consumedTo;
2194	}
2195	}
2196	return { staged, prevEnd };
2197	});
2198	const consumer = yield* Effect.forkChild(consume);
2199	yield* produce;
2200	console.log(`[pump] body ended at +${Date.now() - pumpStarted}ms`);
2201	phases.upload = Date.now() - pumpStarted;
2202	let { staged, prevEnd } = yield* Fiber.join(consumer);
2203	// The tail: chunks too small to hold two whole entries report no
2204	// boundary, so entries after the last settled boundary may remain.
2205	// Scan that final region sequentially (it ends at the trailer).
2206	if (staged < count) {
2207	const last = chunks[chunks.length - 1];
2208	if (last !== undefined) {
2209	const to = last.base + last.payload.length;
2210	if (to > prevEnd) {
2211	const bytes = bytesFrom(prevEnd, to);
2212	const r = yield* hasher
2213	.hashPart(bytes, {
2214	base: prevEnd,
2215	remaining: count - staged,
2216	maxObjectSize: MAX_OBJECT_SIZE,
2217	})
2218	.pipe(Effect.mapError(asIngest));
2219	staged += yield* stageResult(bytes, prevEnd, r);
2220	prevEnd = r.consumedTo;
2221	}
2222	}
2223	}
2224	const consumedTo = prevEnd;
2225	phases.scan = Date.now() - pumpStarted;
2226	phases.chunks = chunks.length;
2227	phases.regions = regionCalls;
2228	if (staged !== count) {
2229	return yield* new PackIngestError({
2230	reason: `pack declared ${count} entries, scanned ${staged}`,
2231	});
2232	}
2233	const total =
2234	source.awaitEnd === undefined
2235	? source.size
2236	: yield* source.awaitEnd.pipe(Effect.mapError(asIngest));
2237	if (consumedTo !== total - 20) {
2238	return yield* new PackIngestError({
2239	reason: `pack has ${total - 20 - consumedTo} unconsumed bytes after ${count} entries`,
2240	});
2241	}
2242	// Everything read so far went through `feed`; the last part's
2243	// bytes did too, so the lag now holds the trailer.
2244	if (lag.length !== 20) {
2245	return yield* new PackIngestError({
2246	reason: "truncated pack trailer",
2247	});
2248	}
2249	const expected = bytesToHex(lag);
2250	const actual = trailerSha.digestHex();
2251	if (expected !== actual) {
2252	return yield* new PackIngestError({
2253	reason: `pack checksum mismatch: expected ${expected}, got ${actual}`,
2254	});
2255	}
2256	// Cross-part and thin deltas: bases are in the retained payloads
2257	// (or in the live store). Fixpoint for chains.
2258	const deltasStarted = Date.now();
2259	phases.unresolved = unresolved.length;
2260	// The resolved pack (DESIGN §22.13): every delta-resolved blob's
2261	// fresh zlib, `typeSize header + zdata` per entry, one object; rows
2262	// point at their zdata like promoted wire-pack rows do. Written in
2263	// two parts so the in-chunk resolutions — most of them — upload
2264	// while the cross-chunk rounds run.
2265	const writeResolvedPack = (suffix: string) =>
2266	Effect.gen(function* () {
2267	if (resolvedBlobs.length === 0 || spill === undefined) return;
2268	const packId = `${spill.packId}${suffix}`;
2269	const key = packKeyOf(spill.repoId, packId);
2270	const pieces: Array<Uint8Array> = [];
2271	const rows: Array<Incoming> = [];
2272	let cursor = 0;
2273	for (const entry of resolvedBlobs) {
2274	const header = encodeTypeSize(entry.type, entry.size);
2275	pieces.push(header, entry.zdata!);
2276	rows.push({
2277	...entry,
2278	pack: { packId, offset: cursor + header.length },
2279	});
2280	cursor += header.length + entry.zdata!.byteLength;
2281	}
2282	resolvedBlobs.length = 0;
2283	const bytes = concatBytes(pieces);
2284	const writeStarted = Date.now();
2285	const writing = yield* Effect.forkDetach(
2286	spill.blobs.put(key, bytes).pipe(
2287	Effect.mapError(
2288	(error) =>
2289	new PackIngestError({
2290	reason: `resolved pack: ${error.reason}`,
2291	}),
2292	),
2293	Effect.provide(RuntimeContext.phantom),
2294	),
2295	);
2296	resolvedKeys.push(key);
2297	resolvedKey = resolvedKeys.join(",");
2298	phases.resolvedBytes = (phases.resolvedBytes ?? 0) + bytes.length;
2299	yield* stage(rows);
2300	return { writing, writeStarted };
2301	});
2302	const early = yield* writeResolvedPack("-r");
2303	const spanOf = (at: number, length: number) =>
2304	Effect.suspend(() => {
2305	const bytes = readSpan?.(at, length);
2306	return bytes === undefined
2307	? source.read(at, length)
2308	: Effect.succeed(bytes);
2309	});
2310	const inflateSpan = (item: Known) =>
2311	item.content !== undefined
2312	? Effect.succeed(item.content)
2313	: item.zdata !== undefined
2314	? Zlib.inflate(item.zdata).pipe(Effect.mapError(asIngest))
2315	: spanOf(item.dataOffset, item.span).pipe(
2316	Effect.flatMap((z) => Zlib.inflate(z)),
2317	Effect.mapError(asIngest),
2318	);
2319	// Rounds (DESIGN §22.13): every delta whose base is known goes into
2320	// byte-bounded batches the hasher resolves in parallel (its bases
2321	// shipped once per batch); a delta on a still-unresolved delta waits
2322	// for the next round. The receiver only assembles bytes it already
2323	// holds.
2324	const batchBudget = hasher.chunkBytes ?? HASH_PART_BYTES;
2325	const baseBytesOf = (item: Known) =>
2326	Effect.gen(function* () {
2327	if (item.content !== undefined) {
2328	return { bytes: item.content, isContent: true };
2329	}
2330	if (item.zdata !== undefined) {
2331	return { bytes: item.zdata, isContent: false };
2332	}
2333	return {
2334	bytes: yield* spanOf(item.dataOffset, item.span),
2335	isContent: false,
2336	};
2337	});
2338	let pending = unresolved;
2339	while (pending.length > 0) {
2340	const next: Array<UnresolvedDelta> = [];
2341	interface Ready {
2342	readonly u: UnresolvedDelta;
2343	readonly base: Known;
2344	readonly baseKey: string;
2345	}
2346	const ready: Array<Ready> = [];
2347	for (const u of pending) {
2348	let base: Known | undefined =
2349	u.baseOffset !== undefined ? known.get(u.baseOffset) : undefined;
2350	if (base === undefined && u.baseOid !== undefined) {
2351	base = knownByOid.get(u.baseOid);
2352	if (base === undefined) {
2353	const live = yield* store.readBase(u.baseOid);
2354	if (live !== undefined) {
2355	base = {
2356	oid: u.baseOid,
2357	type: live.type,
2358	dataOffset: -1,
2359	span: 0,
2360	content: live.content,
2361	};
2362	knownByOid.set(u.baseOid, base);
2363	}
2364	}
2365	}
2366	if (base === undefined) {
2367	next.push(u);
2368	continue;
2369	}
2370	ready.push({
2371	u,
2372	base,
2373	baseKey:
2374	base.dataOffset >= 0
2375	? `o:${base.dataOffset}`
2376	: `oid:${base.oid}`,
2377	});
2378	}
2379	if (ready.length === 0) {
2380	return yield* new PackIngestError({
2381	reason: `delta base not found for ${next.length} entries`,
2382	});
2383	}
2384	// Greedy byte-bounded batches; a base shared by several deltas is
2385	// carried once per batch.
2386	interface Batch {
2387	readonly bases: Array<DeltaBase>;
2388	readonly index: Map<string, number>;
2389	readonly jobs: Array<DeltaJob>;
2390	readonly items: Array<Ready>;
2391	bytes: number;
2392	}
2393	const batches: Array<Batch> = [];
2394	let current: Batch | undefined;
2395	for (const r of ready) {
2396	const delta = yield* spanOf(r.u.dataOffset, r.u.span);
2397	const baseBytes = yield* baseBytesOf(r.base);
2398	const baseCost = current?.index.has(r.baseKey)
2399	? 0
2400	: baseBytes.bytes.length;
2401	if (
2402	current === undefined ||
2403	(current.jobs.length > 0 &&
2404	current.bytes + delta.length + baseCost > batchBudget)
2405	) {
2406	current = {
2407	bases: [],
2408	index: new Map(),
2409	jobs: [],
2410	items: [],
2411	bytes: 0,
2412	};
2413	batches.push(current);
2414	}
2415	let baseIndex = current.index.get(r.baseKey);
2416	if (baseIndex === undefined) {
2417	baseIndex = current.bases.push(baseBytes) - 1;
2418	current.index.set(r.baseKey, baseIndex);
2419	current.bytes += baseBytes.bytes.length;
2420	}
2421	current.jobs.push({
2422	id: current.jobs.length,
2423	type: r.base.type,
2424	base: baseIndex,
2425	delta,
2426	});
2427	current.items.push(r);
2428	current.bytes += delta.length;
2429	}
2430	const resolved = yield* Effect.all(
2431	batches.map((batch) =>
2432	hasher
2433	.resolveDeltas(batch.bases, batch.jobs, {
2434	maxObjectSize: MAX_OBJECT_SIZE,
2435	})
2436	.pipe(
2437	Effect.catchTag("HashError", (error) => {
2438	console.warn(`[pump] ${error.reason}; resolving inline`);
2439	return PartialScan.resolveDeltas(batch.bases, batch.jobs, {
2440	maxObjectSize: MAX_OBJECT_SIZE,
2441	});
2442	}),
2443	Effect.mapError(asIngest),
2444	Effect.map((results) => ({ batch, results })),
2445	),
2446	),
2447	{ concurrency: hasher.concurrency ?? HASH_CONCURRENCY },
2448	);
2449	for (const { batch, results } of resolved) {
2450	const incoming: Array<Incoming> = [];
2451	for (const result of results) {
2452	const r = batch.items[result.id]!;
2453	const item: Known = {
2454	oid: result.oid,
2455	type: r.base.type,
2456	dataOffset: r.u.dataOffset,
2457	span: r.u.span,
2458	zdata: result.zdata,
2459	content: result.content,
2460	};
2461	known.set(r.u.offset, item);
2462	knownByOid.set(result.oid, item);
2463	incoming.push({
2464	oid: result.oid,
2465	type: r.base.type,
2466	size: result.size,
2467	zdata: result.zdata,
2468	span: r.u.span,
2469	dataOffset: r.u.dataOffset,
2470	fromDelta: true,
2471	content: result.content,
2472	});
2473	}
2474	yield* stage(incoming);
2475	}
2476	pending = next;
2477	}
2478	phases.deltas = Date.now() - deltasStarted;
2479	const late = yield* writeResolvedPack("-r2");
2480	yield* flush();
2481	for (const pending of [early, late]) {
2482	if (pending === undefined) continue;
2483	yield* Fiber.join(pending.writing);
2484	phases.resolvedPack = Math.max(
2485	phases.resolvedPack ?? 0,
2486	Date.now() - pending.writeStarted,
2487	);
2488	}
2489	// Settle the spill: every part was written by its hasher; complete
2490	// the object and hand it to the source as the fallback reader.
2491	if (upload !== undefined && spill !== undefined) {
2492	const settled = upload;
2493	const uploaded: Array<UploadedPart> = [];
2494	for (const part of parts) uploaded.push(yield* part);
2495	yield* settled.complete(uploaded).pipe(
2496	Effect.mapError(
2497	(error) =>
2498	new PackIngestError({
2499	reason: `spill complete: ${error.reason}`,
2500	}),
2501	),
2502	Effect.provide(RuntimeContext.phantom),
2503	);
2504	spill.feeder.setFallback(
2505	blobRandomAccess({
2506	blobs: spill.blobs,
2507	key: spill.key,
2508	size: yield* spill.feeder.source.awaitEnd.pipe(
2509	Effect.mapError(asIngest),
2510	),
2511	}),
2512	);
2513	parkedKey = spill.key;
2514	}
2515	return { count };
2516	}).pipe(
2517	// A failed ingest must not leave a dangling multipart upload —
2518	// R2 bills incomplete uploads until aborted or lifecycle-expired.
2519	Effect.onError(() =>
2520	Effect.suspend(() =>
2521	upload === undefined || parkedKey !== undefined
2522	? Effect.void
2523	: upload.abort.pipe(Effect.provide(RuntimeContext.phantom)),
2524	).pipe(Effect.ignore),
2525	),
2526	Effect.onError(() =>
2527	Effect.suspend(() =>
2528	resolvedKeys.length === 0 || options.spill === undefined
2529	? Effect.void
2530	: options.spill.blobs
2531	.delete([...resolvedKeys])
2532	.pipe(Effect.provide(RuntimeContext.phantom)),
2533	).pipe(Effect.ignore),
2534	),
2535	);
2536
2537	const summary =
2538	options.hasher !== undefined
2539	? yield* pumpThroughHasher(options.hasher)
2540	: yield* PackParser.ingestPack({
2541	source,
2542	store,
2543	maxObjectSize: MAX_OBJECT_SIZE,
2544	phases,
2545	sink: (entry) => stage([entry]),
2546	sinkBatch: stage,
2547	}).pipe(
2548	Effect.mapError((error) =>
2549	error._tag === "StoreError" || error._tag === "PackIngestError"
2550	? error
2551	: new PackIngestError({ reason: packParserReason(error) }),
2552	),
2553	);
2554
2555	yield* flush();
2556	if (store.settle !== undefined) {
2557	const at = yield* Effect.sync(() => performance.now());
2558	yield* store.settle;
2559	stageMs += yield* Effect.sync(() => performance.now() - at);
2560	}
2561
2562	return {
2563	stageMs,
2564	phases,
2565	promoted,
2566	parkedKey,
2567	resolvedKey,
2568	objectCount: summary.count,
2569	// The parser verifies the trailer itself; the checksum is not used
2570	// downstream, so it is not re-derived here.
2571	packSha: "",
2572	commits,
2573	// The connectivity check (DESIGN §22.14) needs only the edges that
2574	// leave the push: everything staged here is known present.
2575	referenced: Array.from(referenced).filter((oid) => !stagedOids.has(oid)),
2576	referencedParents: Array.from(referencedParents).filter(
2577	(oid) => !stagedOids.has(oid),
2578	),
2579	} satisfies IngestResult;
2580	});
2581
2582	/** Renders a `PackParser` failure as an in-band `unpack <reason>` string. */
2583	const packParserReason = (error: {
2584	readonly _tag: string;
2585	readonly reason?: string;
2586	readonly expected?: string;
2587	readonly actual?: string;
2588	readonly baseOid?: string;
2589	readonly size?: number;
2590	readonly limit?: number;
2591	}): string => {
2592	switch (error._tag) {
2593	case "PackChecksumMismatch":
2594	return "pack checksum mismatch";
2595	case "MissingDeltaBaseError":
2596	return `missing delta base ${error.baseOid ?? ""}`.trim();
2597	case "ObjectTooLargeError":
2598	return `object too large (${error.size ?? 0} > ${error.limit ?? 0})`;
2599	default:
2600	return error.reason ?? error._tag;
2601	}
2602	};
2603
2604	export const ingestPack = (
2605	pack: Uint8Array,
2606	options: Omit<IngestOptions, "store"> & { readonly store: ObjectStore },
2607	): Effect.Effect<IngestResult, PackIngestError | StoreError> =>
2608	Effect.gen(function* () {
2609	const { store, pushId } = options;
2610	if (pack.length < 12 + 20) {
2611	return yield* new PackIngestError({ reason: "pack too small" });
2612	}
2613	if (
2614	pack[0] !== 0x50 ||
2615	pack[1] !== 0x41 ||
2616	pack[2] !== 0x43 ||
2617	pack[3] !== 0x4b
2618	) {
2619	return yield* new PackIngestError({ reason: "bad pack signature" });
2620	}
2621	const view = new DataView(pack.buffer, pack.byteOffset, pack.byteLength);
2622	const version = view.getUint32(4);
2623	if (version !== 2) {
2624	return yield* new PackIngestError({
2625	reason: `unsupported pack version ${version}`,
2626	});
2627	}
2628	const count = view.getUint32(8);
2629
2630	// Trailer verification first — cheap, and everything after assumes an
2631	// intact buffer.
2632	const trailer = bytesToHex(pack.subarray(pack.length - 20));
2633	const sha = makeSha1();
2634	sha.update(pack.subarray(0, pack.length - 20));
2635	const actual = sha.digestHex();
2636	if (actual !== trailer) {
2637	return yield* new PackIngestError({ reason: "pack checksum mismatch" });
2638	}
2639
2640	const cache = makeContentCache(CACHE_BUDGET);
2641	const entries: Array<RawEntry> = [];
2642	const byOffset = new Map<number, RawEntry>();
2643	const byOid = new Map<Oid, RawEntry>();
2644
2645	const commits: Array<StagedCommitInfo> = [];
2646	const referenced = new Set<Oid>();
2647	const referencedParents = new Set<Oid>();
2648
2649	/** Record graph facts + referenced edges for a resolved object. */
2650	const recordObject = Effect.fn(function* (
2651	oid: Oid,
2652	type: ObjectType,
2653	content: Uint8Array,
2654	) {
2655	switch (type) {
2656	case ObjectType.commit: {
2657	const parsed = yield* parseCommit(content).pipe(
2658	Effect.mapError(
2659	(error) =>
2660	new PackIngestError({
2661	reason: `bad commit ${oid}: ${error.reason}`,
2662	}),
2663	),
2664	);
2665	commits.push({
2666	oid,
2667	tree: parsed.tree,
2668	parents: parsed.parents,
2669	commitTime: parsed.committer.when,
2670	});
2671	referenced.add(parsed.tree);
2672	for (const parent of parsed.parents) {
2673	referencedParents.add(parent);
2674	}
2675	return;
2676	}
2677	case ObjectType.tree: {
2678	const parsed = yield* parseTree(content).pipe(
2679	Effect.mapError(
2680	(error) =>
2681	new PackIngestError({
2682	reason: `bad tree ${oid}: ${error.reason}`,
2683	}),
2684	),
2685	);
2686	for (const entry of parsed) {
2687	if (treeEntryKind(entry.mode) !== "commit") {
2688	referenced.add(entry.oid);
2689	}
2690	}
2691	return;
2692	}
2693	case ObjectType.tag: {
2694	const parsed = yield* parseTag(content).pipe(
2695	Effect.mapError(
2696	(error) =>
2697	new PackIngestError({
2698	reason: `bad tag ${oid}: ${error.reason}`,
2699	}),
2700	),
2701	);
2702	referenced.add(parsed.object);
2703	return;
2704	}
2705	case ObjectType.blob:
2706	return;
2707	}
2708	});
2709
2710	const stage = Effect.fn(function* (
2711	oid: Oid,
2712	type: ObjectType,
2713	content: Uint8Array,
2714	zdata: Uint8Array,
2715	) {
2716	const staged: StagedObject = {
2717	oid,
2718	type,
2719	size: content.length,
2720	zdata,
2721	};
2722	yield* store.insertStaged(pushId, staged);
2723	yield* recordObject(oid, type, content);
2724	});
2725
2726	// ── Pass 1: scan entries, stage non-deltas verbatim ─────────────────────
2727	let offset = 12;
2728	const zlibToIngest = (error: Zlib.ZlibError): PackIngestError =>
2729	new PackIngestError({ reason: error.reason });
2730
2731	for (let i = 0; i < count; i++) {
2732	const entryStart = offset;
2733	let header;
2734	try {
2735	header = decodeTypeSize(pack, offset);
2736	} catch (error) {
2737	return yield* new PackIngestError({
2738	reason: `entry ${i}: ${String(error)}`,
2739	});
2740	}
2741	if (!isDeltaType(header.type)) {
2742	if (header.size > MAX_OBJECT_SIZE) {
2743	return yield* new PackIngestError({ reason: "object too large" });
2744	}
2745	const inflated = yield* Zlib.inflateEntry(pack, header.next, {
2746	maxOutput: header.size,
2747	}).pipe(Effect.mapError(zlibToIngest));
2748	if (inflated.content.length !== header.size) {
2749	return yield* new PackIngestError({
2750	reason: `entry ${i}: size mismatch (${inflated.content.length} != ${header.size})`,
2751	});
2752	}
2753	const type = header.type as ObjectType;
2754	const oid = yield* hashObject(type, inflated.content);
2755	const zdata = pack.subarray(
2756	header.next,
2757	header.next + inflated.bytesConsumed,
2758	);
2759	const entry: RawEntry = {
2760	offset: entryStart,
2761	kind: "object",
2762	type,
2763	declaredSize: header.size,
2764	zstart: header.next,
2765	zlen: inflated.bytesConsumed,
2766	baseOffset: undefined,
2767	baseOid: undefined,
2768	oid,
2769	resolvedType: type,
2770	};
2771	entries.push(entry);
2772	byOffset.set(entryStart, entry);
2773	byOid.set(oid, entry);
2774	cache.set(oid, inflated.content);
2775	yield* stage(oid, type, inflated.content, zdata);
2776	offset = header.next + inflated.bytesConsumed;
2777	} else {
2778	let zstart = header.next;
2779	let baseOffset: number | undefined;
2780	let baseOid: Oid | undefined;
2781	if (header.type === 6) {
2782	let decoded;
2783	try {
2784	decoded = decodeOfsDeltaOffset(pack, header.next);
2785	} catch (error) {
2786	return yield* new PackIngestError({
2787	reason: `entry ${i}: ${String(error)}`,
2788	});
2789	}
2790	baseOffset = entryStart - decoded.value;
2791	if (baseOffset < 12) {
2792	return yield* new PackIngestError({
2793	reason: `entry ${i}: ofs-delta base offset out of range`,
2794	});
2795	}
2796	zstart = decoded.next;
2797	} else {
2798	if (header.next + 20 > pack.length) {
2799	return yield* new PackIngestError({
2800	reason: `entry ${i}: truncated ref-delta base id`,
2801	});
2802	}
2803	baseOid = bytesToHex(pack.subarray(header.next, header.next + 20));
2804	zstart = header.next + 20;
2805	}
2806	const inflated = yield* Zlib.inflateEntry(pack, zstart, {
2807	maxOutput: header.size,
2808	}).pipe(Effect.mapError(zlibToIngest));
2809	const entry: RawEntry = {
2810	offset: entryStart,
2811	kind: header.type === 6 ? "ofs" : "ref",
2812	type: undefined,
2813	declaredSize: header.size,
2814	zstart,
2815	zlen: inflated.bytesConsumed,
2816	baseOffset,
2817	baseOid,
2818	oid: undefined,
2819	resolvedType: undefined,
2820	};
2821	entries.push(entry);
2822	byOffset.set(entryStart, entry);
2823	offset = zstart + inflated.bytesConsumed;
2824	}
2825	if (offset > pack.length - 20) {
2826	return yield* new PackIngestError({
2827	reason: `entry ${i}: pack truncated`,
2828	});
2829	}
2830	}
2831	if (offset !== pack.length - 20) {
2832	return yield* new PackIngestError({
2833	reason: "trailing garbage between entries and pack trailer",
2834	});
2835	}
2836
2837	// ── Pass 2: resolve deltas (bases may chain; thin bases from store) ─────
2838	const contentOf = (
2839	entry: RawEntry,
2840	depth: number,
2841	): Effect.Effect<
2842	{ readonly content: Uint8Array; readonly type: ObjectType },
2843	PackIngestError | StoreError
2844	> =>
2845	Effect.gen(function* () {
2846	if (depth > MAX_DELTA_DEPTH) {
2847	return yield* new PackIngestError({ reason: "delta chain too deep" });
2848	}
2849	if (entry.oid !== undefined && entry.resolvedType !== undefined) {
2850	const cached = cache.get(entry.oid);
2851	if (cached !== undefined) {
2852	return { content: cached, type: entry.resolvedType };
2853	}
2854	}
2855	if (entry.kind === "object") {
2856	const inflated = yield* Zlib.inflateEntry(pack, entry.zstart, {
2857	maxOutput: entry.declaredSize,
2858	}).pipe(Effect.mapError(zlibToIngest));
2859	if (entry.oid !== undefined) cache.set(entry.oid, inflated.content);
2860	return { content: inflated.content, type: entry.type! };
2861	}
2862	// delta — resolve the base first
2863	const base = yield* Effect.gen(function* () {
2864	if (entry.kind === "ofs") {
2865	const baseEntry = byOffset.get(entry.baseOffset!);
2866	if (baseEntry === undefined) {
2867	return yield* new PackIngestError({
2868	reason: "ofs-delta base is not an entry boundary",
2869	});
2870	}
2871	return yield* contentOf(baseEntry, depth + 1);
2872	}
2873	const inPack = byOid.get(entry.baseOid!);
2874	if (inPack !== undefined) {
2875	return yield* contentOf(inPack, depth + 1);
2876	}
2877	// Thin base: must exist in the live store.
2878	const meta = yield* store.getMeta(entry.baseOid!);
2879	if (meta === undefined) {
2880	return yield* new PackIngestError({
2881	reason: `missing thin-pack base ${entry.baseOid}`,
2882	});
2883	}
2884	if (meta.size > MAX_OBJECT_SIZE) {
2885	return yield* new PackIngestError({ reason: "object too large" });
2886	}
2887	const content = yield* store.readContent(entry.baseOid!);
2888	cache.set(entry.baseOid!, content);
2889	return { content, type: meta.type };
2890	});
2891	const delta = yield* Zlib.inflateEntry(pack, entry.zstart, {
2892	maxOutput: entry.declaredSize,
2893	}).pipe(Effect.mapError(zlibToIngest));
2894	const content = yield* applyDelta(base.content, delta.content).pipe(
2895	Effect.mapError(
2896	(error) => new PackIngestError({ reason: error.reason }),
2897	),
2898	);
2899	if (content.length > MAX_OBJECT_SIZE) {
2900	return yield* new PackIngestError({ reason: "object too large" });
2901	}
2902	return { content, type: base.type };
2903	});
2904
2905	for (const entry of entries) {
2906	if (entry.kind === "object" || entry.oid !== undefined) continue;
2907	const resolved = yield* contentOf(entry, 0);
2908	const oid = yield* hashObject(resolved.type, resolved.content);
2909	entry.oid = oid;
2910	entry.resolvedType = resolved.type;
2911	byOid.set(oid, entry);
2912	cache.set(oid, resolved.content);
2913	const zdata = yield* Zlib.deflate(resolved.content).pipe(
2914	Effect.mapError(zlibToIngest),
2915	);
2916	yield* stage(oid, resolved.type, resolved.content, zdata);
2917	}
2918
2919	return {
2920	objectCount: count,
2921	packSha: trailer,
2922	commits,
2923	referenced: Array.from(referenced),
2924	referencedParents: Array.from(referencedParents),
2925	} satisfies IngestResult;
2926	});
2927
2928	// ─────────────────────────────────────────────────────────────────────────────
2929	// Small shared helpers
2930	// ─────────────────────────────────────────────────────────────────────────────
2931
2932	const asOid = (s: string): ApiOid => s as ApiOid;
2933
2934	/** A node of the merge-base paint walk's max-heap (gen desc, time desc). */
2935	export interface WalkHeapNode {
2936	readonly oid: string;
2937	readonly gen: number;
2938	readonly time: number;
2939	}
2940
2941	const heapBefore = (a: WalkHeapNode, b: WalkHeapNode): boolean =>
2942	a.gen !== b.gen ? a.gen > b.gen : a.time > b.time;
2943
2944	/** Pushes onto a binary max-heap keyed `(gen desc, commit_time desc)`. */
2945	export const heapPush = (
2946	heap: Array<WalkHeapNode>,
2947	node: WalkHeapNode,
2948	): void => {
2949	heap.push(node);
2950	let i = heap.length - 1;
2951	while (i > 0) {
2952	const parent = (i - 1) >> 1;
2953	if (!heapBefore(heap[i]!, heap[parent]!)) break;
2954	const tmp = heap[parent]!;
2955	heap[parent] = heap[i]!;
2956	heap[i] = tmp;
2957	i = parent;
2958	}
2959	};
2960
2961	/** Pops the max node. Callers must check `heap.length > 0` first. */
2962	export const heapPop = (heap: Array<WalkHeapNode>): WalkHeapNode => {
2963	const top = heap[0]!;
2964	const last = heap.pop()!;
2965	if (heap.length > 0) {
2966	heap[0] = last;
2967	let i = 0;
2968	for (;;) {
2969	const left = 2 * i + 1;
2970	const right = left + 1;
2971	let max = i;
2972	if (left < heap.length && heapBefore(heap[left]!, heap[max]!)) {
2973	max = left;
2974	}
2975	if (right < heap.length && heapBefore(heap[right]!, heap[max]!)) {
2976	max = right;
2977	}
2978	if (max === i) break;
2979	const tmp = heap[max]!;
2980	heap[max] = heap[i]!;
2981	heap[i] = tmp;
2982	i = max;
2983	}
2984	}
2985	return top;
2986	};
2987
2988	/** Decompresses a request body when it is gzip (header or 1f8b sniff). */
2989	export const gunzipIfNeeded = (
2990	body: Uint8Array,
2991	contentEncoding: string | undefined,
2992	): Effect.Effect<Uint8Array, PackIngestError> => {
2993	const looksGzip = body.length >= 2 && body[0] === 0x1f && body[1] === 0x8b;
2994	if (contentEncoding?.toLowerCase() !== "gzip" && !looksGzip) {
2995	return Effect.succeed(body);
2996	}
2997	if (!looksGzip) {
2998	// Content-Encoding claimed gzip but the magic disagrees — trust bytes.
2999	return Effect.succeed(body);
3000	}
3001	return Effect.tryPromise({
3002	try: () =>
3003	new Response(
3004	new Response(body as BodyInit).body!.pipeThrough(
3005	new DecompressionStream("gzip"),
3006	),
3007	)
3008	.arrayBuffer()
3009	.then((buffer) => new Uint8Array(buffer)),
3010	catch: (error) =>
3011	new PackIngestError({ reason: `gzip decompression failed: ${error}` }),
3012	});
3013	};
3014
3015	const noCache = { "cache-control": "no-cache" } as const;
3016
3017	// ─────────────────────────────────────────────────────────────────────────────
3018	// The Durable Object
3019	// ─────────────────────────────────────────────────────────────────────────────
3020
3021	/**
3022	* The per-repo Durable Object class (modular form). The runtime
3023	* implementation is {@link GitRepoLive}; the hosting Worker provides it via
3024	* `Effect.provide(GitRepoLive)` and consumers `yield* GitRepo` for the
3025	* namespace handle.
3026	*/
3027	export class GitRepo extends Cloudflare.DurableObject<GitRepo, GitRepoShape>()(
3028	"GitRepo",
3029	{
3030	// Every tagged error the RPC surface can fail with: the calling Worker
3031	// reconstructs real class instances from the RPC wire (see the
3032	// `DurableObjectProps.errors` doc), so catchTag/instanceof/HttpApi
3033	// encoding all see the classes the DO actually failed with.
3034	errors: [
3035	ReadOnlyRepo,
3036	RepoNotFound,
3037	RefNotFound,
3038	RefConflict,
3039	ObjectNotFound,
3040	WrongObjectType,
3041	NoMergeBase,
3042	PullNotFound,
3043	PullExists,
3044	PullStateConflict,
3045	BranchMissing,
3046	NothingToMerge,
3047	MergeConflict,
3048	ValidationError,
3049	StoreError,
3050	PackIngestError,
3051	WireProtocolError,
3052	],
3053	},
3054	) {}
3055
3056	/**
3057	* A repository's RPC stub: {@link GitRepoShape} as the Worker calls it,
3058	* with `fetch` taking the request.
3059	*/
3060	export type RepoStub = ReturnType<
3061	Cloudflare.Workers.DurableObject<GitRepoShape>["getByName"]
3062	>;
3063
3064	/** What {@link RepoStore} provides: a repository's stub by id. */
3065	export interface RepoStoreShape {
3066	/** The repository's stub, by its registry id (a ULID). */
3067	readonly getByName: (repoId: string) => RepoStub;
3068	}
3069
3070	/**
3071	* `Git.RepoStore` — the repository block as a service: every route reaches
3072	* a repository through it. `ReposDurableObject` provides it over the
3073	* {@link GitRepo} Durable Object namespace. It is the seam for decorating
3074	* repositories (hook `commitPush`, `updateRef`, `mergePull` to observe ref
3075	* moves) or for implementing {@link GitRepoShape} over another store.
3076	*
3077	* @binding
3078	*/
3079	export class RepoStore extends Context.Service<RepoStore, RepoStoreShape>()(
3080	"alchemy/Git/RepoStore",
3081	) {}
3082
3083	/**
3084	* The Repo DO implementation Layer. Requires the {@link BlobStore}
3085	* service (provide `Git.BlobStoreR2(yourBucket)` — or any other
3086	* implementation — in the same layer graph) and the {@link Registry}
3087	* namespace.
3088	*/
3089	export const GitRepoLive = GitRepo.make(
3090	Effect.gen(function* () {
3091	// ── Outer init: resolve bindings + namespaces (runs at plan time too) ──
3092	const state = yield* Cloudflare.DurableObjectState;
3093	// The building-blocks seam: bulk bytes live behind whatever BlobStore
3094	// the user's layer graph provides (R2 by default, S3, ...).
3095	const blobs: BlobStoreShape = yield* BlobStore;
3096	// The push pipeline's hasher (DESIGN §22.7): a self-binding fan-out in
3097	// production, in-process in tests without the binding.
3098	const hasher = yield* Hasher;
3099	const registry = yield* RegistryStore;
3100	const selfNamespace = yield* Cloudflare.DurableObjectScope;
3101
3102	return Effect.gen(function* () {
3103	// ── Inner init: per-instance construction (runtime only) ────────────
3104	const sql = makeSqlClient(state);
3105	yield* initRepoSchema(sql).pipe(Effect.orDie);
3106	// Isolate-wide, not per repo: the memory it meters is shared.
3107	const pushSemaphore = yield* isolatePushGate;
3108
3109	// ── config helpers ───────────────────────────────────────────────────
3110	const getConfig = (key: string) =>
3111	sql
3112	.first<{ value: string }>(
3113	`SELECT value FROM config WHERE key = ?`,
3114	key,
3115	)
3116	.pipe(Effect.map((row) => row?.value));
3117
3118	const setConfig = (key: string, value: string) =>
3119	sql.run(
3120	`INSERT INTO config (key, value) VALUES (?, ?)
3121	ON CONFLICT (key) DO UPDATE SET value = excluded.value`,
3122	key,
3123	value,
3124	);
3125
3126	/**
3127	* Repo metadata from the `config` table. The `objects` breakdown is
3128	* a GROUP BY over the whole `objects` table — on a large repo that
3129	* scan walks pages dense with inline zdata and was measured at
3130	* ~1.5 s per call (DESIGN §22). It is therefore opt-in: only the
3131	* REST repo-detail responses pay for it; the wire path, auth, and
3132	* every other caller get zeros there and never touch the table.
3133	*/
3134	const readMetaWith = (
3135	withStats: boolean,
3136	): Effect.Effect<RepoMetaData | undefined, StoreError> =>
3137	Effect.gen(function* () {
3138	const rows = yield* sql.all<{ key: string; value: string }>(
3139	`SELECT key, value FROM config`,
3140	);
3141	const map = new Map(rows.map((row) => [row.key, row.value]));
3142	const repoId = map.get("repo_id");
3143	if (repoId === undefined) return undefined;
3144	const stats = withStats
3145	? yield* sql.all<{
3146	bucket: string;
3147	n: number;
3148	bytes: number;
3149	}>(
3150	// Row-resident blobs are "loose" (awaiting compaction);
3151	// row-resident commits/trees/tags are "resident" — they stay
3152	// in SQLite for the life of the repo (Compact.ts BLOB_TYPE).
3153	`SELECT CASE WHEN location = 'row' AND type <> 3 THEN 'resident'
3154	WHEN location = 'row' THEN 'loose'
3155	ELSE location END AS bucket,
3156	COUNT(*) AS n, COALESCE(SUM(zsize), 0) AS bytes
3157	FROM objects WHERE ${LIVE_OBJECTS} GROUP BY bucket`,
3158	)
3159	: [];
3160	const byLocation = new Map(stats.map((row) => [row.bucket, row]));
3161	return {
3162	repoId,
3163	owner: map.get("owner") ?? "",
3164	name: map.get("name") ?? "",
3165	defaultBranch: map.get("default_branch") ?? "main",
3166	description: map.get("description") ?? null,
3167	readOnly: map.get("read_only") === "1",
3168	public: map.get("public") === "1",
3169	forkOf: map.get("fork_of") ?? null,
3170	status: (map.get("status") ?? "ready") as RepoStatus,
3171	createdAt: Number(map.get("created_at") ?? 0),
3172	lastPush:
3173	map.get("last_push") === undefined
3174	? null
3175	: (JSON.parse(map.get("last_push")!) as PushStatsData),
3176	objects: {
3177	loose: byLocation.get("loose")?.n ?? 0,
3178	resident: byLocation.get("resident")?.n ?? 0,
3179	packed: byLocation.get("pack")?.n ?? 0,
3180	r2: byLocation.get("r2")?.n ?? 0,
3181	bytes: stats.reduce((sum, row) => sum + row.bytes, 0),
3182	},
3183	} satisfies RepoMetaData;
3184	});
3185
3186	const readMeta = readMetaWith(false);
3187	const readMetaStats = readMetaWith(true);
3188
3189	const requireFrom = (
3190	read: Effect.Effect<RepoMetaData | undefined, StoreError>,
3191	): Effect.Effect<RepoMetaData, RepoNotFound | StoreError> =>
3192	read.pipe(
3193	Effect.flatMap((meta) =>
3194	meta === undefined
3195	? Effect.fail(new RepoNotFound({ owner: "", repo: "" }))
3196	: Effect.succeed(meta),
3197	),
3198	);
3199	/** Cheap: config only. `objects` reads as zeros. */
3200	const requireMeta = requireFrom(readMeta);
3201	/** The REST detail shape, with the full objects aggregate. */
3202	const requireMetaStats = requireFrom(readMetaStats);
3203
3204	/** The object store, keyed by the current repoId. */
3205	const storeFor = (repoId: string): ObjectStore =>
3206	makeObjectStore({ sql, blobs, repoId });
3207
3208	// ── refs ─────────────────────────────────────────────────────────────
3209	/** Peels a (possibly annotated-tag) oid to its final non-tag target. */
3210	const peelOid = (
3211	repoId: string,
3212	oid: Oid,
3213	): Effect.Effect<Oid | undefined, StoreError> =>
3214	Effect.gen(function* () {
3215	const objects = storeFor(repoId);
3216	let current = oid;
3217	for (let hops = 0; hops < 16; hops++) {
3218	const meta = yield* objects.getMeta(current);
3219	if (meta === undefined) return undefined;
3220	if (meta.type !== ObjectType.tag) {
3221	return hops === 0 ? undefined : current;
3222	}
3223	const content = yield* objects.readContent(current);
3224	const parsed = yield* parseTag(content).pipe(
3225	Effect.mapError(
3226	(error) => new StoreError({ reason: error.reason }),
3227	),
3228	);
3229	current = parsed.object;
3230	}
3231	return current;
3232	});
3233
3234	/**
3235	* Resolves a refname (short or full) or 40-hex oid to an object oid.
3236	* Candidates for short names: `refs/heads/<name>`, `refs/tags/<name>`.
3237	* `undefined` means HEAD (the default branch).
3238	*/
3239	const resolveRevision = Effect.fn(function* (
3240	defaultBranch: string,
3241	rev: string | undefined,
3242	) {
3243	const refName = rev === undefined ? `refs/heads/${defaultBranch}` : rev;
3244	if (isOid(refName)) return refName;
3245	const candidates = refName.startsWith("refs/")
3246	? [refName]
3247	: [`refs/heads/${refName}`, `refs/tags/${refName}`];
3248	for (const candidate of candidates) {
3249	const row = yield* sql.first<RefRow>(
3250	`SELECT name, oid FROM refs WHERE name = ?`,
3251	candidate,
3252	);
3253	if (row !== undefined) return row.oid;
3254	}
3255	return yield* new RefNotFound({ ref: refName });
3256	});
3257
3258	/** Resolves a revision all the way to a COMMIT oid (peels tags). */
3259	const resolveToCommit = Effect.fn(function* (
3260	repoId: string,
3261	defaultBranch: string,
3262	rev: string,
3263	) {
3264	const oid = yield* resolveRevision(defaultBranch, rev);
3265	const objects = storeFor(repoId);
3266	const meta = yield* objects.getMeta(oid);
3267	if (meta === undefined) return yield* new ObjectNotFound({ oid });
3268	if (meta.type === ObjectType.commit) return oid;
3269	if (meta.type === ObjectType.tag) {
3270	const peeled = yield* peelOid(repoId, oid);
3271	if (peeled === undefined) {
3272	return yield* new ObjectNotFound({ oid });
3273	}
3274	const peeledMeta = yield* objects.getMeta(peeled);
3275	if (peeledMeta === undefined) {
3276	return yield* new ObjectNotFound({ oid: peeled });
3277	}
3278	if (peeledMeta.type !== ObjectType.commit) {
3279	return yield* new WrongObjectType({
3280	oid: peeled,
3281	expected: "commit",
3282	actual: objectTypeName(peeledMeta.type),
3283	});
3284	}
3285	return peeled;
3286	}
3287	return yield* new WrongObjectType({
3288	oid,
3289	expected: "commit",
3290	actual: objectTypeName(meta.type),
3291	});
3292	});
3293
3294	const listAllRefs = (
3295	repoId: string,
3296	prefix?: string | undefined,
3297	): Effect.Effect<Array<RefData>, StoreError> =>
3298	Effect.gen(function* () {
3299	const rows =
3300	prefix === undefined || prefix.length === 0
3301	? yield* sql.all<RefRow>(
3302	`SELECT name, oid FROM refs ORDER BY name`,
3303	)
3304	: yield* sql.all<RefRow>(
3305	`SELECT name, oid FROM refs WHERE name >= ? AND name < ? ORDER BY name`,
3306	prefix,
3307	// next lexicographic sibling of the prefix
3308	`${prefix.slice(0, -1)}${String.fromCharCode(prefix.charCodeAt(prefix.length - 1) + 1)}`,
3309	);
3310	const out: Array<RefData> = [];
3311	for (const row of rows) {
3312	const peeled = row.name.startsWith("refs/tags/")
3313	? yield* peelOid(repoId, row.oid)
3314	: undefined;
3315	out.push(
3316	peeled === undefined
3317	? { name: row.name, oid: row.oid }
3318	: { name: row.name, oid: row.oid, peeled },
3319	);
3320	}
3321	return out;
3322	});
3323
3324	/**
3325	* The single transactional ref-CAS path (DESIGN.md §3.4). All ref
3326	* mutations — push commands, REST writes, import finalize — run here.
3327	*/
3328	interface RefResult {
3329	readonly ref: string;
3330	readonly ok: boolean;
3331	readonly reason?: string | undefined;
3332	}
3333	const finalizeRefTxn = (options: {
3334	readonly commands: ReadonlyArray<RefCommand>;
3335	readonly atomic: boolean;
3336	/** Skip the old-oid CAS entirely (import finalize). */
3337	readonly unconditional?: boolean | undefined;
3338	readonly pushId?: string | undefined;
3339	/** Leave the per-row `staged_push` rewrite to `flipPush` (after the response). */
3340	readonly deferFlip?: boolean | undefined;
3341	readonly graph: ReadonlyArray<{
3342	readonly oid: string;
3343	readonly tree: string;
3344	readonly gen: number;
3345	readonly commitTime: number;
3346	readonly parents: ReadonlyArray<string>;
3347	}>;
3348	/**
3349	* Extra statements run inside the transaction body, only when at
3350	* least one command succeeded (right after the graph inserts) —
3351	* e.g. the PR merge stamps its pulls row atomically with the ref
3352	* flip and the staged-object promotion. MUST be fully synchronous.
3353	*/
3354	readonly onCommitted?: ((raw: cf.SqlStorage) => void) | undefined;
3355	}): Effect.Effect<Array<RefResult>, StoreError, RuntimeContext> =>
3356	sql
3357	.transactionSync((raw) => {
3358	if (options.pushId !== undefined) {
3359	const push = raw
3360	.exec<{ state: string }>(
3361	`SELECT state FROM pushes WHERE push_id = ?`,
3362	options.pushId,
3363	)
3364	.toArray()[0];
3365	if (push?.state !== "staging")
3366	return options.commands.map((command) => ({
3367	ref: command.ref,
3368	ok: false,
3369	reason: "push is no longer active",
3370	}));
3371	}
3372	const currentOf = (name: string): string => {
3373	const rows = raw
3374	.exec<RefRow>(`SELECT name, oid FROM refs WHERE name = ?`, name)
3375	.toArray();
3376	return rows.length > 0 ? rows[0]!.oid : ZERO_OID;
3377	};
3378	const results: Array<RefResult> = [];
3379	// atomic: verify every CAS before writing anything
3380	if (options.atomic && options.unconditional !== true) {
3381	const failed = options.commands.some(
3382	(cmd) => currentOf(cmd.ref) !== cmd.oldOid,
3383	);
3384	if (failed) {
3385	for (const cmd of options.commands) {
3386	results.push({
3387	ref: cmd.ref,
3388	ok: false,
3389	reason:
3390	currentOf(cmd.ref) === cmd.oldOid
3391	? "atomic transaction failed"
3392	: "fetch first",
3393	});
3394	}
3395	return results;
3396	}
3397	}
3398	let anyOk = false;
3399	for (const cmd of options.commands) {
3400	if (
3401	options.unconditional !== true &&
3402	currentOf(cmd.ref) !== cmd.oldOid
3403	) {
3404	results.push({
3405	ref: cmd.ref,
3406	ok: false,
3407	reason: "fetch first",
3408	});
3409	continue;
3410	}
3411	if (cmd.newOid === ZERO_OID) {
3412	raw.exec(`DELETE FROM refs WHERE name = ?`, cmd.ref);
3413	} else {
3414	raw.exec(
3415	`INSERT INTO refs (name, oid) VALUES (?, ?)
3416	ON CONFLICT (name) DO UPDATE SET oid = excluded.oid`,
3417	cmd.ref,
3418	cmd.newOid,
3419	);
3420	}
3421	results.push({ ref: cmd.ref, ok: true });
3422	anyOk = true;
3423	}
3424	if (anyOk || options.commands.length === 0) {
3425	if (options.pushId !== undefined) {
3426	// Committing the push row is what makes its objects live
3427	// (LIVE_OBJECTS, DESIGN §22.14); nulling `staged_push` on
3428	// every row is a rewrite of the whole staged set and runs
3429	// after the response (`flipPush`), or inside this
3430	// transaction for callers that want it settled now.
3431	if (options.deferFlip !== true) {
3432	raw.exec(
3433	`UPDATE objects SET staged_push = NULL WHERE staged_push = ?`,
3434	options.pushId,
3435	);
3436	}
3437	raw.exec(
3438	`UPDATE pushes SET state = 'committed' WHERE push_id = ?`,
3439	options.pushId,
3440	);
3441	}
3442	// Multi-row inserts under SQLite's 100-parameter cap: a
3443	// per-row exec was most of finalize once rows were
3444	// coordinates (DESIGN §22.14).
3445	for (let at = 0; at < options.graph.length; at += 25) {
3446	const part = options.graph.slice(at, at + 25);
3447	raw.exec(
3448	`INSERT OR IGNORE INTO commits (oid, tree, gen, commit_time) VALUES ${part.map(() => "(?, ?, ?, ?)").join(", ")}`,
3449	...part.flatMap((commit) => [
3450	commit.oid,
3451	commit.tree,
3452	commit.gen,
3453	commit.commitTime,
3454	]),
3455	);
3456	}
3457	const parentRows = options.graph.flatMap((commit) =>
3458	commit.parents.map((parent, ord) => [commit.oid, parent, ord]),
3459	);
3460	for (let at = 0; at < parentRows.length; at += 33) {
3461	const part = parentRows.slice(at, at + 33);
3462	raw.exec(
3463	`INSERT OR IGNORE INTO commit_parents (oid, parent, ord) VALUES ${part.map(() => "(?, ?, ?)").join(", ")}`,
3464	...part.flat(),
3465	);
3466	}
3467	if (options.onCommitted !== undefined) {
3468	options.onCommitted(raw);
3469	}
3470	// First-branch-push default branch fixup: if the configured
3471	// default branch does not exist but a branch was just created,
3472	// repoint the default at it (prefer main/master).
3473	const configured = raw
3474	.exec<{ value: string }>(
3475	`SELECT value FROM config WHERE key = 'default_branch'`,
3476	)
3477	.toArray()[0]?.value;
3478	if (configured !== undefined) {
3479	const exists = raw
3480	.exec<RefRow>(
3481	`SELECT name, oid FROM refs WHERE name = ?`,
3482	`refs/heads/${configured}`,
3483	)
3484	.toArray();
3485	if (exists.length === 0) {
3486	const branches = raw
3487	.exec<RefRow>(
3488	`SELECT name, oid FROM refs WHERE name LIKE 'refs/heads/%' ORDER BY name`,
3489	)
3490	.toArray()
3491	.map((row) => row.name.slice("refs/heads/".length));
3492	const pick =
3493	branches.find((b) => b === "main") ??
3494	branches.find((b) => b === "master") ??
3495	branches[0];
3496	if (pick !== undefined) {
3497	raw.exec(
3498	`UPDATE config SET value = ? WHERE key = 'default_branch'`,
3499	pick,
3500	);
3501	}
3502	}
3503	}
3504	}
3505	return results;
3506	})
3507	.pipe(
3508	// Any committed ref movement changes what an anonymous reader
3509	// can see — republish the head snapshot (DESIGN.md §21).
3510	Effect.tap((results) =>
3511	results.some((result) => result.ok)
3512	? writeHeadSnapshot
3513	: Effect.void,
3514	),
3515	);
3516
3517	/** Computes gen/commit_time rows for staged commits (§3.6 step 6). */
3518	const computeGraphRows = (
3519	staged: ReadonlyArray<StagedCommitInfo>,
3520	): Effect.Effect<
3521	Array<{
3522	oid: string;
3523	tree: string;
3524	gen: number;
3525	commitTime: number;
3526	parents: ReadonlyArray<string>;
3527	}>,
3528	StoreError
3529	> =>
3530	Effect.gen(function* () {
3531	const stagedByOid = new Map(staged.map((c) => [c.oid, c]));
3532	const externalParents = Array.from(
3533	new Set(
3534	staged.flatMap((c) =>
3535	c.parents.filter((p) => !stagedByOid.has(p)),
3536	),
3537	),
3538	);
3539	const liveGens = new Map<string, number>();
3540	if (externalParents.length > 0) {
3541	const rows = yield* sql.inChunks<{ oid: string; gen: number }>(
3542	(ph) => `SELECT oid, gen FROM commits WHERE oid IN (${ph})`,
3543	externalParents,
3544	);
3545	for (const row of rows) liveGens.set(row.oid, row.gen);
3546	}
3547	const gens = new Map<string, number>();
3548	const genOf = (oid: string, depth: number): number => {
3549	const known = gens.get(oid);
3550	if (known !== undefined) return known;
3551	const live = liveGens.get(oid);
3552	if (live !== undefined) return live;
3553	const commit = stagedByOid.get(oid);
3554	// Missing parent (shallow boundary) counts as gen 0.
3555	if (commit === undefined || depth > 100_000) return 0;
3556	const gen =
3557	commit.parents.length === 0
3558	? 1
3559	: Math.max(...commit.parents.map((p) => genOf(p, depth + 1))) +
3560	1;
3561	gens.set(oid, gen);
3562	return gen;
3563	};
3564	return staged.map((commit) => ({
3565	oid: commit.oid,
3566	tree: commit.tree,
3567	gen: genOf(commit.oid, 0),
3568	commitTime: commit.commitTime,
3569	parents: commit.parents,
3570	}));
3571	});
3572
3573	// ── pull requests ────────────────────────────────────────────────────
3574
3575	/** `main` or `refs/heads/main` → `refs/heads/main`; branches only. */
3576	const normalizeBranchRef = Effect.fn(function* (name: string) {
3577	const full = name.startsWith("refs/") ? name : `refs/heads/${name}`;
3578	if (
3579	!full.startsWith("refs/heads/") ||
3580	full.length <= "refs/heads/".length ||
3581	!REF_NAME_REGEX.test(full)
3582	) {
3583	return yield* new ValidationError({
3584	message: `not a branch name: ${name} (only refs/heads/* is legal)`,
3585	});
3586	}
3587	return full;
3588	});
3589
3590	const pullStateOf = (state: string): "open" | "closed" | "merged" =>
3591	state === "closed" ? "closed" : state === "merged" ? "merged" : "open";
3592
3593	const rowToPull = (row: PullRow): PullData => ({
3594	number: row.number,
3595	title: row.title,
3596	body: row.body,
3597	baseRef: row.base_ref,
3598	headRef: row.head_ref,
3599	state: pullStateOf(row.state),
3600	createdAt: row.created_at,
3601	updatedAt: row.updated_at,
3602	mergedAt: row.merged_at,
3603	mergeCommit: row.merge_commit,
3604	});
3605
3606	const loadPull = Effect.fn(function* (number: number) {
3607	const row = yield* sql.first<PullRow>(
3608	`SELECT * FROM pulls WHERE number = ?`,
3609	number,
3610	);
3611	if (row === undefined) {
3612	return yield* new PullNotFound({ number });
3613	}
3614	return row;
3615	});
3616
3617	/** Result of the bounded two-tip ancestor {@link paintWalk}. */
3618	interface PullCompare {
3619	/** Best common ancestor; `undefined` = none found (see saturated). */
3620	readonly mergeBase: string | undefined;
3621	/** Commits reachable from head but not base. */
3622	readonly aheadBy: number;
3623	/** Commits reachable from base but not head. */
3624	readonly behindBy: number;
3625	/** The walk hit its pop bound — counts/base are unreliable. */
3626	readonly saturated: boolean;
3627	}
3628
3629	/**
3630	* The same gen-ordered merge-base paint walk `compareCommits` runs,
3631	* reduced to what the PR surface needs: the best common ancestor plus
3632	* ahead/behind counts. `mergeBase === head` ⇔ head is reachable from
3633	* base (up-to-date); `mergeBase === base` ⇔ base is reachable from
3634	* head (fast-forwardable) — the gen ordering guarantees a node's
3635	* flags are complete when it pops. A tip missing from the commit
3636	* graph reports `saturated` (uncomputable), never an error.
3637	*/
3638	const paintWalk = Effect.fn(function* (
3639	baseOid: string,
3640	headOid: string,
3641	maxPops: number,
3642	) {
3643	const BASE = 1;
3644	const HEAD = 2;
3645	const BOTH = 3;
3646	const flags = new Map<string, number>();
3647	flags.set(baseOid, BASE);
3648	flags.set(headOid, (flags.get(headOid) ?? 0) | HEAD);
3649	const heap: Array<WalkHeapNode> = [];
3650	const rowOf = (oid: string) =>
3651	sql.first<{ gen: number; commit_time: number }>(
3652	`SELECT gen, commit_time FROM commits WHERE oid = ?`,
3653	oid,
3654	);
3655	for (const oid of new Set([baseOid, headOid])) {
3656	const row = yield* rowOf(oid);
3657	if (row === undefined) {
3658	return {
3659	mergeBase: undefined,
3660	aheadBy: 0,
3661	behindBy: 0,
3662	saturated: true,
3663	} satisfies PullCompare;
3664	}
3665	heapPush(heap, { oid, gen: row.gen, time: row.commit_time });
3666	}
3667	let mergeBase: string | undefined;
3668	let aheadBy = 0;
3669	let behindBy = 0;
3670	let pops = 0;
3671	while (heap.length > 0 && pops < maxPops) {
3672	// stop once nothing in the heap can still change the counts
3673	if (
3674	mergeBase !== undefined &&
3675	heap.every((n) => (flags.get(n.oid)! & BOTH) === BOTH)
3676	) {
3677	break;
3678	}
3679	const node = heapPop(heap);
3680	pops++;
3681	const f = flags.get(node.oid)!;
3682	if ((f & BOTH) === BOTH) {
3683	if (mergeBase === undefined) mergeBase = node.oid;
3684	} else if (f === HEAD) {
3685	aheadBy++;
3686	} else {
3687	behindBy++;
3688	}
3689	const parents = yield* sql.all<{ parent: string }>(
3690	`SELECT parent FROM commit_parents WHERE oid = ? ORDER BY ord`,
3691	node.oid,
3692	);
3693	for (const { parent } of parents) {
3694	const prev = flags.get(parent) ?? 0;
3695	const next = prev | f;
3696	if (next === prev) continue;
3697	flags.set(parent, next);
3698	if (prev === 0) {
3699	const row = yield* rowOf(parent);
3700	if (row === undefined) continue; // shallow boundary
3701	heapPush(heap, {
3702	oid: parent,
3703	gen: row.gen,
3704	time: row.commit_time,
3705	});
3706	}
3707	}
3708	}
3709	return {
3710	mergeBase,
3711	aheadBy,
3712	behindBy,
3713	saturated: pops >= maxPops && heap.length > 0,
3714	} satisfies PullCompare;
3715	});
3716
3717	/** The root tree of a commit, via the commit-graph table. */
3718	const treeOfCommit = Effect.fn(function* (oid: string) {
3719	const row = yield* sql.first<{ tree: string }>(
3720	`SELECT tree FROM commits WHERE oid = ?`,
3721	oid,
3722	);
3723	if (row === undefined) {
3724	return yield* new StoreError({
3725	reason: `commit ${oid} missing from the commit graph`,
3726	});
3727	}
3728	return row.tree;
3729	});
3730
3731	// ── alarm scheduling ─────────────────────────────────────────────────
3732	const armAlarmAt = (time: number) =>
3733	Effect.gen(function* () {
3734	const existing = yield* state.storage.getAlarm();
3735	if (existing === null || existing > time) {
3736	yield* state.storage.setAlarm(time);
3737	}
3738	});
3739
3740	const upsertJob = (kind: string, detail: string | null) =>
3741	sql.run(
3742	`INSERT INTO jobs (kind, state, detail, updated_at) VALUES (?, 'running', ?, ?)
3743	ON CONFLICT (kind) DO UPDATE SET state = 'running', detail = excluded.detail, updated_at = excluded.updated_at`,
3744	kind,
3745	detail,
3746	Date.now(),
3747	);
3748
3749	// ── bootstrap (create / fork / import) ───────────────────────────────
3750	const seedConfig = Effect.fn(function* (
3751	input: InitRepoInput,
3752	status: RepoStatus,
3753	) {
3754	const createdAt = Date.now();
3755	yield* setConfig("repo_id", input.repoId);
3756	yield* setConfig("owner", input.owner.toLowerCase());
3757	yield* setConfig("name", input.name.toLowerCase());
3758	yield* setConfig("default_branch", input.defaultBranch);
3759	if (input.description !== null) {
3760	yield* setConfig("description", input.description);
3761	}
3762	yield* setConfig("read_only", input.readOnly ? "1" : "0");
3763	yield* setConfig("public", input.public ? "1" : "0");
3764	if (input.forkOf !== null) {
3765	yield* setConfig("fork_of", input.forkOf);
3766	}
3767	yield* setConfig("status", status);
3768	yield* readMeta.pipe(
3769	Effect.flatMap((meta) =>
3770	meta === undefined ? Effect.void : syncSummary(meta),
3771	),
3772	Effect.ignore,
3773	);
3774	const existing = yield* getConfig("created_at");
3775	if (existing === undefined) {
3776	yield* setConfig("created_at", String(createdAt));
3777	}
3778	const meta = yield* readMeta;
3779	return meta!;
3780	});
3781
3782	const bootstrap = Effect.fn(function* (
3783	input: InitRepoInput,
3784	status: RepoStatus,
3785	) {
3786	const meta = yield* seedConfig(input, status);
3787	return { meta } satisfies InitRepoResult;
3788	});
3789
3790	// ── wire protocol drivers ────────────────────────────────────────────
3791
3792	/** Parses `/:owner/:repo[.git]/<endpoint>` out of the request URL. */
3793	const parseWirePath = (
3794	url: string,
3795	):
3796	| {
3797	readonly owner: string;
3798	readonly repo: string;
3799	readonly endpoint: string;
3800	readonly service: string | undefined;
3801	}
3802	| undefined => {
3803	const parsed = new URL(url, "http://repo");
3804	const segments = parsed.pathname.split("/").filter(Boolean);
3805	if (segments.length < 3) return undefined;
3806	const owner = decodeURIComponent(segments[0]!).toLowerCase();
3807	let repo = decodeURIComponent(segments[1]!).toLowerCase();
3808	if (repo.endsWith(".git")) repo = repo.slice(0, -4);
3809	const endpoint = segments.slice(2).join("/");
3810	return {
3811	owner,
3812	repo,
3813	endpoint,
3814	service: parsed.searchParams.get("service") ?? undefined,
3815	};
3816	};
3817
3818	const wire503 = HttpServerResponse.empty({
3819	status: 503,
3820	headers: { "retry-after": "10" },
3821	});
3822
3823	/** GET info/refs — the v0 advertisement. */
3824	const handleInfoRefs = (
3825	request: HttpServerRequest.HttpServerRequest,
3826	service: string | undefined,
3827	meta: RepoMetaData,
3828	) =>
3829	Effect.gen(function* () {
3830	if (service !== "git-upload-pack" && service !== "git-receive-pack") {
3831	return HttpServerResponse.text(
3832	"smart HTTP only (dumb protocol not supported)",
3833	{ status: 400 },
3834	);
3835	}
3836	const refs = yield* listAllRefs(meta.repoId);
3837	const body = buildAdvertisement({
3838	service,
3839	defaultBranch: meta.defaultBranch,
3840	refs,
3841	});
3842	return HttpServerResponse.uint8Array(body, {
3843	contentType: `application/x-${service}-advertisement`,
3844	headers: noCache,
3845	});
3846	});
3847
3848	/** POST git-upload-pack — negotiation + pack (DESIGN.md §2.3). */
3849	const handleUploadPack = (
3850	request: HttpServerRequest.HttpServerRequest,
3851	meta: RepoMetaData,
3852	) =>
3853	Effect.gen(function* () {
3854	const rawBody = new Uint8Array(yield* request.arrayBuffer);
3855	const body = yield* gunzipIfNeeded(
3856	rawBody,
3857	request.headers["content-encoding"],
3858	);
3859	const resultType = "application/x-git-upload-pack-result";
3860	const errResponse = (message: string) =>
3861	HttpServerResponse.uint8Array(errPkt(message), {
3862	contentType: resultType,
3863	headers: noCache,
3864	});
3865
3866	const parsed = yield* decodePktLines(body).pipe(
3867	Effect.flatMap(parseUploadPackRequest),
3868	Effect.map(Option.some),
3869	Effect.catchTag("PktLineError", (e) =>
3870	Effect.succeed(Option.none<UploadPackRequest>()).pipe(
3871	Effect.tap(() => Effect.logWarning(`upload-pack: ${e.reason}`)),
3872	),
3873	),
3874	Effect.catchTag("WireProtocolError", (e) =>
3875	Effect.succeed(Option.none<UploadPackRequest>()).pipe(
3876	Effect.tap(() => Effect.logWarning(`upload-pack: ${e.reason}`)),
3877	),
3878	),
3879	);
3880	if (Option.isNone(parsed)) {
3881	return errResponse("malformed upload-pack request");
3882	}
3883	const req = parsed.value;
3884	const objects = storeFor(meta.repoId);
3885	const sideband = req.capabilities.has("side-band-64k");
3886	const noDone = req.capabilities.has("no-done");
3887
3888	const commons = yield* objects.filterExisting(req.haves);
3889
3890	// Negotiation round without done and nothing in common: answer
3891	// NAK and wait for the next round (or done). EXCEPT when the
3892	// request carries `deepen` — the shallow/unshallow section must
3893	// precede the NAK ("fatal: expected shallow/unshallow, got NAK"
3894	// otherwise), so depth-carrying rounds fall through to the
3895	// closure computation below.
3896	if (!req.done && commons.length === 0 && req.depth === undefined) {
3897	return HttpServerResponse.uint8Array(pktText("NAK"), {
3898	contentType: resultType,
3899	headers: noCache,
3900	});
3901	}
3902
3903	// Bundle fast path (DESIGN.md §12.2): a full clone whose wants are
3904	// all covered by the current bundle is served as bytes straight
3905	// from R2 — no closure walk, no per-object reads. `bundleCovers`
3906	// refuses anything with haves/depth/shallow state.
3907	const bundle = yield* readBundle;
3908	if (
3909	req.done &&
3910	bundle !== undefined &&
3911	bundleCovers(bundle, {
3912	wants: req.wants,
3913	haves: req.haves,
3914	depth: req.depth,
3915	clientShallow: req.clientShallow,
3916	})
3917	) {
3918	// The DO does NOT stream the pack: it hands the Worker a
3919	// *marker* naming the R2 object, and the Worker streams those
3920	// bytes to the client itself (DESIGN.md §11, serving plane).
3921	// Auth and planning still happen here — where tokens and refs
3922	// live — but no pack byte transits this object, which is the
3923	// whole point of splitting the planes.
3924	const found = yield* blobs
3925	.head(bundle.key)
3926	.pipe(
3927	Effect.mapError(
3928	(error) =>
3929	new StoreError({ reason: `bundle: ${error.reason}` }),
3930	),
3931	);
3932	if (found !== null) {
3933	return HttpServerResponse.empty({
3934	status: 200,
3935	headers: {
3936	...noCache,
3937	"content-type": resultType,
3938	[BUNDLE_KEY_HEADER]: bundle.key,
3939	[BUNDLE_HASH_HEADER]: bundle.refsHash,
3940	[BUNDLE_COUNT_HEADER]: String(bundle.objectCount),
3941	[BUNDLE_SIDEBAND_HEADER]: sideband ? "1" : "0",
3942	},
3943	});
3944	}
3945	}
3946
3947	// A depth-limited import leaves the repo itself shallow; those
3948	// boundary commits must bound every walk and surface as
3949	// `shallow` lines so clones inherit the shallowness (fsck-clean).
3950	const shallowRoots = JSON.parse(
3951	(yield* getConfig("shallow_roots")) ?? "[]",
3952	) as ReadonlyArray<string>;
3953	const closureResult = yield* Effect.result(
3954	computeClosure({ sql, objects })({
3955	wants: req.wants,
3956	haves: commons,
3957	depth: req.depth,
3958	clientShallow: req.clientShallow,
3959	repoShallow: shallowRoots,
3960	}),
3961	);
3962	if (Result.isFailure(closureResult)) {
3963	const error = closureResult.failure;
3964	return errResponse(
3965	error._tag === "ManifestTooLarge"
3966	? `repository too large to serve in one pack (${error.count} > ${error.cap})`
3967	: error.reason,
3968	);
3969	}
3970	const closure = closureResult.success;
3971
3972	// v0 clients only parse a shallow/unshallow section when THEY
3973	// brought shallowness into the exchange (`deepen` or `shallow`
3974	// lines). If this repo's own boundary truncated the walk for a
3975	// client that did neither, an unsolicited section would derail
3976	// its parser — refuse cleanly instead.
3977	const clientShallowAware =
3978	req.depth !== undefined || req.clientShallow.length > 0;
3979	if (closure.shallow.length > 0 && !clientShallowAware) {
3980	return errResponse(
3981	"repository has shallow history (imported with depth); clone with --depth",
3982	);
3983	}
3984	const head: Array<Uint8Array> = [];
3985	if (clientShallowAware) {
3986	for (const oid of closure.shallow) {
3987	head.push(pktText(`shallow ${oid}`));
3988	}
3989	for (const oid of closure.unshallow) {
3990	head.push(pktText(`unshallow ${oid}`));
3991	}
3992	head.push(flushPkt);
3993	}
3994
3995	// Depth-carrying negotiation round with nothing in common and no
3996	// done yet: the response is the shallow section ALONE (lines +
3997	// flush) — no NAK, no pack. The stateless client buffers this
3998	// response, sends `done`, and expects the NEXT response to begin
3999	// with the shallow section again; a trailing NAK here would sit
4000	// in its buffer and derail that parse ("expected shallow list").
4001	if (!req.done && commons.length === 0) {
4002	return HttpServerResponse.uint8Array(concatBytes(head), {
4003	contentType: resultType,
4004	headers: noCache,
4005	});
4006	}
4007	const lastCommon = commons[commons.length - 1];
4008	if (req.done) {
4009	head.push(
4010	lastCommon === undefined
4011	? pktText("NAK")
4012	: pktText(`ACK ${lastCommon}`),
4013	);
4014	} else {
4015	// !done implies commons.length > 0 (the NAK-only branch above).
4016	const ready = lastCommon ?? ZERO_OID;
4017	for (const common of commons) {
4018	head.push(pktText(`ACK ${common} common`));
4019	}
4020	head.push(pktText(`ACK ${ready} ready`));
4021	if (!noDone) {
4022	// Client will re-POST with done; no pack this round.
4023	return HttpServerResponse.uint8Array(concatBytes(head), {
4024	contentType: resultType,
4025	headers: noCache,
4026	});
4027	}
4028	head.push(pktText(`ACK ${ready}`));
4029	}
4030
4031	// The pack goes out through the native pump (Sideband.ts
4032	// `pumpPackBody`, DESIGN §22.3): the emitter's batched chunks are
4033	// large, so the Effect→ReadableStream bridge is a handful of
4034	// pulls, and framing/flush happen at platform speed.
4035	const source = yield* packStream(closure.entries, objects).pipe(
4036	Stream.toReadableStreamEffect(),
4037	);
4038	const prefix = sideband
4039	? concatBytes([
4040	concatBytes(head),
4041	progressMessage(
4042	`Enumerating objects: ${closure.entries.length}, done.`,
4043	),
4044	])
4045	: concatBytes(head);
4046	return HttpServerResponse.raw(
4047	pumpPackBody({ prefix, source, sideband }),
4048	{ contentType: resultType, headers: noCache },
4049	);
4050	});
4051
4052	// ── alarm jobs ───────────────────────────────────────────────────────
4053	/** Nulls a committed push's `staged_push` column and drops its row. */
4054	const flipPush = (pushId: string) =>
4055	Effect.gen(function* () {
4056	yield* sql.run(
4057	`UPDATE objects SET staged_push = NULL WHERE staged_push = ?`,
4058	pushId,
4059	);
4060	yield* sql.run(
4061	`DELETE FROM pushes WHERE push_id = ? AND state = 'committed'`,
4062	pushId,
4063	);
4064	});
4065
4066	const runGcJob = Effect.gen(function* () {
4067	// Committed pushes whose deferred flip never ran (DESIGN §22.14).
4068	yield* sql.run(
4069	`UPDATE objects SET staged_push = NULL WHERE staged_push IN
4070	(SELECT push_id FROM pushes WHERE state = 'committed')`,
4071	);
4072	yield* sql.run(`DELETE FROM pushes WHERE state = 'committed'`);
4073	const cutoff = Date.now() - STAGING_TTL_MS;
4074	yield* sql.run(
4075	`DELETE FROM objects WHERE staged_push IN
4076	(SELECT push_id FROM pushes WHERE state = 'staging' AND started_at < ?)`,
4077	cutoff,
4078	);
4079	yield* sql.run(
4080	`DELETE FROM pushes WHERE state = 'staging' AND started_at < ?`,
4081	cutoff,
4082	);
4083	yield* sql.run(`DELETE FROM jobs WHERE kind = 'gc'`);
4084	});
4085
4086	/**
4087	* Compaction (DESIGN.md §12.1): moves loose object bytes out of DO
4088	* SQLite into an immutable R2 pack, bounded per run and re-armed
4089	* while more remain.
4090	*/
4091	const runCompactAlarm = Effect.gen(function* () {
4092	const meta = yield* requireMeta;
4093	// Grace-delete packs a PREVIOUS merge unreferenced: only after
4094	// a minute, so any read planned before that row flip has long
4095	// finished its ranged reads (DESIGN.md §21).
4096	const pendingRaw = yield* getConfig("packs_pending_delete");
4097	const pending =
4098	pendingRaw === undefined || pendingRaw === ""
4099	? undefined
4100	: (JSON.parse(pendingRaw) as { keys: Array<string>; at: number });
4101	if (pending !== undefined && Date.now() - pending.at > 60_000) {
4102	yield* blobs
4103	.delete(pending.keys)
4104	.pipe(Effect.catch(() => Effect.void));
4105	yield* setConfig("packs_pending_delete", "");
4106	}
4107	const outcome = yield* runCompactJob({
4108	repoId: meta.repoId,
4109	sql,
4110	blobs,
4111	});
4112	// Restore the geometric invariant over the repo's packs, so pack
4113	// count — and with it read fan-out — stays O(log bytes).
4114	const merge = yield* runGeometricMergeJob({
4115	repoId: meta.repoId,
4116	sql,
4117	blobs,
4118	});
4119	if (merge.pendingDelete.length > 0) {
4120	const carried =
4121	pending !== undefined && !(Date.now() - pending.at > 60_000)
4122	? pending.keys
4123	: [];
4124	yield* setConfig(
4125	"packs_pending_delete",
4126	JSON.stringify({
4127	keys: [...carried, ...merge.pendingDelete],
4128	at: Date.now(),
4129	}),
4130	);
4131	}
4132	if (outcome.more || merge.more) {
4133	yield* armAlarmAt(Date.now() + 1_000);
4134	return;
4135	}
4136	yield* sql.run(`DELETE FROM jobs WHERE kind = 'compact'`);
4137	});
4138
4139	/** The repo's current clone bundle, if one has been cut. */
4140	/**
4141	* Pushes the denormalised display fields to the Registry so listing
4142	* repos never has to wake this DO (DESIGN.md §15 bottleneck 7).
4143	* Best-effort: the DO remains the source of truth, and a failed
4144	* refresh only makes a listing momentarily stale.
4145	*/
4146	const syncSummary = Effect.fn(function* (meta: RepoMetaData) {
4147	yield* registry
4148	.updateSummary(meta.repoId, {
4149	defaultBranch: meta.defaultBranch,
4150	readOnly: meta.readOnly,
4151	status: meta.status,
4152	public: meta.public,
4153	})
4154	.pipe(Effect.ignore);
4155	});
4156
4157	/** Re-reads metadata and refreshes the Registry summary. */
4158	const refreshSummary: Effect.Effect<void, never, RuntimeContext> =
4159	readMeta.pipe(
4160	Effect.flatMap((meta) =>
4161	meta === undefined ? Effect.void : syncSummary(meta),
4162	),
4163	Effect.ignore,
4164	);
4165
4166	const readBundle: Effect.Effect<BundleInfo | undefined, StoreError> =
4167	getConfig("bundle").pipe(
4168	Effect.map((value) =>
4169	value === undefined ? undefined : (JSON.parse(value) as BundleInfo),
4170	),
4171	);
4172
4173	/**
4174	* Republishes the repo's head snapshot (DESIGN.md §21): a tiny
4175	* JSON object in the BlobStore from which the Worker serves
4176	* anonymous public reads — the upload-pack advertisement and
4177	* bundle-covered clones — without waking this DO. Called after
4178	* every commit that changes what an anonymous reader can see.
4179	*
4180	* Best-effort by design: it runs AFTER the SQLite commit (which
4181	* stays authoritative), so a failed write only leaves a stale
4182	* snapshot that the next mutation — or the post-push bundle
4183	* alarm, seconds later — repairs.
4184	*/
4185	const writeHeadSnapshot: Effect.Effect<void, never, RuntimeContext> =
4186	Effect.gen(function* () {
4187	const meta = yield* requireMeta;
4188	const refs = yield* listAllRefs(meta.repoId);
4189	const bundle = yield* readBundle;
4190	const snapshot: HeadSnapshot = {
4191	v: 1,
4192	repoId: meta.repoId,
4193	owner: meta.owner,
4194	name: meta.name,
4195	public: meta.public,
4196	readOnly: meta.readOnly,
4197	defaultBranch: meta.defaultBranch,
4198	refs,
4199	bundle,
4200	};
4201	yield* blobs.put(
4202	headKey(meta.repoId),
4203	utf8Encode(encodeHeadSnapshot(snapshot)),
4204	);
4205	}).pipe(
4206	Effect.catch((error) =>
4207	Effect.logWarning(`head snapshot write failed: ${String(error)}`),
4208	),
4209	Effect.asVoid,
4210	);
4211
4212	/**
4213	* Cuts a clone bundle for the current refs (DESIGN.md §12.2) so future
4214	* clones are served as bytes from R2 instead of a fresh closure walk.
4215	*/
4216	const runBundleAlarm = Effect.gen(function* () {
4217	const meta = yield* requireMeta;
4218	const objects = storeFor(meta.repoId);
4219	const refRows = yield* sql.all<RefRow>(
4220	`SELECT name, oid FROM refs ORDER BY name`,
4221	);
4222	const shallowRoots = JSON.parse(
4223	(yield* getConfig("shallow_roots")) ?? "[]",
4224	) as ReadonlyArray<string>;
4225	const closure = yield* computeClosure({ sql, objects })({
4226	wants: refRows.map((ref) => ref.oid),
4227	haves: [],
4228	clientShallow: [],
4229	repoShallow: shallowRoots,
4230	}).pipe(
4231	Effect.catchTag("ManifestTooLarge", (error) =>
4232	Effect.fail(new StoreError({ reason: `bundle: ${error._tag}` })),
4233	),
4234	);
4235	const info = yield* runBundleJob({
4236	repoId: meta.repoId,
4237	refs: refRows.map((ref) => ({ name: ref.name, oid: ref.oid })),
4238	entries: closure.entries,
4239	packStream: (entries) => packStream(entries, objects),
4240	blobs,
4241	});
4242	if (info !== undefined) {
4243	yield* setConfig("bundle", JSON.stringify(info));
4244	// The snapshot now names a bundle covering the current refs —
4245	// full anonymous clones go DO-less from here (DESIGN.md §21).
4246	yield* writeHeadSnapshot;
4247	}
4248	yield* sql.run(`DELETE FROM jobs WHERE kind = 'bundle'`);
4249	});
4250
4251	const runImportAlarm = (job: JobRow) =>
4252	Effect.gen(function* () {
4253	const meta = yield* requireMeta;
4254	const detail = JSON.parse(job.detail ?? "{}") as {
4255	readonly source: ImportSource;
4256	readonly attempts?: number;
4257	};
4258	const attempts = detail.attempts ?? 0;
4259	const outcome = yield* Effect.result(
4260	runImport({ source: detail.source }).pipe(
4261	Effect.provide(FetchHttpClient.layer),
4262	),
4263	);
4264	if (Result.isFailure(outcome)) {
4265	if (attempts + 1 >= 3) {
4266	yield* sql.run(
4267	`UPDATE jobs SET state = 'failed', detail = ?, updated_at = ? WHERE kind = 'import'`,
4268	JSON.stringify({ ...detail, error: outcome.failure.reason }),
4269	Date.now(),
4270	);
4271	// Give the empty repo back rather than wedging in 'importing'.
4272	yield* setConfig("status", "ready");
4273	yield* refreshSummary;
4274	yield* refreshSummary;
4275	return;
4276	}
4277	yield* sql.run(
4278	`UPDATE jobs SET detail = ?, updated_at = ? WHERE kind = 'import'`,
4279	JSON.stringify({ ...detail, attempts: attempts + 1 }),
4280	Date.now(),
4281	);
4282	yield* armAlarmAt(Date.now() + 15_000);
4283	return;
4284	}
4285	const result = outcome.success;
4286	const objects = storeFor(meta.repoId);
4287	const pushId = yield* ulid();
4288	// Imports use the same staging lifecycle as pushes: finalize
4289	// requires an active row, and GC needs it to reap abandoned data.
4290	yield* sql.run(
4291	`INSERT INTO pushes (push_id, started_at, state) VALUES (?, ?, 'staging')`,
4292	pushId,
4293	Date.now(),
4294	);
4295	yield* upsertJob("gc", null);
4296	yield* armAlarmAt(Date.now() + STAGING_TTL_MS);
4297	let graph: Array<{
4298	oid: string;
4299	tree: string;
4300	gen: number;
4301	commitTime: number;
4302	parents: ReadonlyArray<string>;
4303	}> = [];
4304	if (result.pack !== undefined) {
4305	const ingest = yield* ingestPack(result.pack, {
4306	store: objects,
4307	pushId,
4308	allowMissingParents: true,
4309	}).pipe(
4310	Effect.mapError(
4311	(error) =>
4312	new StoreError({
4313	reason:
4314	error._tag === "PackIngestError"
4315	? `import ingest: ${error.reason}`
4316	: error.reason,
4317	}),
4318	),
4319	);
4320	// Connectivity: hard edges only — shallow imports may lack
4321	// parent commits (they become walk boundaries).
4322	const missing = yield* objects.missingObjects(
4323	[...ingest.referenced],
4324	pushId,
4325	);
4326	if (missing.length > 0) {
4327	return yield* new StoreError({
4328	reason: `import produced ${missing.length} dangling objects`,
4329	});
4330	}
4331	graph = yield* computeGraphRows(ingest.commits);
4332	// A depth-limited import leaves parents un-ingested: record the
4333	// boundary commits so upload-pack serves this repo as shallow
4334	// (walk bound + `shallow` lines — see handleUploadPack).
4335	const parentOids = Array.from(
4336	new Set(ingest.commits.flatMap((commit) => commit.parents)),
4337	);
4338	const missingParents = new Set(
4339	yield* objects.missingObjects(parentOids, pushId),
4340	);
4341	if (missingParents.size > 0) {
4342	const shallowRoots = ingest.commits
4343	.filter((commit) =>
4344	commit.parents.some((parent) => missingParents.has(parent)),
4345	)
4346	.map((commit) => commit.oid);
4347	yield* setConfig("shallow_roots", JSON.stringify(shallowRoots));
4348	}
4349	}
4350	const commands: Array<RefCommand> = result.refs.map((ref) => ({
4351	oldOid: ZERO_OID,
4352	newOid: ref.oid,
4353	ref: ref.name,
4354	}));
4355	const results = yield* finalizeRefTxn({
4356	commands,
4357	atomic: false,
4358	unconditional: true,
4359	pushId,
4360	graph,
4361	});
4362	const rejected = results.find((ref) => !ref.ok);
4363	if (rejected !== undefined) {
4364	return yield* new StoreError({
4365	reason: `import ref ${rejected.ref}: ${rejected.reason}`,
4366	});
4367	}
4368	yield* flipPush(pushId);
4369	if (result.defaultBranch !== null) {
4370	yield* setConfig("default_branch", result.defaultBranch);
4371	}
4372	yield* setConfig("status", "ready");
4373	yield* refreshSummary;
4374	yield* writeHeadSnapshot;
4375	yield* sql.run(`DELETE FROM jobs WHERE kind = 'import'`);
4376	});
4377
4378	const runForkAlarm = (job: JobRow) =>
4379	Effect.gen(function* () {
4380	const detail = JSON.parse(job.detail ?? "{}") as {
4381	readonly parentRepoId: string;
4382	readonly attempts?: number;
4383	};
4384	const parentStub = selfNamespace.getByName(
4385	detail.parentRepoId,
4386	) as unknown as RepoStub;
4387	const applied = yield* Effect.result(
4388	runForkJob({
4389	sql,
4390	snapshot: parentStub.snapshotRows(),
4391	}),
4392	);
4393	if (Result.isFailure(applied)) {
4394	const attempts = (detail.attempts ?? 0) + 1;
4395	if (attempts >= 3) {
4396	yield* sql.run(
4397	`UPDATE jobs SET state = 'failed', detail = ?, updated_at = ? WHERE kind = 'fork'`,
4398	JSON.stringify({ ...detail, error: applied.failure.reason }),
4399	Date.now(),
4400	);
4401	yield* setConfig("status", "ready");
4402	yield* refreshSummary;
4403	yield* refreshSummary;
4404	return;
4405	}
4406	yield* sql.run(
4407	`UPDATE jobs SET detail = ?, updated_at = ? WHERE kind = 'fork'`,
4408	JSON.stringify({ ...detail, attempts }),
4409	Date.now(),
4410	);
4411	yield* armAlarmAt(Date.now() + 15_000);
4412	return;
4413	}
4414	yield* setConfig("status", "ready");
4415	yield* refreshSummary;
4416	yield* writeHeadSnapshot;
4417	yield* sql.run(`DELETE FROM jobs WHERE kind = 'fork'`);
4418	});
4419
4420	/** Returns true when storage was wiped (stop touching SQLite). */
4421	const runPurgeAlarm = Effect.gen(function* () {
4422	const meta = yield* readMeta;
4423	if (meta === undefined) return true;
4424	const registryStub = registry;
4425	const outcome = yield* runPurgeJob({
4426	repoId: meta.repoId,
4427	blobs,
4428	// RuntimeContext is genuinely satisfied here (we run inside the
4429	// DO); discharge the coloring so the purge job's deps stay R=never.
4430	forkCount: registryStub.bumpForkCount(meta.repoId, 0).pipe(
4431	Effect.mapError(
4432	(error) => new StoreError({ reason: String(error) }),
4433	),
4434	Effect.provide(RuntimeContext.phantom),
4435	),
4436	deleteAllStorage: state.storage.deleteAll().pipe(
4437	Effect.mapError(
4438	(error) => new StoreError({ reason: String(error) }),
4439	),
4440	Effect.provide(RuntimeContext.phantom),
4441	),
4442	removeRegistryRow: registryStub.removeRow(meta.repoId).pipe(
4443	Effect.mapError(
4444	(error) => new StoreError({ reason: String(error) }),
4445	),
4446	Effect.provide(RuntimeContext.phantom),
4447	),
4448	});
4449	if (outcome._tag === "continue") {
4450	yield* armAlarmAt(Date.now() + (outcome.forkPinned ? 60_000 : 5_000));
4451	return false;
4452	}
4453	yield* state.storage.deleteAlarm();
4454	return true;
4455	});
4456
4457	// ── the shape ────────────────────────────────────────────────────────
4458	const shape: GitRepoShape = {
4459	initRepo: (input) =>
4460	bootstrap(input, "ready").pipe(Effect.tap(() => writeHeadSnapshot)),
4461
4462	startImport: Effect.fn(function* (input: StartImportInput) {
4463	const result = yield* bootstrap(input, "importing");
4464	yield* upsertJob("import", JSON.stringify({ source: input.source }));
4465	yield* armAlarmAt(Date.now());
4466	return result;
4467	}),
4468
4469	startFork: Effect.fn(function* (input: StartForkInput) {
4470	const result = yield* bootstrap(input, "forking");
4471	yield* upsertJob(
4472	"fork",
4473	JSON.stringify({ parentRepoId: input.parentRepoId }),
4474	);
4475	yield* armAlarmAt(Date.now());
4476	return result;
4477	}),
4478
4479	startCompact: Effect.fn(function* () {
4480	yield* requireMeta;
4481	yield* upsertJob("compact", null);
4482	yield* armAlarmAt(Date.now());
4483	}),
4484
4485	startPurge: Effect.fn(function* () {
4486	const meta = yield* requireMeta;
4487	yield* setConfig("status", "deleting");
4488	// Kill the DO-less fast path NOW — the prefix drain gets the
4489	// rest, but anonymous reads must stop serving immediately.
4490	yield* blobs
4491	.delete(headKey(meta.repoId))
4492	.pipe(Effect.catch(() => Effect.void));
4493	yield* refreshSummary;
4494	yield* upsertJob("purge", null);
4495	yield* armAlarmAt(Date.now());
4496	}),
4497
4498	getRepoMeta: Effect.fn(function* () {
4499	const meta = yield* requireMetaStats;
4500	return meta;
4501	}),
4502
4503	readMeta: Effect.fn(function* () {
4504	return yield* requireMetaStats;
4505	}),
4506
4507	updateRepoMeta: Effect.fn(function* (patch: RepoMetaPatch) {
4508	const meta = yield* requireMeta;
4509	if (patch.defaultBranch !== undefined) {
4510	const ref = yield* sql.first<RefRow>(
4511	`SELECT name, oid FROM refs WHERE name = ?`,
4512	`refs/heads/${patch.defaultBranch}`,
4513	);
4514	if (ref === undefined) {
4515	return yield* new RefNotFound({
4516	ref: `refs/heads/${patch.defaultBranch}`,
4517	});
4518	}
4519	yield* setConfig("default_branch", patch.defaultBranch);
4520	}
4521	if (patch.description !== undefined) {
4522	if (patch.description === null) {
4523	yield* sql.run(`DELETE FROM config WHERE key = 'description'`);
4524	} else {
4525	yield* setConfig("description", patch.description);
4526	}
4527	}
4528	if (patch.readOnly !== undefined) {
4529	yield* setConfig("read_only", patch.readOnly ? "1" : "0");
4530	}
4531	if (patch.public !== undefined) {
4532	yield* setConfig("public", patch.public ? "1" : "0");
4533	}
4534	const updated = yield* readMetaStats;
4535	const result = updated ?? meta;
4536	// Keep the Registry's denormalised copy in step (listing reads it).
4537	yield* syncSummary(result);
4538	yield* writeHeadSnapshot;
4539	return result;
4540	}),
4541
4542	listRefs: Effect.fn(function* (prefix?: string | undefined) {
4543	const meta = yield* requireMeta;
4544	const refs = yield* listAllRefs(meta.repoId, prefix);
4545	const headRef = `refs/heads/${meta.defaultBranch}`;
4546	const headExists = yield* sql.first<RefRow>(
4547	`SELECT name, oid FROM refs WHERE name = ?`,
4548	headRef,
4549	);
4550	return {
4551	head: headExists === undefined ? null : headRef,
4552	refs,
4553	} satisfies RefsPage;
4554	}),
4555
4556	getRef: Effect.fn(function* (name: string) {
4557	const meta = yield* requireMeta;
4558	const row = yield* sql.first<RefRow>(
4559	`SELECT name, oid FROM refs WHERE name = ?`,
4560	name,
4561	);
4562	if (row === undefined) {
4563	return yield* new RefNotFound({ ref: name });
4564	}
4565	const peeled = name.startsWith("refs/tags/")
4566	? yield* peelOid(meta.repoId, row.oid)
4567	: undefined;
4568	return peeled === undefined
4569	? { name: row.name, oid: row.oid }
4570	: ({ name: row.name, oid: row.oid, peeled } satisfies RefData);
4571	}),
4572
4573	updateRef: Effect.fn(function* (input: UpdateRefInput) {
4574	const meta = yield* requireMeta;
4575	if (meta.readOnly) {
4576	return yield* new ReadOnlyRepo();
4577	}
4578	const objects = storeFor(meta.repoId);
4579	const exists = yield* objects.has(input.newOid);
4580	if (!exists) {
4581	return yield* new ObjectNotFound({ oid: input.newOid });
4582	}
4583	yield* sql.transactionSync<void, RefConflict>((raw, rollback) => {
4584	const rows = raw
4585	.exec<RefRow>(
4586	`SELECT name, oid FROM refs WHERE name = ?`,
4587	input.name,
4588	)
4589	.toArray();
4590	const current = rows.length > 0 ? rows[0]!.oid : null;
4591	if (input.expectedOid !== undefined) {
4592	const expected = input.expectedOid; // string | null
4593	if (current !== expected) {
4594	rollback(
4595	new RefConflict({
4596	ref: input.name,
4597	currentOid: current === null ? null : asOid(current),
4598	}),
4599	);
4600	}
4601	}
4602	raw.exec(
4603	`INSERT INTO refs (name, oid) VALUES (?, ?)
4604	ON CONFLICT (name) DO UPDATE SET oid = excluded.oid`,
4605	input.name,
4606	input.newOid,
4607	);
4608	});
4609	yield* writeHeadSnapshot;
4610	return { name: input.name, oid: input.newOid } satisfies RefData;
4611	}),
4612
4613	removeRef: Effect.fn(function* (input: RemoveRefInput) {
4614	const meta = yield* requireMeta;
4615	if (meta.readOnly) {
4616	return yield* new ReadOnlyRepo();
4617	}
4618	yield* sql.transactionSync<void, RefConflict | RefNotFound>(
4619	(raw, rollback) => {
4620	const rows = raw
4621	.exec<RefRow>(
4622	`SELECT name, oid FROM refs WHERE name = ?`,
4623	input.name,
4624	)
4625	.toArray();
4626	if (rows.length === 0) {
4627	rollback(new RefNotFound({ ref: input.name }));
4628	}
4629	const current = rows[0]!.oid;
4630	if (
4631	input.expectedOid !== undefined &&
4632	current !== input.expectedOid
4633	) {
4634	rollback(
4635	new RefConflict({
4636	ref: input.name,
4637	currentOid: asOid(current),
4638	}),
4639	);
4640	}
4641	raw.exec(`DELETE FROM refs WHERE name = ?`, input.name);
4642	},
4643	);
4644	yield* writeHeadSnapshot;
4645	}),
4646
4647	beginPush: Effect.fn(function* (input: BeginPushInput) {
4648	const meta = yield* requireMeta;
4649	if (meta.readOnly) {
4650	return {
4651	_tag: "denied",
4652	reason: "repository is read-only",
4653	} as const;
4654	}
4655	const pushId = yield* ulid();
4656	yield* sql.run(
4657	`INSERT INTO pushes (push_id, started_at, state) VALUES (?, ?, 'staging')`,
4658	pushId,
4659	Date.now(),
4660	);
4661	return { _tag: "ok", pushId } as const;
4662	}),
4663
4664	stagePush: Effect.fn(function* (pushId: string, batch: Uint8Array) {
4665	const meta = yield* requireMeta;
4666	yield* storeFor(meta.repoId).insertStagedBatch(
4667	pushId,
4668	decodeStagedBatch(batch),
4669	);
4670	}),
4671
4672	abortPush: Effect.fn(function* (
4673	pushId: string,
4674	packIds: ReadonlyArray<string> = [],
4675	) {
4676	yield* requireMeta;
4677	return yield* sql.transactionSync((raw) => {
4678	const push = raw
4679	.exec<{ state: string }>(
4680	`SELECT state FROM pushes WHERE push_id = ?`,
4681	pushId,
4682	)
4683	.toArray()[0];
4684	if (push?.state === "committed") return false;
4685	raw.exec(`DELETE FROM objects WHERE staged_push = ?`, pushId);
4686	raw.exec(
4687	`DELETE FROM pushes WHERE push_id = ? AND state = 'staging'`,
4688	pushId,
4689	);
4690	// Another concurrent push may have adopted staged rows. Retain any
4691	// shared pack bytes even though this transaction has been aborted.
4692	for (const packId of packIds) {
4693	if (
4694	raw
4695	.exec(
4696	`SELECT 1 FROM objects WHERE pack_id = ? LIMIT 1`,
4697	packId,
4698	)
4699	.toArray().length > 0
4700	)
4701	return false;
4702	}
4703	return true;
4704	});
4705	}),
4706
4707	validatePush: Effect.fn(function* (input: CommitPushInput) {
4708	const meta = yield* requireMeta;
4709	const referenced = new Set([
4710	...input.referenced,
4711	...input.referencedParents,
4712	]);
4713	for (const command of input.commands)
4714	if (command.newOid !== ZERO_OID) referenced.add(command.newOid);
4715	const missing = yield* storeFor(meta.repoId).missingObjects(
4716	Array.from(referenced),
4717	input.pushId,
4718	);
4719	if (missing.length > 0)
4720	return yield* new StoreError({
4721	reason: `missing objects (${missing.length})`,
4722	});
4723	}),
4724
4725	readPreparedObject: Effect.fn(function* (
4726	pushId: string,
4727	oid: Oid,
4728	maxBytes: number,
4729	) {
4730	const meta = yield* requireMeta;
4731	const push = yield* sql.first<{ state: string }>(
4732	`SELECT state FROM pushes WHERE push_id = ?`,
4733	pushId,
4734	);
4735	if (push?.state !== "staging")
4736	return yield* new StoreError({
4737	reason: "push is no longer active",
4738	});
4739	return yield* storeFor(meta.repoId).readPrepared(
4740	pushId,
4741	oid,
4742	maxBytes,
4743	);
4744	}),
4745
4746	readPushBase: Effect.fn(function* (oid: Oid) {
4747	const meta = yield* requireMeta;
4748	return yield* ingestStoreOf(storeFor(meta.repoId)).readBase(oid);
4749	}),
4750
4751	commitPush: Effect.fn(function* (input: CommitPushInput) {
4752	const meta = yield* requireMeta;
4753	const objects = storeFor(meta.repoId);
4754	const startedAt = yield* Effect.sync(() => performance.now());
4755	const since = (from: number) =>
4756	Effect.sync(() => performance.now() - from);
4757	const allNg = (reason: string) =>
4758	input.commands.map((cmd) => ({
4759	ref: cmd.ref,
4760	ok: false,
4761	reason,
4762	}));
4763	if (meta.readOnly)
4764	return { unpack: "ok", results: allNg("repository is read-only") };
4765	// Full connectivity check (§3.6 step 5) over the staged rows.
4766	const referenced = new Set<string>(input.referenced);
4767	for (const parent of input.referencedParents) referenced.add(parent);
4768	for (const cmd of input.commands) {
4769	if (cmd.newOid !== ZERO_OID) referenced.add(cmd.newOid);
4770	}
4771	const connectivityStarted = yield* Effect.sync(() =>
4772	performance.now(),
4773	);
4774	const missing = yield* objects.missingObjects(
4775	Array.from(referenced),
4776	input.pushId,
4777	);
4778	const connectivityMs = yield* since(connectivityStarted);
4779	if (missing.length > 0) {
4780	return {
4781	unpack: `missing objects (${missing.length})`,
4782	results: allNg("missing necessary objects"),
4783	} satisfies CommitPushResult;
4784	}
4785	const graph = yield* computeGraphRows(input.commits);
4786	const finalizeStarted = yield* Effect.sync(() => performance.now());
4787	const results = yield* finalizeRefTxn({
4788	commands: input.commands,
4789	atomic: input.atomic,
4790	pushId: input.pushId,
4791	deferFlip: true,
4792	graph,
4793	});
4794	const finalizeMs = yield* since(finalizeStarted);
4795	// The per-row flip, off the response path (DESIGN §22.14).
4796	yield* state.waitUntil(flipPush(input.pushId).pipe(Effect.ignore));
4797
4798	yield* setConfig(
4799	"last_push",
4800	JSON.stringify({
4801	objects: input.stats.objects,
4802	bytes: input.stats.bytes,
4803	ingestMs: Math.round(input.stats.ingestMs),
4804	phases:
4805	input.stats.phases === undefined
4806	? undefined
4807	: Object.fromEntries(
4808	Object.entries(input.stats.phases).map(([k, v]) => [
4809	k,
4810	Math.round(v),
4811	]),
4812	),
4813	stageMs: Math.round(input.stats.stageMs),
4814	connectivityMs: Math.round(connectivityMs),
4815	finalizeMs: Math.round(finalizeMs),
4816	totalMs: Math.round(yield* since(startedAt)),
4817	} satisfies PushStatsData),
4818	).pipe(Effect.ignore);
4819
4820	// Post-commit bookkeeping off the response path: staged-row GC
4821	// and, once enough loose bytes have accumulated, compaction into
4822	// an R2 pack (DESIGN.md §12.1).
4823	yield* state.waitUntil(
4824	upsertJob("gc", null).pipe(
4825	Effect.flatMap(() => armAlarmAt(Date.now() + STAGING_TTL_MS)),
4826	Effect.flatMap(() => shouldCompact(sql)),
4827	Effect.flatMap((due) =>
4828	due
4829	? upsertJob("compact", null).pipe(
4830	Effect.flatMap(() => armAlarmAt(Date.now() + 1_000)),
4831	)
4832	: Effect.void,
4833	),
4834	// Re-cut the clone bundle for the new ref snapshot. Debounced
4835	// by the alarm itself: rapid-fire pushes coalesce into one
4836	// run because the job row is upserted, not queued.
4837	Effect.flatMap(() => upsertJob("bundle", null)),
4838	Effect.flatMap(() => armAlarmAt(Date.now() + 2_000)),
4839	Effect.ignore,
4840	),
4841	);
4842	return { unpack: "ok", results } satisfies CommitPushResult;
4843	}),
4844
4845	readObject: Effect.fn(function* (input: ReadObjectInput) {
4846	const meta = yield* requireMeta;
4847	const objects = storeFor(meta.repoId);
4848	const objectMeta = yield* objects.getMeta(input.oid);
4849	if (objectMeta === undefined) {
4850	return yield* new ObjectNotFound({ oid: input.oid });
4851	}
4852	const actual = objectTypeName(objectMeta.type);
4853	if (input.expect !== undefined && actual !== input.expect) {
4854	return yield* new WrongObjectType({
4855	oid: input.oid,
4856	expected: input.expect,
4857	actual,
4858	});
4859	}
4860	const content = yield* objects.readContent(input.oid);
4861	return {
4862	oid: input.oid,
4863	type: actual,
4864	size: objectMeta.size,
4865	content,
4866	} satisfies ObjectData;
4867	}),
4868
4869	readCommitLog: Effect.fn(function* (input: CommitLogInput) {
4870	const meta = yield* requireMeta;
4871	const objects = storeFor(meta.repoId);
4872	const limit = Math.max(1, Math.min(input.limit ?? 20, 100));
4873
4874	// Resolve the starting commit.
4875	const start = yield* resolveRevision(meta.defaultBranch, input.ref);
4876
4877	// Commit-time-ordered walk (v1: recomputed from the start each
4878	// page; the cursor is the last emitted oid).
4879	const frontier: Array<{ oid: string; time: number }> = [];
4880	const visited = new Set<string>();
4881	const items: Array<CommitData> = [];
4882	let skipping = input.cursor !== undefined;
4883	let hasMore = false;
4884
4885	const pushFrontier = Effect.fn(function* (oid: string) {
4886	if (visited.has(oid)) return;
4887	visited.add(oid);
4888	const row = yield* sql.first<{ commit_time: number }>(
4889	`SELECT commit_time FROM commits WHERE oid = ?`,
4890	oid,
4891	);
4892	if (row !== undefined) {
4893	frontier.push({ oid, time: row.commit_time });
4894	}
4895	});
4896	// The start may be a tag — peel to a commit.
4897	const startMeta = yield* objects.getMeta(start);
4898	if (startMeta === undefined) {
4899	return yield* new RefNotFound({ ref: input.ref ?? "HEAD" });
4900	}
4901	const startCommit =
4902	startMeta.type === ObjectType.tag
4903	? yield* peelOid(meta.repoId, start).pipe(
4904	Effect.map((peeled) => peeled ?? start),
4905	)
4906	: start;
4907	yield* pushFrontier(startCommit);
4908
4909	while (frontier.length > 0) {
4910	// pop max commit_time
4911	let best = 0;
4912	for (let i = 1; i < frontier.length; i++) {
4913	if (frontier[i]!.time > frontier[best]!.time) best = i;
4914	}
4915	const [next] = frontier.splice(best, 1);
4916	const oid = next!.oid;
4917	const parents = yield* sql.all<{ parent: string; ord: number }>(
4918	`SELECT parent, ord FROM commit_parents WHERE oid = ? ORDER BY ord`,
4919	oid,
4920	);
4921	for (const parent of parents) {
4922	yield* pushFrontier(parent.parent);
4923	}
4924	if (skipping) {
4925	if (oid === input.cursor) skipping = false;
4926	continue;
4927	}
4928	if (items.length >= limit) {
4929	hasMore = true;
4930	break;
4931	}
4932	const content = yield* objects.readContent(oid);
4933	const parsed = yield* parseCommit(content).pipe(
4934	Effect.mapError(
4935	(error) => new StoreError({ reason: error.reason }),
4936	),
4937	);
4938	items.push({
4939	oid,
4940	tree: parsed.tree,
4941	parents: parsed.parents,
4942	author: {
4943	name: parsed.author.name,
4944	email: parsed.author.email,
4945	date: parsed.author.when,
4946	tz: parsed.author.tz,
4947	},
4948	committer: {
4949	name: parsed.committer.name,
4950	email: parsed.committer.email,
4951	date: parsed.committer.when,
4952	tz: parsed.committer.tz,
4953	},
4954	message: parsed.message,
4955	});
4956	}
4957
4958	return {
4959	items,
4960	nextCursor: hasMore ? items[items.length - 1]!.oid : null,
4961	hasMore,
4962	} satisfies CommitLogPage;
4963	}),
4964
4965	readCommitDiff: Effect.fn(function* (input: { readonly oid: string }) {
4966	const meta = yield* requireMeta;
4967	const objects = storeFor(meta.repoId);
4968
4969	const readCommitOrFail = Effect.fn(function* (oid: string) {
4970	const objectMeta = yield* objects.getMeta(oid);
4971	if (objectMeta === undefined) {
4972	return yield* new ObjectNotFound({ oid });
4973	}
4974	if (objectMeta.type !== ObjectType.commit) {
4975	return yield* new WrongObjectType({
4976	oid,
4977	expected: "commit",
4978	actual: objectTypeName(objectMeta.type),
4979	});
4980	}
4981	const content = yield* objects.readContent(oid);
4982	return yield* parseCommit(content).pipe(
4983	Effect.mapError(
4984	(error) => new StoreError({ reason: error.reason }),
4985	),
4986	);
4987	});
4988
4989	const commit = yield* readCommitOrFail(input.oid);
4990	// First-parent diff (the GitHub default for merge commits).
4991	const parent = commit.parents[0] ?? null;
4992	const parentTree =
4993	parent === null
4994	? undefined
4995	: (yield* readCommitOrFail(parent)).tree;
4996
4997	const result = yield* diffTrees(objects, parentTree, commit.tree);
4998	return {
4999	oid: input.oid,
5000	parent,
5001	files: result.files,
5002	truncated: result.truncated,
5003	} satisfies CommitDiffData;
5004	}),
5005
5006	compareCommits: Effect.fn(function* (input: CompareInput) {
5007	const meta = yield* requireMeta;
5008	const objects = storeFor(meta.repoId);
5009
5010	const baseOid = yield* resolveToCommit(
5011	meta.repoId,
5012	meta.defaultBranch,
5013	input.base,
5014	);
5015	const headOid = yield* resolveToCommit(
5016	meta.repoId,
5017	meta.defaultBranch,
5018	input.head,
5019	);
5020
5021	// The classic merge-base paint walk over the commit-graph tables,
5022	// ordered by the pre-computed generation number (`gen`, maintained
5023	// at ingest): a common ancestor's gen is strictly less than any
5024	// node on a path to it, so processing gen-descending guarantees
5025	// the first both-flagged pop is a best common ancestor. Nodes
5026	// carrying both flags propagate both but are counted as neither.
5027	const BASE = 1;
5028	const HEAD = 2;
5029	const BOTH = 3;
5030	const flags = new Map<string, number>();
5031	flags.set(baseOid, BASE);
5032	flags.set(headOid, (flags.get(headOid) ?? 0) | HEAD);
5033	const heap: Array<WalkHeapNode> = [];
5034	const timeOf = new Map<string, number>();
5035
5036	const rowOf = (oid: string) =>
5037	sql.first<{ gen: number; commit_time: number }>(
5038	`SELECT gen, commit_time FROM commits WHERE oid = ?`,
5039	oid,
5040	);
5041
5042	// seed (base === head short-circuits: mergeBase = both, 0/0)
5043	for (const oid of new Set([baseOid, headOid])) {
5044	const row = yield* rowOf(oid);
5045	if (row === undefined) {
5046	return yield* new ObjectNotFound({ oid });
5047	}
5048	heapPush(heap, { oid, gen: row.gen, time: row.commit_time });
5049	timeOf.set(oid, row.commit_time);
5050	}
5051
5052	let mergeBase: string | undefined;
5053	let aheadBy = 0;
5054	let behindBy = 0;
5055	let pops = 0;
5056	const headSide: Array<string> = [];
5057
5058	while (heap.length > 0 && pops < MAX_COMPARE_WALK) {
5059	// stop when nothing in the heap can still change the counts
5060	if (
5061	mergeBase !== undefined &&
5062	heap.every((n) => (flags.get(n.oid)! & BOTH) === BOTH)
5063	) {
5064	break;
5065	}
5066	const node = heapPop(heap);
5067	pops++;
5068	const f = flags.get(node.oid)!;
5069	if ((f & BOTH) === BOTH) {
5070	if (mergeBase === undefined) mergeBase = node.oid;
5071	} else if (f === HEAD) {
5072	aheadBy++;
5073	headSide.push(node.oid);
5074	} else {
5075	behindBy++;
5076	}
5077	const parents = yield* sql.all<{ parent: string }>(
5078	`SELECT parent FROM commit_parents WHERE oid = ? ORDER BY ord`,
5079	node.oid,
5080	);
5081	for (const { parent } of parents) {
5082	const prev = flags.get(parent) ?? 0;
5083	const next = prev | f;
5084	if (next === prev) continue;
5085	flags.set(parent, next);
5086	if (prev === 0) {
5087	const row = yield* rowOf(parent);
5088	if (row === undefined) continue; // shallow boundary
5089	heapPush(heap, {
5090	oid: parent,
5091	gen: row.gen,
5092	time: row.commit_time,
5093	});
5094	timeOf.set(parent, row.commit_time);
5095	}
5096	// already-heaped nodes just get richer flags; no decrease-key
5097	// needed (flags only grow; both-flagged pops count as neither)
5098	}
5099	}
5100	if (mergeBase === undefined) {
5101	return yield* new NoMergeBase({
5102	base: asOid(baseOid),
5103	head: asOid(headOid),
5104	});
5105	}
5106
5107	// head-side commits, committer-time descending, capped.
5108	headSide.sort((a, b) => (timeOf.get(b) ?? 0) - (timeOf.get(a) ?? 0));
5109	const commitOids = headSide.slice(0, MAX_COMPARE_COMMITS);
5110	const commits: Array<CommitData> = [];
5111	for (const oid of commitOids) {
5112	const content = yield* objects.readContent(oid);
5113	const parsed = yield* parseCommit(content).pipe(
5114	Effect.mapError(
5115	(error) => new StoreError({ reason: error.reason }),
5116	),
5117	);
5118	commits.push({
5119	oid,
5120	tree: parsed.tree,
5121	parents: parsed.parents,
5122	author: {
5123	name: parsed.author.name,
5124	email: parsed.author.email,
5125	date: parsed.author.when,
5126	tz: parsed.author.tz,
5127	},
5128	committer: {
5129	name: parsed.committer.name,
5130	email: parsed.committer.email,
5131	date: parsed.committer.when,
5132	tz: parsed.committer.tz,
5133	},
5134	message: parsed.message,
5135	});
5136	}
5137
5138	// three-dot file diff: mergeBase tree vs head tree
5139	const treeOf = Effect.fn(function* (oid: string) {
5140	const row = yield* sql.first<{ tree: string }>(
5141	`SELECT tree FROM commits WHERE oid = ?`,
5142	oid,
5143	);
5144	if (row === undefined) {
5145	return yield* new StoreError({
5146	reason: `commit ${oid} missing from the commit graph`,
5147	});
5148	}
5149	return row.tree;
5150	});
5151	const mergeBaseTree = yield* treeOf(mergeBase);
5152	const headTree = yield* treeOf(headOid);
5153	const filesResult = yield* diffTrees(
5154	objects,
5155	mergeBaseTree,
5156	headTree,
5157	);
5158
5159	return {
5160	base: baseOid,
5161	head: headOid,
5162	mergeBase,
5163	aheadBy,
5164	behindBy,
5165	commits,
5166	commitsTruncated:
5167	headSide.length > MAX_COMPARE_COMMITS || pops >= MAX_COMPARE_WALK,
5168	files: filesResult.files,
5169	filesTruncated: filesResult.truncated,
5170	} satisfies CompareData;
5171	}),
5172
5173	readFileAtPath: Effect.fn(function* (input: ReadFileInput) {
5174	const meta = yield* requireMeta;
5175	const objects = storeFor(meta.repoId);
5176
5177	const startOid = yield* resolveRevision(
5178	meta.defaultBranch,
5179	input.ref,
5180	);
5181
5182	// Resolve start → commit → root tree (peel tags on the way).
5183	const rootTree = yield* Effect.gen(function* () {
5184	let current = startOid;
5185	for (let hops = 0; hops < 16; hops++) {
5186	const objectMeta = yield* objects.getMeta(current);
5187	if (objectMeta === undefined) {
5188	return yield* new ObjectNotFound({ oid: current });
5189	}
5190	if (objectMeta.type === ObjectType.tree) return current;
5191	const content = yield* objects.readContent(current);
5192	if (objectMeta.type === ObjectType.commit) {
5193	const parsed = yield* parseCommit(content).pipe(
5194	Effect.mapError(
5195	(error) => new StoreError({ reason: error.reason }),
5196	),
5197	);
5198	return parsed.tree;
5199	}
5200	if (objectMeta.type === ObjectType.tag) {
5201	const parsed = yield* parseTag(content).pipe(
5202	Effect.mapError(
5203	(error) => new StoreError({ reason: error.reason }),
5204	),
5205	);
5206	current = parsed.object;
5207	continue;
5208	}
5209	return yield* new ObjectNotFound({ oid: current });
5210	}
5211	return yield* new ObjectNotFound({ oid: current });
5212	});
5213
5214	const segments = input.path.split("/").filter(Boolean);
5215	if (segments.length === 0) {
5216	return yield* new ObjectNotFound({ oid: rootTree });
5217	}
5218	let treeOid = rootTree;
5219	for (let i = 0; i < segments.length; i++) {
5220	const content = yield* objects.readContent(treeOid);
5221	const entries = yield* parseTree(content).pipe(
5222	Effect.mapError(
5223	(error) => new StoreError({ reason: error.reason }),
5224	),
5225	);
5226	const entry = entries.find((e) => e.name === segments[i]);
5227	if (entry === undefined) {
5228	return yield* new ObjectNotFound({ oid: `${input.path}` });
5229	}
5230	const kind = treeEntryKind(entry.mode);
5231	if (i === segments.length - 1) {
5232	if (kind !== "blob") {
5233	return yield* new ObjectNotFound({ oid: input.path });
5234	}
5235	const blobMeta = yield* objects.getMeta(entry.oid);
5236	if (blobMeta === undefined) {
5237	return yield* new ObjectNotFound({ oid: entry.oid });
5238	}
5239	const blob = yield* objects.readContent(entry.oid);
5240	return {
5241	oid: entry.oid,
5242	mode: entry.mode,
5243	size: blobMeta.size,
5244	content: blob,
5245	} satisfies FileData;
5246	}
5247	if (kind !== "tree") {
5248	return yield* new ObjectNotFound({ oid: input.path });
5249	}
5250	treeOid = entry.oid;
5251	}
5252	// unreachable — the loop returns or fails
5253	return yield* new ObjectNotFound({ oid: input.path });
5254	}),
5255
5256	createPull: Effect.fn(function* (input: CreatePullInput) {
5257	yield* requireMeta;
5258	const baseRef = yield* normalizeBranchRef(input.base);
5259	const headRef = yield* normalizeBranchRef(input.head);
5260	if (baseRef === headRef) {
5261	return yield* new ValidationError({
5262	message: "base and head must be different branches",
5263	});
5264	}
5265	for (const ref of [baseRef, headRef]) {
5266	const row = yield* sql.first<RefRow>(
5267	`SELECT name, oid FROM refs WHERE name = ?`,
5268	ref,
5269	);
5270	if (row === undefined) {
5271	return yield* new BranchMissing({ ref });
5272	}
5273	}
5274	const duplicate = yield* sql.first<{ number: number }>(
5275	`SELECT number FROM pulls WHERE state = 'open' AND base_ref = ? AND head_ref = ?`,
5276	baseRef,
5277	headRef,
5278	);
5279	if (duplicate !== undefined) {
5280	return yield* new PullExists({ number: duplicate.number });
5281	}
5282	const now = Date.now();
5283	// Allocate the number and insert the row in ONE transaction —
5284	// the DO's input/output gates make the counter race-free.
5285	const number = yield* sql.transactionSync((raw) => {
5286	const seq = raw
5287	.exec<{ value: string }>(
5288	`SELECT value FROM config WHERE key = 'pull_seq'`,
5289	)
5290	.toArray()[0];
5291	const next = (seq === undefined ? 0 : Number(seq.value)) + 1;
5292	raw.exec(
5293	`INSERT INTO config (key, value) VALUES ('pull_seq', ?)
5294	ON CONFLICT (key) DO UPDATE SET value = excluded.value`,
5295	String(next),
5296	);
5297	raw.exec(
5298	`INSERT INTO pulls (number, title, body, base_ref, head_ref, state, created_at, updated_at)
5299	VALUES (?, ?, ?, ?, ?, 'open', ?, ?)`,
5300	next,
5301	input.title,
5302	input.body ?? null,
5303	baseRef,
5304	headRef,
5305	now,
5306	now,
5307	);
5308	return next;
5309	});
5310	return {
5311	number,
5312	title: input.title,
5313	body: input.body ?? null,
5314	baseRef,
5315	headRef,
5316	state: "open",
5317	createdAt: now,
5318	updatedAt: now,
5319	mergedAt: null,
5320	mergeCommit: null,
5321	} satisfies PullData;
5322	}),
5323
5324	listPulls: Effect.fn(function* (input: ListPullsInput) {
5325	yield* requireMeta;
5326	const state = input.state ?? "open";
5327	const limit = Math.max(1, Math.min(input.limit ?? 20, 100));
5328	const cursor =
5329	input.cursor === undefined ? undefined : Number(input.cursor);
5330	const where: Array<string> = [];
5331	const bindings: Array<string | number> = [];
5332	if (state !== "all") {
5333	where.push("state = ?");
5334	bindings.push(state);
5335	}
5336	if (cursor !== undefined && Number.isFinite(cursor)) {
5337	where.push("number < ?");
5338	bindings.push(cursor);
5339	}
5340	const rows = yield* sql.all<PullRow>(
5341	`SELECT * FROM pulls
5342	${where.length > 0 ? `WHERE ${where.join(" AND ")}` : ""}
5343	ORDER BY number DESC LIMIT ?`,
5344	...bindings,
5345	limit + 1,
5346	);
5347	const items = rows.slice(0, limit).map(rowToPull);
5348	const hasMore = rows.length > limit;
5349	return {
5350	items,
5351	nextCursor: hasMore
5352	? String(items[items.length - 1]!.number)
5353	: null,
5354	hasMore,
5355	} satisfies PullsPage;
5356	}),
5357
5358	getPull: Effect.fn(function* (number: number) {
5359	const meta = yield* requireMeta;
5360	const row = yield* loadPull(number);
5361	const pull = rowToPull(row);
5362	const empty = {
5363	baseOid: null,
5364	headOid: null,
5365	mergeBase: null,
5366	aheadBy: null,
5367	behindBy: null,
5368	mergeable: null,
5369	mergeableReason: null,
5370	} as const;
5371	// A merged PR's record is `mergeCommit` — live fields stay null.
5372	if (pull.state === "merged") {
5373	return { ...pull, ...empty } satisfies PullDetailData;
5374	}
5375	const baseRow = yield* sql.first<RefRow>(
5376	`SELECT name, oid FROM refs WHERE name = ?`,
5377	row.base_ref,
5378	);
5379	const headRow = yield* sql.first<RefRow>(
5380	`SELECT name, oid FROM refs WHERE name = ?`,
5381	row.head_ref,
5382	);
5383	if (baseRow === undefined || headRow === undefined) {
5384	// A branch is gone: everything downstream is uncomputable.
5385	return {
5386	...pull,
5387	...empty,
5388	baseOid: baseRow?.oid ?? null,
5389	headOid: headRow?.oid ?? null,
5390	} satisfies PullDetailData;
5391	}
5392	const baseTip = baseRow.oid;
5393	const headTip = headRow.oid;
5394	const cmp = yield* paintWalk(baseTip, headTip, MAX_PULL_WALK);
5395	if (cmp.saturated) {
5396	return {
5397	...pull,
5398	...empty,
5399	baseOid: baseTip,
5400	headOid: headTip,
5401	mergeableReason: "unknown",
5402	} satisfies PullDetailData;
5403	}
5404	const live = {
5405	...pull,
5406	baseOid: baseTip,
5407	headOid: headTip,
5408	mergeBase: cmp.mergeBase ?? null,
5409	aheadBy: cmp.aheadBy,
5410	behindBy: cmp.behindBy,
5411	};
5412	if (cmp.mergeBase === undefined) {
5413	// Disjoint histories: our trivial-merge rules can never merge.
5414	return {
5415	...live,
5416	mergeable: false,
5417	mergeableReason: "conflict",
5418	} satisfies PullDetailData;
5419	}
5420	if (cmp.mergeBase === headTip) {
5421	return {
5422	...live,
5423	mergeable: false,
5424	mergeableReason: "up-to-date",
5425	} satisfies PullDetailData;
5426	}
5427	if (cmp.mergeBase === baseTip) {
5428	return {
5429	...live,
5430	mergeable: true,
5431	mergeableReason: "ff",
5432	} satisfies PullDetailData;
5433	}
5434	// Trivial-merge dry run: diff intersection only, no tree building.
5435	const objects = storeFor(meta.repoId);
5436	const mbTree = yield* treeOfCommit(cmp.mergeBase);
5437	const baseChanges = yield* diffTrees(
5438	objects,
5439	mbTree,
5440	yield* treeOfCommit(baseTip),
5441	);
5442	const headChanges = yield* diffTrees(
5443	objects,
5444	mbTree,
5445	yield* treeOfCommit(headTip),
5446	);
5447	if (baseChanges.truncated || headChanges.truncated) {
5448	// Can't see every change — conservatively not mergeable.
5449	return {
5450	...live,
5451	mergeable: false,
5452	mergeableReason: "conflict",
5453	} satisfies PullDetailData;
5454	}
5455	const conflicts = conflictingPaths(
5456	baseChanges.files,
5457	headChanges.files,
5458	);
5459	return {
5460	...live,
5461	mergeable: conflicts.length === 0,
5462	mergeableReason:
5463	conflicts.length === 0 ? "merge-commit" : "conflict",
5464	} satisfies PullDetailData;
5465	}),
5466
5467	updatePull: Effect.fn(function* (input: UpdatePullInput) {
5468	yield* requireMeta;
5469	const row = yield* loadPull(input.number);
5470	const current = pullStateOf(row.state);
5471	if (input.state !== undefined && current === "merged") {
5472	return yield* new PullStateConflict({
5473	number: input.number,
5474	state: "merged",
5475	});
5476	}
5477	const title = input.title ?? row.title;
5478	const body = input.body === undefined ? row.body : input.body;
5479	const state = input.state ?? current;
5480	yield* sql.run(
5481	`UPDATE pulls SET title = ?, body = ?, state = ?, updated_at = ? WHERE number = ?`,
5482	title,
5483	body,
5484	state,
5485	Date.now(),
5486	input.number,
5487	);
5488	return rowToPull(yield* loadPull(input.number));
5489	}),
5490
5491	mergePull: Effect.fn(function* (input: MergePullInput) {
5492	const meta = yield* requireMeta;
5493	if (meta.readOnly) {
5494	return yield* new ReadOnlyRepo();
5495	}
5496	const row = yield* loadPull(input.number);
5497	const state = pullStateOf(row.state);
5498	if (state !== "open") {
5499	return yield* new PullStateConflict({
5500	number: input.number,
5501	state,
5502	});
5503	}
5504	// Both branches must exist — the ref CAS below must NEVER
5505	// (re)create a deleted base branch.
5506	const baseRow = yield* sql.first<RefRow>(
5507	`SELECT name, oid FROM refs WHERE name = ?`,
5508	row.base_ref,
5509	);
5510	if (baseRow === undefined) {
5511	return yield* new BranchMissing({ ref: row.base_ref });
5512	}
5513	const headRow = yield* sql.first<RefRow>(
5514	`SELECT name, oid FROM refs WHERE name = ?`,
5515	row.head_ref,
5516	);
5517	if (headRow === undefined) {
5518	return yield* new BranchMissing({ ref: row.head_ref });
5519	}
5520	const baseTip = baseRow.oid;
5521	if (
5522	input.expectedBaseOid !== undefined &&
5523	input.expectedBaseOid !== baseTip
5524	) {
5525	return yield* new RefConflict({
5526	ref: row.base_ref,
5527	currentOid: asOid(baseTip),
5528	});
5529	}
5530	const headTip = headRow.oid;
5531	if (
5532	input.expectedHeadOid !== undefined &&
5533	input.expectedHeadOid !== headTip
5534	) {
5535	return yield* new RefConflict({
5536	ref: row.head_ref,
5537	currentOid: asOid(headTip),
5538	});
5539	}
5540	const cmp = yield* paintWalk(baseTip, headTip, MAX_PULL_MERGE_WALK);
5541	if (cmp.mergeBase === headTip) {
5542	return yield* new NothingToMerge({ number: input.number });
5543	}
5544	const now = Date.now();
5545
5546	// ── fast-forward: base is an ancestor of head ──────────────────
5547	if (cmp.mergeBase === baseTip) {
5548	// No new objects, no graph rows (all present from the push) —
5549	// one transaction re-checks the CAS, flips the ref, and stamps
5550	// the pulls row atomically.
5551	const pull = yield* sql.transactionSync<PullData, RefConflict>(
5552	(raw, rollback) => {
5553	const rows = raw
5554	.exec<RefRow>(
5555	`SELECT name, oid FROM refs WHERE name = ?`,
5556	row.base_ref,
5557	)
5558	.toArray();
5559	const currentOid = rows.length > 0 ? rows[0]!.oid : null;
5560	if (currentOid !== baseTip) {
5561	rollback(
5562	new RefConflict({
5563	ref: row.base_ref,
5564	currentOid:
5565	currentOid === null ? null : asOid(currentOid),
5566	}),
5567	);
5568	}
5569	raw.exec(
5570	`INSERT INTO refs (name, oid) VALUES (?, ?)
5571	ON CONFLICT (name) DO UPDATE SET oid = excluded.oid`,
5572	row.base_ref,
5573	headTip,
5574	);
5575	raw.exec(
5576	`UPDATE pulls SET state = 'merged', merged_at = ?, merge_commit = ?, updated_at = ? WHERE number = ?`,
5577	now,
5578	headTip,
5579	now,
5580	input.number,
5581	);
5582	const updated = raw
5583	.exec<PullRow>(
5584	`SELECT * FROM pulls WHERE number = ?`,
5585	input.number,
5586	)
5587	.toArray()[0]!;
5588	return rowToPull(updated);
5589	},
5590	);
5591	// The base moved: the head snapshot (which serves clones
5592	// without waking this object) must say so.
5593	yield* writeHeadSnapshot;
5594	return {
5595	method: "ff",
5596	oid: headTip,
5597	pull,
5598	} satisfies MergePullResult;
5599	}
5600
5601	// ── trivial three-way merge commit ─────────────────────────────
5602	if (cmp.mergeBase === undefined) {
5603	// Unrelated histories (or a saturated walk): conservative
5604	// conflict — the server never guesses.
5605	return yield* new MergeConflict({
5606	number: input.number,
5607	paths: [],
5608	});
5609	}
5610	const objects = storeFor(meta.repoId);
5611	const mbTree = yield* treeOfCommit(cmp.mergeBase);
5612	const baseTree = yield* treeOfCommit(baseTip);
5613	const baseChanges = yield* diffTrees(objects, mbTree, baseTree);
5614	const headChanges = yield* diffTrees(
5615	objects,
5616	mbTree,
5617	yield* treeOfCommit(headTip),
5618	);
5619	if (baseChanges.truncated || headChanges.truncated) {
5620	// A truncated diff hides changes — conflict detection would be
5621	// unsound, so refuse conservatively.
5622	return yield* new MergeConflict({
5623	number: input.number,
5624	paths: [],
5625	});
5626	}
5627	const conflicts = conflictingPaths(
5628	baseChanges.files,
5629	headChanges.files,
5630	);
5631	if (conflicts.length > 0) {
5632	return yield* new MergeConflict({
5633	number: input.number,
5634	paths: conflicts.slice(0, MAX_CONFLICT_PATHS),
5635	});
5636	}
5637
5638	// Build the merged tree: head's mergeBase→head changes applied
5639	// onto the CURRENT base tree (disjoint by the check above).
5640	const applied = yield* applyTreeChanges(
5641	objects,
5642	baseTree,
5643	headChanges.files,
5644	);
5645	const shortHead = row.head_ref.startsWith("refs/heads/")
5646	? row.head_ref.slice("refs/heads/".length)
5647	: row.head_ref;
5648	const message =
5649	input.message ??
5650	`Merge pull request #${input.number} from ${shortHead}\n`;
5651	const when = Math.floor(now / 1000);
5652	const identity = {
5653	name: "git-service",
5654	email: "git-service@localhost",
5655	when,
5656	tz: "+0000",
5657	};
5658	const commitContent = yield* encodeCommit({
5659	tree: applied.root,
5660	parents: [baseTip, headTip],
5661	author: identity,
5662	committer: identity,
5663	message,
5664	}).pipe(Effect.mapError((e) => new StoreError({ reason: e.reason })));
5665	const mergeOid = yield* hashObject(ObjectType.commit, commitContent);
5666
5667	// Stage under a synthetic push id WITH a pushes staging row first,
5668	// so the staging-GC alarm reaps the objects if we crash (or the
5669	// CAS below loses) — the exact crash-safety story of a push.
5670	const pushId = `pr-${input.number}-${yield* ulid()}`;
5671	yield* sql.run(
5672	`INSERT INTO pushes (push_id, started_at, state) VALUES (?, ?, 'staging')`,
5673	pushId,
5674	now,
5675	);
5676	const zlibToStore = (error: Zlib.ZlibError): StoreError =>
5677	new StoreError({ reason: error.reason });
5678	const staged: Array<StagedObject> = [];
5679	for (const tree of applied.newTrees) {
5680	staged.push({
5681	oid: tree.oid,
5682	type: ObjectType.tree,
5683	size: tree.content.length,
5684	zdata: yield* Zlib.deflate(tree.content).pipe(
5685	Effect.mapError(zlibToStore),
5686	),
5687	});
5688	}
5689	staged.push({
5690	oid: mergeOid,
5691	type: ObjectType.commit,
5692	size: commitContent.length,
5693	zdata: yield* Zlib.deflate(commitContent).pipe(
5694	Effect.mapError(zlibToStore),
5695	),
5696	});
5697	yield* objects.insertStagedBatch(pushId, staged);
5698	// Crash safety: arm the staging-GC alarm (like the push path) so
5699	// a lost CAS below — or a crash before finalize — cannot leak the
5700	// staged objects; a committed push's row is ignored by the GC.
5701	yield* upsertJob("gc", null);
5702	yield* armAlarmAt(Date.now() + STAGING_TTL_MS);
5703
5704	const graph = yield* computeGraphRows([
5705	{
5706	oid: mergeOid,
5707	tree: applied.root,
5708	parents: [baseTip, headTip],
5709	commitTime: when,
5710	},
5711	]);
5712	const results = yield* finalizeRefTxn({
5713	commands: [
5714	{ oldOid: baseTip, newOid: mergeOid, ref: row.base_ref },
5715	],
5716	atomic: true,
5717	pushId,
5718	graph,
5719	// Atomic with the ref flip + staged-object promotion.
5720	onCommitted: (raw) => {
5721	raw.exec(
5722	`UPDATE pulls SET state = 'merged', merged_at = ?, merge_commit = ?, updated_at = ? WHERE number = ?`,
5723	now,
5724	mergeOid,
5725	now,
5726	input.number,
5727	);
5728	},
5729	});
5730	const outcome = results[0];
5731	if (outcome === undefined || !outcome.ok) {
5732	// Base moved between the pre-check and the transaction. The
5733	// staged objects stay parked under `pushId` for the GC alarm.
5734	const current = yield* sql.first<RefRow>(
5735	`SELECT name, oid FROM refs WHERE name = ?`,
5736	row.base_ref,
5737	);
5738	return yield* new RefConflict({
5739	ref: row.base_ref,
5740	currentOid: current === undefined ? null : asOid(current.oid),
5741	});
5742	}
5743	// Post-merge compaction check deliberately skipped: the merge
5744	// adds at most a handful of small trees + one commit (the push
5745	// path's thresholds cover accumulation).
5746	yield* writeHeadSnapshot;
5747	return {
5748	method: "merge-commit",
5749	oid: mergeOid,
5750	pull: rowToPull(yield* loadPull(input.number)),
5751	} satisfies MergePullResult;
5752	}),
5753
5754	snapshotRows: () => snapshotStream(sql),
5755
5756	fetch: Effect.gen(function* () {
5757	const request = yield* HttpServerRequest.HttpServerRequest;
5758	const path = parseWirePath(request.url);
5759	if (path === undefined) {
5760	return HttpServerResponse.text("not found", { status: 404 });
5761	}
5762	const meta = yield* readMeta;
5763	if (meta === undefined) {
5764	return HttpServerResponse.text("repository not found", {
5765	status: 404,
5766	});
5767	}
5768	// Stale-cache fail-safe (DESIGN.md §2.1): this DO only answers
5769	// for its own owner/name.
5770	if (path.owner !== meta.owner || path.repo !== meta.name) {
5771	return HttpServerResponse.text("repository not found", {
5772	status: 404,
5773	});
5774	}
5775	if (meta.status !== "ready") {
5776	return wire503;
5777	}
5778	if (path.endpoint === "info/refs" && request.method === "GET") {
5779	return yield* handleInfoRefs(request, path.service, meta);
5780	}
5781	if (
5782	path.endpoint === "git-upload-pack" &&
5783	request.method === "POST"
5784	) {
5785	return yield* handleUploadPack(request, meta);
5786	}
5787	return HttpServerResponse.text("not found", { status: 404 });
5788	}).pipe(
5789	Effect.catchTag("StoreError", (error) =>
5790	Effect.as(
5791	Effect.logError("git wire: storage failure", error),
5792	HttpServerResponse.text("internal error", { status: 500 }),
5793	),
5794	),
5795	Effect.catchTag("PackIngestError", (error) =>
5796	Effect.succeed(
5797	HttpServerResponse.uint8Array(errPkt(error.reason), {
5798	status: 200,
5799	headers: noCache,
5800	}),
5801	),
5802	),
5803	),
5804
5805	alarm: () =>
5806	Effect.gen(function* () {
5807	const jobs = yield* sql.all<JobRow>(
5808	`SELECT * FROM jobs WHERE state = 'running'`,
5809	);
5810	// `purge` runs FIRST and every job is individually isolated: a
5811	// job that keeps crashing must never starve the others. (It
5812	// did: a fork job dying on every alarm blocked the purge job
5813	// behind it forever, wedging the repo in `status: deleting`
5814	// and reserving its name.)
5815	const order = (kind: string) =>
5816	kind === "purge" ? 0 : kind === "gc" ? 2 : 1;
5817	const ordered = [...jobs].sort(
5818	(a, b) => order(a.kind) - order(b.kind),
5819	);
5820	for (const job of ordered) {
5821	const isolated = Effect.gen(function* () {
5822	switch (job.kind) {
5823	case "purge": {
5824	const wiped = yield* runPurgeAlarm;
5825	return wiped; // storage is gone — stop the whole alarm
5826	}
5827	case "import":
5828	yield* runImportAlarm(job);
5829	return false;
5830	case "fork":
5831	yield* runForkAlarm(job);
5832	return false;
5833	case "gc":
5834	yield* runGcJob;
5835	return false;
5836	case "compact":
5837	yield* runCompactAlarm;
5838	return false;
5839	case "bundle":
5840	yield* runBundleAlarm;
5841	return false;
5842	default:
5843	yield* sql.run(`DELETE FROM jobs WHERE kind = ?`, job.kind);
5844	return false;
5845	}
5846	}).pipe(
5847	Effect.catchCause((cause) =>
5848	Effect.as(
5849	Effect.logError(
5850	`repo alarm job '${job.kind}' failed`,
5851	cause,
5852	),
5853	false,
5854	),
5855	),
5856	);
5857	if (yield* isolated) return;
5858	}
5859	}).pipe(
5860	Effect.catchCause((cause) =>
5861	Effect.gen(function* () {
5862	yield* Effect.logError("repo alarm failed", cause);
5863	// Best-effort retry — never let the alarm defect.
5864	yield* armAlarmAt(Date.now() + 30_000).pipe(Effect.ignore);
5865	}),
5866	),
5867	Effect.asVoid,
5868	),
5869	};
5870
5871	return shape;
5872	});
5873	}),
5874	);
5875
5876	export default GitRepoLive;