1import * as queues from "@distilled.cloud/cloudflare/queues";
2import * as workers from "@distilled.cloud/cloudflare/workers";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as MutableHashMap from "effect/MutableHashMap";
6import * as Option from "effect/Option";
7import * as Schedule from "effect/Schedule";
8import * as Stream from "effect/Stream";
9import { isResolved } from "../../Diff.ts";
10import * as ProviderLayer from "../../Local/ProviderLayer.ts";
11import * as RpcProvider from "../../Local/RpcProvider.ts";
12import { createPhysicalName } from "../../PhysicalName.ts";
13import * as Provider from "../../Provider.ts";
14import { isResourceOfType, Resource } from "../../Resource.ts";
15import { Stack } from "../../Stack.ts";
16import { CloudflareEnvironment } from "../CloudflareEnvironment.ts";
17import { localAccountId } from "../LocalAccount.ts";
18import { detachQueueConsumersOfScript } from "./Consumer.ts";
19import {
20 generateLocalId,
21 isLiveId,
22 LOCAL_PROVIDERS_URL,
23 LocalRuntimeState,
24 localRuntimeServices,
25} from "../LocalRuntime.ts";
26import type { Providers } from "../Providers.ts";
27
28export const isQueue = (value: unknown): value is Queue =>
29 isResourceOfType(value, "Cloudflare.Queues.Queue");
30
31export type QueueProps = {
32 /**
33 * Name of the queue. If omitted, a unique name will be generated.
34 * @default ${app}-${stage}-${id}
35 */
36 name?: string;
37};
38
39export type Queue = Resource<
40 "Cloudflare.Queues.Queue",
41 QueueProps,
42 {
43 queueId: string;
44 queueName: string;
45 accountId: string;
46 },
47 never,
48 Providers
49>;
50
51/**
52 * A Cloudflare Queue for reliable message passing between Workers.
53 *
54 * Queues enable you to send and receive messages with guaranteed delivery.
55 * Create a queue as a resource, then bind it to a Worker to send messages
56 * at runtime. Register a consumer to process messages.
57 * ### Creating a Queue
58 * **Example:** Basic queue
59 * ```typescript
60 * const queue = yield* Cloudflare.Queues.Queue("MyQueue");
61 * ```
62 *
63 * **Example:** Queue with explicit name
64 * ```typescript
65 * const queue = yield* Cloudflare.Queues.Queue("MyQueue", {
66 * name: "my-app-queue",
67 * });
68 * ```
69 *
70 * ### Binding to a Worker
71 * In an Effect-style Worker, use `Cloudflare.Queues.WriteQueue` in
72 * the init phase and provide `Cloudflare.Queues.WriteQueueBinding` in
73 * the runtime layer. The returned `WriteQueueClient` exposes `send`
74 * and `sendBatch`.
75 *
76 * **Example:** Sending messages from a Worker
77 * ```typescript
78 * import * as Cloudflare from "alchemy/Cloudflare";
79 * import * as Effect from "effect/Effect";
80 * import { HttpServerRequest } from "effect/http/HttpServerRequest";
81 * import * as HttpServerResponse from "effect/http/HttpServerResponse";
82 *
83 * export const Queue = Cloudflare.Queues.Queue("Queue");
84 *
85 * export default Cloudflare.Worker(
86 * "Worker",
87 * { main: import.meta.url },
88 * Effect.gen(function* () {
89 * const queue = yield* Cloudflare.Queues.WriteQueue(Queue);
90 *
91 * return {
92 * fetch: Effect.gen(function* () {
93 * const request = yield* HttpServerRequest;
94 * if (request.url === "/queue/send" && request.method === "POST") {
95 * const text = yield* request.text;
96 * yield* queue.send({ text, sentAt: Date.now() }).pipe(Effect.orDie);
97 * return yield* HttpServerResponse.json(
98 * { sent: { text } },
99 * { status: 202 },
100 * );
101 * }
102 * return HttpServerResponse.text("Not Found", { status: 404 });
103 * }),
104 * };
105 * }).pipe(Effect.provide(Cloudflare.Queues.WriteQueueBinding)),
106 * );
107 * ```
108 *
109 * @resource
110 * @product Queues
111 * @category Storage & Databases
112 */
113export const Queue = Resource<Queue>("Cloudflare.Queues.Queue", {
114 aliases: ["Cloudflare.Queue"],
115});
116
117export const ProviderLive = () =>
118 Provider.succeed(Queue, {
119 stables: ["queueId", "accountId"],
120 diff: Effect.fn(function* ({ id, olds = {}, news = {}, output }) {
121 const { accountId } = yield* yield* CloudflareEnvironment;
122 if (!isResolved(news)) return undefined;
123 if ((output?.accountId ?? accountId) !== accountId) {
124 return { action: "replace" } as const;
125 }
126 const oldName =
127 output?.queueName ?? (yield* createQueueName(id, olds.name));
128 // Auto-generated names are engine-owned: the deployed name stays
129 // authoritative even if the generator would name this id differently
130 // today. Only an explicit user-provided name can force a replace.
131 const name = news.name ?? oldName;
132 if (name !== oldName) {
133 return { action: "replace" } as const;
134 }
135 }),
136 reconcile: Effect.fn(function* ({ id, news = {}, output }) {
137 const { accountId } = yield* yield* CloudflareEnvironment;
138 const queueName = yield* createQueueName(id, news.name);
139 const acct = output?.accountId ?? accountId;
140
141 // Observe — re-fetch the cached queue; fall back to a name scan
142 // when the cached id is gone (out-of-band delete or partial
143 // state-persistence failure).
144 let observed:
145 | { queueId?: string | null; queueName?: string | null }
146 | undefined;
147 // A `dev:` id (a mis-stamped legacy local row) is not a real queue id —
148 // skip the lookup (Cloudflare rejects it as a malformed parameter) and
149 // fall through to the name scan.
150 if (output?.queueId && isLiveId(output.queueId)) {
151 observed = yield* queues
152 .getQueue({
153 accountId: acct,
154 queueId: output.queueId,
155 })
156 .pipe(
157 Effect.catchTag(["QueueNotFound", "InvalidRoute"], () =>
158 Effect.succeed(undefined),
159 ),
160 );
161 }
162 if (!observed) {
163 observed = yield* findQueueByName(queueName);
164 }
165
166 // Ensure — create if missing. Cloudflare returns a generic
167 // failure when the queue name is taken; tolerate by adopting
168 // the queue with the same name so reconciles converge after a
169 // crashed peer.
170 if (!observed) {
171 observed = yield* queues
172 .createQueue({
173 accountId: acct,
174 queueName,
175 })
176 .pipe(
177 Effect.catchTag("QueueAlreadyExists", () =>
178 Effect.gen(function* () {
179 const match = yield* findQueueByName(queueName);
180 if (match && match.queueId && match.queueName) {
181 return match;
182 }
183 return yield* Effect.die(
184 `Queue "${queueName}" already exists but could not be found`,
185 );
186 }),
187 ),
188 );
189 }
190
191 // Sync — Cloudflare Queues have no mutable per-queue settings
192 // here (the queue name itself is treated as a replace by diff),
193 // so observed state is the answer.
194 return {
195 queueId: observed.queueId!,
196 queueName: observed.queueName!,
197 accountId: acct,
198 };
199 }),
200 delete: Effect.fn(function* ({ output }) {
201 // A `dev:` id means the physical resource only ever existed locally
202 // (a mis-stamped legacy row) — there is nothing to delete, and the
203 // API would reject the id as a malformed parameter.
204 if (!isLiveId(output.queueId)) return;
205 // Dependents (e.g. R2 event notification configs targeting this
206 // queue) may still be tearing down concurrently — ride out the
207 // dependency violation briefly, then fail loudly instead of
208 // silently leaking the queue. A worker producer binding
209 // (`QueueInUseByWorkerBinding`) gets a shorter window: a sibling
210 // Worker's just-processed delete/unbind takes a few seconds to
211 // propagate to the queues subsystem.
212 const attempt = queues
213 .deleteQueue({
214 accountId: output.accountId,
215 queueId: output.queueId,
216 })
217 .pipe(
218 Effect.retry({
219 while: (e) => e._tag === "QueueInUseByEventNotification",
220 schedule: Schedule.max([
221 Schedule.exponential("1 second"),
222 Schedule.recurs(8),
223 ]),
224 }),
225 Effect.retry({
226 while: (e) => e._tag === "QueueInUseByWorkerBinding",
227 schedule: Schedule.max([
228 Schedule.spaced("2 seconds"),
229 Schedule.recurs(6),
230 ]),
231 }),
232 Effect.catchTag("QueueNotFound", () => Effect.void),
233 );
234 yield* attempt.pipe(
235 // Still pinned by a worker producer binding after the propagation
236 // window: every *tracked* worker is deleted or unbound before its
237 // queues by dependency order, so the referencing script is an
238 // orphaned generation leaked by a pre-stamping dev run (a local-
239 // stamped row whose real script was never deleted). Remove the
240 // scripts that carry this stack+stage's ownership tags and retry;
241 // a script from outside this stack+stage keeps the typed failure.
242 Effect.catchTag("QueueInUseByWorkerBinding", (cause) =>
243 Effect.gen(function* () {
244 const removed = yield* deleteOwnedProducerScripts(
245 output.accountId,
246 output.queueId,
247 );
248 if (removed === 0) return yield* Effect.fail(cause);
249 return yield* attempt;
250 }),
251 ),
252 );
253 }),
254 list: Effect.fn(function* () {
255 const { accountId } = yield* yield* CloudflareEnvironment;
256 return yield* queues.listQueues.pages({ accountId }).pipe(
257 Stream.runCollect,
258 Effect.map((chunk) =>
259 Array.from(chunk).flatMap((page) =>
260 (page.result ?? [])
261 .filter(
262 (q): q is typeof q & { queueId: string; queueName: string } =>
263 q.queueId != null && q.queueName != null,
264 )
265 .map((q) => ({
266 queueId: q.queueId,
267 queueName: q.queueName,
268 accountId,
269 })),
270 ),
271 ),
272 );
273 }),
274 read: Effect.fn(function* ({ id, output, olds }) {
275 const { accountId } = yield* yield* CloudflareEnvironment;
276 if (output?.queueId && isLiveId(output.queueId)) {
277 return yield* queues
278 .getQueue({
279 accountId: output.accountId,
280 queueId: output.queueId,
281 })
282 .pipe(
283 Effect.map((queue) => ({
284 queueId: queue.queueId!,
285 queueName: queue.queueName!,
286 accountId: output.accountId,
287 })),
288 Effect.catchTag(["QueueNotFound", "InvalidRoute"], () =>
289 Effect.succeed(undefined),
290 ),
291 );
292 }
293 const queueName = yield* createQueueName(id, olds?.name);
294 const match = yield* findQueueByName(queueName);
295 if (match && match.queueId && match.queueName) {
296 return {
297 queueId: match.queueId,
298 queueName: match.queueName,
299 accountId,
300 };
301 }
302 return undefined;
303 }),
304 });
305
306const createQueueName = (id: string, name: string | undefined) =>
307 Effect.gen(function* () {
308 if (name) return name;
309 return (yield* createPhysicalName({
310 id,
311 maxLength: 63,
312 })).toLowerCase();
313 });
314
315/**
316 * Delete the worker scripts that hold a producer binding on `queueId` AND
317 * carry this stack+stage's `alchemy:` ownership tags. Returns how many
318 * scripts were deleted.
319 *
320 * Used by the live queue delete when `QueueInUseByWorkerBinding` persists
321 * past the propagation-lag retry window: tracked workers are always
322 * deleted/unbound before their queues, so an own-stack script still
323 * binding the queue at that point is an orphaned generation (leaked by a
324 * pre-stamping dev run that rewrote the worker's row as local). Scripts
325 * without our ownership tags are left alone.
326 */
327const deleteOwnedProducerScripts = Effect.fn(function* (
328 accountId: string,
329 queueId: string,
330) {
331 const stack = yield* Stack;
332 const queue = yield* queues
333 .getQueue({ accountId, queueId })
334 .pipe(
335 Effect.catchTag(["QueueNotFound", "InvalidRoute"], () =>
336 Effect.succeed(undefined),
337 ),
338 );
339 const producerScripts = Array.from(
340 new Set(
341 (queue?.producers ?? []).flatMap((producer) =>
342 producer.type === "worker" &&
343 "scriptName" in producer &&
344 producer.scriptName
345 ? [producer.scriptName]
346 : [],
347 ),
348 ),
349 );
350 let removed = 0;
351 for (const scriptName of producerScripts) {
352 const settings = yield* workers
353 .getScriptScriptAndVersionSetting({ accountId, scriptName })
354 .pipe(
355 Effect.catchTag(["WorkerNotFound", "WorkerHasNoVersions"], () =>
356 Effect.succeed(undefined),
357 ),
358 );
359 const tags = new Set(settings?.tags ?? []);
360 if (
361 !tags.has(`alchemy:stack:${stack.name}`) ||
362 !tags.has(`alchemy:stage:${stack.stage}`)
363 ) {
364 continue;
365 }
366 yield* Effect.logWarning(
367 `Cloudflare Queue delete: removing orphaned worker script ` +
368 `"${scriptName}" that still binds queue ${queueId} (leaked by a ` +
369 `pre-providerMode dev run)`,
370 );
371 yield* workers.deleteScript({ accountId, scriptName, force: true }).pipe(
372 // The orphan may also be registered as a queue consumer — detach
373 // its consumers and retry, mirroring the Worker provider's own
374 // delete recovery.
375 Effect.catchTag("QueueConsumerConflict", () =>
376 detachQueueConsumersOfScript(accountId, scriptName).pipe(
377 Effect.andThen(
378 workers.deleteScript({ accountId, scriptName, force: true }),
379 ),
380 ),
381 ),
382 Effect.catchTag("WorkerNotFound", () => Effect.void),
383 );
384 removed++;
385 }
386 return removed;
387});
388
389// Cloudflare's `listQueues` accepts no name/prefix filter, so
390// adoption-by-name has to scan every page. Use the paginated
391// `.items` stream off the un-yielded operation method.
392const findQueueByName = Effect.fn(function* (queueName: string) {
393 const { accountId } = yield* yield* CloudflareEnvironment;
394 return yield* queues.listQueues.items({ accountId }).pipe(
395 Stream.filter((q) => q.queueName === queueName),
396 Stream.runHead,
397 Effect.map(Option.getOrUndefined),
398 );
399});
400
401export const ProviderLocal = () =>
402 RpcProvider.effect(
403 Queue,
404 LOCAL_PROVIDERS_URL,
405 Effect.gen(function* () {
406 const localRuntimeState = yield* LocalRuntimeState;
407 return {
408 stables: ["accountId"],
409 diff: Effect.fn(function* ({ id, olds = {}, news = {}, output }) {
410 const accountId = yield* localAccountId;
411 if (!output?.queueId) return { action: "update" };
412 // A real (non-`dev:`) queueId on a local-mode row is legacy damage:
413 // pre-stamping dev runs preserved the live id, which the worker
414 // binding then treats as an `Alchemy.remote()` queue and fails on
415 // the missing producer shim. Replace so the new generation mints a
416 // true local identity (delete best-effort removes the stray live
417 // queue).
418 if (isLiveId(output.queueId)) {
419 return { action: "replace" };
420 }
421 if (!isResolved(news)) return undefined;
422 const name = yield* createQueueName(id, news.name);
423 const oldName = output?.queueName
424 ? yield* createQueueName(id, olds.name)
425 : yield* createQueueName(id, olds.name);
426 if (name !== oldName || output.accountId !== accountId) {
427 return { action: "replace" };
428 }
429 // If the resource is a noop, add it to the local runtime state so it's available downstream.
430 // We do it here instead of in the reconcile function so it doesn't appear as an update.
431 MutableHashMap.set(localRuntimeState.queues, output.queueId, output);
432 return { action: "noop" };
433 }),
434 read: Effect.fn(function* ({ output }) {
435 if (!output?.queueId) return undefined;
436 return MutableHashMap.get(
437 localRuntimeState.queues,
438 output.queueId,
439 ).pipe(Option.getOrUndefined);
440 }),
441 reconcile: Effect.fn(function* ({ id, news = {}, output }) {
442 const accountId = yield* localAccountId;
443 const queue: Queue["Attributes"] = {
444 // Never carry a real (non-`dev:`) id forward onto a local row —
445 // the worker binding would treat it as an `Alchemy.remote()`
446 // queue and fail on the missing producer shim.
447 queueId:
448 output?.queueId && !isLiveId(output.queueId)
449 ? output.queueId
450 : generateLocalId(),
451 queueName: yield* createQueueName(id, news.name),
452 accountId: output?.accountId ?? accountId,
453 };
454 MutableHashMap.set(localRuntimeState.queues, queue.queueId, queue);
455 return queue;
456 }),
457 delete: Effect.fn(function* ({ output }) {
458 MutableHashMap.remove(localRuntimeState.queues, output.queueId);
459 // Legacy local-mode rows written before providerMode stamping can
460 // carry a real queue's id — remove the live queue too so migrating
461 // the row to a true local identity doesn't leak it.
462 if (isLiveId(output.queueId)) {
463 yield* queues
464 .deleteQueue({
465 accountId: output.accountId,
466 queueId: output.queueId,
467 })
468 .pipe(
469 Effect.retry({
470 while: (e) => e._tag === "QueueInUseByEventNotification",
471 schedule: Schedule.max([
472 Schedule.exponential("1 second"),
473 Schedule.recurs(8),
474 ]),
475 }),
476 Effect.catchTag(
477 ["QueueNotFound", "InvalidRoute"],
478 () => Effect.void,
479 ),
480 );
481 }
482 }),
483 };
484 }),
485 );
486
487export const QueueProvider = () =>
488 ProviderLayer.dual(Queue, {
489 local: () => ProviderLocal().pipe(Layer.provide(localRuntimeServices())),
490 live: () => ProviderLive(),
491 });