1import type * as lambda from "aws-lambda";
2import type * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import * as Stream from "effect/Stream";
5import * as Binding from "../../Binding.ts";
6import type { Queue } from "./Queue.ts";
7
8export type SQSRecord = lambda.SQSRecord;
9
10export interface MessagesProps extends QueueEventSourceProps {
11 /**
12 * How long each `ReceiveMessage` long-polls for messages when using the
13 * process (run) path (e.g. `"20 seconds"`). Rounded to whole seconds on
14 * the wire.
15 * @default 20 seconds
16 */
17 waitTime?: Duration.Input;
18 /**
19 * Maximum number of messages to receive per poll when using the process (run) path.
20 * @default 10
21 */
22 maxNumberOfMessages?: number;
23}
24
25type MessagesHandler<Req> = (
26 stream: Stream.Stream<SQSRecord>,
27) => Effect.Effect<void, never, Req>;
28
29/**
30 * Subscribe an Effect handler to messages produced by an SQS {@link Queue}.
31 *
32 * @param queue The SQS queue to consume messages from.
33 * @param props Optional event-source configuration.
34 * @param process The handler invoked with a stream of SQS records (last argument).
35 *
36 * @example
37 * ```typescript
38 * yield* SQS.consumeQueueMessages(queue, (records) =>
39 * records.pipe(Stream.runForEach((record) => Effect.log(record.body))),
40 * );
41 * ```
42 *
43 * @example With batching configuration
44 * ```typescript
45 * yield* SQS.consumeQueueMessages(queue, { batchSize: 10 }, (records) =>
46 * records.pipe(
47 * Stream.map((record) => ({ MessageBody: record.body })),
48 * Stream.run(sink),
49 * Effect.orDie,
50 * ),
51 * );
52 * ```
53 */
54export function consumeQueueMessages<
55 Q extends Queue,
56 Req = never,
57 ResourceReq = never,
58>(
59 queue: Q | Effect.Effect<Q, never, ResourceReq>,
60 process: MessagesHandler<Req>,
61): Effect.Effect<void, never, QueueEventSource | ResourceReq>;
62export function consumeQueueMessages<
63 Q extends Queue,
64 Req = never,
65 ResourceReq = never,
66>(
67 queue: Q | Effect.Effect<Q, never, ResourceReq>,
68 props: MessagesProps,
69 process: MessagesHandler<Req>,
70): Effect.Effect<void, never, QueueEventSource | ResourceReq>;
71export function consumeQueueMessages<
72 Q extends Queue,
73 Req = never,
74 ResourceReq = never,
75>(
76 queue: Q | Effect.Effect<Q, never, ResourceReq>,
77 propsOrProcess: MessagesProps | MessagesHandler<Req>,
78 maybeProcess?: MessagesHandler<Req>,
79): Effect.Effect<void, never, QueueEventSource | ResourceReq> {
80 const [props, process] =
81 typeof propsOrProcess === "function"
82 ? [{} as MessagesProps, propsOrProcess]
83 : [propsOrProcess, maybeProcess!];
84 // Accept the queue or the Effect that declares it, like bindings do.
85 const resolved = Effect.isEffect(queue) ? queue : Effect.succeed(queue);
86 return resolved.pipe(
87 Effect.flatMap((value) =>
88 QueueEventSource.use((source) => source(value, props, process)),
89 ),
90 );
91}
92
93/**
94 * Event source connecting an SQS {@link Queue} to the hosting compute
95 * (Lambda function or ServerHost process).
96 *
97 * The contract is a `Binding.Service`; the host-specific implementation
98 * layers are `Lambda.QueueEventSource` (event-source mapping + runtime
99 * dispatch) and `Server.SQSQueueEventSource` (long-poll receive loop).
100 * Consume it through the {@link consumeQueueMessages} helper.
101 * ### Consuming a Queue
102 * **Example:** Consume Messages in a Lambda Function
103 * ```typescript
104 * export default WorkerFunction.make(
105 * { main: import.meta.url },
106 * Effect.gen(function* () {
107 * const queue = yield* SQS.Queue("Jobs");
108 *
109 * // registers the event-source mapping and the runtime dispatcher
110 * yield* SQS.consumeQueueMessages(queue, { batchSize: 10 }, (records) =>
111 * records.pipe(
112 * Stream.runForEach((record) => Effect.log(record.body)),
113 * ),
114 * );
115 * }).pipe(Effect.provide(Lambda.QueueEventSource)),
116 * );
117 * ```
118 *
119 * @binding
120 */
121export interface QueueEventSource extends Binding.Service<
122 QueueEventSource,
123 "AWS.SQS.QueueEventSource",
124 QueueEventSourceService
125> {}
126
127export const QueueEventSource = Binding.Service<QueueEventSource>(
128 "AWS.SQS.QueueEventSource",
129);
130
131export interface QueueEventSourceProps {
132 /**
133 * The maximum number of records in each batch that Lambda pulls from the queue.
134 * @default 10
135 */
136 batchSize?: number;
137 /**
138 * The maximum amount of time that Lambda spends gathering records before
139 * invoking the function (e.g. `"5 seconds"`). Rounded to whole seconds on
140 * the wire.
141 * @default 0
142 */
143 maximumBatchingWindow?: Duration.Input;
144}
145
146export type QueueEventSourceService = <Req = never>(
147 bucket: Queue,
148 props: MessagesProps,
149 process: (
150 stream: Stream.Stream<SQSRecord>,
151 ) => Effect.Effect<void, never, Req>,
152) => Effect.Effect<void, never, never>;