1import * as Cloudflare from "alchemy/Cloudflare";
2import * as Test from "alchemy/Test/Bun";
3import { expect } from "bun:test";
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 Stack from "../alchemy.run.ts";
9
10// A fresh workers.dev URL transiently 404s/5xxs while the edge converges.
11// `Test.getWhenReady` / `Test.executeWhenReady` retry through that window and
12// return the first non-cold-start response to assert on.
13const { getWhenReady, executeWhenReady } = Test;
14
15const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
16 providers: Cloudflare.providers(),
17 state: Cloudflare.state(),
18});
19
20const stack = beforeAll(deploy(Stack));
21afterAll.skipIf(!!process.env.NO_DESTROY)(destroy(Stack));
22
23test(
24 "deploys and exposes a url",
25 Effect.gen(function* () {
26 const { url } = yield* stack;
27 expect(url).toBeString();
28 }),
29);
30
31test(
32 "echoes env.API_KEY",
33 Effect.gen(function* () {
34 const { url } = yield* stack;
35
36 const response = yield* getWhenReady(`${url}/api-key`);
37 expect(response.status).toBe(200);
38 const body = yield* response.text;
39 expect(body).toBe("SOME_API_KEY");
40 }),
41 // `getWhenReady` rides out the edge cold-start window with exponential
42 // backoff (up to 20 attempts), which routinely exceeds bun's default 5s
43 // test timeout on a fresh workers.dev URL. Match the other HTTP cases.
44 { timeout: 120_000 },
45);
46
47/**
48 * Native (async) queue handler round-trip. The async worker exports
49 * a plain `queue(batch, env)` handler that writes each message body
50 * to R2 at `/queue/<id>`. POST /queue/send enqueues a message;
51 * GET /<path> reads from R2, so we read /queue/<id> back.
52 *
53 * Pairs with the cloudflare-worker example, which exercises the
54 * Effect-style `Cloudflare.Queues.consumeQueueMessages(Queue, handler)` path
55 * against the same producer/consumer round-trip.
56 */
57test(
58 "native queue() handler round-trip",
59 Effect.gen(function* () {
60 const { url } = yield* stack;
61 const text = `hello-${Date.now()}`;
62 type Message = { id: string; text: string; sentAt: number };
63
64 const send = Effect.gen(function* () {
65 const sendResponse = yield* executeWhenReady(
66 HttpClientRequest.post(
67 `${url}/queue/send?text=${encodeURIComponent(text)}`,
68 ),
69 );
70 expect(sendResponse.status).toBe(202);
71 const { sent } = (yield* sendResponse.json) as { sent: Message };
72 expect(sent.id).toBeTypeOf("string");
73 return sent;
74 });
75
76 const sent: Message[] = [yield* send];
77
78 // First delivery on a freshly created queue+consumer can lag well past
79 // the settled steady state (consumer attachment propagates through
80 // Cloudflare's queue subsystem asynchronously). Poll for ANY of the
81 // messages we sent, re-sending every ~20s in case an early message was
82 // published into the propagation window; the consumer persists each
83 // message to R2 keyed by its id, so the first one to land wins.
84 let polls = 0;
85 const findConsumed = Effect.gen(function* () {
86 polls++;
87 if (polls % 10 === 0 && sent.length < 5) {
88 sent.push(yield* send);
89 }
90 for (const message of sent) {
91 const resultResponse = yield* HttpClient.get(
92 `${url}/queue/${message.id}`,
93 );
94 if (resultResponse.status === 200) {
95 const body = yield* resultResponse.text;
96 if (body) return JSON.parse(body) as Message;
97 }
98 }
99 return undefined;
100 });
101
102 const consumed = yield* findConsumed.pipe(
103 Effect.repeat({
104 schedule: Schedule.spaced("2 seconds"),
105 until: (message) => message !== undefined,
106 times: 75,
107 }),
108 );
109
110 expect(consumed).toBeDefined();
111 expect(sent.map((message) => message.id)).toContain(consumed!.id);
112 expect(consumed!.text).toBe(text);
113 }),
114 { timeout: 180_000 },
115);