1import { Credentials } from "@distilled.cloud/gcp/Credentials";
2import type * as storage from "@distilled.cloud/gcp/storage_v1";
3import * as Data from "effect/Data";
4import * as Effect from "effect/Effect";
5import * as Schedule from "effect/Schedule";
6import * as HttpClient from "effect/http/HttpClient";
7import * as HttpClientRequest from "effect/http/HttpClientRequest";
8import type * as HttpClientResponse from "effect/http/HttpClientResponse";
9
10/**
11 * Object content transfer over the Cloud Storage JSON API media endpoints
12 * (`alt=media` download, `uploadType=multipart` upload). The generated
13 * storage SDK only covers the metadata surface — `objects.get` returns
14 * metadata and `objects.insert` without the `/upload` path is rejected —
15 * so content bindings speak these two endpoints directly.
16 *
17 * NOT exported from `index.ts`.
18 */
19
20const API = "https://storage.googleapis.com";
21
22/** The object (or bucket) does not exist. */
23export class ObjectNotFound extends Data.TaggedError(
24 "GCP.Storage.ObjectNotFound",
25)<{
26 bucket: string;
27 object: string;
28}> {}
29
30/** Cloud Storage rejected the request. */
31export class ObjectRequestFailed extends Data.TaggedError(
32 "GCP.Storage.ObjectRequestFailed",
33)<{
34 bucket: string;
35 object: string;
36 status: number;
37 message: string;
38}> {}
39
40export interface ObjectContent {
41 /** Raw object bytes. */
42 body: Uint8Array;
43 /** `Content-Type` the object was stored with. */
44 contentType: string | undefined;
45 /** Object generation served. */
46 generation: string | undefined;
47}
48
49export interface PutObjectContent {
50 /** Object name (key). */
51 name: string;
52 /** Object content. Strings are encoded as UTF-8. */
53 body: string | Uint8Array;
54 /** @default "application/octet-stream" (or "text/plain; charset=utf-8" for strings) */
55 contentType?: string;
56 /** Custom metadata stored on the object. */
57 metadata?: Record<string, string>;
58 /** `Cache-Control` served with the object. */
59 cacheControl?: string;
60 /** Only write when the live generation matches (`0` = only if absent). */
61 ifGenerationMatch?: string;
62}
63
64const isRetryable = (error: ObjectNotFound | ObjectRequestFailed) =>
65 error._tag === "GCP.Storage.ObjectRequestFailed" &&
66 (error.status === 429 || error.status >= 500);
67
68const failureOf = (
69 bucket: string,
70 object: string,
71 response: HttpClientResponse.HttpClientResponse,
72) =>
73 Effect.gen(function* () {
74 if (response.status === 404) {
75 return yield* new ObjectNotFound({ bucket, object });
76 }
77 const text = yield* response.text.pipe(Effect.orElseSucceed(() => ""));
78 return yield* new ObjectRequestFailed({
79 bucket,
80 object,
81 status: response.status,
82 message: text.slice(0, 500),
83 });
84 });
85
86const transportFailure =
87 (bucket: string, object: string) => (cause: { message?: string }) =>
88 new ObjectRequestFailed({
89 bucket,
90 object,
91 status: 0,
92 message: cause.message ?? String(cause),
93 });
94
95/**
96 * Build the download / upload callables once, closing over the ambient
97 * `Credentials` and `HttpClient` (resolved at Layer construction, like a
98 * distilled operation).
99 */
100export const makeObjectMedia = Effect.gen(function* () {
101 const credentials = yield* Credentials;
102 const http = yield* HttpClient.HttpClient;
103
104 const authorized = (request: HttpClientRequest.HttpClientRequest) =>
105 credentials.pipe(
106 Effect.map((config) =>
107 request.pipe(HttpClientRequest.bearerToken(config.accessToken)),
108 ),
109 );
110
111 const download = (options: {
112 bucket: string;
113 object: string;
114 generation?: string;
115 }) =>
116 Effect.gen(function* () {
117 const request = yield* authorized(
118 HttpClientRequest.get(
119 `${API}/storage/v1/b/${encodeURIComponent(options.bucket)}/o/${encodeURIComponent(options.object)}`,
120 ).pipe(
121 HttpClientRequest.setUrlParams({
122 alt: "media",
123 ...(options.generation === undefined
124 ? {}
125 : { generation: options.generation }),
126 }),
127 ),
128 );
129 const response = yield* http
130 .execute(request)
131 .pipe(
132 Effect.mapError(transportFailure(options.bucket, options.object)),
133 );
134 if (response.status !== 200) {
135 return yield* failureOf(options.bucket, options.object, response);
136 }
137 const body = yield* response.arrayBuffer.pipe(
138 Effect.mapError(transportFailure(options.bucket, options.object)),
139 );
140 return {
141 body: new Uint8Array(body),
142 contentType: response.headers["content-type"],
143 generation: response.headers["x-goog-generation"],
144 } satisfies ObjectContent;
145 }).pipe(
146 Effect.retry({
147 while: isRetryable,
148 times: 4,
149 schedule: Schedule.exponential("200 millis"),
150 }),
151 );
152
153 const upload = (bucket: string, options: PutObjectContent) =>
154 Effect.gen(function* () {
155 const bytes =
156 typeof options.body === "string"
157 ? new TextEncoder().encode(options.body)
158 : options.body;
159 const contentType =
160 options.contentType ??
161 (typeof options.body === "string"
162 ? "text/plain; charset=utf-8"
163 : "application/octet-stream");
164 const boundary = yield* Effect.sync(
165 () => `alchemy-${crypto.randomUUID()}`,
166 );
167 const metadata = JSON.stringify({
168 name: options.name,
169 contentType,
170 metadata: options.metadata,
171 cacheControl: options.cacheControl,
172 });
173 const encoder = new TextEncoder();
174 const head = encoder.encode(
175 `--${boundary}\r\nContent-Type: application/json; charset=UTF-8\r\n\r\n${metadata}\r\n` +
176 `--${boundary}\r\nContent-Type: ${contentType}\r\n\r\n`,
177 );
178 const tail = encoder.encode(`\r\n--${boundary}--\r\n`);
179 const payload = new Uint8Array(head.length + bytes.length + tail.length);
180 payload.set(head, 0);
181 payload.set(bytes, head.length);
182 payload.set(tail, head.length + bytes.length);
183
184 const request = yield* authorized(
185 HttpClientRequest.post(
186 `${API}/upload/storage/v1/b/${encodeURIComponent(bucket)}/o`,
187 ).pipe(
188 HttpClientRequest.setUrlParams({
189 uploadType: "multipart",
190 ...(options.ifGenerationMatch === undefined
191 ? {}
192 : { ifGenerationMatch: options.ifGenerationMatch }),
193 }),
194 HttpClientRequest.bodyUint8Array(
195 payload,
196 `multipart/related; boundary=${boundary}`,
197 ),
198 ),
199 );
200 const response = yield* http
201 .execute(request)
202 .pipe(Effect.mapError(transportFailure(bucket, options.name)));
203 if (response.status !== 200) {
204 return yield* failureOf(bucket, options.name, response);
205 }
206 const json = yield* response.json.pipe(
207 Effect.mapError(transportFailure(bucket, options.name)),
208 );
209 // The upload endpoint answers with the same `Object` resource
210 // `objects.insert` documents.
211 return json as storage.Storage_Object;
212 }).pipe(
213 Effect.retry({
214 while: isRetryable,
215 times: 4,
216 schedule: Schedule.exponential("200 millis"),
217 }),
218 );
219
220 return { download, upload };
221});