1import * as Cloudflare from "@/Cloudflare/index.ts";
2import * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import * as Stream from "effect/Stream";
5import { HttpServerRequest } from "effect/http/HttpServerRequest";
6import * as HttpServerResponse from "effect/http/HttpServerResponse";
7
8/**
9 * Per-key counter that the queue handler increments on every
10 * message it processes, and that the test polls back via the
11 * worker's `GET /count?name=...` route. Persists in DO storage so
12 * the count survives DO hibernation.
13 */
14export class Counter extends Cloudflare.DurableObject<Counter>()(
15 "Counter",
16 Effect.gen(function* () {
17 return Effect.gen(function* () {
18 const state = yield* Cloudflare.DurableObjectState;
19 let count = (yield* state.storage.get<number>("count")) ?? 0;
20 const lastBodies =
21 (yield* state.storage.get<string[]>("lastBodies")) ?? [];
22 return {
23 record: Effect.fn(function* (body: string) {
24 count += 1;
25 lastBodies.push(body);
26 yield* state.storage.put("count", count);
27 yield* state.storage.put("lastBodies", lastBodies);
28 }),
29 snapshot: () =>
30 Effect.succeed({
31 count,
32 lastBodies,
33 }),
34 };
35 });
36 }),
37) {}
38
39/**
40 * The queue resources the worker produces to and consumes from. The
41 * test deploys this stack, sends N messages via `POST /send?name=K`,
42 * waits for the consumer to push them through to the Counter DO,
43 * and polls `GET /count?name=K` until the count matches.
44 *
45 * `Cloudflare.Queues.consumeQueueMessages(...)` (in QueueWorker below)
46 * auto-creates the matching `Cloudflare.Queues.Consumer` resource at
47 * deploy time, so this fixture has no separate consumer wiring.
48 */
49export const RoundTripQueue = Cloudflare.Queues.Queue("RoundTripQueue");
50export const SecondaryRoundTripQueue = Cloudflare.Queues.Queue(
51 "SecondaryRoundTripQueue",
52);
53
54interface QueueMessageBody {
55 name: string;
56 text: string;
57}
58
59export default class QueueWorker extends Cloudflare.Worker<QueueWorker>()(
60 "QueueRoundTripWorker",
61 {
62 main: import.meta.url,
63 },
64 Effect.gen(function* () {
65 const counters = yield* Counter;
66 const queueResource = yield* RoundTripQueue;
67 const queue = yield* Cloudflare.Queues.WriteQueue(queueResource);
68 const secondaryQueueResource = yield* SecondaryRoundTripQueue;
69 const secondaryQueue = yield* Cloudflare.Queues.WriteQueue(
70 secondaryQueueResource,
71 );
72
73 // Effect-style queue consumer. The handler delegates to the
74 // Counter DO so the test can verify the message landed by
75 // polling the DO's snapshot.
76 //
77 // Mixed `Duration.Input` forms are intentional: the e2e test
78 // exercises that a `Duration` value (`maxWaitTime`) and a
79 // string (`retryDelay: "1 second"`) both type-check at the
80 // `Cloudflare.Queues.consumeQueueMessages(...)` call site and survive the convert-
81 // and-forward path into Cloudflare's Consumer settings.
82 // Values are kept small so the round-trip latency stays well
83 // under the test's 240s timeout.
84 yield* Cloudflare.Queues.consumeQueueMessages<QueueMessageBody>(
85 queueResource,
86 {
87 batchSize: 10,
88 maxRetries: 3,
89 maxWaitTime: Duration.millis(500),
90 retryDelay: "1 second",
91 },
92 (stream) =>
93 Stream.runForEach(stream, (msg) =>
94 counters.getByName(msg.body.name).record(msg.body.text),
95 ),
96 );
97
98 // A second queue subscription on the same Worker verifies that
99 // dispatch reaches every listener for the `queue` event. Each
100 // listener still scopes itself by queue name internally.
101 yield* Cloudflare.Queues.consumeQueueMessages<QueueMessageBody>(
102 secondaryQueueResource,
103 {
104 batchSize: 1,
105 maxRetries: 2,
106 retryDelay: "2 seconds",
107 },
108 (stream) =>
109 Stream.runForEach(stream, (msg) =>
110 counters.getByName(msg.body.name).record(msg.body.text),
111 ),
112 );
113
114 return {
115 fetch: Effect.gen(function* () {
116 const request = yield* HttpServerRequest;
117 const url = new URL(request.url, "http://x");
118
119 if (request.method === "POST" && url.pathname === "/send") {
120 const name = url.searchParams.get("name") ?? "default";
121 const text = yield* request.text;
122 yield* queue.send({ name, text }).pipe(Effect.orDie);
123 return yield* HttpServerResponse.json(
124 { sent: { name, text } },
125 { status: 202 },
126 );
127 }
128
129 if (request.method === "POST" && url.pathname === "/send-secondary") {
130 const name = url.searchParams.get("name") ?? "default";
131 const text = yield* request.text;
132 yield* secondaryQueue.send({ name, text }).pipe(Effect.orDie);
133 return yield* HttpServerResponse.json(
134 { sent: { name, text } },
135 { status: 202 },
136 );
137 }
138
139 if (request.method === "GET" && url.pathname === "/count") {
140 const name = url.searchParams.get("name") ?? "default";
141 const snapshot = yield* counters.getByName(name).snapshot();
142 return yield* HttpServerResponse.json(snapshot);
143 }
144
145 return HttpServerResponse.text("Not Found", { status: 404 });
146 }),
147 };
148 }).pipe(
149 Effect.provide(Cloudflare.Queues.WriteQueueBinding),
150 Effect.provide(Cloudflare.Queues.EventSourceLive),
151 ),
152) {}