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 { createPhysicalName } from "../../PhysicalName.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
10import { AWSEnvironment } from "../Environment.ts";
11import type { Providers } from "../Providers.ts";
12import { syncManagedIntegrationsTags, toTagRecord } from "./internal.ts";
13
14export interface DestinationProps {
15 /**
16 * Name of the destination. If omitted, a unique name is generated from the
17 * app, stage, and logical ID. Changing the name replaces the destination.
18 */
19 name?: string;
20 /**
21 * ARN of the delivery destination that receives events and notifications —
22 * e.g. a Kinesis Data Stream ARN.
23 */
24 deliveryDestinationArn: string;
25 /**
26 * Type of the delivery destination.
27 * @default "KINESIS"
28 */
29 deliveryDestinationType?: mi.DeliveryDestinationType;
30 /**
31 * ARN of the IAM role that grants Managed integrations permission to write
32 * to the delivery destination.
33 */
34 roleArn: string;
35 /**
36 * Description of the destination.
37 */
38 description?: string;
39 /**
40 * User-defined tags to apply to the destination.
41 */
42 tags?: Record<string, string>;
43}
44
45export interface Destination extends Resource<
46 "AWS.IoTManagedIntegrations.Destination",
47 DestinationProps,
48 {
49 /** Name of the destination (its identifier). */
50 destinationName: string;
51 /** ARN of the delivery destination (e.g. Kinesis stream). */
52 deliveryDestinationArn: string;
53 /** Type of the delivery destination. */
54 deliveryDestinationType: mi.DeliveryDestinationType;
55 /** ARN of the IAM role used to deliver to the destination. */
56 roleArn: string;
57 /** Description of the destination. */
58 description: string | undefined;
59 /** Tags applied to the destination (user + internal). */
60 tags: Record<string, string>;
61 },
62 never,
63 Providers
64> {}
65
66/**
67 * An AWS IoT Managed Integrations notification destination. Managed
68 * integrations delivers lifecycle events and device notifications to the
69 * destination (currently a Kinesis Data Stream) using the provided IAM role.
70 *
71 * IoT Managed Integrations is a regional service available in a limited set
72 * of regions (e.g. `eu-west-1`, `ca-central-1`).
73 *
74 * ### Creating Destinations
75 * **Example:** Kinesis Destination
76 * ```typescript
77 * const stream = yield* Kinesis.Stream("Events", {});
78 * const role = yield* IAM.Role("DeliveryRole", {
79 * assumeRolePolicyDocument: {
80 * Version: "2012-10-17",
81 * Statement: [
82 * {
83 * Effect: "Allow",
84 * Principal: { Service: "iotmanagedintegrations.amazonaws.com" },
85 * Action: ["sts:AssumeRole"],
86 * },
87 * ],
88 * },
89 * });
90 * const destination = yield* Destination("EventDestination", {
91 * deliveryDestinationArn: stream.streamArn,
92 * roleArn: role.roleArn,
93 * description: "Managed integrations device events",
94 * });
95 * ```
96 *
97 * @resource
98 */
99export const Destination = Resource<Destination>(
100 "AWS.IoTManagedIntegrations.Destination",
101);
102
103export const DestinationProvider = () =>
104 Provider.effect(
105 Destination,
106 Effect.gen(function* () {
107 const toName = (id: string, props: { name?: string } = {}) =>
108 props.name
109 ? Effect.succeed(props.name)
110 : createPhysicalName({ id, maxLength: 128 });
111
112 const observe = (name: string) =>
113 mi
114 .getDestination({ Name: name })
115 .pipe(
116 Effect.catchTag("ResourceNotFoundException", () =>
117 Effect.succeed(undefined),
118 ),
119 );
120
121 // GetDestination does not return an ARN; the tag APIs need one, so
122 // construct it from the ambient account/region.
123 const destinationArn = Effect.fn(function* (name: string) {
124 const { accountId, region } = yield* AWSEnvironment.current;
125 return `arn:aws:iotmanagedintegrations:${region}:${accountId}:destination/${name}`;
126 });
127
128 const toAttributes = Effect.fn(function* (
129 destination: mi.GetDestinationResponse,
130 ) {
131 if (
132 destination.Name === undefined ||
133 destination.DeliveryDestinationArn === undefined ||
134 destination.DeliveryDestinationType === undefined ||
135 destination.RoleArn === undefined
136 ) {
137 return yield* Effect.fail(
138 new Error("destination response is missing required fields"),
139 );
140 }
141 return {
142 destinationName: destination.Name,
143 deliveryDestinationArn: destination.DeliveryDestinationArn,
144 deliveryDestinationType: destination.DeliveryDestinationType,
145 roleArn: destination.RoleArn,
146 description: destination.Description,
147 tags: toTagRecord(destination.Tags),
148 };
149 });
150
151 return {
152 stables: ["destinationName"],
153 diff: Effect.fn(function* ({ id, olds, news }) {
154 if (!isResolved(news)) return;
155 if (
156 (yield* toName(id, olds ?? {})) !== (yield* toName(id, news ?? {}))
157 ) {
158 return { action: "replace" } as const;
159 }
160 }),
161 read: Effect.fn(function* ({ id, olds, output }) {
162 const name =
163 output?.destinationName ?? (yield* toName(id, olds ?? {}));
164 const destination = yield* observe(name);
165 if (destination === undefined) return undefined;
166 const attrs = yield* toAttributes(destination);
167 return (yield* hasAlchemyTags(id, attrs.tags))
168 ? attrs
169 : Unowned(attrs);
170 }),
171 reconcile: Effect.fn(function* ({ id, news, output, session }) {
172 const name = output?.destinationName ?? (yield* toName(id, news));
173 const internalTags = yield* createInternalTags(id);
174 const desiredTags = { ...internalTags, ...news.tags };
175 const desiredType = news.deliveryDestinationType ?? "KINESIS";
176
177 // Observe — cloud state is authoritative.
178 let destination = yield* observe(name);
179
180 // Ensure — create if missing; tolerate a concurrent-create race.
181 if (destination === undefined) {
182 yield* mi
183 .createDestination({
184 Name: name,
185 DeliveryDestinationArn: news.deliveryDestinationArn,
186 DeliveryDestinationType: desiredType,
187 RoleArn: news.roleArn,
188 Description: news.description,
189 Tags: desiredTags,
190 })
191 .pipe(Effect.catchTag("ConflictException", () => Effect.void));
192 destination = yield* observe(name);
193 if (destination === undefined) {
194 return yield* Effect.fail(
195 new Error(`destination '${name}' vanished after create`),
196 );
197 }
198 }
199
200 // Sync mutable settings — apply only the delta.
201 if (
202 destination.DeliveryDestinationArn !==
203 news.deliveryDestinationArn ||
204 destination.DeliveryDestinationType !== desiredType ||
205 destination.RoleArn !== news.roleArn ||
206 (destination.Description ?? undefined) !==
207 (news.description ?? undefined)
208 ) {
209 yield* mi.updateDestination({
210 Name: name,
211 DeliveryDestinationArn: news.deliveryDestinationArn,
212 DeliveryDestinationType: desiredType,
213 RoleArn: news.roleArn,
214 Description: news.description,
215 });
216 }
217
218 // Sync tags — diff against OBSERVED cloud tags.
219 yield* syncManagedIntegrationsTags(
220 yield* destinationArn(name),
221 toTagRecord(destination.Tags),
222 desiredTags,
223 );
224
225 // Return fresh attributes.
226 const final = yield* observe(name);
227 if (final === undefined) {
228 return yield* Effect.fail(
229 new Error(`destination '${name}' vanished during reconcile`),
230 );
231 }
232 const attrs = yield* toAttributes(final);
233 yield* session.note(attrs.destinationName);
234 return attrs;
235 }),
236 // Enumerate every destination in the account/region; fetch each one to
237 // resolve its tags (summaries omit them).
238 list: () =>
239 Effect.gen(function* () {
240 const summaries = yield* mi.listDestinations.items({}).pipe(
241 Stream.runCollect,
242 Effect.map((chunk) => Array.from(chunk)),
243 );
244 const destinations = yield* Effect.forEach(
245 summaries.filter(
246 (s): s is mi.DestinationSummary & { Name: string } =>
247 s.Name !== undefined,
248 ),
249 (summary) => observe(summary.Name),
250 { concurrency: 5 },
251 );
252 return yield* Effect.forEach(
253 destinations.filter(
254 (d): d is mi.GetDestinationResponse => d !== undefined,
255 ),
256 toAttributes,
257 );
258 }),
259 delete: Effect.fn(function* ({ output }) {
260 yield* mi
261 .deleteDestination({ Name: output.destinationName })
262 .pipe(
263 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
264 );
265 }),
266 };
267 }),
268 );