1import { Action } from "@/Action";
2import * as Cloudflare from "@/Cloudflare";
3import { CloudflareEnvironment } from "@/Cloudflare/CloudflareEnvironment";
4import * as Test from "@/Test/Alchemy";
5import { poll } from "@/Util/poll.ts";
6import * as queues from "@distilled.cloud/cloudflare/queues";
7import { expect } from "alchemy-test";
8import * as Effect from "effect/Effect";
9import { MinimumLogLevel } from "effect/References";
10import * as Schedule from "effect/Schedule";
11
12const { test } = Test.make({ providers: Cloudflare.providers() });
13
14const logLevel = Effect.provideService(
15 MinimumLogLevel,
16 process.env.DEBUG ? "Debug" : "Info",
17);
18
19// Sending to a Queue inside an Action via `WriteQueueLocal` — the local
20// (current-credentials) implementation of the `WriteQueue` binding. Exercises
21// both the binding client (send/sendBatch) and the accessor mechanism
22// (`yield* queue.queueId` resolved at apply time). The producer binding is
23// send-only, so we verify out-of-band by pulling the messages back off the
24// queue through an HTTP-pull consumer.
25test.provider(
26 "WriteQueueLocal: send and sendBatch to a queue from an Action",
27 (stack) =>
28 Effect.gen(function* () {
29 const { accountId } = yield* yield* CloudflareEnvironment;
30
31 yield* stack.destroy();
32
33 const marker = yield* Effect.sync(() => crypto.randomUUID());
34
35 const out = yield* stack.deploy(
36 Effect.gen(function* () {
37 const queue = yield* Cloudflare.Queues.Queue("LocalWriteQueue");
38
39 const Seed = Action(
40 "Seed",
41 Effect.gen(function* () {
42 const q = yield* Cloudflare.Queues.WriteQueue(queue);
43 // Accessor — resolved at apply time against the tracker.
44 const queueId = yield* queue.queueId;
45
46 return Effect.fn(function* () {
47 const id = yield* queueId;
48
49 // Register an HTTP-pull consumer so the test can pull the
50 // messages back out-of-band. A brand-new queue can briefly
51 // 404 the consumer create under load — ride out the lag.
52 yield* queues
53 .createConsumer({
54 accountId,
55 queueId: id,
56 type: "http_pull",
57 })
58 .pipe(
59 Effect.retry({
60 while: (e) => e._tag === "QueueNotFound",
61 schedule: Schedule.exponential("500 millis"),
62 times: 8,
63 }),
64 Effect.catchTag("ConsumerAlreadyExists", () => Effect.void),
65 );
66
67 yield* q.send({ marker, seq: 1 });
68 yield* q.sendBatch([
69 { body: { marker, seq: 2 } },
70 { body: { marker, seq: 3 } },
71 ]);
72
73 return { queueId: id };
74 });
75 }).pipe(Effect.provide(Cloudflare.Queues.WriteQueueLocal)),
76 );
77
78 return yield* Seed({});
79 }),
80 );
81
82 expect(out.queueId).toBeTruthy();
83
84 // Out-of-band verification: pull the messages back off the queue and
85 // assert all three payloads (carrying our unique marker) arrive. Pulls
86 // lease a subset per call, so accumulate distinct bodies across polls.
87 const collected = new Set<string>();
88 yield* poll({
89 description: "pull sent messages back off the queue",
90 effect: Effect.gen(function* () {
91 const res = yield* queues.pullMessage({
92 accountId,
93 queueId: out.queueId,
94 batchSize: 10,
95 visibilityTimeoutMs: 2_000,
96 });
97 for (const m of res.messages ?? []) {
98 if (m.body) collected.add(m.body);
99 }
100 const acks = (res.messages ?? [])
101 .filter((m): m is typeof m & { leaseId: string } => !!m.leaseId)
102 .map((m) => ({ leaseId: m.leaseId }));
103 if (acks.length > 0) {
104 yield* queues.ackMessage({ accountId, queueId: out.queueId, acks });
105 }
106 return collected.size;
107 }),
108 predicate: (size) => size >= 3,
109 schedule: Schedule.max([
110 Schedule.spaced("2 seconds"),
111 Schedule.recurs(30),
112 ]),
113 });
114
115 expect(collected.size).toBeGreaterThanOrEqual(3);
116 for (const body of collected) {
117 expect(body).toContain(marker);
118 }
119
120 yield* stack.destroy();
121 }).pipe(logLevel),
122 {
123 tags: ["provider:cloudflare", "provider:cloudflare:queue", "live"],
124 timeout: 120_000,
125 },
126);