1import * as Context from "effect/Context";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4
5/**
6 * The JSON message delivered to the handler. This is exactly the payload
7 * produced by the rule's SQL `SELECT` — for a `SELECT *` rule it is the raw
8 * MQTT message body, and for enriched SQL it also carries the selected
9 * metadata (`topic()`, `timestamp()`, ...).
10 */
11export type IoTMessage = Record<string, unknown>;
12
13export interface TopicRuleEventSourceProps {
14 /**
15 * Explicit rule name. Rule names may only contain `[a-zA-Z0-9_]`.
16 * If omitted, a unique name is generated.
17 */
18 ruleName?: string;
19
20 /**
21 * Override the rule's SQL statement. When omitted, the rule uses
22 * `SELECT * FROM '${topicFilter}'`.
23 */
24 sql?: string;
25
26 /**
27 * The SQL rules engine version.
28 * @default "2016-03-23"
29 */
30 awsIotSqlVersion?: string;
31}
32
33type MessagesHandler<Req> = (
34 stream: Stream.Stream<IoTMessage>,
35) => Effect.Effect<void, never, Req>;
36
37/**
38 * Invoke an Effect handler for every MQTT message matching an IoT topic
39 * filter, by creating an IoT {@link TopicRule} with a Lambda action targeting
40 * the current function.
41 *
42 * Provide the host-specific implementation layer on the Function —
43 * `AWS.Lambda.TopicRuleEventSource` creates the topic rule and dispatches
44 * matching invocations to the handler.
45 *
46 * @param topicFilter The MQTT topic (filter) to subscribe to, e.g.
47 * `sensors/+/telemetry`.
48 * @param props Optional rule configuration.
49 * @param process The handler invoked with a stream of matching messages.
50 *
51 * @example Consume MQTT messages in a Lambda
52 * ```typescript
53 * export default IngestFunction.make(
54 * { main: import.meta.url },
55 * Effect.gen(function* () {
56 * const table = yield* Table;
57 * const putItem = yield* AWS.DynamoDB.PutItem(table);
58 *
59 * yield* AWS.IoT.consumeTopicMessages("sensors/+/telemetry", (messages) =>
60 * messages.pipe(
61 * Stream.runForEach((message) =>
62 * putItem({
63 * Item: { pk: { S: String(message.deviceId) } },
64 * }),
65 * ),
66 * Effect.orDie,
67 * ),
68 * );
69 *
70 * return {};
71 * }).pipe(
72 * Effect.provide(
73 * Layer.mergeAll(AWS.Lambda.TopicRuleEventSource, AWS.DynamoDB.PutItemHttp),
74 * ),
75 * ),
76 * );
77 * ```
78 *
79 * @example Enrich the payload with rule SQL
80 * ```typescript
81 * yield* AWS.IoT.consumeTopicMessages(
82 * "sensors/+/telemetry",
83 * { sql: "SELECT *, topic(2) AS deviceId FROM 'sensors/+/telemetry'" },
84 * (messages) =>
85 * messages.pipe(
86 * Stream.runForEach((message) => Effect.log(message.deviceId)),
87 * Effect.orDie,
88 * ),
89 * );
90 * ```
91 */
92export function consumeTopicMessages<Req = never>(
93 topicFilter: string,
94 process: MessagesHandler<Req>,
95): Effect.Effect<void, never, TopicRuleEventSource>;
96export function consumeTopicMessages<Req = never>(
97 topicFilter: string,
98 props: TopicRuleEventSourceProps,
99 process: MessagesHandler<Req>,
100): Effect.Effect<void, never, TopicRuleEventSource>;
101export function consumeTopicMessages<Req = never>(
102 topicFilter: string,
103 propsOrProcess: TopicRuleEventSourceProps | MessagesHandler<Req>,
104 maybeProcess?: MessagesHandler<Req>,
105): Effect.Effect<void, never, TopicRuleEventSource> {
106 const [props, process] =
107 typeof propsOrProcess === "function"
108 ? [{} as TopicRuleEventSourceProps, propsOrProcess]
109 : [propsOrProcess, maybeProcess!];
110 return TopicRuleEventSource.use((source) =>
111 source(topicFilter, props, process),
112 );
113}
114
115/**
116 * Event source connecting an IoT topic filter to the hosting compute.
117 *
118 * The contract is a Context service consumed via
119 * {@link consumeTopicMessages}; the Lambda implementation layer is
120 * `AWS.Lambda.TopicRuleEventSource`, which deploys an IoT {@link TopicRule}
121 * with a Lambda action (plus the invoke permission) and streams matching
122 * messages into the registered handler.
123 *
124 * @binding
125 */
126export class TopicRuleEventSource extends Context.Service<
127 TopicRuleEventSource,
128 TopicRuleEventSourceService
129>()("AWS.IoT.TopicRuleEventSource") {}
130
131export type TopicRuleEventSourceService = <Req = never>(
132 topicFilter: string,
133 props: TopicRuleEventSourceProps,
134 process: (
135 stream: Stream.Stream<IoTMessage>,
136 ) => Effect.Effect<void, never, Req>,
137) => Effect.Effect<void, never, never>;