1import * as iotw from "@distilled.cloud/aws/iot-wireless";
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 {
10 createInternalTags,
11 createTagsList,
12 hasAlchemyTags,
13} from "../../Tags.ts";
14import type { Providers } from "../Providers.ts";
15import { readIotWirelessTags, syncIotWirelessTags } from "./internal.ts";
16
17export interface DestinationProps {
18 /**
19 * Name of the destination. Names are unique per account/region and may
20 * contain only alphanumerics, dashes, and underscores. If omitted, a
21 * deterministic physical name is generated. Changing the name replaces
22 * the destination.
23 */
24 name?: string;
25 /**
26 * How `expression` is interpreted: `RuleName` routes uplinks to an AWS
27 * IoT rule; `MqttTopic` publishes them to an MQTT topic.
28 */
29 expressionType: iotw.ExpressionType;
30 /**
31 * The IoT rule name or MQTT topic uplink messages are routed to. The
32 * referenced rule does not need to exist when the destination is created.
33 */
34 expression: string;
35 /**
36 * Human-readable description of the destination.
37 */
38 description?: string;
39 /**
40 * ARN of the IAM role IoT Wireless assumes to deliver messages to the
41 * rule or topic. The role's trust policy must allow
42 * `iotwireless.amazonaws.com` to assume it.
43 */
44 roleArn: string;
45 /**
46 * Tags applied to the destination. Alchemy ownership tags are merged in
47 * automatically.
48 */
49 tags?: Record<string, string>;
50}
51
52export interface Destination extends Resource<
53 "AWS.IoTWireless.Destination",
54 DestinationProps,
55 {
56 /** Name of the destination (its unique identifier). */
57 destinationName: string;
58 /** ARN of the destination. */
59 destinationArn: string;
60 /** How the expression is interpreted (`RuleName` or `MqttTopic`). */
61 expressionType: iotw.ExpressionType;
62 /** The IoT rule name or MQTT topic uplinks are routed to. */
63 expression: string;
64 /** ARN of the delivery IAM role. */
65 roleArn: string;
66 },
67 never,
68 Providers
69> {}
70
71/**
72 * An AWS IoT Core for LoRaWAN destination — the routing rule that delivers
73 * uplink messages from wireless devices to an AWS IoT rule or MQTT topic.
74 *
75 * The destination name is its identity (changing it replaces the
76 * destination); the expression, expression type, description, role, and
77 * tags all update in place.
78 * ### Creating Destinations
79 * **Example:** Route uplinks to an IoT rule
80 * ```typescript
81 * import * as IoTWireless from "alchemy/AWS/IoTWireless";
82 *
83 * const destination = yield* IoTWireless.Destination("Uplinks", {
84 * expressionType: "RuleName",
85 * expression: "process_sensor_uplinks",
86 * roleArn: deliveryRole.roleArn,
87 * });
88 * ```
89 *
90 * **Example:** Publish uplinks straight to an MQTT topic
91 * ```typescript
92 * const destination = yield* IoTWireless.Destination("Uplinks", {
93 * expressionType: "MqttTopic",
94 * expression: "sensors/uplinks",
95 * roleArn: deliveryRole.roleArn,
96 * });
97 * ```
98 *
99 * ### Delivery Role
100 * **Example:** IAM role IoT Wireless assumes for delivery
101 * ```typescript
102 * const deliveryRole = yield* IAM.Role("IotWirelessDelivery", {
103 * assumeRolePolicyDocument: {
104 * Version: "2012-10-17",
105 * Statement: [{
106 * Effect: "Allow",
107 * Principal: { Service: "iotwireless.amazonaws.com" },
108 * Action: ["sts:AssumeRole"],
109 * }],
110 * },
111 * policies: [{
112 * policyName: "deliver",
113 * policyDocument: {
114 * Version: "2012-10-17",
115 * Statement: [{
116 * Effect: "Allow",
117 * Action: ["iot:DescribeEndpoint", "iot:Publish"],
118 * Resource: ["*"],
119 * }],
120 * },
121 * }],
122 * });
123 * ```
124 *
125 * ### Consuming Uplinks in a Function
126 * Uplinks are delivered through AWS IoT Core. For a `RuleName` destination,
127 * `IoTWireless.consumeUplinks` (see {@link DestinationEventSource}) creates
128 * the named IoT rule targeting the current Lambda and invokes the handler
129 * for every uplink. Alternatively, point an `MqttTopic` destination at a
130 * topic and consume it with `AWS.IoT.consumeTopicMessages`.
131 * **Example:** Route Device Uplinks into a Lambda
132 * ```typescript
133 * const destination = yield* IoTWireless.Destination("Uplinks", {
134 * expressionType: "RuleName",
135 * expression: "sensor_uplinks",
136 * roleArn: deliveryRole.roleArn,
137 * });
138 *
139 * // inside the Function effect (provide Lambda.WirelessDestinationEventSource):
140 * yield* IoTWireless.consumeUplinks(destination, (uplinks) =>
141 * uplinks.pipe(
142 * Stream.runForEach((uplink) => processUplink(uplink)),
143 * Effect.orDie,
144 * ),
145 * );
146 * ```
147 *
148 * @resource
149 */
150export const Destination = Resource<Destination>("AWS.IoTWireless.Destination");
151
152export const DestinationProvider = () =>
153 Provider.effect(
154 Destination,
155 Effect.gen(function* () {
156 const createName = Effect.fn(function* (
157 id: string,
158 props: { name?: string },
159 ) {
160 return (
161 props.name ?? (yield* createPhysicalName({ id, maxLength: 128 }))
162 );
163 });
164
165 const getDestination = (name: string) =>
166 iotw
167 .getDestination({ Name: name })
168 .pipe(
169 Effect.catchTag("ResourceNotFoundException", () =>
170 Effect.succeed(undefined),
171 ),
172 );
173
174 const toAttrs = Effect.fn(function* (
175 destination: iotw.GetDestinationResponse,
176 name: string,
177 ) {
178 if (destination.Arn === undefined) {
179 return yield* Effect.fail(
180 new Error(
181 `IoT Wireless destination '${name}' returned without Arn`,
182 ),
183 );
184 }
185 return {
186 destinationName: destination.Name ?? name,
187 destinationArn: destination.Arn,
188 expressionType: destination.ExpressionType ?? "RuleName",
189 expression: destination.Expression ?? "",
190 roleArn: destination.RoleArn ?? "",
191 };
192 });
193
194 return Destination.Provider.of({
195 stables: ["destinationName", "destinationArn"],
196
197 list: () =>
198 iotw.listDestinations.pages({}).pipe(
199 Stream.runCollect,
200 Effect.map((chunk) =>
201 Array.from(chunk)
202 .flatMap((page) => page.DestinationList ?? [])
203 .flatMap((d) =>
204 d.Name !== undefined && d.Arn !== undefined
205 ? [
206 {
207 destinationName: d.Name,
208 destinationArn: d.Arn,
209 expressionType: d.ExpressionType ?? "RuleName",
210 expression: d.Expression ?? "",
211 roleArn: d.RoleArn ?? "",
212 },
213 ]
214 : [],
215 ),
216 ),
217 ),
218
219 read: Effect.fn(function* ({ id, olds, output }) {
220 const name =
221 output?.destinationName ?? (yield* createName(id, olds ?? {}));
222 const destination = yield* getDestination(name);
223 if (destination === undefined) return undefined;
224 const attrs = yield* toAttrs(destination, name);
225 const tags = yield* readIotWirelessTags(attrs.destinationArn);
226 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
227 }),
228
229 // The name is the destination's identity — changing it replaces the
230 // resource. Everything else updates in place via UpdateDestination.
231 diff: Effect.fn(function* ({ id, news, olds }) {
232 if (!isResolved(news)) return undefined;
233 const oldName = yield* createName(id, olds ?? {});
234 const newName = yield* createName(id, news);
235 if (oldName !== newName) return { action: "replace" } as const;
236 }),
237
238 reconcile: Effect.fn(function* ({ id, news, output, session }) {
239 const name = output?.destinationName ?? (yield* createName(id, news));
240 const internalTags = yield* createInternalTags(id);
241 const desiredTags = { ...news.tags, ...internalTags };
242
243 // 1. OBSERVE — cloud state is authoritative.
244 let destination = yield* getDestination(name);
245
246 // 2. ENSURE — create if missing; a Conflict means a peer created
247 // it concurrently, so fall through to the sync step.
248 if (destination === undefined) {
249 yield* session.note(`creating destination ${name}`);
250 yield* iotw
251 .createDestination({
252 Name: name,
253 ExpressionType: news.expressionType,
254 Expression: news.expression,
255 Description: news.description,
256 RoleArn: news.roleArn,
257 Tags: createTagsList(desiredTags),
258 })
259 .pipe(
260 Effect.asVoid,
261 Effect.catchTag("ConflictException", () => Effect.void),
262 );
263 destination = yield* getDestination(name);
264 }
265 if (destination === undefined) {
266 return yield* Effect.fail(
267 new Error(
268 `IoT Wireless destination '${name}' not found after create`,
269 ),
270 );
271 }
272
273 // 3. SYNC — apply the expression/description/role delta from
274 // OBSERVED state; skip the API call entirely on no-op.
275 const expressionDelta =
276 destination.Expression !== news.expression ||
277 destination.ExpressionType !== news.expressionType;
278 const descriptionDelta =
279 news.description !== undefined &&
280 destination.Description !== news.description;
281 const roleDelta = destination.RoleArn !== news.roleArn;
282 if (expressionDelta || descriptionDelta || roleDelta) {
283 yield* iotw.updateDestination({
284 Name: name,
285 ExpressionType: news.expressionType,
286 Expression: news.expression,
287 Description: news.description,
288 RoleArn: news.roleArn,
289 });
290 destination = yield* getDestination(name);
291 }
292 if (destination === undefined) {
293 return yield* Effect.fail(
294 new Error(
295 `IoT Wireless destination '${name}' vanished during update`,
296 ),
297 );
298 }
299
300 // 3b. SYNC tags — diff against OBSERVED cloud tags.
301 const attrs = yield* toAttrs(destination, name);
302 yield* syncIotWirelessTags(attrs.destinationArn, desiredTags);
303
304 yield* session.note(attrs.destinationArn);
305 return attrs;
306 }),
307
308 delete: Effect.fn(function* ({ output }) {
309 yield* iotw.deleteDestination({ Name: output.destinationName }).pipe(
310 Effect.asVoid,
311 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
312 );
313 }),
314 });
315 }),
316 );