1import type * as cf from "@cloudflare/workers-types";
2import * as Cause from "effect/Cause";
3import * as Context from "effect/Context";
4import * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";
6import * as Layer from "effect/Layer";
7import * as Stream from "effect/Stream";
8import * as Namespace from "../../Namespace.ts";
9import { RuntimeContext } from "../../RuntimeContext.ts";
10import type { FunctionContext } from "../../Serverless/Function.ts";
11import * as DurationUtil from "../../Util/Duration.ts";
12import { isWorkerEvent, Worker } from "../Workers/Worker.ts";
13import type { Queue } from "./Queue.ts";
14import { Consumer } from "./Consumer.ts";
15
16/**
17 * Subscriber settings — the same shape Cloudflare's `Consumer`
18 * accepts. `consumeQueueMessages(queue, props, handler)` passes these
19 * through to the auto-created `Cloudflare.Queues.Consumer` so a single
20 * call captures both runtime and deploy-time intent.
21 */
22export interface MessagesProps {
23 /** Maximum messages per batch. */
24 batchSize?: number;
25 /** Maximum concurrent invocations. */
26 maxConcurrency?: number;
27 /** Maximum delivery attempts before dead-lettering. */
28 maxRetries?: number;
29 /**
30 * Wait time before flushing a partial batch. Rounded up to whole
31 * milliseconds when forwarded to Cloudflare.
32 */
33 maxWaitTime?: Duration.Input;
34 /**
35 * Backoff applied to a retry. Rounded up to whole seconds when
36 * forwarded to Cloudflare.
37 */
38 retryDelay?: Duration.Input;
39 /** Optional dead-letter queue name. */
40 deadLetterQueue?: string;
41}
42
43/**
44 * Convert a {@link MessagesProps} (with `Duration.Input` time fields)
45 * into the numeric settings shape Cloudflare's `Consumer` API
46 * expects. `maxWaitTime` is rounded up to whole milliseconds and
47 * `retryDelay` to whole seconds.
48 *
49 * Exposed for testing and for callers that want to mirror the
50 * conversion when wiring `Consumer` directly.
51 */
52export const toConsumerSettings = (props: MessagesProps) => ({
53 batchSize: props.batchSize,
54 maxConcurrency: props.maxConcurrency,
55 maxRetries: props.maxRetries,
56 maxWaitTimeMs: DurationUtil.toMillis(props.maxWaitTime),
57 retryDelay: DurationUtil.toSeconds(props.retryDelay),
58});
59
60/**
61 * A single queue message handed to the subscribe handler. Mirrors
62 * Cloudflare's runtime `Message<Body>` shape so per-message
63 * `ack()` / `retry()` semantics match the platform docs.
64 */
65export type Message<Body = unknown> = cf.Message<Body>;
66
67/**
68 * Subscribe to a Cloudflare Queue with an Effect stream handler.
69 *
70 * Mirrors `AWS.SQS.consumeQueueMessages(queue, handler)` on the
71 * Cloudflare side. Wires both halves of the consumer in one call:
72 *
73 * - **Runtime**: registers a `queue` event listener on the Worker.
74 * Each batch is piped through `process` as a `Stream.Stream`.
75 * - **Deploy-time**: yields a `Cloudflare.Queues.Consumer` resource
76 * so Cloudflare actually dispatches messages from `queue` to
77 * this Worker. No manual `Consumer` wiring needed in
78 * `alchemy.run.ts`.
79 *
80 * Acking semantics: if `process` succeeds, every message in the
81 * batch is `ack()`ed; if it fails, every message is `retry()`ed
82 * and Cloudflare applies `maxRetries` / `retryDelay` from the
83 * settings before dead-lettering. Per-message control is still
84 * available by calling `msg.ack()` / `msg.retry()` inside the
85 * handler.
86 * **Example:** Example
87 * ```typescript
88 * import * as Cloudflare from "alchemy/Cloudflare";
89 * import * as Duration from "effect/Duration";
90 * import * as Effect from "effect/Effect";
91 * import * as Stream from "effect/Stream";
92 *
93 * yield* Cloudflare.Queues.consumeQueueMessages<MyEvent>(
94 * queueResource,
95 * {
96 * batchSize: 25,
97 * maxRetries: 3,
98 * maxWaitTime: "5 seconds",
99 * retryDelay: Duration.seconds(30),
100 * },
101 * (stream) =>
102 * Stream.runForEach(stream, (msg) =>
103 * Effect.log(`event ${msg.body.id}`),
104 * ),
105 * );
106 * ```
107 *
108 * **Example:** Example
109 * ```typescript
110 * // Without options — handler is the second argument.
111 * yield* Cloudflare.Queues.consumeQueueMessages<MyEvent>(queueResource, (stream) =>
112 * Stream.runForEach(stream, (msg) => Effect.log(`event ${msg.body.id}`)),
113 * );
114 * ```
115 *
116 * @binding
117 * @product Queues
118 * @category Storage & Databases
119 */
120export function consumeQueueMessages<Body = unknown>(
121 queue: Queue,
122 process: (
123 stream: Stream.Stream<Message<Body>>,
124 ) => Effect.Effect<void, unknown, any>,
125): Effect.Effect<void, never, EventSource>;
126export function consumeQueueMessages<Body = unknown>(
127 queue: Queue,
128 props: MessagesProps,
129 process: (
130 stream: Stream.Stream<Message<Body>>,
131 ) => Effect.Effect<void, unknown, any>,
132): Effect.Effect<void, never, EventSource>;
133export function consumeQueueMessages<Body = unknown>(
134 queue: Queue,
135 propsOrProcess:
136 | MessagesProps
137 | ((
138 stream: Stream.Stream<Message<Body>>,
139 ) => Effect.Effect<void, unknown, any>),
140 maybeProcess?: (
141 stream: Stream.Stream<Message<Body>>,
142 ) => Effect.Effect<void, unknown, any>,
143): Effect.Effect<void, never, EventSource> {
144 const [props, process] =
145 typeof propsOrProcess === "function"
146 ? [{} as MessagesProps, propsOrProcess]
147 : [propsOrProcess, maybeProcess!];
148 return EventSource.use((source) => source<Body>(queue, props, process));
149}
150
151// `Req` is the handler's requirements. The service registers the
152// handler with the Worker's runtime context, where the runtime
153// machinery provides bindings and `WorkerEnvironment` when the
154// dispatch fires — so the requirement is satisfied at handler
155// invocation, NOT at subscribe time. We drop `Req` from the return
156// to keep init effects clean (mirrors `AWS.SQS.EventSourceService`).
157export type EventSourceService = <Body = unknown>(
158 queue: Queue,
159 props: MessagesProps,
160 process: (
161 stream: Stream.Stream<Message<Body>>,
162 ) => Effect.Effect<void, unknown, any>,
163) => Effect.Effect<void, never, never>;
164
165/**
166 * Service tag for the Cloudflare Queue event source. Provided by
167 * {@link EventSourceLive} on the Worker's runtime layer.
168 */
169export class EventSource extends Context.Service<
170 EventSource,
171 EventSourceService
172>()("Cloudflare.Queues.EventSource") {}
173
174/**
175 * Runtime layer for {@link consumeQueueMessages}. Wires each
176 * `consumeQueueMessages(queue, handler)` call in the Worker init phase to
177 * a `queue` event listener on the runtime context, and (at deploy
178 * time) yields the matching `Cloudflare.Queues.Consumer` resource so
179 * Cloudflare dispatches messages from the queue to this Worker.
180 *
181 * Provide alongside other Cloudflare runtime layers (e.g.
182 * `WriteQueueBinding`) on the Worker effect.
183 */
184export const EventSourceLive = Layer.effect(
185 EventSource,
186 Effect.gen(function* () {
187 const host = yield* Worker;
188 return Effect.fn(function* <Body, Req>(
189 queue: Queue,
190 props: MessagesProps,
191 process: (
192 stream: Stream.Stream<Message<Body>>,
193 ) => Effect.Effect<void, unknown, any>,
194 ) {
195 // Deploy-time: yield the Consumer resource as a sibling of the
196 // Worker so Cloudflare dispatches messages from the queue to it.
197 // Skipped once running inside the deployed Worker (the global
198 // guard), where the only work is registering the runtime handler
199 // below. Namespaced under the host so the Consumer's logical
200 // identity matches the previous Binding.Policy.
201 if (!globalThis.__ALCHEMY_RUNTIME__) {
202 yield* Namespace.push(
203 host.LogicalId,
204 Effect.gen(function* () {
205 // The engine creates / updates / destroys the Consumer
206 // alongside the Worker's lifecycle; the consumer's
207 // reconciler waits for the Worker upload to expose the
208 // `queue` handler before completing (see PR #257 for the
209 // 11001 retry).
210 yield* Consumer(`${queue.LogicalId}Consumer`, {
211 queueId: queue.queueId,
212 scriptName: host.workerName,
213 settings: toConsumerSettings(props),
214 deadLetterQueue: props.deadLetterQueue,
215 });
216 }),
217 );
218 }
219
220 // Resolve the runtime context per-call rather than at layer
221 // construction. Capturing it on the layer would leak the
222 // requirement past `PlatformServices` exclusion when the
223 // Worker typechecks its init effect.
224 const ctx = (yield* RuntimeContext) as unknown as FunctionContext;
225 // Capture the queue-name accessor once; the listener body
226 // re-resolves it per event via `yield* QueueName`. A worker
227 // can consume multiple queues — each subscribe registers its
228 // own listener and they all see every queue event, so the
229 // queue-name match is what scopes the handler.
230 const QueueName = yield* queue.queueName;
231
232 yield* ctx.listen<void, Req>((event) => {
233 if (!isWorkerEvent(event) || event.type !== "queue") return;
234 const batch = event.input as cf.MessageBatch<Body>;
235
236 return Effect.gen(function* () {
237 const queueName = yield* QueueName;
238 if (batch.queue !== queueName) return;
239
240 yield* process(Stream.fromIterable(batch.messages)).pipe(
241 Effect.tap(() =>
242 Effect.sync(() => {
243 for (const msg of batch.messages) msg.ack();
244 }),
245 ),
246 Effect.onError((cause) =>
247 Effect.sync(() => {
248 // Surface the failure so the operator sees what
249 // tripped the retry path; without this the only
250 // signal is the message reappearing on the next
251 // attempt.
252 console.error(
253 `[EventSource] handler failed on queue ` +
254 `"${queueName}": ${Cause.pretty(cause)}`,
255 );
256 for (const msg of batch.messages) msg.retry();
257 }),
258 ),
259 Effect.catchCause(() => Effect.void),
260 );
261 });
262 });
263 }) as EventSourceService;
264 }),
265);