1import * as Cloudflare from "@/Cloudflare/index.ts";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import { HttpServerRequest } from "effect/http/HttpServerRequest";
5import * as HttpServerResponse from "effect/http/HttpServerResponse";
6import { DedicatedQueue } from "./dedicated-consumer-queue.ts";
7
8/** Records the bodies the queue handler observed, so the test can poll them. */
9export class Inbox extends Cloudflare.DurableObject<Inbox>()(
10 "DedicatedInbox",
11 Effect.gen(function* () {
12 return Effect.gen(function* () {
13 const state = yield* Cloudflare.DurableObjectState;
14 const bodies = (yield* state.storage.get<string[]>("bodies")) ?? [];
15 return {
16 record: Effect.fn(function* (body: string) {
17 bodies.push(body);
18 yield* state.storage.put("bodies", bodies);
19 }),
20 snapshot: () => Effect.succeed({ bodies }),
21 };
22 });
23 }),
24) {}
25
26/**
27 * A Worker that ONLY consumes — it never binds the queue as a producer, so
28 * the queue's name reaches it exclusively through the
29 * `DedicatedQueue_queueName` env binding that `consumeQueueMessages` reads
30 * to scope its handler. This is the topology of #1243.
31 *
32 * `GET /binding` returns that binding RAW from `WorkerEnvironment` (not
33 * through the unpacking `ctx.get` accessor) so the test can pin the wire
34 * format Cloudflare actually stores.
35 */
36export default class ConsumerWorker extends Cloudflare.Worker<ConsumerWorker>()(
37 "dedicated-consumer-worker",
38 { main: import.meta.url },
39 Effect.gen(function* () {
40 const inbox = yield* Inbox;
41 const queue = yield* DedicatedQueue;
42 const env = yield* Cloudflare.WorkerEnvironment;
43
44 yield* Cloudflare.Queues.consumeQueueMessages<{ text: string }>(
45 queue,
46 (stream) =>
47 Stream.runForEach(stream, (msg) =>
48 inbox.getByName("default").record(msg.body.text),
49 ),
50 );
51
52 return {
53 fetch: Effect.gen(function* () {
54 const request = yield* HttpServerRequest;
55 const url = new URL(request.url, "http://x");
56 if (url.pathname === "/binding") {
57 return yield* HttpServerResponse.json({
58 queueName: (env as Record<string, unknown>)
59 .DedicatedQueue_queueName,
60 });
61 }
62 if (url.pathname === "/received") {
63 return yield* HttpServerResponse.json(
64 yield* inbox.getByName("default").snapshot(),
65 );
66 }
67 return HttpServerResponse.text("Not Found", { status: 404 });
68 }),
69 };
70 }).pipe(Effect.provide(Cloudflare.Queues.EventSourceLive)),
71) {}