1/**
2 * `Git.BlobStoreS3` — AWS S3 implementation of {@link BlobStore}.
3 *
4 * Runs the git service's bulk bytes (packs, clone bundles, oversize
5 * objects, push spill) on an S3 bucket while compute stays on Cloudflare.
6 * It is built over the S3 bindings (`AWS.S3.GetObject`, `PutObject`, …),
7 * so one provision serves the Worker and the Repo DO alike, exactly like
8 * {@link BlobStoreR2}: the bucket is an `AWS.S3.Bucket` resource, the
9 * Worker gets a least-privilege IAM identity minted for it, and every
10 * request is signed with credentials assumed at runtime, in the bucket's
11 * region.
12 *
13 * Latency note: every blob operation crosses the internet to S3 (no
14 * Cloudflare-internal path), so serving-path reads are slower than R2.
15 * The window-coalescing in `ObjectStore`/`PackSource` keeps request
16 * counts low, but R2 remains the recommended default for
17 * Cloudflare-hosted assemblies.
18 *
19 * ### Providing the store
20 * **Example:** Packs on S3, compute on Cloudflare
21 * ```typescript
22 * const GitObjects = AWS.S3.Bucket("GitObjects");
23 *
24 * const GitLive = Git.ApiLive.pipe(
25 * Layer.provide(Git.ApiHandlersLive),
26 * Layer.provide(Git.ReposDurableObject),
27 * Layer.provide(Git.RegistryDurableObject),
28 * Layer.provide(Git.BlobStoreS3(GitObjects)),
29 * );
30 * ```
31 */
32import * as Effect from "effect/Effect";
33import * as Layer from "effect/Layer";
34import * as Option from "effect/Option";
35import * as Stream from "effect/Stream";
36import { AbortMultipartUpload } from "../AWS/S3/AbortMultipartUpload.ts";
37import { AbortMultipartUploadHttp } from "../AWS/S3/AbortMultipartUploadHttp.ts";
38import { CompleteMultipartUpload } from "../AWS/S3/CompleteMultipartUpload.ts";
39import { CompleteMultipartUploadHttp } from "../AWS/S3/CompleteMultipartUploadHttp.ts";
40import { CreateMultipartUpload } from "../AWS/S3/CreateMultipartUpload.ts";
41import { CreateMultipartUploadHttp } from "../AWS/S3/CreateMultipartUploadHttp.ts";
42import { DeleteObjects } from "../AWS/S3/DeleteObjects.ts";
43import { DeleteObjectsHttp } from "../AWS/S3/DeleteObjectsHttp.ts";
44import { GetObject } from "../AWS/S3/GetObject.ts";
45import { GetObjectHttp } from "../AWS/S3/GetObjectHttp.ts";
46import { HeadObject } from "../AWS/S3/HeadObject.ts";
47import { HeadObjectHttp } from "../AWS/S3/HeadObjectHttp.ts";
48import { ListObjectsV2 } from "../AWS/S3/ListObjectsV2.ts";
49import { ListObjectsV2Http } from "../AWS/S3/ListObjectsV2Http.ts";
50import { PutObject } from "../AWS/S3/PutObject.ts";
51import { PutObjectHttp } from "../AWS/S3/PutObjectHttp.ts";
52import { UploadPart } from "../AWS/S3/UploadPart.ts";
53import { UploadPartHttp } from "../AWS/S3/UploadPartHttp.ts";
54import {
55 BlobStore,
56 BlobStoreError,
57 type BlobBody,
58 type BlobMeta,
59 orderedParts,
60 type BlobMultipart,
61 type BlobStoreShape,
62} from "./BlobStore.ts";
63
64const s3Error =
65 (what: string) =>
66 (error: { readonly _tag?: string; readonly message?: string }) =>
67 new BlobStoreError({
68 reason: `${what}: ${error._tag ?? "S3Error"}${
69 error.message ? `: ${error.message}` : ""
70 }`,
71 });
72
73const collectBytes = (
74 stream: Stream.Stream<Uint8Array, Error>,
75 what: string,
76): Effect.Effect<Uint8Array, BlobStoreError> =>
77 Stream.runCollect(stream).pipe(
78 Effect.mapError((error) => s3Error(what)({ message: String(error) })),
79 Effect.map((chunks) => {
80 const total = chunks.reduce((n, c) => n + c.length, 0);
81 const out = new Uint8Array(total);
82 let at = 0;
83 for (const chunk of chunks) {
84 out.set(chunk, at);
85 at += chunk.length;
86 }
87 return out;
88 }),
89 );
90
91/**
92 * S3-backed {@link BlobStore} using the given bucket. Accepts an
93 * `AWS.S3.Bucket` resource or its declaration Effect, like {@link BlobStoreR2}.
94 * See the module doc for identity and latency semantics.
95 *
96 * @layer
97 * @provides Git.BlobStore
98 */
99export const BlobStoreS3 = (
100 bucket: Parameters<typeof GetObject>[0],
101): Layer.Layer<BlobStore> =>
102 Layer.effect(
103 BlobStore,
104 Effect.gen(function* () {
105 const getObject = yield* GetObject(bucket);
106 const putObject = yield* PutObject(bucket);
107 const headObject = yield* HeadObject(bucket);
108 const deleteObjects = yield* DeleteObjects(bucket);
109 const listObjectsV2 = yield* ListObjectsV2(bucket);
110 const createMultipart = yield* CreateMultipartUpload(bucket);
111 const uploadPart = yield* UploadPart(bucket);
112 const completeMultipart = yield* CompleteMultipartUpload(bucket);
113 const abortMultipart = yield* AbortMultipartUpload(bucket);
114
115 return {
116 get: (key, range) =>
117 getObject({
118 Key: key,
119 Range:
120 range === undefined
121 ? undefined
122 : `bytes=${range.offset}-${range.offset + range.length - 1}`,
123 }).pipe(
124 Effect.map((output): BlobBody | null => {
125 const body = output.Body ?? Stream.empty;
126 return {
127 size: output.ContentLength ?? 0,
128 bytes: collectBytes(body, `get ${key}`),
129 stream: body.pipe(
130 Stream.mapError((error) =>
131 s3Error(`stream ${key}`)({ message: String(error) }),
132 ),
133 ),
134 };
135 }),
136 Effect.catchTag("NoSuchKey", () => Effect.succeed(null)),
137 Effect.mapError(s3Error(`get ${key}`)),
138 ),
139
140 put: (key, body, opts) =>
141 putObject({
142 Key: key,
143 Body: body,
144 ContentLength:
145 body instanceof Uint8Array ? body.length : opts?.contentLength,
146 }).pipe(Effect.mapError(s3Error(`put ${key}`)), Effect.asVoid),
147
148 head: (key) =>
149 headObject({ Key: key }).pipe(
150 Effect.map((output): BlobMeta => ({
151 key,
152 size: output.ContentLength ?? 0,
153 })),
154 Effect.catchTag("NotFound", () => Effect.succeed(null)),
155 Effect.mapError(s3Error(`head ${key}`)),
156 ),
157
158 multipart: (key) =>
159 createMultipart({ Key: key }).pipe(
160 Effect.mapError(s3Error(`multipart ${key}`)),
161 Effect.map((created): BlobMultipart => {
162 const UploadId = created.UploadId ?? "";
163 return {
164 uploadId: UploadId,
165 uploadPart: (partNumber, part) =>
166 uploadPart({
167 Key: key,
168 UploadId,
169 PartNumber: partNumber,
170 Body: part,
171 ContentLength: part.length,
172 }).pipe(
173 Effect.mapError(s3Error(`part ${partNumber} of ${key}`)),
174 Effect.map((uploaded) => ({
175 partNumber,
176 etag: uploaded.ETag ?? "",
177 })),
178 ),
179 complete: (parts) =>
180 completeMultipart({
181 Key: key,
182 UploadId,
183 MultipartUpload: {
184 Parts: orderedParts(parts).map((p) => ({
185 PartNumber: p.partNumber,
186 ETag: p.etag,
187 })),
188 },
189 }).pipe(
190 Effect.mapError(s3Error(`complete ${key}`)),
191 Effect.asVoid,
192 ),
193 abort: abortMultipart({ Key: key, UploadId }).pipe(
194 Effect.mapError(s3Error(`abort ${key}`)),
195 Effect.asVoid,
196 ),
197 };
198 }),
199 ),
200 uploadPart: (key, uploadId, partNumber, part) =>
201 uploadPart({
202 Key: key,
203 UploadId: uploadId,
204 PartNumber: partNumber,
205 Body: part,
206 ContentLength: part.length,
207 }).pipe(
208 Effect.mapError(s3Error(`part ${partNumber} of ${key}`)),
209 Effect.map((uploaded) => ({
210 partNumber,
211 etag: uploaded.ETag ?? "",
212 })),
213 ),
214 delete: (keys) => {
215 const list = typeof keys === "string" ? [keys] : [...keys];
216 if (list.length === 0) return Effect.void;
217 return deleteObjects({
218 Delete: {
219 Objects: list.map((Key) => ({ Key })),
220 Quiet: true,
221 },
222 }).pipe(Effect.mapError(s3Error("delete")), Effect.asVoid);
223 },
224
225 list: (prefix) =>
226 Stream.paginate(
227 undefined as string | undefined,
228 (ContinuationToken) =>
229 listObjectsV2({
230 Prefix: prefix,
231 ContinuationToken,
232 MaxKeys: 1000,
233 }).pipe(
234 Effect.mapError(s3Error(`list ${prefix}`)),
235 Effect.map((page) => {
236 const metas = (page.Contents ?? []).flatMap(
237 (object): Array<BlobMeta> =>
238 object.Key === undefined
239 ? []
240 : [{ key: object.Key, size: object.Size ?? 0 }],
241 );
242 return [
243 metas,
244 page.IsTruncated && page.NextContinuationToken
245 ? Option.some(page.NextContinuationToken)
246 : Option.none<string>(),
247 ] as const;
248 }),
249 ),
250 ),
251 } satisfies BlobStoreShape;
252 }),
253 ).pipe(
254 Layer.provide([
255 GetObjectHttp,
256 PutObjectHttp,
257 HeadObjectHttp,
258 DeleteObjectsHttp,
259 ListObjectsV2Http,
260 CreateMultipartUploadHttp,
261 UploadPartHttp,
262 CompleteMultipartUploadHttp,
263 AbortMultipartUploadHttp,
264 ]),
265 ) as never;