1import * as Cloudflare from "@/Cloudflare";
2import { CloudflareEnvironment } from "@/Cloudflare/CloudflareEnvironment";
3import { generateLocalId, isLiveId } from "@/Cloudflare/LocalRuntime";
4import * as Provider from "@/Provider";
5import { poll } from "@/Util/poll.ts";
6import { State } from "@/State";
7import type { CreatedResourceState } from "@/State/ResourceState";
8import * as Test from "@/Test/Alchemy";
9import * as queues from "@distilled.cloud/cloudflare/queues";
10import { expect } from "alchemy-test";
11import * as Effect from "effect/Effect";
12import * as Exit from "effect/Exit";
13import { MinimumLogLevel } from "effect/References";
14import * as Schedule from "effect/Schedule";
15import * as HttpClient from "effect/http/HttpClient";
16import ConsumerWorker from "./fixtures/dedicated-consumer-worker.ts";
17import ProducerWorker from "./fixtures/dedicated-producer-worker.ts";
18
19const { test } = Test.make({ providers: Cloudflare.providers() });
20
21const logLevel = Effect.provideService(
22 MinimumLogLevel,
23 process.env.DEBUG ? "Debug" : "Info",
24);
25
26/**
27 * Seed a `created` Queue state row whose `queueId` is a `dev:` mock id —
28 * i.e. the shape `alchemy dev` persists when a queue is "created" locally
29 * against the in-memory local runtime instead of Cloudflare. Dev runs stamp
30 * `providerMode: "local"` on every commit, so the seeded row carries it.
31 */
32const seedDevQueue = (input: {
33 stackName: string;
34 stage: string;
35 fqn: string;
36 queueId: string;
37 queueName: string;
38 accountId: string;
39}) =>
40 Effect.gen(function* () {
41 const state = yield* yield* State;
42 yield* state.set({
43 stack: input.stackName,
44 stage: input.stage,
45 fqn: input.fqn,
46 value: {
47 kind: "resource",
48 status: "created",
49 resourceType: "Cloudflare.Queues.Queue",
50 namespace: undefined,
51 fqn: input.fqn,
52 logicalId: input.fqn,
53 instanceId: "00000000000000000000000000000001",
54 providerVersion: 0,
55 providerMode: "local",
56 bindings: [],
57 downstream: [],
58 props: {},
59 attr: {
60 queueId: input.queueId,
61 queueName: input.queueName,
62 accountId: input.accountId,
63 },
64 } satisfies CreatedResourceState,
65 });
66 });
67
68/**
69 * Promotion: a queue that was "created" in dev (its persisted `queueId`
70 * is a `dev:` mock id) must be promoted to a real Cloudflare queue on the
71 * first live deploy.
72 *
73 * The row is stamped `providerMode: "local"`, which differs from the
74 * deploy's resolved mode (`live`), so the engine plans a REPLACEMENT: the
75 * live provider creates a real queue and the old local-mode generation is
76 * deleted via the local provider (no cloud call for the `dev:` id). The
77 * resulting `queueId` must be a live id and resolvable on Cloudflare.
78 */
79test.provider(
80 "promotes a dev queue to a live queue on deploy",
81 (stack) =>
82 Effect.gen(function* () {
83 const { accountId } = yield* yield* CloudflareEnvironment;
84
85 yield* stack.destroy();
86
87 const devQueueId = generateLocalId();
88 yield* seedDevQueue({
89 stackName: stack.name,
90 stage: stack.stage,
91 fqn: "Q",
92 queueId: devQueueId,
93 queueName: "dev-placeholder-name",
94 accountId,
95 });
96
97 const deployed = yield* stack.deploy(
98 Effect.gen(function* () {
99 const queue = yield* Cloudflare.Queues.Queue("Q");
100 return { queue };
101 }),
102 );
103
104 // The dev id has been replaced with a real Cloudflare queue id.
105 expect(isLiveId(deployed.queue.queueId)).toBe(true);
106 expect(deployed.queue.queueId).not.toEqual(devQueueId);
107
108 // The promoted queue is a real, resolvable Cloudflare resource. A
109 // brand-new queue can briefly 404 from this out-of-band read under load,
110 // so ride out the read-after-create lag before asserting.
111 const live = yield* queues
112 .getQueue({
113 accountId,
114 queueId: deployed.queue.queueId,
115 })
116 .pipe(
117 Effect.retry({
118 while: (e) => e._tag === "QueueNotFound",
119 schedule: Schedule.exponential("500 millis"),
120 times: 8,
121 }),
122 );
123 expect(live.queueId).toEqual(deployed.queue.queueId);
124
125 // And the persisted state now carries the live id, not the dev one.
126 const persisted = yield* Effect.gen(function* () {
127 const state = yield* yield* State;
128 return yield* state.get({
129 stack: stack.name,
130 stage: stack.stage,
131 fqn: "Q",
132 });
133 });
134 expect((persisted as any)?.attr?.queueId).toEqual(deployed.queue.queueId);
135
136 yield* stack.destroy();
137
138 // After destroy the promoted (live) queue is gone on Cloudflare.
139 const exit = yield* Effect.exit(
140 queues.getQueue({ accountId, queueId: deployed.queue.queueId }),
141 );
142 expect(Exit.isFailure(exit)).toBe(true);
143 }).pipe(logLevel),
144 { tags: ["provider:cloudflare", "provider:cloudflare:queue", "live"] },
145);
146
147// Canonical `list()` test (account-scoped collection): deploy a real
148// queue, resolve the provider from context via `findProvider`, call
149// `list()`, and assert the deployed queue appears in the exhaustively-
150// paginated result.
151test.provider(
152 "list enumerates the deployed queue",
153 (stack) =>
154 Effect.gen(function* () {
155 yield* stack.destroy();
156
157 const deployed = yield* stack.deploy(
158 Effect.gen(function* () {
159 return yield* Cloudflare.Queues.Queue("ListQueue");
160 }),
161 );
162
163 const provider = yield* Provider.findProvider(Cloudflare.Queues.Queue);
164
165 // A just-created queue can lag the account-wide list under load — poll
166 // until it shows up (bounded) instead of asserting on the first read.
167 const all = yield* poll({
168 description: "list() includes the deployed queue",
169 effect: provider.list(),
170 predicate: (all) => all.some((q) => q.queueId === deployed.queueId),
171 schedule: Schedule.max([
172 Schedule.spaced("2 seconds"),
173 Schedule.recurs(20),
174 ]),
175 });
176
177 expect(all.some((q) => q.queueId === deployed.queueId)).toBe(true);
178
179 yield* stack.destroy();
180 }).pipe(logLevel),
181 { tags: ["provider:cloudflare", "provider:cloudflare:queue", "live"] },
182);
183
184/**
185 * Suppressed delete: a queue that only ever existed in dev (its
186 * persisted `queueId` is a `dev:` mock id) has no Cloudflare counterpart.
187 * Destroying it must NOT issue a `deleteQueue` against Cloudflare — the
188 * `dev:` id is not a valid queue id and the request URL would be
189 * malformed. The row is stamped `providerMode: "local"`, so the engine
190 * resolves the LOCAL provider's `delete` for it (an in-memory registry
191 * removal), and destroy succeeds with the state row removed cleanly.
192 */
193test.provider(
194 "suppresses deletion of a dev-only queue",
195 (stack) =>
196 Effect.gen(function* () {
197 const { accountId } = yield* yield* CloudflareEnvironment;
198
199 yield* stack.destroy();
200
201 const devQueueId = generateLocalId();
202 yield* seedDevQueue({
203 stackName: stack.name,
204 stage: stack.stage,
205 fqn: "Q",
206 queueId: devQueueId,
207 queueName: "dev-placeholder-name",
208 accountId,
209 });
210
211 // Destroy must not attempt a (malformed) live delete against the dev id.
212 const exit = yield* Effect.exit(stack.destroy());
213 expect(Exit.isSuccess(exit)).toBe(true);
214
215 // The dev-only resource is removed from state.
216 const persisted = yield* Effect.gen(function* () {
217 const state = yield* yield* State;
218 return yield* state.get({
219 stack: stack.name,
220 stage: stack.stage,
221 fqn: "Q",
222 });
223 });
224 expect(persisted).toBeUndefined();
225 }).pipe(logLevel),
226 { tags: ["provider:cloudflare", "provider:cloudflare:queue", "live"] },
227);
228
229/**
230 * Regression test for #1243 — producer and consumer split across two
231 * Workers, so the consuming Worker has no producer binding and learns its
232 * queue's name only from the `DedicatedQueue_queueName` env binding.
233 *
234 * Before the `packEnvValue` fix, that binding deployed as the JSON-packed
235 * `"the-name"` (quote characters on the wire): the raw-binding assertion
236 * below failed, and any consumer of the raw value — including the
237 * reporter's — could never match Cloudflare's bare `MessageBatch.queue`.
238 * The test pins both halves: the binding is the bare name, and the
239 * handler receives the message end-to-end.
240 */
241test.provider.skipIf(!!process.env.FAST)(
242 "dedicated consumer worker deploys a bare queue-name binding and receives messages",
243 (stack) =>
244 Effect.gen(function* () {
245 yield* stack.destroy();
246
247 const out = yield* stack.deploy(
248 Effect.gen(function* () {
249 const producer = yield* ProducerWorker;
250 const consumer = yield* ConsumerWorker;
251 return {
252 producer: producer.url.as<string>(),
253 consumer: consumer.url.as<string>(),
254 };
255 }),
256 );
257
258 const client = yield* HttpClient.HttpClient;
259 // Fresh workers.dev URLs 404 for a few seconds; retry through it.
260 const get = (url: string) =>
261 client.get(url).pipe(
262 Effect.flatMap((res) =>
263 res.status < 300
264 ? Effect.succeed(res)
265 : Effect.fail(new Error(`Worker not ready: ${res.status}`)),
266 ),
267 Effect.retry({
268 schedule: Schedule.max([
269 Schedule.min([
270 Schedule.exponential("500 millis"),
271 Schedule.spaced("3 seconds"),
272 ]),
273 Schedule.recurs(10),
274 ]),
275 }),
276 Effect.orDie,
277 );
278
279 // The raw env binding is the bare queue name — no quote characters
280 // on the wire (#1243: it deployed as `"the-name"`).
281 const binding = (yield* (yield* get(`${out.consumer}/binding`)).json) as {
282 queueName: string;
283 };
284 expect(binding.queueName).not.toMatch(/^"/);
285 expect(binding.queueName).toMatch(/queue/);
286
287 yield* get(`${out.producer}/send?text=dedicated`);
288
289 const received = yield* get(`${out.consumer}/received`).pipe(
290 Effect.flatMap((res) => res.json),
291 Effect.map((body) => (body as { bodies?: string[] })?.bodies ?? []),
292 Effect.repeat({
293 schedule: Schedule.spaced("4 seconds"),
294 until: (bodies) => bodies.includes("dedicated"),
295 times: 10,
296 }),
297 Effect.orDie,
298 );
299 expect(received).toContain("dedicated");
300
301 yield* stack.destroy();
302 }).pipe(logLevel),
303 {
304 tags: [
305 "provider:cloudflare",
306 "provider:cloudflare:queue",
307 "provider:cloudflare:worker",
308 "live",
309 ],
310 timeout: 120_000,
311 },
312);