1import * as sitewise from "@distilled.cloud/aws/iotsitewise";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import * as EffectStream from "effect/Stream";
6import { Unowned } from "../../AdoptPolicy.ts";
7import { isResolved } from "../../Diff.ts";
8import { createPhysicalName } from "../../PhysicalName.ts";
9import * as Provider from "../../Provider.ts";
10import { Resource } from "../../Resource.ts";
11import { createInternalTags, hasAlchemyTags, type Tags } from "../../Tags.ts";
12import type { Providers } from "../Providers.ts";
13import { fetchSiteWiseTags, syncSiteWiseTags } from "./internal.ts";
14
15export type AssetState = sitewise.AssetState;
16
17export interface AssetProps {
18 /**
19 * A unique name for the asset.
20 * @default ${app}-${stage}-${id}
21 */
22 assetName?: string;
23 /**
24 * The ID of the asset model from which to create the asset.
25 * Changing the model replaces the asset.
26 */
27 assetModelId: string;
28 /**
29 * A description for the asset.
30 */
31 assetDescription?: string;
32 /**
33 * Tags to associate with the asset.
34 */
35 tags?: Record<string, string>;
36}
37
38export interface Asset extends Resource<
39 "AWS.IoTSiteWise.Asset",
40 AssetProps,
41 {
42 /**
43 * Service-assigned UUID of the asset.
44 */
45 assetId: string;
46 /**
47 * ARN of the asset.
48 */
49 assetArn: string;
50 /**
51 * The asset's name.
52 */
53 assetName: string;
54 /**
55 * The ID of the asset model the asset was created from.
56 */
57 assetModelId: string;
58 /**
59 * Current lifecycle state of the asset.
60 */
61 state: AssetState;
62 /**
63 * Current tags reported for the asset.
64 */
65 tags: Record<string, string>;
66 },
67 never,
68 Providers
69> {}
70
71/**
72 * An AWS IoT SiteWise asset — an instance of an asset model representing
73 * a physical piece of industrial equipment or a logical grouping.
74 *
75 * Asset creation and updates are asynchronous (`CREATING`/`UPDATING` →
76 * `ACTIVE`); the provider waits for the asset to converge to `ACTIVE`
77 * before returning.
78 *
79 * ### Creating Assets
80 * **Example:** Asset from an Asset Model
81 * ```typescript
82 * import * as AWS from "alchemy/AWS";
83 *
84 * const model = yield* AWS.IoTSiteWise.AssetModel("PumpModel", {
85 * assetModelProperties: [
86 * { name: "Temperature", dataType: "DOUBLE", type: { measurement: {} } },
87 * ],
88 * });
89 *
90 * const asset = yield* AWS.IoTSiteWise.Asset("Pump1", {
91 * assetModelId: model.assetModelId,
92 * assetDescription: "Pump #1 on the west line",
93 * });
94 * ```
95 *
96 * @resource
97 */
98export const Asset = Resource<Asset>("AWS.IoTSiteWise.Asset");
99
100const createAssetName = (
101 id: string,
102 props: { assetName?: string | undefined },
103) =>
104 props.assetName
105 ? Effect.succeed(props.assetName)
106 : createPhysicalName({ id, maxLength: 256 });
107
108interface AssetState_ {
109 attrs: Asset["Attributes"];
110 described: sitewise.DescribeAssetResponse;
111}
112
113const readAssetById = Effect.fn(function* (assetId: string) {
114 const described = yield* sitewise
115 .describeAsset({ assetId, excludeProperties: true })
116 .pipe(
117 Effect.catchTag("ResourceNotFoundException", () =>
118 Effect.succeed(undefined),
119 ),
120 );
121 if (!described || described.assetStatus.state === "DELETING") {
122 return undefined;
123 }
124 const state: AssetState_ = {
125 described,
126 attrs: {
127 assetId: described.assetId,
128 assetArn: described.assetArn,
129 assetName: described.assetName,
130 assetModelId: described.assetModelId,
131 state: described.assetStatus.state,
132 tags: yield* fetchSiteWiseTags(described.assetArn),
133 },
134 };
135 return state;
136});
137
138const findAssetByName = Effect.fn(function* (
139 name: string,
140 assetModelId: string,
141) {
142 const summaries = yield* sitewise.listAssets.pages({ assetModelId }).pipe(
143 EffectStream.runCollect,
144 Effect.map((chunk) =>
145 Array.from(chunk).flatMap((page) => page.assetSummaries),
146 ),
147 // The model itself may not exist yet on a greenfield deploy.
148 Effect.catchTag("ResourceNotFoundException", () => Effect.succeed([])),
149 );
150 const match = summaries.find(
151 (summary) => summary.name === name && summary.status.state !== "DELETING",
152 );
153 if (!match) return undefined;
154 return yield* readAssetById(match.id);
155});
156
157/**
158 * An asset still transitioning toward the awaited state — retried by
159 * {@link waitForAssetState}'s bounded schedule.
160 */
161class AssetNotReady extends Data.TaggedError("AssetNotReady")<{
162 readonly assetId: string;
163 readonly state: string | undefined;
164}> {}
165
166/**
167 * An asset whose asynchronous provisioning converged to the terminal
168 * `FAILED` state.
169 */
170export class AssetProvisioningFailed extends Data.TaggedError(
171 "AssetProvisioningFailed",
172)<{
173 readonly assetId: string;
174 readonly message: string | undefined;
175}> {}
176
177// Explicitly-typed retry wrapper — see AssetModel.ts for why inline
178// Effect.retry is forbidden in provider lifecycle code.
179const retryWhileNotReady = <A, E extends { readonly _tag: string }, R>(
180 self: Effect.Effect<A, E, R>,
181): Effect.Effect<A, E, R> =>
182 Effect.retry(self, {
183 while: (e) => e._tag === "AssetNotReady",
184 schedule: Schedule.max([Schedule.spaced("3 seconds"), Schedule.recurs(30)]),
185 });
186
187const retryThroughConflictingOperation = <
188 A,
189 E extends { readonly _tag: string },
190 R,
191>(
192 self: Effect.Effect<A, E, R>,
193): Effect.Effect<A, E, R> =>
194 Effect.retry(self, {
195 while: (e) => e._tag === "ConflictingOperationException",
196 schedule: Schedule.max([Schedule.spaced("5 seconds"), Schedule.recurs(10)]),
197 });
198
199const waitForAssetState = (assetId: string, target: "ACTIVE" | "DELETED") =>
200 retryWhileNotReady(
201 Effect.gen(function* () {
202 const described = yield* sitewise
203 .describeAsset({ assetId, excludeProperties: true })
204 .pipe(
205 Effect.catchTag("ResourceNotFoundException", () =>
206 Effect.succeed(undefined),
207 ),
208 );
209 if (target === "DELETED") {
210 if (described === undefined) return;
211 return yield* Effect.fail(
212 new AssetNotReady({ assetId, state: described.assetStatus.state }),
213 );
214 }
215 if (described?.assetStatus.state === "ACTIVE") return;
216 if (described?.assetStatus.state === "FAILED") {
217 return yield* Effect.fail(
218 new AssetProvisioningFailed({
219 assetId,
220 message: described.assetStatus.error?.message,
221 }),
222 );
223 }
224 return yield* Effect.fail(
225 new AssetNotReady({ assetId, state: described?.assetStatus.state }),
226 );
227 }),
228 );
229
230export const AssetProvider = () =>
231 Provider.effect(
232 Asset,
233 Effect.gen(function* () {
234 return {
235 stables: ["assetId", "assetArn"],
236 list: () =>
237 Effect.gen(function* () {
238 // ListAssets requires an assetModelId filter — enumerate models
239 // first, then each model's assets.
240 const models = yield* sitewise.listAssetModels.pages({}).pipe(
241 EffectStream.runCollect,
242 Effect.map((chunk) =>
243 Array.from(chunk).flatMap((page) => page.assetModelSummaries),
244 ),
245 );
246 const summaries = yield* Effect.forEach(
247 models.map((model) => model.id),
248 (assetModelId) =>
249 sitewise.listAssets.pages({ assetModelId }).pipe(
250 EffectStream.runCollect,
251 Effect.map((chunk) =>
252 Array.from(chunk).flatMap((page) => page.assetSummaries),
253 ),
254 Effect.catchTag("ResourceNotFoundException", () =>
255 Effect.succeed([]),
256 ),
257 ),
258 { concurrency: 3 },
259 );
260 const hydrated = yield* Effect.forEach(
261 summaries.flat().map((summary) => summary.id),
262 (assetId) => readAssetById(assetId),
263 { concurrency: 5 },
264 );
265 return hydrated.flatMap((state) =>
266 state === undefined ? [] : [state.attrs],
267 );
268 }),
269 read: Effect.fn(function* ({ id, olds, output }) {
270 const state = output?.assetId
271 ? yield* readAssetById(output.assetId)
272 : olds?.assetModelId
273 ? yield* findAssetByName(
274 yield* createAssetName(id, olds),
275 olds.assetModelId,
276 )
277 : undefined;
278 if (!state) return undefined;
279 return (yield* hasAlchemyTags(id, state.attrs.tags as Tags))
280 ? state.attrs
281 : Unowned(state.attrs);
282 }),
283 diff: Effect.fn(function* ({ news, olds }) {
284 if (!isResolved(news)) return;
285 if (olds === undefined) return;
286 // An asset cannot move to a different model.
287 if (olds.assetModelId !== news.assetModelId) {
288 return { action: "replace" } as const;
289 }
290 }),
291 reconcile: Effect.fn(function* ({ id, news, output, session }) {
292 if (!news) {
293 return yield* Effect.fail(
294 new Error("IoT SiteWise Asset requires props"),
295 );
296 }
297 const name = yield* createAssetName(id, news);
298 const internalTags = yield* createInternalTags(id);
299 const desiredTags = { ...internalTags, ...news.tags };
300
301 // Observe — prefer the cached id; fall back to a name lookup so
302 // a create whose state failed to persist is adopted.
303 let state = output?.assetId
304 ? yield* readAssetById(output.assetId)
305 : yield* findAssetByName(name, news.assetModelId);
306
307 // Ensure — create if missing, then wait for ACTIVE. A model that
308 // just reached ACTIVE can still briefly reject creates with
309 // ConflictingOperationException.
310 if (state === undefined) {
311 const created = yield* retryThroughConflictingOperation(
312 sitewise.createAsset({
313 assetName: name,
314 assetModelId: news.assetModelId,
315 assetDescription: news.assetDescription,
316 tags: desiredTags,
317 }),
318 );
319 yield* session.note(
320 `Creating asset ${name} (${created.assetId})...`,
321 );
322 yield* waitForAssetState(created.assetId, "ACTIVE");
323 state = yield* readAssetById(created.assetId);
324 if (state === undefined) {
325 return yield* Effect.fail(
326 new Error(`failed to read created asset ${name}`),
327 );
328 }
329 }
330
331 // Sync — name/description drift via UpdateAsset.
332 const described = state.described;
333 if (
334 name !== described.assetName ||
335 (news.assetDescription ?? "") !== (described.assetDescription ?? "")
336 ) {
337 yield* retryThroughConflictingOperation(
338 sitewise.updateAsset({
339 assetId: state.attrs.assetId,
340 assetName: name,
341 assetDescription: news.assetDescription,
342 }),
343 );
344 yield* waitForAssetState(state.attrs.assetId, "ACTIVE");
345 yield* session.note(`Updated asset ${name}`);
346 }
347
348 // Sync tags — diff against observed cloud tags.
349 yield* syncSiteWiseTags(
350 state.attrs.assetArn,
351 state.attrs.tags,
352 desiredTags,
353 );
354
355 yield* session.note(state.attrs.assetArn);
356
357 const final = yield* readAssetById(state.attrs.assetId);
358 if (!final) {
359 return yield* Effect.fail(
360 new Error(`failed to read reconciled asset ${name}`),
361 );
362 }
363 return final.attrs;
364 }),
365 delete: Effect.fn(function* ({ output }) {
366 // Deleting an asset that is still associated to a parent (or mid
367 // transition) raises ConflictingOperationException — retry
368 // through the window, then wait until fully gone so the asset
369 // model can be deleted after it.
370 yield* retryThroughConflictingOperation(
371 sitewise
372 .deleteAsset({ assetId: output.assetId })
373 .pipe(
374 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
375 ),
376 );
377 yield* waitForAssetState(output.assetId, "DELETED");
378 }),
379 };
380 }),
381 );