1/**
2 * `Git.BlobStore` — the swappable bulk-byte store (RFC "Git Building
3 * Blocks" §3.1).
4 *
5 * Immutable, content-addressed bulk bytes: compacted packs, clone
6 * bundles, oversize loose objects, and spilled push bodies. The contract
7 * is everything the pack plane needs and nothing more — ranged reads,
8 * whole-object writes with a known length, multipart writes of unknown
9 * length, delete, and prefix listing (GC only, never on a serving path).
10 *
11 * Implementations shipped:
12 *
13 * - {@link BlobStoreR2} — Cloudflare R2 over the existing
14 * `ReadWriteBucket` capability (the default).
15 *
16 * Multipart rules are the R2/S3 intersection: uniform part sizes
17 * (≥ 5 MiB) except the last part.
18 *
19 * ### Providing the store
20 * **Example:** R2 (the default)
21 * ```typescript
22 * const GitLive = Git.ApiLive.pipe(
23 * Layer.provide(Git.ApiHandlersLive),
24 * Layer.provide(Git.BlobStoreR2(MyBucket)),
25 * // ...
26 * );
27 * ```
28 */
29import * as Cloudflare from "../Cloudflare/index.ts";
30import type { R2Error, ReadWriteBucketClient } from "../Cloudflare/R2/index.ts";
31import type { RuntimeContext } from "../RuntimeContext.ts";
32import * as Context from "effect/Context";
33import * as Data from "effect/Data";
34import * as Effect from "effect/Effect";
35import * as Layer from "effect/Layer";
36import * as Option from "effect/Option";
37import * as Stream from "effect/Stream";
38
39/** Failure of a {@link BlobStore} operation. */
40export class BlobStoreError extends Data.TaggedError("BlobStoreError")<{
41 readonly reason: string;
42}> {}
43
44/** A read blob: size plus the bytes as a one-shot buffer or a stream. */
45export interface BlobBody {
46 readonly size: number;
47 readonly bytes: Effect.Effect<Uint8Array, BlobStoreError>;
48 readonly stream: Stream.Stream<Uint8Array, BlobStoreError>;
49 /**
50 * The store's native body stream when it has one, for serving paths
51 * that pipe bytes to the client with no per-chunk Effect work (DESIGN
52 * §22). Consume either this or `stream`, never both.
53 */
54 readonly readable?: ReadableStream<Uint8Array> | undefined;
55}
56
57/** One uploaded part of a multipart write, as the store identifies it. */
58export interface UploadedPart {
59 readonly partNumber: number;
60 readonly etag: string;
61}
62
63/**
64 * An in-flight multipart write. Parts must be uniform in size (≥ 5 MiB)
65 * except the last — the R2/S3 intersection. Parts may be uploaded from
66 * ANY isolate that knows the key and `uploadId` (`BlobStoreShape.uploadPart`
67 * — the push pipeline's hasher isolates do, DESIGN §22.10); the creator
68 * completes with the collected parts, in any order.
69 */
70export interface BlobMultipart {
71 readonly uploadId: string;
72 readonly uploadPart: (
73 partNumber: number,
74 part: Uint8Array,
75 ) => Effect.Effect<UploadedPart, BlobStoreError>;
76 readonly complete: (
77 parts: ReadonlyArray<UploadedPart>,
78 ) => Effect.Effect<void, BlobStoreError>;
79 readonly abort: Effect.Effect<void, BlobStoreError>;
80}
81
82/** Sorts parts by number: stores validate the uniform-part rule against the list as given. */
83export const orderedParts = (
84 parts: ReadonlyArray<UploadedPart>,
85): Array<UploadedPart> =>
86 [...parts].sort((a, b) => a.partNumber - b.partNumber);
87
88/** One listed blob (GC/purge walks). */
89export interface BlobMeta {
90 readonly key: string;
91 readonly size: number;
92}
93
94/** The service shape — see the module doc for semantics. */
95export interface BlobStoreShape {
96 /** Ranged read; omit `range` for the whole object. `null` = missing. */
97 readonly get: (
98 key: string,
99 range?: { readonly offset: number; readonly length: number },
100 ) => Effect.Effect<BlobBody | null, BlobStoreError, RuntimeContext>;
101 /** Whole-object write. `contentLength` is required for streams. */
102 readonly put: (
103 key: string,
104 body: Uint8Array | Stream.Stream<Uint8Array, BlobStoreError>,
105 options?: { readonly contentLength?: number | undefined },
106 ) => Effect.Effect<void, BlobStoreError, RuntimeContext>;
107 /** Existence/size check without reading the body. */
108 readonly head: (
109 key: string,
110 ) => Effect.Effect<BlobMeta | null, BlobStoreError, RuntimeContext>;
111 /** Begin a multipart write (spilled push bodies of unknown length). */
112 readonly multipart: (
113 key: string,
114 ) => Effect.Effect<BlobMultipart, BlobStoreError, RuntimeContext>;
115 /**
116 * Uploads one part of a multipart write created elsewhere (by key and
117 * `uploadId`) — the resume path the push pipeline's hasher isolates use.
118 */
119 readonly uploadPart: (
120 key: string,
121 uploadId: string,
122 partNumber: number,
123 part: Uint8Array,
124 ) => Effect.Effect<UploadedPart, BlobStoreError, RuntimeContext>;
125 readonly delete: (
126 keys: string | ReadonlyArray<string>,
127 ) => Effect.Effect<void, BlobStoreError, RuntimeContext>;
128 /** Prefix listing — GC/purge only, never on a serving path. */
129 readonly list: (
130 prefix: string,
131 ) => Stream.Stream<BlobMeta, BlobStoreError, RuntimeContext>;
132}
133
134/**
135 * The swappable bulk-byte store of the git service.
136 *
137 * @binding
138 */
139export class BlobStore extends Context.Service<BlobStore, BlobStoreShape>()(
140 "alchemy/Git/BlobStore",
141) {}
142
143// ─────────────────────────────────────────────────────────────────────────────
144// R2 implementation
145// ─────────────────────────────────────────────────────────────────────────────
146
147const r2Error = (what: string) => (error: R2Error) =>
148 new BlobStoreError({ reason: `${what}: ${error.message}` });
149
150/**
151 * Builds the R2-backed shape from an already-resolved bucket client.
152 * Used directly by the Repo DO (which resolves its own binding) and by
153 * the {@link BlobStoreR2} layer.
154 */
155export const makeBlobStoreR2 = (
156 bucket: ReadWriteBucketClient,
157): BlobStoreShape => ({
158 get: (key, range) =>
159 bucket.get(key, range === undefined ? undefined : { range }).pipe(
160 Effect.mapError(r2Error(`get ${key}`)),
161 Effect.map((object) =>
162 object === null
163 ? null
164 : ({
165 size: object.size,
166 bytes: object
167 .bytes()
168 .pipe(Effect.mapError(r2Error(`read ${key}`))),
169 stream: object.body.pipe(
170 Stream.mapError(r2Error(`stream ${key}`)),
171 ),
172 readable: object.readable,
173 } satisfies BlobBody),
174 ),
175 ),
176 put: (key, body, options) => {
177 // A stream body's own BlobStoreError joins R2's error union — map
178 // only the R2 side, pass ours through.
179 const write: Effect.Effect<
180 unknown,
181 R2Error | BlobStoreError,
182 RuntimeContext
183 > =
184 body instanceof Uint8Array
185 ? bucket.put(key, body)
186 : bucket.put(key, body, {
187 contentLength: options?.contentLength ?? 0,
188 });
189 return write.pipe(
190 Effect.mapError((error) =>
191 error instanceof BlobStoreError ? error : r2Error(`put ${key}`)(error),
192 ),
193 Effect.asVoid,
194 );
195 },
196 head: (key) =>
197 bucket.head(key).pipe(
198 Effect.mapError(r2Error(`head ${key}`)),
199 Effect.map((object) =>
200 object === null ? null : { key, size: object.size },
201 ),
202 ),
203 multipart: (key) =>
204 bucket.createMultipartUpload(key).pipe(
205 Effect.mapError(r2Error(`multipart ${key}`)),
206 Effect.map((upload): BlobMultipart => ({
207 uploadId: upload.uploadId,
208 uploadPart: (partNumber, part) =>
209 upload
210 .uploadPart(partNumber, part)
211 .pipe(Effect.mapError(r2Error(`part ${partNumber} of ${key}`))),
212 complete: (parts) =>
213 upload
214 .complete(orderedParts(parts))
215 .pipe(Effect.mapError(r2Error(`complete ${key}`)), Effect.asVoid),
216 abort: upload
217 .abort()
218 .pipe(Effect.mapError(r2Error(`abort ${key}`)), Effect.asVoid),
219 })),
220 ),
221 uploadPart: (key, uploadId, partNumber, part) =>
222 bucket.resumeMultipartUpload(key, uploadId).pipe(
223 Effect.flatMap((upload) => upload.uploadPart(partNumber, part)),
224 Effect.mapError(r2Error(`part ${partNumber} of ${key}`)),
225 ),
226 delete: (keys) =>
227 bucket
228 .delete(typeof keys === "string" ? keys : [...keys])
229 .pipe(Effect.mapError(r2Error("delete")), Effect.asVoid),
230 list: (prefix) =>
231 Stream.paginate(undefined as string | undefined, (cursor) =>
232 bucket.list({ prefix, limit: 1000, cursor }).pipe(
233 Effect.mapError(r2Error(`list ${prefix}`)),
234 Effect.map((page) => {
235 const metas = page.objects.map((object): BlobMeta => ({
236 key: object.key,
237 size: object.size,
238 }));
239 return [
240 metas,
241 page.truncated ? Option.some(page.cursor) : Option.none<string>(),
242 ] as const;
243 }),
244 ),
245 ),
246});
247
248/**
249 * R2-backed {@link BlobStore}: packs, bundles, oversize objects, and
250 * push spill live in the given bucket. Registers the bucket binding on
251 * the host Worker.
252 *
253 * @layer
254 * @provides Git.BlobStore
255 */
256export const BlobStoreR2 = (
257 bucket: Parameters<typeof Cloudflare.R2.ReadWriteBucket>[0],
258): Layer.Layer<BlobStore> =>
259 Layer.effect(
260 BlobStore,
261 Effect.gen(function* () {
262 const client = yield* Cloudflare.R2.ReadWriteBucket(bucket);
263 return makeBlobStoreR2(client);
264 }),
265 ).pipe(Layer.provide(Cloudflare.R2.ReadWriteBucketBinding)) as never;