1/** Git operations without HTTP requests, user services, or authorization callbacks. */
2import * as Context from "effect/Context";
3import * as Effect from "effect/Effect";
4import * as Fiber from "effect/Fiber";
5import * as Layer from "effect/Layer";
6import * as Option from "effect/Option";
7import * as Semaphore from "effect/Semaphore";
8import { RuntimeContext } from "../RuntimeContext.ts";
9import { PushDenied, RefConflict, RepoNotFound } from "./Api/Schema.ts";
10import { BlobStore } from "./BlobStore.ts";
11import { Hasher } from "./Hasher/Hasher.ts";
12import { Operations, OperationsLive } from "./Operations.ts";
13import { StoreError } from "./Protocol/Store.ts";
14import {
15 incomingStates,
16 invalidPush,
17 preparedStates,
18 type PreparedPush,
19 type PushInput,
20 type RefUpdate,
21} from "./Push.ts";
22import { encodeStagedBatch } from "./PushWire.ts";
23import { ulid } from "./RegistryObject.ts";
24import {
25 MAX_PACK_BYTES,
26 PUSH_WAIT_TIMEOUT,
27 RepoStore,
28 ingestPackFrom,
29 isolatePushGate,
30 pushPermitsFor,
31 type CommitPushInput,
32 type IngestResult,
33 type IngestStore,
34 type RepoMetaData,
35} from "./RepoObject.ts";
36import { incomingKey, wirePackId } from "./Store/Keys.ts";
37import { sliceRandomAccess } from "./Store/PackSource.ts";
38
39const path = (repo: RepoMetaData) => ({ owner: repo.owner, repo: repo.name });
40
41const asStoreError = (error: {
42 readonly _tag: string;
43 readonly reason?: string;
44}) =>
45 error instanceof StoreError
46 ? error
47 : new StoreError({
48 reason: `${error._tag}${error.reason === undefined ? "" : `: ${error.reason}`}`,
49 });
50
51const scopedMutation = <A, E, R>(
52 updates: ReadonlyArray<RefUpdate>,
53 commit: Effect.Effect<A, E, R>,
54) =>
55 Effect.gen(function* () {
56 let active = true;
57 yield* Effect.addFinalizer(() =>
58 Effect.sync(() => {
59 active = false;
60 }),
61 );
62 return Object.freeze({
63 updates: Object.freeze(
64 updates.map((update) => Object.freeze({ ...update })),
65 ),
66 commit: Effect.suspend<A, E | StoreError, R>(() => {
67 if (!active) return Effect.fail(invalidPush());
68 active = false;
69 return commit;
70 }),
71 });
72 });
73
74const make = Effect.gen(function* () {
75 const operations = yield* Operations;
76 const repos = yield* RepoStore;
77 const blobs = yield* BlobStore;
78 const hasher = yield* Hasher;
79 const gate = yield* isolatePushGate;
80 const owner = {};
81
82 const get = (path: { readonly owner: string; readonly repo: string }) =>
83 Effect.gen(function* () {
84 const name = {
85 owner: path.owner.toLowerCase(),
86 repo: path.repo.toLowerCase().replace(/\.git$/, ""),
87 };
88 const entry = yield* operations.resolveCached(name.owner, name.repo);
89 if (entry === undefined || entry.deletedAt !== null)
90 return yield* new RepoNotFound(name);
91 return yield* repos.getByName(entry.repoId).getRepoMeta();
92 });
93
94 const preparePush = (repo: RepoMetaData, input: PushInput) =>
95 Effect.gen(function* () {
96 const source = incomingStates.get(input);
97 if (source === undefined || !source.active || source.claimed)
98 return yield* invalidPush();
99 source.claimed = true;
100 const { feeder, receiving, packStart } = source;
101 const stub = repos.getByName(repo.repoId);
102 const staging: Array<Fiber.Fiber<void, StoreError>> = [];
103 let active = true;
104 let committed = false;
105 let ingest: IngestResult | undefined;
106 const receiveId = yield* ulid();
107 const parkedKey = incomingKey(repo.repoId, receiveId);
108 // Registered before ingest: cancellation and policy failure share the same owner.
109 const begun = yield* Effect.acquireRelease(
110 stub.beginPush({ commands: input.updates }),
111 (begun) =>
112 Effect.gen(function* () {
113 active = false;
114 if (begun._tag === "denied") return;
115 yield* Fiber.interruptAll(staging.splice(0));
116 if (committed) return;
117 // Ask durable state before deleting bytes: an interrupted RPC may have committed.
118 const keys = [
119 parkedKey,
120 ...(ingest?.resolvedKey?.split(",") ?? []),
121 ].filter(Boolean);
122 const packIds = keys.map((key) =>
123 wirePackId(key.slice(key.lastIndexOf("/") + 1, -5)),
124 );
125 const aborted = yield* stub
126 .abortPush(begun.pushId, packIds)
127 .pipe(Effect.orElseSucceed(() => false));
128 if (aborted) {
129 yield* blobs.delete(keys).pipe(Effect.ignore);
130 }
131 }),
132 );
133 if (begun._tag === "denied")
134 return yield* new PushDenied({
135 ref: input.updates[0]?.ref ?? "",
136 reason: begun.reason,
137 });
138 const pushId = begun.pushId;
139 const permits = pushPermitsFor(source.declaredBytes);
140 const acquired = yield* Effect.acquireRelease(
141 Semaphore.take(gate, permits).pipe(
142 Effect.timeoutOption(PUSH_WAIT_TIMEOUT),
143 ),
144 (acquired) =>
145 Option.isSome(acquired)
146 ? Semaphore.release(gate, permits)
147 : Effect.void,
148 );
149 if (Option.isNone(acquired))
150 return yield* new StoreError({ reason: "push admission timed out" });
151 const started = Date.now();
152 const stageGate = yield* Semaphore.make(6);
153 const bases = new Map<
154 string,
155 Effect.Success<ReturnType<typeof stub.readPushBase>>
156 >();
157 const store: IngestStore = {
158 insertStagedBatch: (id, objects) =>
159 Effect.gen(function* () {
160 const encoded = encodeStagedBatch(objects);
161 const fiber = yield* Effect.forkDetach(
162 Semaphore.withPermits(
163 stageGate,
164 1,
165 )(
166 stub
167 .stagePush(id, encoded)
168 .pipe(
169 Effect.mapError(asStoreError),
170 Effect.provide(RuntimeContext.phantom),
171 ),
172 ),
173 );
174 staging.push(fiber);
175 }),
176 settle: Effect.gen(function* () {
177 for (const fiber of staging) yield* Fiber.join(fiber);
178 staging.length = 0;
179 }),
180 readBase: (oid) =>
181 bases.has(oid)
182 ? Effect.succeed(bases.get(oid))
183 : stub.readPushBase(oid).pipe(
184 Effect.mapError(asStoreError),
185 Effect.tap((value) =>
186 Effect.sync(() => {
187 bases.set(oid, value);
188 }),
189 ),
190 Effect.provide(RuntimeContext.phantom),
191 ),
192 };
193 const probe = yield* feeder.source.read(packStart, 12);
194 if (probe.length > 0 && probe.length < 12) {
195 return yield* new StoreError({ reason: "truncated pack header" });
196 }
197 const hasPack = probe.length === 12;
198 if (hasPack) {
199 ingest = yield* ingestPackFrom(
200 sliceRandomAccess(feeder.source, packStart),
201 {
202 store,
203 pushId,
204 hasher,
205 spill: {
206 body: feeder.source,
207 feeder,
208 packStart,
209 blobs,
210 key: parkedKey,
211 packId: wirePackId(receiveId),
212 repoId: repo.repoId,
213 threshold: MAX_PACK_BYTES,
214 },
215 },
216 );
217 }
218 if (store.settle) yield* store.settle;
219 const received = yield* Fiber.join(receiving).pipe(
220 Effect.flatMap(Effect.fromResult),
221 );
222 const commitInput: CommitPushInput = {
223 pushId,
224 commands: input.updates,
225 atomic: input.atomic,
226 commits: ingest?.commits ?? [],
227 referenced: ingest?.referenced ?? [],
228 referencedParents: ingest?.referencedParents ?? [],
229 promoted: ingest?.promoted ?? 0,
230 stats: {
231 objects: ingest?.objectCount ?? 0,
232 bytes: hasPack ? received.total - packStart : 0,
233 ingestMs: Date.now() - started,
234 stageMs: ingest?.stageMs ?? 0,
235 phases: ingest?.phases,
236 },
237 };
238 // Connectivity is checked before application validation, then again at commit.
239 yield* stub.validatePush(commitInput).pipe(Effect.mapError(asStoreError));
240 const prepared: PreparedPush = Object.freeze({
241 repo: Object.freeze({ ...repo }),
242 updates: input.updates,
243 readObject: (oid: string, maxBytes = 1024 * 1024) =>
244 Effect.suspend(() =>
245 !active || !source.active
246 ? Effect.fail(invalidPush())
247 : stub
248 .readPreparedObject(pushId, oid, maxBytes)
249 .pipe(Effect.mapError(asStoreError)),
250 ),
251 });
252 preparedStates.set(prepared, {
253 owner,
254 commit: Effect.suspend(() => {
255 if (!active || !source.active) return Effect.fail(invalidPush());
256 active = false;
257 return stub.commitPush(commitInput).pipe(
258 Effect.mapError(asStoreError),
259 Effect.tap((result) =>
260 Effect.sync(() => {
261 committed = result.results.some((ref) => ref.ok);
262 }),
263 ),
264 );
265 }),
266 });
267 return prepared;
268 });
269
270 const commitPush = (prepared: PreparedPush) =>
271 Effect.suspend(() => {
272 const state = preparedStates.get(prepared);
273 return state?.owner === owner ? state.commit : Effect.fail(invalidPush());
274 });
275
276 const prepareRefUpdate = (
277 repo: RepoMetaData,
278 input: {
279 readonly ref: string;
280 readonly newOid: string;
281 readonly expectedOid?: string | null;
282 },
283 ) =>
284 Effect.gen(function* () {
285 const stub = repos.getByName(repo.repoId);
286 const current = yield* stub.getRef(input.ref).pipe(
287 Effect.map((ref) => ref.oid),
288 Effect.catchTag("RefNotFound", () => Effect.succeed("0".repeat(40))),
289 );
290 if (
291 input.expectedOid !== undefined &&
292 input.expectedOid !== (current === "0".repeat(40) ? null : current)
293 )
294 return yield* new RefConflict({
295 ref: input.ref,
296 currentOid: current as import("./Api.ts").Oid,
297 });
298 const update = { ref: input.ref, oldOid: current, newOid: input.newOid };
299 const commit = stub.updateRef({
300 name: input.ref,
301 newOid: input.newOid,
302 expectedOid: current === "0".repeat(40) ? null : current,
303 });
304 return yield* scopedMutation([update], commit);
305 });
306
307 const prepareRefRemoval = (
308 repo: RepoMetaData,
309 input: { readonly ref: string; readonly expectedOid?: string },
310 ) =>
311 Effect.gen(function* () {
312 const stub = repos.getByName(repo.repoId);
313 const current = yield* stub.getRef(input.ref);
314 if (input.expectedOid !== undefined && input.expectedOid !== current.oid)
315 return yield* new RefConflict({
316 ref: input.ref,
317 currentOid: current.oid as import("./Api.ts").Oid,
318 });
319 return yield* scopedMutation(
320 [{ ref: input.ref, oldOid: current.oid, newOid: "0".repeat(40) }],
321 stub.removeRef({ name: input.ref, expectedOid: current.oid }),
322 );
323 });
324
325 const prepareMerge = (
326 repo: RepoMetaData,
327 input: Parameters<ReturnType<typeof repos.getByName>["mergePull"]>[0],
328 ) =>
329 Effect.gen(function* () {
330 const stub = repos.getByName(repo.repoId);
331 const pull = yield* stub.getPull(input.number);
332 return yield* scopedMutation(
333 [
334 {
335 ref: pull.baseRef,
336 oldOid: pull.baseOid ?? "0".repeat(40),
337 newOid: pull.headOid ?? "0".repeat(40),
338 },
339 ],
340 stub.mergePull({
341 ...input,
342 expectedBaseOid: pull.baseOid ?? "0".repeat(40),
343 expectedHeadOid:
344 input.expectedHeadOid ?? pull.headOid ?? "0".repeat(40),
345 }),
346 );
347 });
348
349 // These operate on repository data; authorization is composed by the caller.
350 return {
351 repositories: {
352 get,
353 create: (
354 payload: Parameters<typeof operations.repos.create>[0]["payload"],
355 ) => operations.repos.create({ payload }),
356 list: (query: Parameters<typeof operations.repos.list>[0]["query"]) =>
357 operations.repos.list({ query }),
358 update: (
359 repo: RepoMetaData,
360 payload: Parameters<typeof operations.repos.update>[0]["payload"],
361 ) => operations.repos.update({ params: path(repo), payload }),
362 remove: (repo: RepoMetaData) =>
363 operations.repos.delete({ params: path(repo) }),
364 fork: (
365 repo: RepoMetaData,
366 payload: Parameters<typeof operations.repos.fork>[0]["payload"],
367 ) => operations.repos.fork({ params: path(repo), payload }),
368 import: (
369 payload: Parameters<typeof operations.repos.import>[0]["payload"],
370 ) => operations.repos.import({ payload }),
371 compact: (repo: RepoMetaData) =>
372 operations.repos.compact({ params: path(repo) }),
373 },
374 refs: {
375 list: (repo: RepoMetaData) => repos.getByName(repo.repoId).listRefs(),
376 get: (repo: RepoMetaData, name: string) =>
377 repos.getByName(repo.repoId).getRef(name),
378 update: (
379 repo: RepoMetaData,
380 input: Parameters<ReturnType<typeof repos.getByName>["updateRef"]>[0],
381 ) => repos.getByName(repo.repoId).updateRef(input),
382 remove: (
383 repo: RepoMetaData,
384 input: Parameters<ReturnType<typeof repos.getByName>["removeRef"]>[0],
385 ) => repos.getByName(repo.repoId).removeRef(input),
386 },
387 pulls: {
388 create: (
389 repo: RepoMetaData,
390 payload: Parameters<typeof operations.pulls.create>[0]["payload"],
391 ) => operations.pulls.create({ params: path(repo), payload }),
392 list: (
393 repo: RepoMetaData,
394 query: Parameters<typeof operations.pulls.list>[0]["query"],
395 ) => operations.pulls.list({ params: path(repo), query }),
396 update: (
397 repo: RepoMetaData,
398 number: number,
399 payload: Parameters<typeof operations.pulls.update>[0]["payload"],
400 ) =>
401 operations.pulls.update({ params: { ...path(repo), number }, payload }),
402 get: (repo: RepoMetaData, number: number) =>
403 repos.getByName(repo.repoId).getPull(number),
404 merge: (
405 repo: RepoMetaData,
406 input: Parameters<ReturnType<typeof repos.getByName>["mergePull"]>[0],
407 ) => repos.getByName(repo.repoId).mergePull(input),
408 },
409 objects: {
410 commit: (repo: RepoMetaData, oid: string) =>
411 operations.objects.commit({
412 params: { ...path(repo), oid: oid as import("./Api.ts").Oid },
413 }),
414 file: (
415 repo: RepoMetaData,
416 input: Parameters<
417 ReturnType<typeof repos.getByName>["readFileAtPath"]
418 >[0],
419 ) => repos.getByName(repo.repoId).readFileAtPath(input),
420 log: (
421 repo: RepoMetaData,
422 query: Parameters<typeof operations.objects.log>[0]["query"],
423 ) => operations.objects.log({ params: path(repo), query }),
424 tree: (
425 repo: RepoMetaData,
426 oid: Parameters<typeof operations.objects.tree>[0]["params"]["oid"],
427 ) => operations.objects.tree({ params: { ...path(repo), oid } }),
428 diff: (
429 repo: RepoMetaData,
430 oid: Parameters<typeof operations.objects.diff>[0]["params"]["oid"],
431 ) => operations.objects.diff({ params: { ...path(repo), oid } }),
432 compare: (
433 repo: RepoMetaData,
434 query: Parameters<typeof operations.objects.compare>[0]["query"],
435 ) => operations.objects.compare({ params: path(repo), query }),
436 read: (repo: RepoMetaData, oid: string) =>
437 repos.getByName(repo.repoId).readPushBase(oid),
438 },
439 preparePush,
440 commitPush,
441 prepareRefUpdate,
442 prepareRefRemoval,
443 prepareMerge,
444 };
445});
446
447/** Git operations usable from HTTP handlers, jobs, and ordinary Effects. */
448export class Engine extends Context.Service<
449 Engine,
450 Effect.Success<typeof make>
451>()("alchemy/Git/Engine") {}
452
453/**
454 * Provides the Git engine using the application's repository, registry, blob, and hasher layers.
455 *
456 * @layer
457 * @provides Git.Engine
458 */
459export const EngineLive = Layer.effect(Engine, make).pipe(
460 Layer.provideMerge(OperationsLive),
461);