1import * as Credentials from "@distilled.cloud/aws/Credentials";
2import * as Endpoint from "@distilled.cloud/aws/Endpoint";
3import { presignS3Url } from "@distilled.cloud/aws/Presign";
4import * as Region from "@distilled.cloud/aws/Region";
5import * as S3 from "@distilled.cloud/aws/s3";
6import * as Data from "effect/Data";
7import * as Effect from "effect/Effect";
8import * as Layer from "effect/Layer";
9import * as Redacted from "effect/Redacted";
10import * as HttpClient from "effect/http/HttpClient";
11
12export interface StorageConfig {
13 /** Branch-specific, path-style S3 endpoint. */
14 endpoint: string;
15 /** Signing region returned by Neon. */
16 region: string;
17 /** Branch credential token ID, not an AWS account key. */
18 accessKeyId: string;
19 /** Branch credential S3 secret. */
20 secretAccessKey: Redacted.Redacted<string>;
21}
22
23export class StoragePaginationError extends Data.TaggedError(
24 "StoragePaginationError",
25)<{ message: string }> {}
26export class StorageDeleteError extends Data.TaggedError("StorageDeleteError")<{
27 message: string;
28}> {}
29
30/** Internal S3 transport. Endpoint overrides retain the bucket in the request path. */
31export const storageLayer = (config: StorageConfig) =>
32 Layer.mergeAll(
33 Endpoint.of(config.endpoint),
34 Region.of(config.region as Region.RegionName),
35 Layer.succeed(
36 Credentials.Credentials,
37 Effect.succeed({
38 accessKeyId: Redacted.make(config.accessKeyId),
39 secretAccessKey: config.secretAccessKey,
40 sessionToken: undefined,
41 region: config.region as Region.RegionName,
42 }),
43 ),
44 );
45
46export type StoragePutOptions = Pick<
47 S3.PutObjectRequest,
48 | "ContentType"
49 | "CacheControl"
50 | "ContentDisposition"
51 | "ContentEncoding"
52 | "ContentLanguage"
53 | "Metadata"
54>;
55
56/** Construct clients once; no connections or disposable resources are retained. */
57export const makeStorageClient = Effect.fn(function* (
58 config: StorageConfig,
59 bucket: string,
60) {
61 const http = yield* HttpClient.HttpClient;
62 const layer = Layer.mergeAll(
63 storageLayer(config),
64 Layer.succeed(HttpClient.HttpClient, http),
65 );
66 const provide = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
67 effect.pipe(Effect.provide(layer));
68 return {
69 get: (key: string) =>
70 provide(S3.getObject({ Bucket: bucket, Key: key })).pipe(
71 Effect.catchTag("NoSuchKey", () => Effect.succeed(undefined)),
72 ),
73 head: (key: string) =>
74 provide(S3.headObject({ Bucket: bucket, Key: key })).pipe(
75 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
76 ),
77 list: (
78 options: {
79 prefix?: string;
80 delimiter?: string;
81 cursor?: string;
82 limit?: number;
83 } = {},
84 ) =>
85 provide(
86 S3.listObjectsV2({
87 Bucket: bucket,
88 Prefix: options.prefix,
89 Delimiter: options.delimiter,
90 ContinuationToken: options.cursor,
91 MaxKeys: options.limit,
92 }),
93 ),
94 put: (
95 key: string,
96 body: string | Uint8Array,
97 options: StoragePutOptions = {},
98 ) =>
99 provide(
100 S3.putObject({ Bucket: bucket, Key: key, Body: body, ...options }),
101 ),
102 delete: (key: string) =>
103 provide(S3.deleteObject({ Bucket: bucket, Key: key })),
104 deleteMany: (keys: string[]) =>
105 Effect.gen(function* () {
106 const deleted: S3.DeletedObject[] = [];
107 // Large Neon batches can exceed the gateway's 60-second request limit.
108 for (let offset = 0; offset < keys.length; offset += 100) {
109 const result = yield* provide(
110 S3.deleteObjects({
111 Bucket: bucket,
112 Delete: {
113 Objects: keys
114 .slice(offset, offset + 100)
115 .map((Key) => ({ Key })),
116 },
117 }),
118 );
119 if (result.Errors?.length)
120 return yield* new StorageDeleteError({
121 message: `S3 refused ${result.Errors.length} object deletions`,
122 });
123 deleted.push(...(result.Deleted ?? []));
124 }
125 return { Deleted: deleted } satisfies S3.DeleteObjectsOutput;
126 }),
127 createMultipartUpload: (key: string, options: StoragePutOptions = {}) =>
128 provide(
129 S3.createMultipartUpload({ Bucket: bucket, Key: key, ...options }),
130 ),
131 uploadPart: (
132 key: string,
133 uploadId: string,
134 partNumber: number,
135 body: Uint8Array,
136 ) =>
137 provide(
138 S3.uploadPart({
139 Bucket: bucket,
140 Key: key,
141 UploadId: uploadId,
142 PartNumber: partNumber,
143 Body: body,
144 }),
145 ),
146 completeMultipartUpload: (
147 key: string,
148 uploadId: string,
149 parts: S3.CompletedPart[],
150 ) =>
151 provide(
152 S3.completeMultipartUpload({
153 Bucket: bucket,
154 Key: key,
155 UploadId: uploadId,
156 MultipartUpload: { Parts: parts },
157 }),
158 ),
159 abortMultipartUpload: (key: string, uploadId: string) =>
160 provide(
161 S3.abortMultipartUpload({
162 Bucket: bucket,
163 Key: key,
164 UploadId: uploadId,
165 }),
166 ).pipe(Effect.catchTag("NoSuchUpload", () => Effect.void)),
167 listMultipartUploads: (keyMarker?: string, uploadIdMarker?: string) =>
168 provide(
169 S3.listMultipartUploads({
170 Bucket: bucket,
171 KeyMarker: keyMarker,
172 UploadIdMarker: uploadIdMarker,
173 }),
174 ),
175 listParts: (key: string, uploadId: string, partNumberMarker?: string) =>
176 provide(
177 S3.listParts({
178 Bucket: bucket,
179 Key: key,
180 UploadId: uploadId,
181 PartNumberMarker: partNumberMarker,
182 }),
183 ),
184 presign: (
185 key: string,
186 method: "GET" | "PUT",
187 options: { expiresIn?: number; contentType?: string } = {},
188 ) => provide(presignS3Url({ bucket, key, method, ...options })),
189 getTags: () =>
190 provide(S3.getBucketTagging({ Bucket: bucket })).pipe(
191 Effect.catchTag("NoSuchTagSet", () => Effect.succeed({ TagSet: [] })),
192 ),
193 putTags: (tags: Record<string, string>) =>
194 provide(
195 S3.putBucketTagging({
196 Bucket: bucket,
197 Tagging: {
198 TagSet: Object.entries(tags).map(([Key, Value]) => ({
199 Key,
200 Value,
201 })),
202 },
203 }),
204 ),
205 getCors: () =>
206 provide(S3.getBucketCors({ Bucket: bucket })).pipe(
207 Effect.catchTag("NoSuchCORSConfiguration", () =>
208 Effect.succeed({ CORSRules: [] }),
209 ),
210 ),
211 putCors: (rules: S3.CORSRule[]) =>
212 provide(
213 S3.putBucketCors({
214 Bucket: bucket,
215 CORSConfiguration: { CORSRules: rules },
216 }),
217 ),
218 deleteCors: () => provide(S3.deleteBucketCors({ Bucket: bucket })),
219 deleteBucket: () =>
220 provide(S3.deleteBucket({ Bucket: bucket })).pipe(
221 Effect.catchTag("NoSuchBucket", () => Effect.void),
222 ),
223 };
224});
225
226export type StorageClient = Effect.Success<
227 ReturnType<typeof makeStorageClient>
228>;
229
230/** Empty only this bucket and branch, checking pagination and partial failures. */
231export const emptyStorageBucket = Effect.fn(function* (client: StorageClient) {
232 let cursor: string | undefined;
233 const seen = new Set<string>();
234 do {
235 const page = yield* client.list({ cursor, limit: 1000 });
236 const keys = (page.Contents ?? []).flatMap((object) =>
237 object.Key === undefined ? [] : [object.Key],
238 );
239 if (keys.length) yield* client.deleteMany(keys);
240 cursor = page.IsTruncated ? page.NextContinuationToken : undefined;
241 if (page.IsTruncated && (!cursor || seen.has(cursor))) {
242 return yield* new StoragePaginationError({
243 message: "Object listing did not advance",
244 });
245 }
246 if (cursor) seen.add(cursor);
247 } while (cursor);
248 let keyMarker: string | undefined;
249 let uploadMarker: string | undefined;
250 const uploadsSeen = new Set<string>();
251 for (;;) {
252 const page = yield* client.listMultipartUploads(keyMarker, uploadMarker);
253 for (const upload of page.Uploads ?? []) {
254 if (upload.Key && upload.UploadId)
255 yield* client.abortMultipartUpload(upload.Key, upload.UploadId);
256 }
257 if (!page.IsTruncated) break;
258 keyMarker = page.NextKeyMarker;
259 uploadMarker = page.NextUploadIdMarker;
260 const marker = `${keyMarker ?? ""}/${uploadMarker ?? ""}`;
261 if (!keyMarker || uploadsSeen.has(marker)) {
262 return yield* new StoragePaginationError({
263 message: "Multipart listing did not advance",
264 });
265 }
266 uploadsSeen.add(marker);
267 }
268});