1import * as iot from "@distilled.cloud/aws/iot";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import { createPhysicalName } from "../../PhysicalName.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { hasAlchemyTags } from "../../Tags.ts";
10import { AWSEnvironment } from "../Environment.ts";
11import type { Providers } from "../Providers.ts";
12import { readIotTags, sanitizeRuleName, syncIotTags } from "./internal.ts";
13
14export type Action = iot.Action;
15
16export interface TopicRuleProps {
17 /**
18 * Name of the rule. Rule names may only contain letters, numbers, and
19 * underscores (`[a-zA-Z0-9_]`). If omitted, a unique name is generated.
20 * Changing it replaces the rule.
21 */
22 ruleName?: string;
23
24 /**
25 * The SQL statement used to query the topic, e.g.
26 * `SELECT * FROM 'my/topic'`.
27 */
28 sql: string;
29
30 /**
31 * The actions associated with the rule (Lambda, SQS, SNS, republish, ...).
32 */
33 actions: Action[];
34
35 /**
36 * A textual description of the rule.
37 */
38 description?: string;
39
40 /**
41 * Whether the rule is disabled.
42 * @default false
43 */
44 ruleDisabled?: boolean;
45
46 /**
47 * The version of the SQL rules engine to use (`2015-10-08` or `2016-03-23`).
48 * @default "2016-03-23"
49 */
50 awsIotSqlVersion?: string;
51
52 /**
53 * The action to take when an error occurs.
54 */
55 errorAction?: Action;
56
57 /**
58 * User tags to attach to the rule.
59 */
60 tags?: Record<string, string>;
61}
62
63export interface TopicRule extends Resource<
64 "AWS.IoT.TopicRule",
65 TopicRuleProps,
66 {
67 /** The name of the rule. */
68 ruleName: string;
69 /** The ARN of the rule. */
70 ruleArn: string;
71 },
72 never,
73 Providers
74> {}
75
76/**
77 * An AWS IoT topic rule — evaluates an SQL statement against messages
78 * published to MQTT topics and routes matching messages to one or more
79 * actions (invoke a Lambda, enqueue to SQS, republish, etc.).
80 *
81 * ### Creating a Rule
82 * **Example:** Route Messages to a Lambda
83 * ```typescript
84 * const rule = yield* TopicRule("ingest", {
85 * sql: "SELECT * FROM 'sensors/+/telemetry'",
86 * actions: [{ lambda: { functionArn: yield* fn.functionArn } }],
87 * });
88 * ```
89 *
90 * @resource
91 */
92export const TopicRule = Resource<TopicRule>("AWS.IoT.TopicRule");
93
94export const TopicRuleProvider = () =>
95 Provider.effect(
96 TopicRule,
97 Effect.gen(function* () {
98 const createName = Effect.fn(function* (
99 id: string,
100 props: TopicRuleProps,
101 ) {
102 return (
103 props.ruleName ??
104 sanitizeRuleName(
105 yield* createPhysicalName({ id, delimiter: "_", maxLength: 128 }),
106 )
107 );
108 });
109
110 const ruleArnOf = (accountId: string, region: string, ruleName: string) =>
111 `arn:aws:iot:${region}:${accountId}:rule/${ruleName}`;
112
113 const buildPayload = (props: TopicRuleProps): iot.TopicRulePayload => ({
114 sql: props.sql,
115 description: props.description,
116 actions: props.actions,
117 ruleDisabled: props.ruleDisabled,
118 awsIotSqlVersion: props.awsIotSqlVersion ?? "2016-03-23",
119 errorAction: props.errorAction,
120 });
121
122 return TopicRule.Provider.of({
123 stables: ["ruleName", "ruleArn"],
124 list: () =>
125 iot.listTopicRules.pages({}).pipe(
126 Stream.runCollect,
127 Effect.map((chunk) =>
128 Array.from(chunk).flatMap((page) =>
129 (page.rules ?? [])
130 .filter((r) => r.ruleName != null && r.ruleArn != null)
131 .map((r) => ({
132 ruleName: r.ruleName!,
133 ruleArn: r.ruleArn!,
134 })),
135 ),
136 ),
137 ),
138 read: Effect.fn(function* ({ id, olds, output }) {
139 const { accountId, region } = yield* AWSEnvironment.current;
140 const ruleName =
141 output?.ruleName ?? (yield* createName(id, olds ?? {}));
142 const found = yield* iot
143 .getTopicRule({ ruleName })
144 .pipe(
145 Effect.catchTag("TopicRuleNotFound", () =>
146 Effect.succeed(undefined),
147 ),
148 );
149 if (!found) return undefined;
150 const ruleArn =
151 found.ruleArn ?? ruleArnOf(accountId, region, ruleName);
152 const attrs = { ruleName, ruleArn };
153 const tags = yield* readIotTags(ruleArn);
154 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
155 }),
156 diff: Effect.fn(function* ({ id, news, olds }) {
157 if (!isResolved(news)) return undefined;
158 const oldName = yield* createName(id, olds);
159 const newName = yield* createName(id, news);
160 if (oldName !== newName) return { action: "replace" } as const;
161 return undefined;
162 }),
163 reconcile: Effect.fn(function* ({ id, news, output, session }) {
164 const { accountId, region } = yield* AWSEnvironment.current;
165 const ruleName = output?.ruleName ?? (yield* createName(id, news));
166 const ruleArn = ruleArnOf(accountId, region, ruleName);
167 const payload = buildPayload(news);
168
169 // OBSERVE
170 const live = yield* iot
171 .getTopicRule({ ruleName })
172 .pipe(
173 Effect.catchTag("TopicRuleNotFound", () =>
174 Effect.succeed(undefined),
175 ),
176 );
177
178 // ENSURE / SYNC — createTopicRule for a new rule, replaceTopicRule
179 // (a full upsert of the payload) for an existing one.
180 if (live === undefined) {
181 yield* iot
182 .createTopicRule({ ruleName, topicRulePayload: payload })
183 .pipe(
184 Effect.catchTag("ResourceAlreadyExistsException", () =>
185 iot.replaceTopicRule({ ruleName, topicRulePayload: payload }),
186 ),
187 );
188 } else {
189 yield* iot.replaceTopicRule({
190 ruleName,
191 topicRulePayload: payload,
192 });
193 }
194
195 // SYNC tags
196 yield* syncIotTags(ruleArn, id, news.tags);
197
198 yield* session.note(ruleName);
199 return { ruleName, ruleArn };
200 }),
201 delete: Effect.fn(function* ({ output }) {
202 yield* iot
203 .deleteTopicRule({ ruleName: output.ruleName })
204 .pipe(Effect.catchTag("TopicRuleNotFound", () => Effect.void));
205 }),
206 });
207 }),
208 );