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 * as HttpClientResponse from "effect/http/HttpClientResponse";
11import QueueWorker from "./round-trip-worker.ts";
12
13const { test } = Test.make({ providers: Cloudflare.providers() });
14
15const logLevel = Effect.provideService(
16 MinimumLogLevel,
17 process.env.DEBUG ? "Debug" : "Info",
18);
19
20class CountMismatch extends Data.TaggedError("CountMismatch")<{
21 expected: number;
22 actual: number;
23}> {}
24
25/**
26 * End-to-end Cloudflare Queue round-trip via
27 * `Cloudflare.Queues.consumeQueueMessages(queue, handler)`.
28 *
29 * Stack:
30 *
31 * - `Counter` Durable Object (per-key count + last-bodies tail).
32 * - `RoundTripQueue` and `SecondaryRoundTripQueue`
33 * (Cloudflare.Queues.Queue).
34 * - `QueueRoundTripWorker` — exposes:
35 * - `POST /send?name=K` → enqueues a message via the
36 * `Cloudflare.Queues.WriteQueue` producer.
37 * - `POST /send-secondary?name=K` → enqueues to a second
38 * Queue bound to the same Worker.
39 * - subscribe handlers → increment the named Counter DO
40 * and stores the body, via
41 * `Cloudflare.Queues.consumeQueueMessages(RoundTripQueue, handler)`.
42 * - `GET /count?name=K` → reads the DO snapshot.
43 * - `Cloudflare.Queues.Consumer` is auto-created by the policy
44 * side of `consumeQueueMessages(...)` — there is no explicit
45 * `Consumer(...)` yield in the stack.
46 *
47 * The test sends N messages, then polls `/count?name=K` with
48 * exponential backoff until the DO reports `count >= N`. The
49 * round-trip proves: producer binding writes, Cloudflare dispatches
50 * to the registered consumer, the subscribe handler runs, the DO
51 * RPC stub from inside the queue handler works, and the test
52 * client can read the resulting DO state.
53 *
54 * The second queue catches regressions where Worker dispatch stops
55 * after the first registered queue listener. The listener generated
56 * by `consumeQueueMessages(...)` performs its queue-name check inside
57 * the returned Effect, so dispatch must invoke every listener for
58 * the event type.
59 */
60test.provider.skipIf(!!process.env.FAST)(
61 "send → subscribe handlers → DO state → polled by test client",
62 (stack) =>
63 Effect.gen(function* () {
64 yield* stack.destroy();
65
66 const out = yield* stack.deploy(
67 Effect.gen(function* () {
68 // The Worker's init body yields Counter and
69 // both Queue resources internally — yielding QueueWorker
70 // is enough to bring the whole stack (Queues +
71 // QueueConsumers + Counter DO + Worker) into the plan.
72 const worker = yield* QueueWorker;
73 return { url: worker.url };
74 }),
75 );
76 const url = out.url;
77 expect(url).toBeTypeOf("string");
78 const baseUrl = url as string;
79
80 // Use a unique counter key per test run so we don't
81 // accumulate state from prior runs (the DO survives across
82 // deploys when the namespace logical id is stable).
83 const name = `roundtrip-${Math.random().toString(36).slice(2, 8)}`;
84 const secondaryName = `roundtrip-secondary-${Math.random()
85 .toString(36)
86 .slice(2, 8)}`;
87 const messages = ["alpha", "beta", "gamma", "delta"];
88 const secondaryMessages = ["one", "two"];
89
90 const sendMessage = (
91 pathname: string,
92 counterName: string,
93 text: string,
94 ) =>
95 HttpClient.execute(
96 HttpClientRequest.post(
97 `${baseUrl}${pathname}?name=${encodeURIComponent(counterName)}`,
98 ).pipe(HttpClientRequest.bodyText(text)),
99 ).pipe(
100 Effect.flatMap((res) =>
101 res.status === 202
102 ? Effect.succeed(res)
103 : Effect.fail(new Error(`Worker not ready: ${res.status}`)),
104 ),
105 Effect.retry({
106 // Bound fresh workers.dev readiness to roughly 25 seconds.
107 schedule: Schedule.max([
108 Schedule.min([
109 Schedule.exponential("500 millis"),
110 Schedule.spaced("3 seconds"),
111 ]),
112 Schedule.recurs(10),
113 ]),
114 }),
115 );
116
117 for (const text of messages) {
118 // Cloudflare's edge takes a few seconds to start serving a fresh
119 // workers.dev URL — retry until the worker returns 202.
120 const sendResponse = yield* sendMessage("/send", name, text);
121 expect(sendResponse.status).toBe(202);
122 const sent = (yield* sendResponse.json) as {
123 sent: { name: string; text: string };
124 };
125 expect(sent.sent.name).toBe(name);
126 expect(sent.sent.text).toBe(text);
127 }
128
129 for (const text of secondaryMessages) {
130 const sendResponse = yield* sendMessage(
131 "/send-secondary",
132 secondaryName,
133 text,
134 );
135 expect(sendResponse.status).toBe(202);
136 const sent = (yield* sendResponse.json) as {
137 sent: { name: string; text: string };
138 };
139 expect(sent.sent.name).toBe(secondaryName);
140 expect(sent.sent.text).toBe(text);
141 }
142
143 // GET /count is idempotent, so we retry on *any* failure — not
144 // just CountMismatch. A fresh/hibernating DO can briefly return a
145 // 500 ("Internal Server Error", which isn't valid JSON and would
146 // otherwise surface as a decode error), and edge propagation can
147 // 404 the first calls; both are transient and must be retried.
148 const readSnapshot = (counterName: string, expected: number) =>
149 HttpClient.get(
150 `${baseUrl}/count?name=${encodeURIComponent(counterName)}`,
151 ).pipe(
152 Effect.flatMap(HttpClientResponse.filterStatusOk),
153 Effect.flatMap((res) => res.json),
154 Effect.flatMap((body) => {
155 const snap = body as { count: number; lastBodies: string[] };
156 return snap.count >= expected
157 ? Effect.succeed(snap)
158 : Effect.fail(
159 new CountMismatch({
160 expected,
161 actual: snap.count,
162 }),
163 );
164 }),
165 Effect.retry({
166 // Bound consumer catch-up to roughly 35 seconds.
167 schedule: Schedule.max([
168 Schedule.min([
169 Schedule.exponential("500 millis"),
170 Schedule.spaced("4 seconds"),
171 ]),
172 Schedule.recurs(10),
173 ]),
174 }),
175 );
176
177 // Poll the DO snapshot until each consumer has caught up.
178 const [snapshot, secondarySnapshot] = yield* Effect.all(
179 [
180 readSnapshot(name, messages.length),
181 readSnapshot(secondaryName, secondaryMessages.length),
182 ],
183 { concurrency: "unbounded" },
184 );
185
186 // The DO observed every message. Cloudflare Queues are
187 // *at-least-once*: a message can be delivered (and thus recorded)
188 // more than once, so `lastBodies` may legitimately contain
189 // duplicates and `count` may exceed `messages.length`. Order is
190 // also best-effort (batches dispatch in parallel). So assert every
191 // expected message was observed at least once (set containment),
192 // not exact multiset equality — the latter flakes whenever
193 // Cloudflare redelivers a message.
194 expect([...new Set(snapshot.lastBodies)].sort()).toEqual(
195 [...messages].sort(),
196 );
197 expect(secondarySnapshot.count).toBeGreaterThanOrEqual(
198 secondaryMessages.length,
199 );
200 expect([...new Set(secondarySnapshot.lastBodies)].sort()).toEqual(
201 [...secondaryMessages].sort(),
202 );
203
204 yield* stack.destroy();
205 }).pipe(logLevel),
206 {
207 tags: [
208 "provider:cloudflare",
209 "provider:cloudflare:queue",
210 "provider:cloudflare:worker",
211 "live",
212 ],
213 timeout: 120_000,
214 },
215);