1import { DurableObject } from "cloudflare:workers";
2import type { WorkerEnv } from "../alchemy.run.ts";
3
4/**
5 * Type of messages we send on the queue — kept minimal so the example stays
6 * self-contained. Both producer (`env.Queue.send(...)`) and consumer (the
7 * `queue()` handler below) use this shape.
8 */
9interface Message {
10 id: string;
11 text: string;
12 sentAt: number;
13}
14
15export default {
16 async fetch(request: Request, env: WorkerEnv) {
17 const url = new URL(request.url);
18 const path = url.pathname;
19
20 // Echo back env.API_KEY so the integ test can verify the env round-trip.
21 if (request.method === "GET" && path === "/api-key") {
22 return new Response(env.API_KEY, {
23 headers: { "content-type": "text/plain" },
24 });
25 }
26
27 // Queue producer — POST /queue/send?text=...
28 //
29 // Exercises Cloudflare.Queues.WriteQueue by calling `env.Queue.send(...)`.
30 // The message is persisted by the consumer handler into R2 at /queue/<id>
31 // so the integ test can read it back and assert the full round-trip.
32 if (request.method === "POST" && path === "/queue/send") {
33 const text = url.searchParams.get("text") ?? "hello queue";
34 const msg: Message = {
35 id: crypto.randomUUID(),
36 text,
37 sentAt: Date.now(),
38 };
39 await env.Queue.send(msg);
40 return new Response(JSON.stringify({ sent: msg }), {
41 status: 202,
42 headers: { "content-type": "application/json" },
43 });
44 }
45
46 if (request.method === "GET") {
47 return new Response((await env.Bucket.get(path))?.body ?? null);
48 } else if (request.method === "PUT") {
49 const object = (await env.Bucket.put(path, request.body))!;
50 return new Response(
51 JSON.stringify({
52 key: object.key,
53 size: object.size,
54 }),
55 { status: 201 },
56 );
57 } else if (request.method === "POST") {
58 const counter = env.Counter.getByName("counter");
59 return new Response(JSON.stringify({ count: await counter.increment() }));
60 }
61 return env.ASSETS.fetch(request);
62 },
63
64 /**
65 * Queue consumer handler — invoked by Cloudflare when messages accumulate
66 * on the Queue registered as a Consumer for this worker.
67 *
68 * Persists each message body into R2 under `/queue/<id>` so the integ test
69 * (or a manual `GET /queue/<id>`) can verify the round-trip succeeded.
70 * `msg.ack()` marks the message as consumed so it isn't redelivered.
71 */
72 async queue(batch: MessageBatch<Message>, env: WorkerEnv) {
73 for (const msg of batch.messages) {
74 await env.Bucket.put(`/queue/${msg.body.id}`, JSON.stringify(msg.body), {
75 httpMetadata: { contentType: "application/json" },
76 });
77 msg.ack();
78 }
79 },
80};
81
82export class Counter extends DurableObject {
83 private counter = 0;
84
85 async increment() {
86 return ++this.counter;
87 }
88}