1import * as Binding from "alchemy/Binding";
2import type * as lambda from "aws-lambda";
3import type * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import type { Queue } from "./Queue.ts";
6
7export type SQSRecord = lambda.SQSRecord;
8
9export type SQSEvent = lambda.SQSEvent;
10
11export interface MessagesProps extends QueueEventSourceProps {
12 /**
13 * How long each `ReceiveMessage` long-polls for messages when using the
14 * process (run) path (e.g. `"20 seconds"`). Rounded to whole seconds on
15 * the wire.
16 * @default 20 seconds
17 */
18 waitTime?: Duration.Input;
19 /**
20 * Maximum number of messages to receive per poll when using the process (run) path.
21 * @default 10
22 */
23 maxNumberOfMessages?: number;
24}
25
26/**
27 * Lambda's partial-batch response for SQS. Every message listed in
28 * `batchItemFailures` (by `itemIdentifier` = `messageId`) is retried; every
29 * other message in the batch is deleted.
30 */
31export type SQSBatchResponse = lambda.SQSBatchResponse;
32
33/**
34 * Handler for one delivered batch of SQS messages: receives the
35 * {@link SQSEvent} and returns Lambda's {@link SQSBatchResponse} (or `void`
36 * when every message succeeded). A failure or defect retries the whole batch.
37 */
38export type QueueMessagesHandler<E = never, Req = never> = (
39 event: SQSEvent,
40) => Effect.Effect<SQSBatchResponse | void, E, Req>;
41
42/**
43 * Subscribe an Effect handler to batches of messages delivered from an SQS
44 * {@link Queue}.
45 *
46 * The handler receives the {@link SQSEvent} and returns Lambda's
47 * {@link SQSBatchResponse}: list the `messageId` of each message that failed
48 * in `batchItemFailures` and only those are retried. Return `void` when every
49 * message succeeded. If the handler fails, the whole batch is retried.
50 *
51 * Each batch runs inside a `process <queue>` span (`SpanKind.CONSUMER`) linked
52 * to the `traceparent` message attribute of each message that carries one.
53 *
54 * @param queue The SQS queue to consume messages from.
55 * @param props Optional event-source configuration.
56 * @param process The batch handler (last argument).
57 *
58 * @example Report only the failed messages
59 * ```typescript
60 * yield* SQS.consumeQueueMessages(
61 * queue,
62 * Effect.fn(function* (event) {
63 * const failed = yield* Effect.filter(
64 * event.Records,
65 * (record) =>
66 * Effect.try(() => JSON.parse(record.body)).pipe(
67 * Effect.flatMap((order) => putItem({ Item: { id: { S: order.id } } })),
68 * Effect.as(false),
69 * Effect.catchCause(() => Effect.succeed(true)),
70 * ),
71 * { concurrency: "unbounded" },
72 * );
73 * return {
74 * batchItemFailures: failed.map((record) => ({ itemIdentifier: record.messageId })),
75 * };
76 * }),
77 * );
78 * ```
79 *
80 * @example Process the whole batch at once
81 * ```typescript
82 * const sink = yield* SQS.QueueSink(results);
83 *
84 * yield* SQS.consumeQueueMessages(
85 * queue,
86 * { batchSize: 10 },
87 * Effect.fn(function* (event) {
88 * // a failure retries the whole batch
89 * yield* Stream.fromIterable(event.Records).pipe(
90 * Stream.map((record) => ({ MessageBody: record.body })),
91 * Stream.run(sink),
92 * );
93 * }),
94 * );
95 * ```
96 */
97export function consumeQueueMessages<Q extends Queue, E = never, Req = never, ResourceReq = never>(
98 queue: Q | Effect.Effect<Q, never, ResourceReq>,
99 process: QueueMessagesHandler<E, Req>,
100): Effect.Effect<void, never, QueueEventSource | ResourceReq>;
101export function consumeQueueMessages<Q extends Queue, E = never, Req = never, ResourceReq = never>(
102 queue: Q | Effect.Effect<Q, never, ResourceReq>,
103 props: MessagesProps,
104 process: QueueMessagesHandler<E, Req>,
105): Effect.Effect<void, never, QueueEventSource | ResourceReq>;
106export function consumeQueueMessages<Q extends Queue, E = never, Req = never, ResourceReq = never>(
107 queue: Q | Effect.Effect<Q, never, ResourceReq>,
108 propsOrProcess: MessagesProps | QueueMessagesHandler<E, Req>,
109 maybeProcess?: QueueMessagesHandler<E, Req>,
110): Effect.Effect<void, never, QueueEventSource | ResourceReq> {
111 const [props, process] =
112 typeof propsOrProcess === "function"
113 ? [{} as MessagesProps, propsOrProcess]
114 : [propsOrProcess, maybeProcess!];
115 // Accept the queue or the Effect that declares it, like bindings do.
116 const resolved = Effect.isEffect(queue) ? queue : Effect.succeed(queue);
117 return resolved.pipe(
118 Effect.flatMap((value) => QueueEventSource.use((source) => source(value, props, process))),
119 );
120}
121
122/**
123 * Event source connecting an SQS {@link Queue} to the hosting compute
124 * (Lambda function or ServerHost process).
125 *
126 * The contract is a `Binding.Service`; the host-specific implementation
127 * layers are `Lambda.QueueEventSource` (event-source mapping + runtime
128 * dispatch) and `AWS.Server.SQSQueueEventSource` (long-poll receive loop).
129 * Consume it through the {@link consumeQueueMessages} helper.
130 * ### Consuming a Queue
131 * **Example:** Consume Messages in a Lambda Function
132 * ```typescript
133 * export default WorkerFunction.make(
134 * { main: import.meta.url },
135 * Effect.gen(function* () {
136 * const queue = yield* SQS.Queue("Jobs");
137 *
138 * // registers the event-source mapping and the runtime dispatcher
139 * yield* SQS.consumeQueueMessages(
140 * queue,
141 * { batchSize: 10 },
142 * Effect.fn(function* (event) {
143 * for (const record of event.Records) {
144 * yield* Effect.log(record.body);
145 * }
146 * }),
147 * );
148 * }).pipe(Effect.provide(Lambda.QueueEventSource)),
149 * );
150 * ```
151 *
152 * @binding
153 */
154export interface QueueEventSource extends Binding.Service<
155 QueueEventSource,
156 "AWS.SQS.QueueEventSource",
157 QueueEventSourceService
158> {}
159
160export const QueueEventSource = Binding.Service<QueueEventSource>("AWS.SQS.QueueEventSource");
161
162export interface QueueEventSourceProps {
163 /**
164 * The maximum number of records in each batch that Lambda pulls from the queue.
165 * @default 10
166 */
167 batchSize?: number;
168 /**
169 * The maximum amount of time that Lambda spends gathering records before
170 * invoking the function (e.g. `"5 seconds"`). Rounded to whole seconds on
171 * the wire.
172 * @default 0
173 */
174 maximumBatchingWindow?: Duration.Input;
175}
176
177export type QueueEventSourceService = <E = never, Req = never>(
178 queue: Queue,
179 props: MessagesProps,
180 process: QueueMessagesHandler<E, Req>,
181) => Effect.Effect<void, never, never>;