1import * as mi from "@distilled.cloud/aws/iot-managed-integrations";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import * as Provider from "../../Provider.ts";
7import { Resource } from "../../Resource.ts";
8import { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
9import { AWSEnvironment } from "../Environment.ts";
10import type { Providers } from "../Providers.ts";
11import { syncManagedIntegrationsTags, toTagRecord } from "./internal.ts";
12
13export interface NotificationConfigurationProps {
14 /**
15 * Event type to route, e.g. `DEVICE_STATE`, `DEVICE_LIFE_CYCLE`,
16 * `DEVICE_EVENT`, `DEVICE_COMMAND_REQUEST`, `DEVICE_OTA`,
17 * `CONNECTOR_ASSOCIATION`, `ACCOUNT_ASSOCIATION`,
18 * `CONNECTOR_ERROR_REPORT`. The event type is the configuration's
19 * identifier — at most one configuration exists per event type, and
20 * changing it replaces the configuration.
21 */
22 eventType: mi.EventType;
23 /**
24 * Name of the {@link Destination} that receives events of this type.
25 */
26 destinationName: string;
27 /**
28 * User-defined tags to apply to the notification configuration.
29 */
30 tags?: Record<string, string>;
31}
32
33export interface NotificationConfiguration extends Resource<
34 "AWS.IoTManagedIntegrations.NotificationConfiguration",
35 NotificationConfigurationProps,
36 {
37 /** Event type routed by this configuration (its identifier). */
38 eventType: mi.EventType;
39 /** Name of the destination that receives the events. */
40 destinationName: string;
41 /** Constructed ARN of the notification configuration. */
42 notificationConfigurationArn: string;
43 /** Tags applied to the notification configuration (user + internal). */
44 tags: Record<string, string>;
45 },
46 never,
47 Providers
48> {}
49
50/**
51 * An AWS IoT Managed Integrations notification configuration — routes one
52 * event type (device state, lifecycle, OTA, discovery, association events)
53 * to a {@link Destination}.
54 *
55 * Managed integrations has no direct Lambda invocation path: events flow
56 * `NotificationConfiguration -> Destination -> Kinesis Data Stream`, so pair
57 * this resource with a Kinesis-backed destination and consume the stream
58 * with the Kinesis stream event source (`Kinesis.consumeStreamRecords`).
59 *
60 * IoT Managed Integrations is a regional service available in a limited set
61 * of regions (e.g. `eu-west-1`, `ca-central-1`).
62 *
63 * ### Routing Events
64 * **Example:** Route Device State Events to a Kinesis Destination
65 * ```typescript
66 * const destination = yield* Destination("EventDestination", {
67 * deliveryDestinationArn: stream.streamArn,
68 * roleArn: role.roleArn,
69 * });
70 * const routing = yield* NotificationConfiguration("DeviceState", {
71 * eventType: "DEVICE_STATE",
72 * destinationName: destination.destinationName,
73 * });
74 * ```
75 *
76 * **Example:** Route Lifecycle Events with Tags
77 * ```typescript
78 * const routing = yield* NotificationConfiguration("Lifecycle", {
79 * eventType: "DEVICE_LIFE_CYCLE",
80 * destinationName: destination.destinationName,
81 * tags: { team: "iot" },
82 * });
83 * ```
84 *
85 * @resource
86 */
87export const NotificationConfiguration = Resource<NotificationConfiguration>(
88 "AWS.IoTManagedIntegrations.NotificationConfiguration",
89);
90
91export const NotificationConfigurationProvider = () =>
92 Provider.effect(
93 NotificationConfiguration,
94 Effect.gen(function* () {
95 const observe = (eventType: mi.EventType) =>
96 mi
97 .getNotificationConfiguration({ EventType: eventType })
98 .pipe(
99 Effect.catchTag("ResourceNotFoundException", () =>
100 Effect.succeed(undefined),
101 ),
102 );
103
104 // GetNotificationConfiguration does not return an ARN; the tag APIs
105 // need one, so construct it from the ambient account/region (same
106 // approach as the Destination provider).
107 const configurationArn = Effect.fn(function* (eventType: mi.EventType) {
108 const { accountId, region } = yield* AWSEnvironment.current;
109 return `arn:aws:iotmanagedintegrations:${region}:${accountId}:notification-configuration/${eventType}`;
110 });
111
112 const toAttributes = Effect.fn(function* (
113 configuration: mi.GetNotificationConfigurationResponse,
114 ) {
115 if (
116 configuration.EventType === undefined ||
117 configuration.DestinationName === undefined
118 ) {
119 return yield* Effect.fail(
120 new Error(
121 "notification configuration response is missing EventType or DestinationName",
122 ),
123 );
124 }
125 return {
126 eventType: configuration.EventType,
127 destinationName: configuration.DestinationName,
128 notificationConfigurationArn: yield* configurationArn(
129 configuration.EventType,
130 ),
131 tags: toTagRecord(configuration.Tags),
132 };
133 });
134
135 return {
136 stables: ["eventType", "notificationConfigurationArn"],
137 diff: Effect.fn(function* ({ olds, news }) {
138 if (!isResolved(news)) return;
139 if (olds !== undefined && olds.eventType !== news.eventType) {
140 return { action: "replace" } as const;
141 }
142 }),
143 read: Effect.fn(function* ({ id, olds, output }) {
144 const eventType = output?.eventType ?? olds?.eventType;
145 if (eventType === undefined) return undefined;
146 const configuration = yield* observe(eventType);
147 if (configuration === undefined) return undefined;
148 const attrs = yield* toAttributes(configuration);
149 return (yield* hasAlchemyTags(id, attrs.tags))
150 ? attrs
151 : Unowned(attrs);
152 }),
153 reconcile: Effect.fn(function* ({ id, news, session }) {
154 const internalTags = yield* createInternalTags(id);
155 const desiredTags = { ...internalTags, ...news.tags };
156
157 // Observe — cloud state is authoritative; the event type is the
158 // identifier so no cached output is needed to find it.
159 let configuration = yield* observe(news.eventType);
160
161 // Ensure — create if missing; tolerate a concurrent-create race.
162 if (configuration === undefined) {
163 yield* mi
164 .createNotificationConfiguration({
165 EventType: news.eventType,
166 DestinationName: news.destinationName,
167 Tags: desiredTags,
168 })
169 .pipe(Effect.catchTag("ConflictException", () => Effect.void));
170 configuration = yield* observe(news.eventType);
171 if (configuration === undefined) {
172 return yield* Effect.fail(
173 new Error(
174 `notification configuration '${news.eventType}' vanished after create`,
175 ),
176 );
177 }
178 }
179
180 // Sync the destination — apply only the delta.
181 if (configuration.DestinationName !== news.destinationName) {
182 yield* mi.updateNotificationConfiguration({
183 EventType: news.eventType,
184 DestinationName: news.destinationName,
185 });
186 configuration = {
187 ...configuration,
188 DestinationName: news.destinationName,
189 };
190 }
191
192 // Sync tags — diff against OBSERVED cloud tags.
193 yield* syncManagedIntegrationsTags(
194 yield* configurationArn(news.eventType),
195 toTagRecord(configuration.Tags),
196 desiredTags,
197 );
198
199 const attrs = yield* toAttributes(configuration);
200 yield* session.note(attrs.eventType);
201 return { ...attrs, tags: desiredTags };
202 }),
203 // Enumerate every notification configuration; fetch each one to
204 // resolve tags (summaries omit them).
205 list: () =>
206 Effect.gen(function* () {
207 const summaries = yield* mi.listNotificationConfigurations
208 .items({})
209 .pipe(
210 Stream.runCollect,
211 Effect.map((chunk) => Array.from(chunk)),
212 );
213 const configurations = yield* Effect.forEach(
214 summaries.filter(
215 (
216 s,
217 ): s is mi.NotificationConfigurationSummary & {
218 EventType: mi.EventType;
219 } => s.EventType !== undefined,
220 ),
221 (summary) => observe(summary.EventType),
222 { concurrency: 5 },
223 );
224 return yield* Effect.forEach(
225 configurations.filter(
226 (c): c is mi.GetNotificationConfigurationResponse =>
227 c !== undefined,
228 ),
229 toAttributes,
230 );
231 }),
232 delete: Effect.fn(function* ({ output }) {
233 yield* mi
234 .deleteNotificationConfiguration({ EventType: output.eventType })
235 .pipe(
236 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
237 );
238 }),
239 };
240 }),
241 );