1import * as Cloudflare from "@/Cloudflare";
2import * as Test from "@/Test/Alchemy";
3import { expect } from "alchemy-test";
4import * as Data from "effect/Data";
5import * as Effect from "effect/Effect";
6import { MinimumLogLevel } from "effect/References";
7import * as Schedule from "effect/Schedule";
8import * as HttpClient from "effect/http/HttpClient";
9import * as HttpClientRequest from "effect/http/HttpClientRequest";
10import type * as HttpClientResponse from "effect/http/HttpClientResponse";
11import { createHash } from "node:crypto";
12import Stack from "./fixtures/stack.ts";
13
14const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
15 providers: Cloudflare.providers(),
16 state: Cloudflare.state(),
17});
18
19const HOOK_TIMEOUT = 300_000;
20const TEST_TIMEOUT = 120_000;
21
22const logLevel = Effect.provideService(
23 MinimumLogLevel,
24 process.env.DEBUG ? "Debug" : "Info",
25);
26
27class WorkerNotReady extends Data.TaggedError("WorkerNotReady")<{
28 status: number;
29 body: string;
30}> {}
31
32// Bounded spaced schedule — caps total wait so a genuine failure
33// surfaces fast instead of an uncapped exponential blowing past the
34// test timeout while riding out cold-start propagation.
35const ready = Schedule.max([Schedule.spaced("2 seconds"), Schedule.recurs(30)]);
36
37/** Retry an HTTP call until it returns 200 (rides out cold-start 404s). */
38const untilOk = <E, R>(
39 eff: Effect.Effect<HttpClientResponse.HttpClientResponse, E, R>,
40) =>
41 eff.pipe(
42 Effect.flatMap((res) =>
43 res.status === 200
44 ? Effect.succeed(res)
45 : res.text.pipe(
46 Effect.flatMap((body) =>
47 Effect.fail(new WorkerNotReady({ status: res.status, body })),
48 ),
49 ),
50 ),
51 Effect.retry({
52 while: (e): e is WorkerNotReady => e instanceof WorkerNotReady,
53 schedule: ready,
54 }),
55 );
56
57class ValueMismatch extends Data.TaggedError("ValueMismatch")<{
58 expected: string;
59 actual: string | null;
60}> {}
61
62const retryMismatch = <A, E, R>(
63 eff: Effect.Effect<A, E, R>,
64): Effect.Effect<A, E, R> =>
65 eff.pipe(
66 Effect.retry({
67 while: (e: E) => e instanceof ValueMismatch,
68 schedule: ready,
69 }),
70 );
71
72/** GET `${base}/get?key=` and retry until the value matches (read-after-write propagation). */
73const expectValue = (base: string, key: string, expected: string) =>
74 untilOk(HttpClient.get(`${base}/get?key=${encodeURIComponent(key)}`)).pipe(
75 Effect.flatMap((res) => res.json),
76 Effect.flatMap((body) => {
77 const actual = (body as { value: string | null }).value;
78 return actual === expected
79 ? Effect.succeed(actual)
80 : Effect.fail(new ValueMismatch({ expected, actual }));
81 }),
82 retryMismatch,
83 );
84
85/** GET `/get` and retry until the object is gone (`value === null`). */
86const expectMissing = (base: string, key: string) =>
87 untilOk(HttpClient.get(`${base}/get?key=${encodeURIComponent(key)}`)).pipe(
88 Effect.flatMap((res) => res.json),
89 Effect.flatMap((body) => {
90 const actual = (body as { value: string | null }).value;
91 return actual === null
92 ? Effect.succeed(null)
93 : Effect.fail(new ValueMismatch({ expected: "<missing>", actual }));
94 }),
95 retryMismatch,
96 );
97
98/** HEAD-equivalent: `/head` returns `{ exists, size }` (metadata only, no body). */
99const headObject = (base: string, key: string) =>
100 untilOk(HttpClient.get(`${base}/head?key=${encodeURIComponent(key)}`)).pipe(
101 Effect.flatMap((res) => res.json),
102 Effect.map((body) => body as { exists: boolean; size: number | null }),
103 );
104
105/** `/list?prefix=` and retry until `key` appears (list is eventually consistent). */
106const expectListed = (base: string, prefix: string, key: string) =>
107 untilOk(
108 HttpClient.get(`${base}/list?prefix=${encodeURIComponent(prefix)}`),
109 ).pipe(
110 Effect.flatMap((res) => res.json),
111 Effect.flatMap((body) => {
112 const keys = (body as { keys: string[] }).keys;
113 return keys.includes(key)
114 ? Effect.succeed(keys)
115 : Effect.fail(
116 new ValueMismatch({ expected: key, actual: keys.join(",") }),
117 );
118 }),
119 retryMismatch,
120 );
121
122const put = (base: string, key: string, value: string) =>
123 untilOk(
124 HttpClient.execute(
125 HttpClientRequest.put(`${base}/put?key=${encodeURIComponent(key)}`).pipe(
126 HttpClientRequest.bodyText(value),
127 ),
128 ),
129 );
130
131/** PUT `/put-stream?key=&sha256=`: the body is streamed into R2 under a declared hash. */
132const putStream = (base: string, key: string, body: string, sha256: string) =>
133 HttpClient.execute(
134 HttpClientRequest.put(
135 `${base}/put-stream?key=${encodeURIComponent(key)}&sha256=${sha256}`,
136 ).pipe(HttpClientRequest.bodyText(body)),
137 );
138
139const del = (base: string, key: string) =>
140 untilOk(
141 HttpClient.execute(
142 HttpClientRequest.make("DELETE")(
143 `${base}/del?key=${encodeURIComponent(key)}`,
144 ),
145 ),
146 );
147
148const delMany = (base: string, keys: string[]) =>
149 untilOk(
150 HttpClient.execute(
151 HttpClientRequest.make("DELETE")(
152 `${base}/del-many?keys=${encodeURIComponent(keys.join(","))}`,
153 ),
154 ),
155 );
156
157/**
158 * Drive every client method through `fetch`: `put` → `get` → `head` →
159 * `list` → `delete` (single) → `delete` (batch), reading back through
160 * `readBase` and writing through `writeBase`. All workers share one bucket,
161 * so keys are namespaced by `label` to keep the four runs independent.
162 */
163const post = (base: string, path: string, key: string) =>
164 untilOk(
165 HttpClient.execute(
166 HttpClientRequest.post(`${base}/${path}?key=${encodeURIComponent(key)}`),
167 ),
168 );
169
170const exercise = (
171 label: string,
172 writeBase: string,
173 readBase: string,
174 /** Multipart upload is only supported over the native binding. */
175 multipart: boolean,
176) =>
177 Effect.gen(function* () {
178 const prefix = `${label}/`;
179 const k1 = `${prefix}k1`;
180 const v1 = `${label}-value`;
181
182 // put + get (read-after-write)
183 expect((yield* put(writeBase, k1, v1)).status).toBe(200);
184 expect(yield* expectValue(readBase, k1, v1)).toBe(v1);
185
186 // head — metadata reflects the written object
187 const meta = yield* headObject(readBase, k1);
188 expect(meta.exists).toBe(true);
189 expect(meta.size).toBe(new TextEncoder().encode(v1).length);
190
191 // list — the key shows up under its prefix
192 expect(yield* expectListed(readBase, prefix, k1)).toContain(k1);
193
194 // delete (single) — head/get then report it gone
195 yield* del(writeBase, k1);
196 yield* expectMissing(readBase, k1);
197 expect((yield* headObject(readBase, k1)).exists).toBe(false);
198
199 // delete (missing key) — idempotent in BOTH implementations: the native
200 // binding resolves on a key that was never there, so the HTTP client must
201 // not surface R2's `NoSuchKey` either.
202 expect((yield* del(writeBase, `${prefix}never-written`)).status).toBe(200);
203
204 // delete (batch) — write two, delete both in one call
205 const k2 = `${prefix}k2`;
206 const k3 = `${prefix}k3`;
207 yield* put(writeBase, k2, "v2");
208 yield* put(writeBase, k3, "v3");
209 expect(yield* expectValue(readBase, k2, "v2")).toBe("v2");
210 yield* delMany(writeBase, [k2, k3]);
211 yield* expectMissing(readBase, k2);
212 yield* expectMissing(readBase, k3);
213
214 if (multipart) {
215 // 5 MiB first part + 4-byte tail. `head` reads back the size without
216 // downloading the payload.
217 const expectedSize = 5 * 1024 * 1024 + 4;
218
219 // create → uploadPart → complete
220 const mk = `${prefix}mpu`;
221 const created = yield* (yield* post(writeBase, "mpu", mk)).json;
222 expect((created as { size: number }).size).toBe(expectedSize);
223 expect((yield* headObject(readBase, mk)).size).toBe(expectedSize);
224
225 // create → uploadPart → abort (object never materializes)
226 const ak = `${prefix}mpu-abort`;
227 yield* post(writeBase, "mpu-abort", ak);
228 expect((yield* headObject(readBase, ak)).exists).toBe(false);
229
230 // create → resumeMultipartUpload → uploadPart → complete
231 const rk = `${prefix}mpu-resume`;
232 const resumed = yield* (yield* post(writeBase, "mpu-resume", rk)).json;
233 expect((resumed as { size: number }).size).toBe(expectedSize);
234 expect((yield* headObject(readBase, rk)).size).toBe(expectedSize);
235 }
236 });
237
238/**
239 * Deploys six Workers that all bind one shared R2 bucket — read /
240 * write / read-write, each over the native Worker binding
241 * (`*BucketBinding`) and over a scoped HTTP API token (`*BucketHttp`)
242 * — via {@link Stack}, then drives each binding flavor over `fetch` in
243 * its own test:
244 *
245 * - write through the Write worker, read it back through the Read
246 * worker (cross-worker, proving both halves agree on the bucket);
247 * - round-trip a key through the ReadWrite worker by itself.
248 *
249 * The stack lives in `fixtures/stack.ts` so it can also be inspected
250 * directly, e.g. `alchemy logs --tail --stage test --config ./test/Cloudflare/R2/fixtures/stack.ts`.
251 */
252const stack = beforeAll(deploy(Stack), { timeout: HOOK_TIMEOUT });
253afterAll.skipIf(!!process.env.NO_DESTROY)(destroy(Stack), {
254 timeout: HOOK_TIMEOUT,
255});
256
257// ── Native Worker binding ── write through the Write worker, read back through
258// the Read worker (cross-worker, proving both halves agree on the bucket).
259test(
260 "native binding: write + read across separate workers",
261 Effect.gen(function* () {
262 const out = yield* stack;
263 yield* exercise("bind", out.writeBinding, out.readBinding, true);
264 }).pipe(logLevel),
265 {
266 tags: [
267 "provider:cloudflare",
268 "provider:cloudflare:r2",
269 "provider:cloudflare:worker",
270 "live",
271 ],
272 timeout: TEST_TIMEOUT,
273 },
274);
275
276// The ReadWrite worker round-trips a key by itself over the native binding.
277test(
278 "native binding: read-write round-trip in one worker",
279 Effect.gen(function* () {
280 const out = yield* stack;
281 yield* exercise(
282 "rw-bind",
283 out.readWriteBinding,
284 out.readWriteBinding,
285 true,
286 );
287 }).pipe(logLevel),
288 {
289 tags: [
290 "provider:cloudflare",
291 "provider:cloudflare:r2",
292 "provider:cloudflare:worker",
293 "live",
294 ],
295 timeout: TEST_TIMEOUT,
296 },
297);
298
299// A `put` given an Effect `Stream` goes through a `FixedLengthStream`, and
300// the options passed alongside it must still reach R2. Without them R2 has
301// nothing to verify and stores whatever bytes arrive under the declared
302// hash, which lets a caller plant content under a hash it does not have.
303test(
304 "native binding: a streamed put is verified against its declared sha256",
305 Effect.gen(function* () {
306 const out = yield* stack;
307 const value = "stream-value";
308 const sha256 = yield* Effect.sync(() =>
309 createHash("sha256").update(value).digest("hex"),
310 );
311
312 // Matching hash: stored and readable.
313 const ok = yield* untilOk(
314 putStream(out.writeBinding, "stream/ok", value, sha256),
315 );
316 expect(((yield* ok.json) as { stored: boolean }).stored).toBe(true);
317 expect(yield* expectValue(out.readBinding, "stream/ok", value)).toBe(value);
318
319 // Mismatched hash: R2 rejects the body and nothing is stored. The key is
320 // cleared first so a previous run cannot satisfy the final check.
321 yield* del(out.writeBinding, "stream/bad");
322 yield* expectMissing(out.readBinding, "stream/bad");
323 const bad = yield* putStream(
324 out.writeBinding,
325 "stream/bad",
326 value,
327 "0".repeat(64),
328 );
329 expect(bad.status).toBe(400);
330 expect(((yield* bad.json) as { stored: boolean }).stored).toBe(false);
331 expect((yield* headObject(out.readBinding, "stream/bad")).exists).toBe(
332 false,
333 );
334 }).pipe(logLevel),
335 {
336 tags: [
337 "provider:cloudflare",
338 "provider:cloudflare:r2",
339 "provider:cloudflare:worker",
340 "live",
341 ],
342 timeout: TEST_TIMEOUT,
343 },
344);
345
346// ── Scoped HTTP API token ── same matrix over the `*BucketHttp` clients
347// (multipart is unsupported over the HTTP API, so it is skipped here).
348test(
349 "http token: write + read across separate workers",
350 Effect.gen(function* () {
351 const out = yield* stack;
352 yield* exercise("http", out.writeHttp, out.readHttp, false);
353 }).pipe(logLevel),
354 {
355 tags: [
356 "provider:cloudflare",
357 "provider:cloudflare:r2",
358 "provider:cloudflare:worker",
359 "live",
360 ],
361 timeout: TEST_TIMEOUT,
362 },
363);
364
365test(
366 "http token: read-write round-trip in one worker",
367 Effect.gen(function* () {
368 const out = yield* stack;
369 yield* exercise("rw-http", out.readWriteHttp, out.readWriteHttp, false);
370 }).pipe(logLevel),
371 {
372 tags: [
373 "provider:cloudflare",
374 "provider:cloudflare:r2",
375 "provider:cloudflare:worker",
376 "live",
377 ],
378 timeout: TEST_TIMEOUT,
379 },
380);