1/**
2 * Runtime half of the Prisma Object Store bindings: an S3-compatible client
3 * pointed at a bucket key's endpoint. Internal — the capability modules
4 * (`ReadBucket.ts`, `WriteBucket.ts`, `ReadWriteBucket.ts`) build their public
5 * clients on top of these primitives.
6 */
7import { Credentials, Endpoint, Presign, Region } from "@distilled.cloud/aws";
8import type * as S3 from "@distilled.cloud/aws/s3";
9import * as Effect from "effect/Effect";
10import * as Layer from "effect/Layer";
11import * as Redacted from "effect/Redacted";
12import * as Stream from "effect/Stream";
13import * as FetchHttpClient from "effect/http/FetchHttpClient";
14import type { HttpClient } from "effect/http/HttpClient";
15import type { RuntimeContext } from "../../RuntimeContext.ts";
16import {
17 BucketError,
18 type BucketCredentials,
19 type BucketObject,
20 type BucketObjectBody,
21 type BucketRange,
22} from "../BucketTypes.ts";
23
24/**
25 * SigV4 credential scope used for Prisma Object Store requests. Prisma
26 * addresses buckets through a single regionless endpoint and does not report
27 * a region on a bucket key, so requests are signed under the `auto` scope
28 * that S3-compatible stores conventionally accept.
29 */
30export const BUCKET_SIGNING_REGION = "auto";
31
32export type PresignRequest =
33 | {
34 method: "GET";
35 key: string;
36 expiresIn?: number | undefined;
37 /** Signed `response-content-type` override for the download. */
38 responseContentType?: string | undefined;
39 }
40 | {
41 method: "PUT";
42 key: string;
43 expiresIn?: number | undefined;
44 /** `Content-Type` header the uploader must send, signed into the URL. */
45 contentType?: string | undefined;
46 };
47
48/**
49 * The transport a capability client is built against: the resolved bucket
50 * name, a way to run a distilled S3 operation with the bucket key's
51 * credentials, and SigV4 query-string presigning against the same endpoint.
52 */
53export interface BucketAccess {
54 bucketName: Effect.Effect<string, never, RuntimeContext>;
55 authorize: <A, E>(
56 effect: Effect.Effect<
57 A,
58 E,
59 Credentials.Credentials | Region.Region | HttpClient
60 >,
61 ) => Effect.Effect<A, E, RuntimeContext>;
62 presign: (
63 request: PresignRequest,
64 ) => Effect.Effect<string, BucketError, RuntimeContext>;
65}
66
67export const toBucketError = (error: unknown): BucketError =>
68 new BucketError({
69 message:
70 typeof error === "object" && error !== null && "message" in error
71 ? String((error as { message: unknown }).message)
72 : "Unknown Prisma Object Store error",
73 cause: error instanceof Error ? error : new Error(String(error)),
74 });
75
76const signingContext = (credentials: BucketCredentials) =>
77 Effect.all([
78 credentials.endpoint,
79 credentials.accessKeyId,
80 credentials.secretAccessKey,
81 ]).pipe(
82 Effect.map(([endpoint, accessKeyId, secretAccessKey]) =>
83 Layer.mergeAll(
84 Layer.succeed(
85 Credentials.Credentials,
86 Effect.succeed({
87 accessKeyId: Redacted.make(accessKeyId),
88 secretAccessKey,
89 sessionToken: undefined,
90 region: BUCKET_SIGNING_REGION,
91 }),
92 ),
93 Layer.succeed(Region.Region, Effect.succeed(BUCKET_SIGNING_REGION)),
94 Layer.succeed(Endpoint.Endpoint, Effect.succeed(endpoint)),
95 FetchHttpClient.layer,
96 ),
97 ),
98 );
99
100/**
101 * Build the transport for a bound bucket key. Requests are signed with the
102 * key's own credentials against its endpoint; because a custom endpoint is
103 * set, the S3 client addresses the bucket path-style rather than through the
104 * AWS virtual-host rules.
105 */
106export const makeBucketAccess = (
107 credentials: BucketCredentials,
108): BucketAccess => {
109 const context = signingContext(credentials);
110 return {
111 bucketName: credentials.bucketName,
112 authorize: (effect) =>
113 context.pipe(Effect.flatMap((layer) => Effect.provide(effect, layer))),
114 presign: (request) =>
115 Effect.all([context, credentials.bucketName]).pipe(
116 Effect.flatMap(([layer, bucket]) =>
117 Presign.presignS3Url({
118 method: request.method,
119 bucket,
120 key: request.key,
121 region: BUCKET_SIGNING_REGION,
122 expiresIn: request.expiresIn,
123 contentType:
124 request.method === "PUT" ? request.contentType : undefined,
125 responseContentType:
126 request.method === "GET"
127 ? request.responseContentType
128 : undefined,
129 }).pipe(Effect.provide(layer)),
130 ),
131 Effect.mapError(toBucketError),
132 ),
133 };
134};
135
136/** Render a {@link BucketRange} as an HTTP `Range` header value. */
137export const rangeHeader = (range: BucketRange | undefined) => {
138 if (range === undefined) return undefined;
139 const end =
140 range.length === undefined ? "" : String(range.offset + range.length - 1);
141 return `bytes=${range.offset}-${end}`;
142};
143
144const stripQuotes = (etag: string | undefined) =>
145 etag === undefined ? "" : etag.replaceAll('"', "");
146
147const definedMetadata = (
148 metadata: { [key: string]: string | undefined } | undefined,
149): Record<string, string> => {
150 const result: Record<string, string> = {};
151 for (const [key, value] of Object.entries(metadata ?? {})) {
152 if (value !== undefined) result[key] = value;
153 }
154 return result;
155};
156
157export const objectFrom = (
158 key: string,
159 attributes: {
160 ContentLength?: number | undefined;
161 ETag?: string | undefined;
162 LastModified?: Date | undefined;
163 ContentType?: string | undefined;
164 Metadata?: { [key: string]: string | undefined } | undefined;
165 },
166): BucketObject => ({
167 key,
168 size: attributes.ContentLength ?? 0,
169 etag: stripQuotes(attributes.ETag),
170 lastModified: attributes.LastModified,
171 contentType: attributes.ContentType,
172 metadata: definedMetadata(attributes.Metadata),
173});
174
175export const objectFromListEntry = (entry: S3.Object): BucketObject => ({
176 key: entry.Key ?? "",
177 size: entry.Size ?? 0,
178 etag: stripQuotes(entry.ETag),
179 lastModified: entry.LastModified,
180 contentType: undefined,
181 metadata: {},
182});
183
184const concatBytes = (chunks: readonly Uint8Array[]): Uint8Array => {
185 const total = chunks.reduce((sum, chunk) => sum + chunk.byteLength, 0);
186 const result = new Uint8Array(total);
187 let offset = 0;
188 for (const chunk of chunks) {
189 result.set(chunk, offset);
190 offset += chunk.byteLength;
191 }
192 return result;
193};
194
195/**
196 * Wrap a `GetObject` response as a {@link BucketObjectBody}. The body is a
197 * single-consumption stream, so `body` and the buffering accessors are three
198 * views of the same bytes and only one of them may be read.
199 */
200export const objectBodyFrom = (
201 key: string,
202 response: S3.GetObjectOutput,
203): BucketObjectBody => {
204 const body = (response.Body ?? Stream.empty).pipe(
205 Stream.mapError(toBucketError),
206 );
207 const bytes = () => Stream.runCollect(body).pipe(Effect.map(concatBytes));
208 const text = () => Stream.mkString(Stream.decodeText(body));
209 return {
210 ...objectFrom(key, response),
211 body,
212 bytes,
213 arrayBuffer: () =>
214 bytes().pipe(
215 Effect.map((collected) => {
216 const buffer = new ArrayBuffer(collected.byteLength);
217 new Uint8Array(buffer).set(collected);
218 return buffer;
219 }),
220 ),
221 text,
222 json: <T>() =>
223 text().pipe(
224 Effect.flatMap((decoded) =>
225 Effect.try({
226 try: () => JSON.parse(decoded) as T,
227 catch: toBucketError,
228 }),
229 ),
230 ),
231 };
232};