1import * as iotfleetwise from "@distilled.cloud/aws/iotfleetwise";
2import type * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import type * as Redacted from "effect/Redacted";
5import * as Stream 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 } from "../../Tags.ts";
12import { toWireMillis, toWireSeconds } from "../../Util/Duration.ts";
13import type { Providers } from "../Providers.ts";
14import {
15 inFleetWiseRegion,
16 readFleetWiseTags,
17 retryObservation,
18 stableEquals,
19 syncFleetWiseTags,
20 toFleetWiseTagList,
21} from "./internal.ts";
22
23/**
24 * Time-based collection: collect the signals on a fixed cadence.
25 */
26export interface TimeBasedCollectionScheme {
27 /**
28 * How often to collect the signals, e.g. `"10 seconds"` or
29 * `Duration.seconds(10)`. A bare number is milliseconds. Sent to
30 * FleetWise as whole milliseconds (`periodMs`, 50 ms - 60 s).
31 */
32 period: Duration.Input;
33}
34
35/**
36 * Condition-based collection: collect the signals whenever the expression
37 * evaluates true.
38 */
39export interface ConditionBasedCollectionScheme {
40 /**
41 * The logical expression that triggers collection, e.g.
42 * `$variable.\`Vehicle.OutsideAirTemperature\` >= 105.0`.
43 */
44 expression: string | Redacted.Redacted<string>;
45 /**
46 * Minimum time between two triggers, e.g. `"1 second"`. A bare number
47 * is milliseconds. Sent to FleetWise as whole milliseconds
48 * (`minimumTriggerIntervalMs`).
49 * @default 0
50 */
51 minimumTriggerInterval?: Duration.Input;
52 /**
53 * Whether to trigger on every evaluation (`"ALWAYS"`) or only when the
54 * expression flips to true (`"RISING_EDGE"`).
55 * @default "ALWAYS"
56 */
57 triggerMode?: iotfleetwise.TriggerMode;
58 /**
59 * Version of the condition language.
60 * @default 1
61 */
62 conditionLanguageVersion?: number;
63}
64
65/**
66 * When to collect data — exactly one of a time-based or condition-based
67 * scheme.
68 */
69export type CollectionScheme =
70 | {
71 timeBasedCollectionScheme: TimeBasedCollectionScheme;
72 conditionBasedCollectionScheme?: never;
73 }
74 | {
75 timeBasedCollectionScheme?: never;
76 conditionBasedCollectionScheme: ConditionBasedCollectionScheme;
77 };
78
79/** A signal the campaign collects from vehicles. */
80export interface SignalInformation {
81 /** The name of the signal, e.g. `Vehicle.Speed`. */
82 name: string;
83 /** Maximum number of samples collected per trigger. */
84 maxSampleCount?: number;
85 /**
86 * Minimum time between two samples of this signal, e.g. `"500 millis"`.
87 * A bare number is milliseconds. Sent to FleetWise as whole milliseconds
88 * (`minimumSamplingIntervalMs`).
89 */
90 minimumSamplingInterval?: Duration.Input;
91 /** The ID of the data partition the signal is stored in. */
92 dataPartitionId?: string;
93}
94
95/** Time-based on-demand fetch: run the fetch actions on a fixed cadence. */
96export interface TimeBasedSignalFetchConfig {
97 /**
98 * How often to fetch the signal, e.g. `"1 minute"`. A bare number is
99 * milliseconds. Sent to FleetWise as whole milliseconds
100 * (`executionFrequencyMs`).
101 */
102 executionFrequency: Duration.Input;
103}
104
105/** How an on-demand signal fetch is triggered. */
106export type SignalFetchConfig =
107 | {
108 timeBased: TimeBasedSignalFetchConfig;
109 conditionBased?: never;
110 }
111 | {
112 timeBased?: never;
113 conditionBased: iotfleetwise.ConditionBasedSignalFetchConfig;
114 };
115
116/** A signal fetched on demand by actuator-triggered fetch configs. */
117export interface SignalFetchInformation {
118 /** The fully-qualified name of the signal to fetch. */
119 fullyQualifiedName: string;
120 /** When the fetch runs — time-based or condition-based. */
121 signalFetchConfig: SignalFetchConfig;
122 /**
123 * Version of the condition language used by condition-based configs.
124 * @default 1
125 */
126 conditionLanguageVersion?: number;
127 /** The actions (expressions) evaluated when the fetch triggers. */
128 actions: (string | Redacted.Redacted<string>)[];
129}
130
131const toWireCollectionScheme = (
132 scheme: CollectionScheme,
133): iotfleetwise.CollectionScheme =>
134 scheme.timeBasedCollectionScheme !== undefined
135 ? {
136 timeBasedCollectionScheme: {
137 periodMs: toWireMillis(scheme.timeBasedCollectionScheme.period)!,
138 },
139 }
140 : {
141 conditionBasedCollectionScheme: {
142 expression: scheme.conditionBasedCollectionScheme.expression,
143 minimumTriggerIntervalMs: toWireMillis(
144 scheme.conditionBasedCollectionScheme.minimumTriggerInterval,
145 ),
146 triggerMode: scheme.conditionBasedCollectionScheme.triggerMode,
147 conditionLanguageVersion:
148 scheme.conditionBasedCollectionScheme.conditionLanguageVersion,
149 },
150 };
151
152const toWireSignalsToCollect = (
153 signals: SignalInformation[] | undefined,
154): iotfleetwise.SignalInformation[] | undefined =>
155 signals?.map((signal) => ({
156 name: signal.name,
157 maxSampleCount: signal.maxSampleCount,
158 minimumSamplingIntervalMs: toWireMillis(signal.minimumSamplingInterval),
159 dataPartitionId: signal.dataPartitionId,
160 }));
161
162const toWireSignalsToFetch = (
163 signals: SignalFetchInformation[] | undefined,
164): iotfleetwise.SignalFetchInformation[] | undefined =>
165 signals?.map((signal) => ({
166 fullyQualifiedName: signal.fullyQualifiedName,
167 signalFetchConfig:
168 signal.signalFetchConfig.timeBased !== undefined
169 ? {
170 timeBased: {
171 executionFrequencyMs: toWireMillis(
172 signal.signalFetchConfig.timeBased.executionFrequency,
173 )!,
174 },
175 }
176 : { conditionBased: signal.signalFetchConfig.conditionBased },
177 conditionLanguageVersion: signal.conditionLanguageVersion,
178 actions: signal.actions,
179 }));
180
181export interface CampaignProps {
182 /**
183 * Name of the campaign. Must be 1-100 characters of `[a-zA-Z0-9:_-]`.
184 * If omitted, a deterministic physical name is generated. Changing the
185 * name replaces the campaign.
186 */
187 campaignName?: string;
188 /**
189 * Human-readable description of the campaign. Updated in place.
190 */
191 description?: string;
192 /**
193 * ARN of the {@link SignalCatalog} the collected signals come from.
194 * Changing it replaces the campaign.
195 */
196 signalCatalogArn: string;
197 /**
198 * ARN of the {@link Fleet} or {@link Vehicle} the campaign deploys to.
199 * Changing it replaces the campaign.
200 */
201 targetArn: string;
202 /**
203 * When to collect data — a time-based or condition-based scheme.
204 * Changing it replaces the campaign.
205 */
206 collectionScheme: CollectionScheme;
207 /**
208 * Signals to collect from vehicles. Changing them replaces the
209 * campaign.
210 */
211 signalsToCollect?: SignalInformation[];
212 /**
213 * Destinations (S3, Timestream or MQTT topic) the collected data is
214 * delivered to. Changing them replaces the campaign.
215 */
216 dataDestinationConfigs?: iotfleetwise.DataDestinationConfig[];
217 /**
218 * Time the campaign starts collecting data, as an ISO-8601 string.
219 * Changing it replaces the campaign.
220 * @default now
221 */
222 startTime?: string;
223 /**
224 * Time the campaign expires, as an ISO-8601 string. Changing it
225 * replaces the campaign.
226 * @default 253402214400 (9999-12-31)
227 */
228 expiryTime?: string;
229 /**
230 * How long to keep collecting data after a condition-based trigger
231 * fires, e.g. `"30 seconds"` or `Duration.seconds(30)`. Sent to
232 * FleetWise as whole seconds (a bare number is milliseconds).
233 * Changing it replaces the campaign.
234 * @default 0
235 */
236 postTriggerCollectionDuration?: Duration.Input;
237 /**
238 * Whether to send active diagnostic trouble codes — `"OFF"` or
239 * `"SEND_ACTIVE_DTCS"`. Changing it replaces the campaign.
240 * @default "OFF"
241 */
242 diagnosticsMode?: iotfleetwise.DiagnosticsMode;
243 /**
244 * Whether to store collected data offline when a vehicle lacks
245 * connectivity — `"OFF"` or `"TO_DISK"`. Changing it replaces the
246 * campaign.
247 * @default "OFF"
248 */
249 spoolingMode?: iotfleetwise.SpoolingMode;
250 /**
251 * Whether to compress vehicle-to-cloud transmissions — `"OFF"` or
252 * `"SNAPPY"`. Changing it replaces the campaign.
253 * @default "OFF"
254 */
255 compression?: iotfleetwise.Compression;
256 /**
257 * Vehicle attribute node paths added as extra dimensions on the
258 * collected data, e.g. `["Vehicle.VIN"]`. Updated in place.
259 */
260 dataExtraDimensions?: string[];
261 /**
262 * Data partitions for on-vehicle storage (requires spooling to disk).
263 * Changing them replaces the campaign.
264 */
265 dataPartitions?: iotfleetwise.DataPartition[];
266 /**
267 * Signals fetched on demand by actuator-triggered fetch configs.
268 * Changing them replaces the campaign.
269 */
270 signalsToFetch?: SignalFetchInformation[];
271 /**
272 * Whether the provider approves the campaign (transitioning it from
273 * `WAITING_FOR_APPROVAL` to `RUNNING`) after creation.
274 * @default false
275 */
276 autoApprove?: boolean;
277 /**
278 * User-defined tags for the campaign.
279 */
280 tags?: Record<string, string>;
281}
282
283export interface Campaign extends Resource<
284 "AWS.IoTFleetWise.Campaign",
285 CampaignProps,
286 {
287 /** The name of the campaign. */
288 campaignName: string;
289 /** The ARN of the campaign. */
290 campaignArn: string;
291 /** The current status of the campaign (`RUNNING`, `SUSPENDED`, ...). */
292 status: string;
293 /** The signal catalog the campaign collects from. */
294 signalCatalogArn: string | undefined;
295 /** The fleet or vehicle the campaign targets. */
296 targetArn: string | undefined;
297 },
298 never,
299 Providers
300> {}
301
302/**
303 * An AWS IoT FleetWise campaign — the orchestration of data-collection
304 * rules that the Edge Agent uses to decide which signals to collect from
305 * a {@link Fleet} or {@link Vehicle} and where to deliver them.
306 *
307 * Campaigns are created in `WAITING_FOR_APPROVAL` status; set
308 * `autoApprove: true` to have the provider approve them into `RUNNING`.
309 * Only the description and extra dimensions are mutable — every other
310 * change replaces the campaign. AWS IoT FleetWise is allowlist-gated and
311 * offered in `us-east-1`/`eu-central-1` only.
312 * ### Creating a Campaign
313 * **Example:** Time-Based Collection to S3
314 * ```typescript
315 * const campaign = yield* Campaign("SpeedTelemetry", {
316 * signalCatalogArn: catalog.signalCatalogArn,
317 * targetArn: fleet.fleetArn,
318 * collectionScheme: {
319 * timeBasedCollectionScheme: { period: "10 seconds" },
320 * },
321 * signalsToCollect: [{ name: "Vehicle.Speed" }],
322 * dataDestinationConfigs: [
323 * { s3Config: { bucketArn: bucket.bucketArn } },
324 * ],
325 * autoApprove: true,
326 * });
327 * ```
328 *
329 * **Example:** Condition-Based Collection
330 * ```typescript
331 * const campaign = yield* Campaign("HardBraking", {
332 * signalCatalogArn: catalog.signalCatalogArn,
333 * targetArn: fleet.fleetArn,
334 * collectionScheme: {
335 * conditionBasedCollectionScheme: {
336 * expression: "$variable.`Vehicle.Speed` > 120.0",
337 * minimumTriggerInterval: "5 seconds",
338 * triggerMode: "RISING_EDGE",
339 * },
340 * },
341 * signalsToCollect: [
342 * { name: "Vehicle.Speed", minimumSamplingInterval: "500 millis" },
343 * ],
344 * postTriggerCollectionDuration: "30 seconds",
345 * dataDestinationConfigs: [
346 * { s3Config: { bucketArn: bucket.bucketArn } },
347 * ],
348 * });
349 * ```
350 *
351 * @resource
352 */
353export const Campaign = Resource<Campaign>("AWS.IoTFleetWise.Campaign");
354
355export const CampaignProvider = () =>
356 Provider.effect(
357 Campaign,
358 Effect.gen(function* () {
359 const toName = (id: string, props: { campaignName?: string }) =>
360 props.campaignName
361 ? Effect.succeed(props.campaignName)
362 : createPhysicalName({ id, maxLength: 100 });
363
364 const readCampaign = Effect.fn(function* (name: string) {
365 return yield* iotfleetwise.getCampaign({ name }).pipe(
366 inFleetWiseRegion,
367 Effect.catchTag("ResourceNotFoundException", () =>
368 Effect.succeed(undefined),
369 ),
370 );
371 });
372
373 // Campaign provisioning is asynchronous (status CREATING). Bounded
374 // wait until the service settles it into an actionable status.
375 const waitUntilSettled = Effect.fn(function* (name: string) {
376 return yield* readCampaign(name).pipe(
377 Effect.flatMap((campaign) => {
378 if (campaign === undefined) {
379 return Effect.fail(new Error(`Campaign '${name}' not found`));
380 }
381 if (campaign.status === "CREATING") {
382 return Effect.fail(
383 new Error(`Campaign '${name}' still creating`),
384 );
385 }
386 return Effect.succeed(campaign);
387 }),
388 retryObservation,
389 );
390 });
391
392 const toAttrs = Effect.fn(function* (
393 campaign: iotfleetwise.GetCampaignResponse,
394 ) {
395 if (campaign.name === undefined || campaign.arn === undefined) {
396 return yield* Effect.fail(
397 new Error(`Campaign '${campaign.name}' is missing its ARN`),
398 );
399 }
400 return {
401 campaignName: campaign.name,
402 campaignArn: campaign.arn,
403 status: campaign.status ?? "CREATING",
404 signalCatalogArn: campaign.signalCatalogArn,
405 targetArn: campaign.targetArn,
406 };
407 });
408
409 return {
410 stables: ["campaignName", "campaignArn"],
411
412 diff: Effect.fn(function* ({ id, olds, news }) {
413 if (!isResolved(news)) return undefined;
414 if ((yield* toName(id, olds)) !== (yield* toName(id, news))) {
415 return { action: "replace" } as const;
416 }
417 // Only description and dataExtraDimensions are mutable; every
418 // other property is fixed at creation.
419 const createOnly = (props: CampaignProps) => ({
420 signalCatalogArn: props.signalCatalogArn,
421 targetArn: props.targetArn,
422 // Durations are compared in wire units so that equivalent
423 // inputs ("10 seconds" vs 10_000) never force a replacement.
424 collectionScheme: toWireCollectionScheme(props.collectionScheme),
425 signalsToCollect: toWireSignalsToCollect(props.signalsToCollect),
426 dataDestinationConfigs: props.dataDestinationConfigs,
427 startTime: props.startTime,
428 expiryTime: props.expiryTime,
429 postTriggerCollectionDuration: toWireSeconds(
430 props.postTriggerCollectionDuration,
431 ),
432 diagnosticsMode: props.diagnosticsMode,
433 spoolingMode: props.spoolingMode,
434 compression: props.compression,
435 dataPartitions: props.dataPartitions,
436 signalsToFetch: toWireSignalsToFetch(props.signalsToFetch),
437 });
438 if (!stableEquals(createOnly(olds), createOnly(news))) {
439 return { action: "replace" } as const;
440 }
441 }),
442
443 read: Effect.fn(function* ({ id, olds, output }) {
444 const name = output?.campaignName ?? (yield* toName(id, olds ?? {}));
445 const found = yield* readCampaign(name);
446 if (found?.arn === undefined) return undefined;
447 const attrs = yield* toAttrs(found);
448 const tags = yield* readFleetWiseTags(found.arn);
449 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
450 }),
451
452 reconcile: Effect.fn(function* ({ id, news, output, session }) {
453 const name = output?.campaignName ?? (yield* toName(id, news));
454 const internalTags = yield* createInternalTags(id);
455 const desiredTags = { ...internalTags, ...news.tags };
456
457 // 1. Observe — cloud state is authoritative.
458 let observed = yield* readCampaign(name);
459
460 // 2. Ensure — create if missing; tolerate the AlreadyExists race.
461 if (observed === undefined) {
462 yield* iotfleetwise
463 .createCampaign({
464 name,
465 description: news.description,
466 signalCatalogArn: news.signalCatalogArn,
467 targetArn: news.targetArn,
468 collectionScheme: toWireCollectionScheme(news.collectionScheme),
469 signalsToCollect: toWireSignalsToCollect(news.signalsToCollect),
470 dataDestinationConfigs: news.dataDestinationConfigs,
471 startTime:
472 news.startTime !== undefined
473 ? new Date(news.startTime)
474 : undefined,
475 expiryTime:
476 news.expiryTime !== undefined
477 ? new Date(news.expiryTime)
478 : undefined,
479 postTriggerCollectionDuration: toWireSeconds(
480 news.postTriggerCollectionDuration,
481 ),
482 diagnosticsMode: news.diagnosticsMode,
483 spoolingMode: news.spoolingMode,
484 compression: news.compression,
485 dataExtraDimensions: news.dataExtraDimensions,
486 dataPartitions: news.dataPartitions,
487 signalsToFetch: toWireSignalsToFetch(news.signalsToFetch),
488 tags: toFleetWiseTagList(desiredTags),
489 })
490 .pipe(
491 inFleetWiseRegion,
492 Effect.catchTag("ConflictException", () => Effect.void),
493 );
494 }
495 observed = yield* waitUntilSettled(name);
496
497 // 3. Sync mutable aspects — description / dataExtraDimensions via
498 // the UPDATE action, applied only on an observed delta.
499 const descriptionChanged =
500 news.description !== undefined &&
501 news.description !== observed.description;
502 const dimensionsChanged =
503 news.dataExtraDimensions !== undefined &&
504 !stableEquals(
505 observed.dataExtraDimensions ?? [],
506 news.dataExtraDimensions,
507 );
508 if (descriptionChanged || dimensionsChanged) {
509 yield* iotfleetwise
510 .updateCampaign({
511 name,
512 action: "UPDATE",
513 description: descriptionChanged ? news.description : undefined,
514 dataExtraDimensions: dimensionsChanged
515 ? news.dataExtraDimensions
516 : undefined,
517 })
518 .pipe(inFleetWiseRegion);
519 }
520
521 // 3b. Approve a waiting campaign when requested.
522 if (
523 news.autoApprove === true &&
524 observed.status === "WAITING_FOR_APPROVAL"
525 ) {
526 yield* iotfleetwise
527 .updateCampaign({ name, action: "APPROVE" })
528 .pipe(inFleetWiseRegion);
529 }
530
531 // 3c. Sync tags against OBSERVED cloud tags.
532 const arn = observed.arn;
533 if (arn !== undefined) {
534 yield* syncFleetWiseTags(arn, desiredTags);
535 }
536
537 observed = yield* waitUntilSettled(name);
538 yield* session.note(name);
539 return yield* toAttrs(observed);
540 }),
541
542 delete: Effect.fn(function* ({ output }) {
543 // Deleting a campaign suspends data collection and removes it
544 // from vehicles; deleting a missing campaign is success.
545 yield* iotfleetwise
546 .deleteCampaign({ name: output.campaignName })
547 .pipe(
548 inFleetWiseRegion,
549 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
550 );
551 }),
552
553 list: () =>
554 iotfleetwise.listCampaigns.items({}).pipe(
555 Stream.runCollect,
556 Effect.map((chunk) =>
557 Array.from(chunk).flatMap((summary) =>
558 summary.name !== undefined && summary.arn !== undefined
559 ? [
560 {
561 campaignName: summary.name,
562 campaignArn: summary.arn,
563 status: summary.status ?? "CREATING",
564 signalCatalogArn: summary.signalCatalogArn,
565 targetArn: summary.targetArn,
566 },
567 ]
568 : [],
569 ),
570 ),
571 inFleetWiseRegion,
572 ),
573 };
574 }),
575 );