1import * as Queue from "../../core/bindings/queue/Queue.ts";
2import cloudflareVitePlugin from "../plugin.ts";
3import * as fs from "node:fs/promises";
4import * as path from "node:path";
5import * as vite from "vite";
6import { afterEach, describe, expect, test } from "vitest";
7
8/**
9 * A Worker that both produces to and consumes from one queue.
10 *
11 * `POST /send` enqueues a message. `GET /received` reports what the `queue()`
12 * handler has seen. The handler accumulates on `globalThis` — the dev server
13 * runs one Worker instance, so the producer request, the delivery and the
14 * readback all share an isolate.
15 */
16const workerSource = `
17export default {
18 async fetch(request, env) {
19 const url = new URL(request.url);
20 if (url.pathname === "/send") {
21 await env.QUEUE.send({ id: url.searchParams.get("id") });
22 return new Response("sent");
23 }
24 return Response.json(globalThis.__received ?? []);
25 },
26
27 async queue(batch) {
28 globalThis.__received ??= [];
29 for (const message of batch.messages) {
30 globalThis.__received.push(message.body.id);
31 message.ack();
32 }
33 },
34};
35`;
36
37const QUEUE_NAME = "vite-plugin-queue-consumer-test";
38
39/** Under `.cache` so a crashed run cannot leave untracked files behind. */
40const TMP_ROOT = path.resolve(import.meta.dirname, "../.cache/test-roots");
41
42const cleanups: Array<() => Promise<void>> = [];
43
44afterEach(async () => {
45 for (const cleanup of cleanups.splice(0).reverse()) {
46 await cleanup();
47 }
48});
49
50async function startDevServer() {
51 await fs.mkdir(TMP_ROOT, { recursive: true });
52 const root = await fs.mkdtemp(path.join(TMP_ROOT, "queue-consumer-"));
53 cleanups.push(() => fs.rm(root, { recursive: true, force: true }));
54
55 const entry = path.join(root, "worker.js");
56 await fs.writeFile(entry, workerSource);
57
58 const server = await vite.createServer({
59 root,
60 configFile: false,
61 logLevel: "silent",
62 server: { port: 0 },
63 plugins: [
64 cloudflareVitePlugin({
65 main: entry,
66 compatibilityDate: "2026-03-10",
67 worker: {
68 name: "vite-plugin-queue-consumer-test",
69 bindings: [Queue.local({ binding: "QUEUE", queueName: QUEUE_NAME })],
70 // The worker consumes the same queue it produces to, so the producer
71 // binding must resolve to a local broker rather than the dev-registry
72 // proxy. That only happens if this option reaches `runtime.start`.
73 queueConsumers: [
74 { queueName: QUEUE_NAME, maxBatchSize: 1, maxBatchTimeout: 0 },
75 ],
76 },
77 }),
78 ],
79 });
80 cleanups.push(() => server.close());
81 await server.listen();
82
83 const url = server.resolvedUrls?.local[0];
84 if (!url) {
85 throw new Error("Dev server did not report a local URL");
86 }
87
88 return {
89 send: (id: string) =>
90 fetch(new URL(`/send?id=${id}`, url), {
91 method: "POST",
92 signal: AbortSignal.timeout(10_000),
93 }),
94 received: async (): Promise<Array<string>> => {
95 const response = await fetch(new URL("/received", url), {
96 signal: AbortSignal.timeout(10_000),
97 });
98 return (await response.json()) as Array<string>;
99 },
100 };
101}
102
103/** Delivery is asynchronous, so poll rather than assert on the first read. */
104async function receivedEventually(
105 received: () => Promise<Array<string>>,
106 count: number,
107): Promise<Array<string>> {
108 const deadline = Date.now() + 30_000;
109 let last = await received();
110 while (last.length < count && Date.now() < deadline) {
111 await new Promise((resolve) => setTimeout(resolve, 250));
112 last = await received();
113 }
114 return last;
115}
116
117describe("queue consumers in dev", () => {
118 /**
119 * Regression test for a dropped option: `serve()` built its `runtime.start`
120 * call by listing worker options one at a time and omitted `queueConsumers`.
121 * The type allowed it, so callers passed it and it was silently discarded.
122 *
123 * The worker then started with no consumers, its producer binding resolved to
124 * the dev-registry `ExternalQueueConsumer` instead of a local broker, and that
125 * accepts-and-drops:
126 *
127 * [registry] No consumer registered for queue "…". Accepting and dropping message.
128 *
129 * Two things failed, and this test covers both: `queue()` never ran, and
130 * `send()` never settled — so the request hung rather than erroring.
131 */
132 test("delivers a message to the queue() handler of the worker that produced it", async () => {
133 const { send, received } = await startDevServer();
134
135 // `send()` not settling was half the fault, so await the producer request
136 // rather than fire-and-forget. The AbortSignal turns a hang into a failure.
137 const response = await send("first");
138 expect(response.status).toBe(200);
139
140 expect(await receivedEventually(received, 1)).toEqual(["first"]);
141 });
142
143 test("delivers every message when several are produced", async () => {
144 const { send, received } = await startDevServer();
145
146 await send("a");
147 await send("b");
148 await send("c");
149
150 expect((await receivedEventually(received, 3)).sort()).toEqual([
151 "a",
152 "b",
153 "c",
154 ]);
155 });
156});