1import * as Cloudflare from "@/Cloudflare/index.ts";
2import * as Alchemy from "@/index.ts";
3import * as Test from "@/Test/Alchemy";
4import * as queues from "@distilled.cloud/cloudflare/queues";
5import { expect } from "alchemy-test";
6import { CloudflareEnvironment } from "@/Cloudflare/CloudflareEnvironment.ts";
7import * as Data from "effect/Data";
8import * as Effect from "effect/Effect";
9import { MinimumLogLevel } from "effect/References";
10import * as Schedule from "effect/Schedule";
11import * as HttpClient from "effect/http/HttpClient";
12import * as pathe from "pathe";
13
14// `dev: true` runs local providers behind the RPC sidecar proxy by default,
15// matching the process topology of the real `alchemy dev` command (see
16// MakeOptions.sidecar in Test/Core.ts). For Queue and Consumer this matters
17// doubly: both are RPC-backed providers, so under the proxy their entire
18// lifecycle (broker registration, consumer wiring, worker restart hooks)
19// runs in the sidecar process.
20const { test } = Test.make({
21 providers: Cloudflare.providers(),
22 dev: true,
23});
24
25const logLevel = Effect.provideService(
26 MinimumLogLevel,
27 process.env.DEBUG ? "Debug" : "Info",
28);
29
30class WorkerNotReady extends Data.TaggedError("WorkerNotReady")<{
31 status: number;
32}> {}
33
34const getJsonReady = (url: string) =>
35 Effect.gen(function* () {
36 const client = yield* HttpClient.HttpClient;
37 const res = yield* client.get(url).pipe(
38 Effect.flatMap((res) =>
39 res.status === 200
40 ? Effect.succeed(res)
41 : Effect.fail(new WorkerNotReady({ status: res.status })),
42 ),
43 Effect.retry({
44 while: (e): e is WorkerNotReady => e instanceof WorkerNotReady,
45 // Cap the backoff: an uncapped exponential over 10 recurs sums to
46 // ~8.5 minutes and turns a persistent non-200 into an apparent hang.
47 schedule: Schedule.max([
48 Schedule.min([
49 Schedule.exponential("500 millis"),
50 Schedule.spaced("2 seconds"),
51 ]),
52 Schedule.recurs(10),
53 ]),
54 }),
55 );
56 return yield* res.json;
57 }).pipe(Effect.orDie);
58
59/**
60 * Under `alchemy dev` the Queue resource is emulated (a `dev:` queueId) and
61 * the worker's producer binding targets the local broker. The fixture both
62 * produces and consumes, so this pins end-to-end local delivery: send over
63 * the binding, broker dispatches to the `queue()` handler, poll the
64 * recorded bodies back over HTTP.
65 */
66test.provider(
67 "local queue delivers produced messages to the consumer",
68 (stack) =>
69 Effect.gen(function* () {
70 yield* stack.destroy();
71
72 const deployed = yield* stack.deploy(
73 Effect.gen(function* () {
74 const queue = yield* Cloudflare.Queues.Queue("LocalQueue");
75 const worker = yield* Cloudflare.Worker("queue-local-worker", {
76 main: pathe.resolve(
77 import.meta.dirname,
78 "fixtures/queue-local-worker.ts",
79 ),
80 env: { QUEUE: queue },
81 });
82 yield* Cloudflare.Queues.Consumer("LocalConsumer", {
83 queueId: queue.queueId,
84 scriptName: worker.workerName,
85 });
86 return { queue, worker };
87 }),
88 );
89
90 // The local provider fabricates a `dev:` id — proof no cloud call ran
91 // — and the worker serves from the local dev proxy.
92 expect(deployed.queue.queueId).toMatch(/^dev:/);
93 expect(deployed.worker.url).toMatch(/^http:\/\/localhost:\d+$/);
94
95 const sent = (yield* getJsonReady(
96 `${deployed.worker.url}/send?text=local-hello`,
97 )) as { sent: string };
98 expect(sent.sent).toBe("local-hello");
99
100 // Poll until the broker delivers to the fixture's queue() handler.
101 const received = yield* getJsonReady(
102 `${deployed.worker.url}/received`,
103 ).pipe(
104 Effect.map((body) => (body as { received: string[] }).received),
105 Effect.repeat({
106 schedule: Schedule.spaced("500 millis"),
107 until: (received) => received.includes("local-hello"),
108 times: 30,
109 }),
110 );
111 expect(received).toContain("local-hello");
112
113 yield* stack.destroy();
114 }).pipe(logLevel),
115 {
116 tags: [
117 "provider:cloudflare",
118 "provider:cloudflare:queue",
119 "provider:cloudflare:worker",
120 "local",
121 ],
122 timeout: 120_000,
123 },
124);
125
126/**
127 * Consumer `settings` reach the local broker with the field names and units
128 * the runtime expects (`batchSize` → `maxBatchSize`, `maxWaitTimeMs` (ms) →
129 * `maxBatchTimeout` (s)). With `batchSize: 2` and a 2s wait, five quick
130 * sends must arrive in batches of at most 2 — under the broker's defaults
131 * (batch of 5, 1s flush) they'd land as one batch of 5. The trailing
132 * single-message batch flushing within the poll window pins the ms→s
133 * conversion: an unconverted 2000 would stall it for over half an hour.
134 */
135test.provider(
136 "local consumer settings control broker batch size and flush timeout",
137 (stack) =>
138 Effect.gen(function* () {
139 yield* stack.destroy();
140
141 const deployed = yield* stack.deploy(
142 Effect.gen(function* () {
143 const queue = yield* Cloudflare.Queues.Queue("BatchSettingsQueue");
144 const worker = yield* Cloudflare.Worker("queue-batch-local-worker", {
145 main: pathe.resolve(
146 import.meta.dirname,
147 "fixtures/queue-local-worker.ts",
148 ),
149 env: { QUEUE: queue },
150 });
151 yield* Cloudflare.Queues.Consumer("BatchSettingsConsumer", {
152 queueId: queue.queueId,
153 scriptName: worker.workerName,
154 settings: { batchSize: 2, maxWaitTimeMs: 2000 },
155 });
156 return { queue, worker };
157 }),
158 );
159
160 expect(deployed.queue.queueId).toMatch(/^dev:/);
161
162 for (let i = 0; i < 5; i++) {
163 yield* getJsonReady(`${deployed.worker.url}/send?text=batch-${i}`);
164 }
165
166 // All five arrive: two full batches immediately, the leftover after
167 // the 2s flush timeout.
168 const received = yield* getJsonReady(
169 `${deployed.worker.url}/received`,
170 ).pipe(
171 Effect.map((body) => (body as { received: string[] }).received),
172 Effect.repeat({
173 schedule: Schedule.spaced("500 millis"),
174 until: (received) => received.length >= 5,
175 times: 30,
176 }),
177 );
178 expect(received.length).toBe(5);
179
180 const { batches } = (yield* getJsonReady(
181 `${deployed.worker.url}/batches`,
182 )) as { batches: number[] };
183 expect(Math.max(...batches)).toBe(2);
184 expect(batches.reduce((a, b) => a + b, 0)).toBe(5);
185
186 yield* stack.destroy();
187 }).pipe(logLevel),
188 {
189 tags: [
190 "provider:cloudflare",
191 "provider:cloudflare:queue",
192 "provider:cloudflare:worker",
193 "local",
194 ],
195 timeout: 120_000,
196 },
197);
198
199/**
200 * Regression test for #988: a consumer configured with a dead-letter queue
201 * that the worker does not itself consume. The DLQ binding resolves through
202 * the local runtime's registry proxy; the worker previously failed to start
203 * with `WorkerdStartFailed` because the DLQ binding referenced the
204 * `cloudflare-runtime:registry-proxy` service before it was defined.
205 */
206test.provider(
207 "local queue consumer with an unconsumed dead-letter queue starts and delivers",
208 (stack) =>
209 Effect.gen(function* () {
210 yield* stack.destroy();
211
212 const deployed = yield* stack.deploy(
213 Effect.gen(function* () {
214 const queue = yield* Cloudflare.Queues.Queue("DlqSourceQueue");
215 const dlq = yield* Cloudflare.Queues.Queue("DlqTargetQueue");
216 const worker = yield* Cloudflare.Worker("queue-dlq-local-worker", {
217 main: pathe.resolve(
218 import.meta.dirname,
219 "fixtures/queue-local-worker.ts",
220 ),
221 env: { QUEUE: queue },
222 });
223 yield* Cloudflare.Queues.Consumer("DlqLocalConsumer", {
224 queueId: queue.queueId,
225 scriptName: worker.workerName,
226 deadLetterQueue: dlq.queueName,
227 settings: { maxRetries: 1 },
228 });
229 return { queue, dlq, worker };
230 }),
231 );
232
233 expect(deployed.queue.queueId).toMatch(/^dev:/);
234 expect(deployed.dlq.queueId).toMatch(/^dev:/);
235
236 // The worker starts (the DLQ binding resolves) and delivers normally.
237 const sent = (yield* getJsonReady(
238 `${deployed.worker.url}/send?text=dlq-hello`,
239 )) as { sent: string };
240 expect(sent.sent).toBe("dlq-hello");
241
242 const received = yield* getJsonReady(
243 `${deployed.worker.url}/received`,
244 ).pipe(
245 Effect.map((body) => (body as { received: string[] }).received),
246 Effect.repeat({
247 schedule: Schedule.spaced("500 millis"),
248 until: (received) => received.includes("dlq-hello"),
249 times: 30,
250 }),
251 );
252 expect(received).toContain("dlq-hello");
253
254 yield* stack.destroy();
255 }).pipe(logLevel),
256 {
257 tags: [
258 "provider:cloudflare",
259 "provider:cloudflare:queue",
260 "provider:cloudflare:worker",
261 "local",
262 ],
263 timeout: 120_000,
264 },
265);
266
267/**
268 * The full `Alchemy.remote()` queue roundtrip in dev: the local worker
269 * produces through the deployed shim into the REAL queue, and its local
270 * `queue()` handler receives the messages back through the runtime's pull
271 * loop (the Consumer's local provider attaches an `http_pull` consumer to
272 * the real queue; the pull loop drains it into the local broker).
273 */
274test.provider(
275 "Alchemy.remote() queue round-trips: local produce, real queue, local consume",
276 (stack) =>
277 Effect.gen(function* () {
278 yield* stack.destroy();
279
280 const deployed = yield* stack.deploy(
281 Effect.gen(function* () {
282 const queue = yield* Cloudflare.Queues.Queue(
283 "LiveRoundtripQueue",
284 ).pipe(Alchemy.remote());
285 const worker = yield* Cloudflare.Worker("queue-roundtrip-worker", {
286 main: pathe.resolve(
287 import.meta.dirname,
288 "fixtures/queue-local-worker.ts",
289 ),
290 env: { QUEUE: queue },
291 });
292 yield* Cloudflare.Queues.Consumer("LiveRoundtripConsumer", {
293 queueId: queue.queueId,
294 scriptName: worker.workerName,
295 });
296 return { queue, worker };
297 }),
298 );
299
300 // The queue is real; the worker (and its queue() handler) is local.
301 expect(deployed.queue.queueId).not.toMatch(/^dev:/);
302
303 // Produce through the shim into the real queue.
304 const sent = (yield* getJsonReady(
305 `${deployed.worker.url}/send?text=roundtrip-hello`,
306 )) as { sent: string };
307 expect(sent.sent).toBe("roundtrip-hello");
308
309 // The pull loop drains the real queue into the local broker, which
310 // delivers to the fixture's queue() handler.
311 const received = yield* getJsonReady(
312 `${deployed.worker.url}/received`,
313 ).pipe(
314 Effect.map((body) => (body as { received: string[] }).received),
315 Effect.repeat({
316 schedule: Schedule.spaced("4 seconds"),
317 until: (received) => received.includes("roundtrip-hello"),
318 times: 10,
319 }),
320 );
321 expect(received).toContain("roundtrip-hello");
322
323 yield* stack.destroy();
324
325 // Destroy removed the real queue (with its pull consumer and shim).
326 const { accountId } = yield* yield* CloudflareEnvironment;
327 const gone = yield* queues
328 .getQueue({ accountId, queueId: deployed.queue.queueId })
329 .pipe(
330 Effect.as(false),
331 Effect.catchTag("QueueNotFound", () => Effect.succeed(true)),
332 );
333 expect(gone).toBe(true);
334 }).pipe(logLevel),
335 {
336 tags: [
337 "provider:cloudflare",
338 "provider:cloudflare:queue",
339 "provider:cloudflare:worker",
340 "live",
341 ],
342 timeout: 120_000,
343 },
344);
345
346/**
347 * `Alchemy.remote()` queues in dev: Cloudflare's remote-binding (preview)
348 * sessions reject queue bindings at the platform level
349 * (cloudflare/workers-sdk#9929), so production goes through the deployed
350 * shim instead — the eval registers a real shim worker holding the actual
351 * queue binding, and the local worker's producer binding relays the queue
352 * wire protocol to it over HTTPS with a minted bearer token.
353 *
354 * The proof is out-of-band: an `http_pull` consumer attached directly via
355 * the cloud API pulls the messages back from the REAL queue.
356 */
357test.provider(
358 "Alchemy.remote() queue in dev produces through the deployed shim",
359 (stack) =>
360 Effect.gen(function* () {
361 yield* stack.destroy();
362
363 const deployed = yield* stack.deploy(
364 Effect.gen(function* () {
365 const queue = yield* Cloudflare.Queues.Queue("LiveDevQueue").pipe(
366 Alchemy.remote(),
367 );
368 const worker = yield* Cloudflare.Worker("queue-live-worker", {
369 main: pathe.resolve(
370 import.meta.dirname,
371 "fixtures/queue-local-worker.ts",
372 ),
373 env: { QUEUE: queue },
374 });
375 return { queue, worker };
376 }),
377 );
378
379 // The queue is real (live provider), the worker is local.
380 expect(deployed.queue.queueId).not.toMatch(/^dev:/);
381 const { accountId } = yield* yield* CloudflareEnvironment;
382
383 // Out-of-band http_pull consumer so the test can pull from the real
384 // queue (a local queue() handler cannot receive from a live queue —
385 // Cloudflare pushes to deployed consumers only).
386 yield* queues.createConsumer({
387 accountId,
388 queueId: deployed.queue.queueId,
389 type: "http_pull",
390 });
391
392 // Produce through the local worker: binding → forwarder → shim →
393 // queue. The first send rides out the shim's workers.dev propagation
394 // (the forwarder retries 404/503 internally; the outer retry covers
395 // the tail).
396 const sent = (yield* getJsonReady(
397 `${deployed.worker.url}/send?text=live-hello`,
398 )) as { sent: string };
399 expect(sent.sent).toBe("live-hello");
400 const batch = (yield* getJsonReady(
401 `${deployed.worker.url}/sendbatch?text=live-batch`,
402 )) as { sent: number };
403 expect(batch.sent).toBe(2);
404
405 // Pull the messages back from the REAL queue. Pulled messages are
406 // leased (invisible to subsequent pulls until the timeout), so
407 // accumulate bodies across pulls instead of expecting one batch to
408 // carry everything.
409 const seen = new Set<string>();
410 const bodies = yield* queues
411 .pullMessage({
412 accountId,
413 queueId: deployed.queue.queueId,
414 batchSize: 10,
415 visibilityTimeoutMs: 1000,
416 })
417 .pipe(
418 Effect.map((res) => {
419 for (const m of res.messages ?? []) {
420 seen.add(String(m.body ?? ""));
421 }
422 return Array.from(seen);
423 }),
424 Effect.repeat({
425 schedule: Schedule.spaced("2 seconds"),
426 until: (bodies) =>
427 bodies.some((b) => b.includes("live-hello")) &&
428 bodies.some((b) => b.includes("live-batch-a")) &&
429 bodies.some((b) => b.includes("live-batch")),
430 times: 20,
431 }),
432 );
433 expect(bodies.some((b) => b.includes("live-hello"))).toBe(true);
434 expect(bodies.some((b) => b.includes("live-batch-a"))).toBe(true);
435
436 yield* stack.destroy();
437
438 // Destroy removed the real queue (and with it the shim + consumer).
439 const gone = yield* queues
440 .getQueue({ accountId, queueId: deployed.queue.queueId })
441 .pipe(
442 Effect.as(false),
443 Effect.catchTag("QueueNotFound", () => Effect.succeed(true)),
444 );
445 expect(gone).toBe(true);
446 }).pipe(logLevel),
447 {
448 tags: [
449 "provider:cloudflare",
450 "provider:cloudflare:queue",
451 "provider:cloudflare:worker",
452 "live",
453 ],
454 timeout: 120_000,
455 },
456);