1import * as Context from "effect/Context";
2import type * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import * as Stream from "effect/Stream";
5import type { Broker } from "./Broker.ts";
6
7/**
8 * A single message delivered by an Amazon MQ event source. The shape differs
9 * slightly between ActiveMQ (`aws:mq`) and RabbitMQ (`aws:rmq`) — the common
10 * fields are typed here and the raw payload is available for engine-specific
11 * access.
12 */
13export interface MQMessage {
14 /** Base64-encoded message body. */
15 readonly data?: string;
16 /** ActiveMQ message id. */
17 readonly messageID?: string;
18 /** ActiveMQ message type (e.g. `jms/text-message`). */
19 readonly messageType?: string;
20 /** ActiveMQ destination the message was published to. */
21 readonly destination?: { readonly physicalName?: string };
22 /** Whether the message was redelivered. */
23 readonly redelivered?: boolean;
24 /** RabbitMQ basic properties. */
25 readonly basicProperties?: Record<string, unknown>;
26 /** Any additional engine-specific fields. */
27 readonly [key: string]: unknown;
28}
29
30/**
31 * The raw ActiveMQ (`aws:mq`) event Lambda receives.
32 */
33export interface ActiveMQEvent {
34 readonly eventSource: "aws:mq";
35 readonly eventSourceArn: string;
36 readonly messages: MQMessage[];
37}
38
39/**
40 * The raw RabbitMQ (`aws:rmq`) event Lambda receives.
41 */
42export interface RabbitMQEvent {
43 readonly eventSource: "aws:rmq";
44 readonly eventSourceArn: string;
45 readonly rmqMessagesByQueue: { readonly [queue: string]: MQMessage[] };
46}
47
48export type MQEvent = ActiveMQEvent | RabbitMQEvent;
49
50export interface BrokerEventSourceProps {
51 /**
52 * Destination queue name(s) on the broker to consume. ActiveMQ supports a
53 * single queue; RabbitMQ supports multiple.
54 */
55 queues: string[];
56 /**
57 * ARN of a Secrets Manager secret holding the broker credentials as
58 * `{ "username": "...", "password": "..." }`. Lambda uses these
59 * (`BASIC_AUTH`) to connect to the broker.
60 */
61 credentialsSecretArn: string;
62 /**
63 * The maximum number of messages in each batch that Lambda pulls from the
64 * broker.
65 * @default 100
66 */
67 batchSize?: number;
68 /**
69 * The maximum time Lambda spends gathering records before invoking the
70 * function (e.g. `"5 seconds"`). Rounded to whole seconds on the wire.
71 */
72 maximumBatchingWindow?: Duration.Input;
73 /**
74 * Whether the event source mapping is active.
75 * @default true
76 */
77 enabled?: boolean;
78}
79
80type MessagesHandler<Req> = (
81 stream: Stream.Stream<MQMessage>,
82) => Effect.Effect<void, never, Req>;
83
84/**
85 * Subscribe an Effect handler to messages produced by an Amazon MQ
86 * {@link Broker}. Lambda polls the named broker queue(s) using the supplied
87 * Secrets Manager credentials and invokes the handler with a stream of
88 * messages.
89 *
90 * Call it in the function's init phase and provide the host implementation
91 * layer (`Lambda.BrokerEventSource`) on the function Effect — the
92 * event-source mapping, IAM grants, and runtime dispatch are registered
93 * automatically.
94 *
95 * @example
96 * ```typescript
97 * // init — inside the Lambda function's setup Effect
98 * const broker = yield* MQ.Broker("Orders", {
99 * engineType: "ACTIVEMQ",
100 * engineVersion: "5.18",
101 * hostInstanceType: "mq.t3.micro",
102 * users: [{ username: "app", password: appPassword }],
103 * });
104 *
105 * yield* MQ.consumeBrokerMessages(
106 * broker,
107 * {
108 * queues: ["orders"],
109 * credentialsSecretArn: secret.secretArn,
110 * batchSize: 10,
111 * },
112 * (messages) =>
113 * messages.pipe(
114 * Stream.runForEach((message) =>
115 * // message bodies arrive base64-encoded
116 * Effect.log(atob(message.data ?? "")),
117 * ),
118 * ),
119 * );
120 * ```
121 *
122 * Provide the Lambda implementation layer on the function:
123 * ```typescript
124 * export default OrdersFunction.make(
125 * { main },
126 * Effect.gen(function* () {
127 * // ... consumeBrokerMessages(...) as above
128 * return { fetch: Effect.succeed(HttpServerResponse.text("ok")) };
129 * }).pipe(Effect.provide(Lambda.BrokerEventSource)),
130 * );
131 * ```
132 *
133 * @param broker The Amazon MQ broker to consume from.
134 * @param props Queues, credentials secret, and batching configuration.
135 * @param process The handler invoked with a stream of MQ messages.
136 */
137export function consumeBrokerMessages<B extends Broker, Req = never>(
138 broker: B,
139 props: BrokerEventSourceProps,
140 process: MessagesHandler<Req>,
141): Effect.Effect<void, never, BrokerEventSource> {
142 return BrokerEventSource.use((source) => source(broker, props, process));
143}
144
145/**
146 * Event source connecting an Amazon MQ {@link Broker} to the hosting Lambda
147 * function. Prefer the {@link consumeBrokerMessages} helper; this service is
148 * the underlying contract, implemented by the `Lambda.BrokerEventSource`
149 * layer (which registers the event-source mapping, IAM grants, and runtime
150 * dispatch).
151 *
152 * **Example:** Example
153 * ```typescript
154 * // equivalent to consumeBrokerMessages(broker, props, process)
155 * yield* BrokerEventSource.use((source) =>
156 * source(broker, props, process),
157 * );
158 * ```
159 *
160 * @binding
161 */
162export class BrokerEventSource extends Context.Service<
163 BrokerEventSource,
164 BrokerEventSourceService
165>()("AWS.MQ.BrokerEventSource") {}
166
167export type BrokerEventSourceService = <Req = never>(
168 broker: Broker,
169 props: BrokerEventSourceProps,
170 process: (
171 stream: Stream.Stream<MQMessage>,
172 ) => Effect.Effect<void, never, Req>,
173) => Effect.Effect<void, never, never>;