1import * as iam from "@distilled.cloud/aws/iam";
2import * as analytics from "@distilled.cloud/aws/kinesis-analytics-v2";
3import * as Data from "effect/Data";
4import type * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";
6import * as Schedule from "effect/Schedule";
7import * as Stream from "effect/Stream";
8import { Unowned } from "../../AdoptPolicy.ts";
9import { isResolved } from "../../Diff.ts";
10import { createPhysicalName } from "../../PhysicalName.ts";
11import * as Provider from "../../Provider.ts";
12import { Resource } from "../../Resource.ts";
13import {
14 createInternalTags,
15 createTagsList,
16 diffTags,
17 hasAlchemyTags,
18 type Tags,
19} from "../../Tags.ts";
20import { toWireMillis } from "../../Util/Duration.ts";
21import { AWSEnvironment } from "../Environment.ts";
22import type { PolicyStatement } from "../IAM/Policy.ts";
23import type { Providers } from "../Providers.ts";
24
25/**
26 * Runtime environment of a Managed Service for Apache Flink application,
27 * e.g. `"FLINK-1_20"`.
28 */
29export type RuntimeEnvironment = analytics.RuntimeEnvironment;
30
31/**
32 * Lifecycle status of the application.
33 */
34export type ApplicationStatus = analytics.ApplicationStatus;
35
36/**
37 * Mode of the application. `STREAMING` runs the packaged Flink job;
38 * `INTERACTIVE` backs a Studio (Zeppelin) notebook.
39 */
40export type ApplicationMode = "STREAMING" | "INTERACTIVE";
41
42export interface ApplicationCodeProps {
43 /**
44 * ARN of the S3 bucket that contains the application code package
45 * (a zip/jar built for the configured Flink runtime).
46 */
47 bucketArn: string;
48 /**
49 * Object key of the application code package within the bucket.
50 */
51 fileKey: string;
52 /**
53 * Version of the S3 object containing the application code.
54 * @default the latest object version
55 */
56 objectVersion?: string;
57}
58
59export interface CheckpointConfigurationProps {
60 /**
61 * Whether to use the `DEFAULT` Flink checkpointing behavior or the
62 * `CUSTOM` values supplied in this configuration.
63 */
64 configurationType: "DEFAULT" | "CUSTOM";
65 /**
66 * Whether checkpointing is enabled. Only used with `CUSTOM`.
67 */
68 checkpointingEnabled?: boolean;
69 /**
70 * Interval between checkpoints, e.g. `"1 minute"` or
71 * `Duration.seconds(30)`. The API stores whole milliseconds. Only used
72 * with `CUSTOM`.
73 */
74 checkpointInterval?: Duration.Input;
75 /**
76 * Minimum pause between checkpoint operations, e.g. `"5 seconds"`. The
77 * API stores whole milliseconds. Only used with `CUSTOM`.
78 */
79 minPauseBetweenCheckpoints?: Duration.Input;
80}
81
82export interface MonitoringConfigurationProps {
83 /**
84 * Whether to use the `DEFAULT` monitoring behavior or the `CUSTOM`
85 * values supplied in this configuration.
86 */
87 configurationType: "DEFAULT" | "CUSTOM";
88 /**
89 * Granularity of the CloudWatch metrics emitted by the application.
90 * Only used with `CUSTOM`.
91 */
92 metricsLevel?: "APPLICATION" | "TASK" | "OPERATOR" | "PARALLELISM";
93 /**
94 * Log verbosity of the application. Only used with `CUSTOM`.
95 */
96 logLevel?: "INFO" | "WARN" | "ERROR" | "DEBUG";
97}
98
99export interface ParallelismConfigurationProps {
100 /**
101 * Whether to use the `DEFAULT` parallelism behavior or the `CUSTOM`
102 * values supplied in this configuration.
103 */
104 configurationType: "DEFAULT" | "CUSTOM";
105 /**
106 * Initial number of parallel tasks the application can perform.
107 * Only used with `CUSTOM`.
108 */
109 parallelism?: number;
110 /**
111 * Number of parallel tasks per Kinesis Processing Unit.
112 * Only used with `CUSTOM`.
113 */
114 parallelismPerKPU?: number;
115 /**
116 * Whether the service automatically scales the application's parallelism.
117 * Only used with `CUSTOM`.
118 */
119 autoScalingEnabled?: boolean;
120}
121
122export interface FlinkConfigurationProps {
123 /**
124 * Checkpointing (fault tolerance) settings for the Flink job.
125 */
126 checkpointConfiguration?: CheckpointConfigurationProps;
127 /**
128 * CloudWatch metrics/logging verbosity settings for the Flink job.
129 */
130 monitoringConfiguration?: MonitoringConfigurationProps;
131 /**
132 * Parallelism (KPU scaling) settings for the Flink job.
133 */
134 parallelismConfiguration?: ParallelismConfigurationProps;
135}
136
137export interface PropertyGroupProps {
138 /**
139 * Identifier of the property group, e.g. `"kinesis.analytics.flink.run.options"`
140 * or an application-defined group name.
141 */
142 propertyGroupId: string;
143 /**
144 * Key-value runtime properties exposed to the application through this group.
145 */
146 propertyMap: Record<string, string>;
147}
148
149export interface ApplicationVpcProps {
150 /**
151 * IDs of the subnets the application's ENIs are placed in.
152 */
153 subnetIds: string[];
154 /**
155 * IDs of the security groups attached to the application's ENIs.
156 */
157 securityGroupIds: string[];
158}
159
160export interface ApplicationProps {
161 /**
162 * Name of the application. Changing the name replaces the application.
163 * @default ${app}-${id}-${stage}-${instanceId}
164 */
165 applicationName?: string;
166 /**
167 * Description of the application. The API offers no way to change the
168 * description in place, so changing it replaces the application.
169 */
170 description?: string;
171 /**
172 * Runtime environment of the application, e.g. `"FLINK-1_20"`.
173 * Updated in place (Flink version upgrade).
174 */
175 runtimeEnvironment: RuntimeEnvironment;
176 /**
177 * Mode of the application. Changing the mode replaces the application.
178 * @default "STREAMING"
179 */
180 applicationMode?: ApplicationMode;
181 /**
182 * ARN of the IAM role the service assumes to read the code object and
183 * access sources/sinks.
184 * @default a role is auto-created granting read access to the code
185 * bucket, CloudWatch Logs delivery, and (when `vpc` is set) the EC2
186 * permissions required to manage the application's network interfaces.
187 */
188 serviceExecutionRole?: string;
189 /**
190 * Location of the application code package (zip/jar) in S3.
191 */
192 code: ApplicationCodeProps;
193 /**
194 * Runtime properties exposed to the application as property groups.
195 */
196 environmentProperties?: PropertyGroupProps[];
197 /**
198 * Flink-specific settings (checkpointing, monitoring, parallelism).
199 */
200 flinkConfiguration?: FlinkConfigurationProps;
201 /**
202 * Whether snapshots (savepoints) are enabled for the application.
203 * @default false
204 */
205 snapshotsEnabled?: boolean;
206 /**
207 * VPC configuration. When set the application's ENIs are placed in the
208 * given subnets so the Flink job can reach VPC-private resources.
209 */
210 vpc?: ApplicationVpcProps;
211 /**
212 * Whether the application should be running. When `true` the reconciler
213 * starts the application and waits for `RUNNING`; when `false` (default)
214 * a running application is force-stopped back to `READY`.
215 *
216 * Starting requires the code object to be a real Flink application jar —
217 * a placeholder zip passes creation but fails to start.
218 * @default false
219 */
220 start?: boolean;
221 /**
222 * Start of the 8-hour daily window in which the service may apply
223 * maintenance (patching) to the application, as `"HH:mm"` UTC — e.g.
224 * `"02:00"`. When omitted the service-assigned window is left unchanged.
225 */
226 maintenanceWindowStartTime?: string;
227 /**
228 * Tags to apply to the application.
229 */
230 tags?: Record<string, string>;
231}
232
233export interface Application extends Resource<
234 "AWS.KinesisAnalyticsV2.Application",
235 ApplicationProps,
236 {
237 /**
238 * Physical name of the application.
239 */
240 applicationName: string;
241 /**
242 * ARN of the application.
243 */
244 applicationArn: string;
245 /**
246 * Current lifecycle status of the application.
247 */
248 applicationStatus: ApplicationStatus;
249 /**
250 * Current version of the application's configuration. Incremented by
251 * every configuration update.
252 */
253 applicationVersionId: number;
254 /**
255 * Runtime environment of the application.
256 */
257 runtimeEnvironment: RuntimeEnvironment;
258 /**
259 * Mode of the application.
260 */
261 applicationMode: ApplicationMode | undefined;
262 /**
263 * ARN of the IAM role the service assumes.
264 */
265 serviceExecutionRole: string | undefined;
266 /**
267 * Name of the auto-created IAM role, when one was synthesized for this
268 * application. `undefined` when the caller supplied `serviceExecutionRole`.
269 */
270 roleName: string | undefined;
271 /**
272 * ID of the application's VPC configuration, when one is attached.
273 */
274 vpcConfigurationId: string | undefined;
275 /**
276 * ARN of the S3 bucket currently holding the application code.
277 */
278 codeBucketArn: string | undefined;
279 /**
280 * Object key of the application code currently configured.
281 */
282 codeFileKey: string | undefined;
283 /**
284 * Object version of the application code currently configured.
285 */
286 codeObjectVersion: string | undefined;
287 /**
288 * Start of the daily maintenance window currently configured, as
289 * `"HH:mm"` UTC.
290 */
291 maintenanceWindowStartTime: string | undefined;
292 /**
293 * End of the daily maintenance window currently configured, as
294 * `"HH:mm"` UTC (always 8 hours after the start).
295 */
296 maintenanceWindowEndTime: string | undefined;
297 /**
298 * Current tags reported for the application.
299 */
300 tags: Record<string, string>;
301 },
302 never,
303 Providers
304> {}
305
306/**
307 * A Managed Service for Apache Flink (Kinesis Data Analytics v2)
308 * application.
309 *
310 * `Application` owns the application definition — runtime environment, code
311 * location in S3, runtime properties, Flink settings, optional VPC
312 * connectivity and tags — and converges each aspect in place via
313 * `UpdateApplication`. Unless you supply `serviceExecutionRole`, an IAM role
314 * is auto-created granting the service read access to the code bucket and
315 * CloudWatch Logs delivery.
316 *
317 * The application is created in `READY` and does not run (or bill KPUs)
318 * until started. Set `start: true` to have the reconciler start the job and
319 * wait for `RUNNING` — this requires the code object to be a real Flink
320 * application jar.
321 * ### Creating Applications
322 * **Example:** Flink application from S3 code
323 * ```typescript
324 * import * as AWS from "alchemy/AWS";
325 *
326 * const bucket = yield* AWS.S3.Bucket("FlinkCode");
327 * const app = yield* AWS.KinesisAnalyticsV2.Application("Enrichment", {
328 * runtimeEnvironment: "FLINK-1_20",
329 * code: {
330 * bucketArn: bucket.bucketArn,
331 * fileKey: "jobs/enrichment-1.0.jar",
332 * },
333 * });
334 * ```
335 *
336 * **Example:** Runtime properties and parallelism
337 * ```typescript
338 * const app = yield* AWS.KinesisAnalyticsV2.Application("Enrichment", {
339 * runtimeEnvironment: "FLINK-1_20",
340 * code: { bucketArn: bucket.bucketArn, fileKey: "jobs/enrichment-1.0.jar" },
341 * environmentProperties: [
342 * {
343 * propertyGroupId: "EnrichmentProperties",
344 * propertyMap: { "input.stream": "clickstream", "region": "us-west-2" },
345 * },
346 * ],
347 * flinkConfiguration: {
348 * parallelismConfiguration: {
349 * configurationType: "CUSTOM",
350 * parallelism: 2,
351 * parallelismPerKPU: 1,
352 * autoScalingEnabled: false,
353 * },
354 * },
355 * snapshotsEnabled: true,
356 * });
357 * ```
358 *
359 * ### VPC Connectivity
360 * **Example:** Place the application in a VPC
361 * ```typescript
362 * const app = yield* AWS.KinesisAnalyticsV2.Application("Enrichment", {
363 * runtimeEnvironment: "FLINK-1_20",
364 * code: { bucketArn: bucket.bucketArn, fileKey: "jobs/enrichment-1.0.jar" },
365 * vpc: {
366 * subnetIds: [subnetA.subnetId, subnetB.subnetId],
367 * securityGroupIds: [sg.securityGroupId],
368 * },
369 * });
370 * ```
371 *
372 * ### Maintenance Window
373 * **Example:** Pin the daily maintenance window
374 * ```typescript
375 * const app = yield* AWS.KinesisAnalyticsV2.Application("Enrichment", {
376 * runtimeEnvironment: "FLINK-1_20",
377 * code: { bucketArn: bucket.bucketArn, fileKey: "jobs/enrichment-1.0.jar" },
378 * maintenanceWindowStartTime: "02:00",
379 * });
380 * ```
381 *
382 * ### Running the Application
383 * **Example:** Start the Flink job and keep it running
384 * ```typescript
385 * const app = yield* AWS.KinesisAnalyticsV2.Application("Enrichment", {
386 * runtimeEnvironment: "FLINK-1_20",
387 * code: { bucketArn: bucket.bucketArn, fileKey: "jobs/enrichment-1.0.jar" },
388 * start: true,
389 * });
390 * ```
391 *
392 * @resource
393 */
394export const Application = Resource<Application>(
395 "AWS.KinesisAnalyticsV2.Application",
396);
397
398/**
399 * Validation error raised before any AWS call when the props are invalid.
400 */
401export class ApplicationValidationError extends Data.TaggedError(
402 "ApplicationValidationError",
403)<{
404 readonly message: string;
405}> {}
406
407/**
408 * The application did not settle into a stable status (READY / RUNNING /
409 * ROLLED_BACK) within the bounded wait.
410 */
411export class ApplicationNotStable extends Data.TaggedError(
412 "ApplicationNotStable",
413)<{
414 readonly applicationName: string;
415 readonly status: string;
416}> {}
417
418/**
419 * `start: true` was requested but the application failed to reach `RUNNING`
420 * — either the start rolled back (bad jar, bad configuration) or the
421 * bounded wait elapsed.
422 */
423export class ApplicationStartFailed extends Data.TaggedError(
424 "ApplicationStartFailed",
425)<{
426 readonly applicationName: string;
427 readonly status: string;
428}> {}
429
430class ApplicationStatusPending extends Data.TaggedError(
431 "ApplicationStatusPending",
432)<{
433 readonly status: string;
434}> {}
435
436class ApplicationStillExists extends Data.TaggedError(
437 "ApplicationStillExists",
438) {}
439
440const createApplicationName = (
441 id: string,
442 props: { applicationName?: string | undefined },
443) =>
444 Effect.gen(function* () {
445 if (props.applicationName) {
446 return props.applicationName;
447 }
448 return yield* createPhysicalName({ id, maxLength: 128 });
449 });
450
451/**
452 * Data-FIRST `Effect.retry(self, options)` wrapped with an explicit return
453 * annotation — inlining `Effect.retry` with an options object in provider
454 * lifecycle code leaves `Retry.Return`'s conditional type unresolved in the
455 * provider's inferred layer type, which TypeScript's declaration emit widens
456 * to an `unknown` R — poisoning `AWS.providers()` for every downstream
457 * consumer (see `retryThroughDeletionWindow` in SecretsManager/Secret.ts).
458 *
459 * Retries `InvalidArgumentException`s that mention the IAM role — a freshly
460 * created role takes a few seconds to become assumable by the
461 * kinesisanalytics service principal, and `CreateApplication` surfaces that
462 * propagation window as a role-related `InvalidArgumentException`.
463 */
464const retryThroughRolePropagation = <A, E extends { _tag: string }, R>(
465 self: Effect.Effect<A, E, R>,
466): Effect.Effect<A, E, R> =>
467 Effect.retry(self, {
468 while: (e) =>
469 e._tag === "InvalidArgumentException" &&
470 /role|assume|trust|principal/i.test(
471 (e as { message?: string }).message ?? "",
472 ),
473 schedule: Schedule.max([Schedule.fixed("2 seconds"), Schedule.recurs(15)]),
474 });
475
476/**
477 * Retries `ResourceInUseException` / `ConcurrentModificationException`
478 * (application transitioning between statuses, or a concurrent operation in
479 * flight) on a bounded schedule. Explicitly annotated for the same
480 * declaration-emit reason as {@link retryThroughRolePropagation}.
481 */
482const retryWhileInUse = <A, E extends { _tag: string }, R>(
483 self: Effect.Effect<A, E, R>,
484): Effect.Effect<A, E, R> =>
485 Effect.retry(self, {
486 while: (e) =>
487 e._tag === "ResourceInUseException" ||
488 e._tag === "ConcurrentModificationException",
489 schedule: Schedule.max([Schedule.fixed("2 seconds"), Schedule.recurs(30)]),
490 });
491
492const toTagRecord = (
493 tags: ReadonlyArray<{ Key: string; Value?: string }> | undefined,
494) =>
495 Object.fromEntries(
496 (tags ?? [])
497 .filter(
498 (tag): tag is { Key: string; Value: string } =>
499 typeof tag.Value === "string",
500 )
501 .map((tag) => [tag.Key, tag.Value]),
502 );
503
504const toAttrs = ({
505 detail,
506 tags,
507 roleName,
508}: {
509 detail: analytics.ApplicationDetail;
510 tags: Record<string, string>;
511 roleName: string | undefined;
512}): Application["Attributes"] => {
513 const config = detail.ApplicationConfigurationDescription;
514 const code =
515 config?.ApplicationCodeConfigurationDescription?.CodeContentDescription
516 ?.S3ApplicationCodeLocationDescription;
517 return {
518 applicationName: detail.ApplicationName,
519 applicationArn: detail.ApplicationARN,
520 applicationStatus: detail.ApplicationStatus,
521 applicationVersionId: detail.ApplicationVersionId,
522 runtimeEnvironment: detail.RuntimeEnvironment,
523 applicationMode: detail.ApplicationMode as ApplicationMode | undefined,
524 serviceExecutionRole: detail.ServiceExecutionRole,
525 roleName,
526 vpcConfigurationId:
527 config?.VpcConfigurationDescriptions?.[0]?.VpcConfigurationId,
528 codeBucketArn: code?.BucketARN,
529 codeFileKey: code?.FileKey,
530 codeObjectVersion: code?.ObjectVersion,
531 maintenanceWindowStartTime:
532 detail.ApplicationMaintenanceConfigurationDescription
533 ?.ApplicationMaintenanceWindowStartTime,
534 maintenanceWindowEndTime:
535 detail.ApplicationMaintenanceConfigurationDescription
536 ?.ApplicationMaintenanceWindowEndTime,
537 tags,
538 };
539};
540
541const describeApplicationDetail = Effect.fn(function* (
542 applicationName: string,
543) {
544 const response = yield* analytics
545 .describeApplication({ ApplicationName: applicationName })
546 .pipe(
547 Effect.catchTag("ResourceNotFoundException", () =>
548 Effect.succeed(undefined),
549 ),
550 );
551 return response?.ApplicationDetail;
552});
553
554const readApplication = Effect.fn(function* ({
555 applicationName,
556 roleName,
557}: {
558 applicationName: string;
559 roleName: string | undefined;
560}) {
561 const detail = yield* describeApplicationDetail(applicationName);
562 if (!detail) {
563 return undefined;
564 }
565 // The application can vanish between the describe above and the tag read
566 // (e.g. a concurrent destroy) — a typed `ResourceNotFoundException` here
567 // just means it's gone.
568 const tagsResponse = yield* analytics
569 .listTagsForResource({ ResourceARN: detail.ApplicationARN })
570 .pipe(
571 Effect.catchTag("ResourceNotFoundException", () =>
572 Effect.succeed(undefined),
573 ),
574 );
575 if (!tagsResponse) {
576 return undefined;
577 }
578 return toAttrs({
579 detail,
580 tags: toTagRecord(tagsResponse.Tags),
581 roleName,
582 });
583});
584
585// Configuration updates settle fast (the app stays READY); 2s × 45 bounds
586// the wait at 90s. RUNNING is also stable — a started app being updated
587// passes through UPDATING and returns to RUNNING.
588const waitForApplicationStable = (applicationName: string) =>
589 Effect.gen(function* () {
590 const detail = yield* describeApplicationDetail(applicationName);
591 if (!detail) {
592 return yield* Effect.fail(
593 new ApplicationNotStable({ applicationName, status: "MISSING" }),
594 );
595 }
596 if (
597 detail.ApplicationStatus !== "READY" &&
598 detail.ApplicationStatus !== "RUNNING" &&
599 detail.ApplicationStatus !== "ROLLED_BACK"
600 ) {
601 return yield* Effect.fail(
602 new ApplicationStatusPending({ status: detail.ApplicationStatus }),
603 );
604 }
605 return detail;
606 }).pipe(
607 Effect.retry({
608 while: (e: { _tag: string }) => e._tag === "ApplicationStatusPending",
609 schedule: Schedule.max([
610 Schedule.fixed("2 seconds"),
611 Schedule.recurs(45),
612 ]),
613 }),
614 Effect.catchTag("ApplicationStatusPending", (e) =>
615 Effect.fail(
616 new ApplicationNotStable({ applicationName, status: e.status }),
617 ),
618 ),
619 );
620
621// A Flink job start submits the jar to a fresh cluster — several minutes is
622// normal. 10s × 57 bounds the wait at 570s. A start that rolls back
623// (ROLLED_BACK / back to READY after STARTING) fails fast with a typed error.
624const waitForApplicationRunning = (applicationName: string) =>
625 Effect.gen(function* () {
626 const detail = yield* describeApplicationDetail(applicationName);
627 const status = detail?.ApplicationStatus ?? "MISSING";
628 if (status === "RUNNING") {
629 return detail!;
630 }
631 if (status === "STARTING" || status === "UPDATING") {
632 return yield* Effect.fail(new ApplicationStatusPending({ status }));
633 }
634 return yield* Effect.fail(
635 new ApplicationStartFailed({ applicationName, status }),
636 );
637 }).pipe(
638 Effect.retry({
639 while: (e: { _tag: string }) => e._tag === "ApplicationStatusPending",
640 schedule: Schedule.max([
641 Schedule.fixed("10 seconds"),
642 Schedule.recurs(57),
643 ]),
644 }),
645 Effect.catchTag("ApplicationStatusPending", (e) =>
646 Effect.fail(
647 new ApplicationStartFailed({ applicationName, status: e.status }),
648 ),
649 ),
650 );
651
652// Force-stop transitions through FORCE_STOPPING/STOPPING back to READY;
653// 10s × 30 bounds the wait at 300s.
654const waitForApplicationStopped = (applicationName: string) =>
655 Effect.gen(function* () {
656 const detail = yield* describeApplicationDetail(applicationName);
657 const status = detail?.ApplicationStatus ?? "MISSING";
658 if (
659 status === "READY" ||
660 status === "ROLLED_BACK" ||
661 status === "MISSING"
662 ) {
663 return;
664 }
665 return yield* Effect.fail(new ApplicationStatusPending({ status }));
666 }).pipe(
667 Effect.retry({
668 while: (e: { _tag: string }) => e._tag === "ApplicationStatusPending",
669 schedule: Schedule.max([
670 Schedule.fixed("10 seconds"),
671 Schedule.recurs(30),
672 ]),
673 }),
674 Effect.catchTag("ApplicationStatusPending", (e) =>
675 Effect.fail(
676 new ApplicationNotStable({ applicationName, status: e.status }),
677 ),
678 ),
679 );
680
681// Deletion is quick for READY apps but a previously-running app halts its
682// job first; 3s × 50 bounds the wait at 150s.
683const waitForApplicationDeleted = (applicationName: string) =>
684 Effect.gen(function* () {
685 const detail = yield* describeApplicationDetail(applicationName);
686 if (detail !== undefined) {
687 return yield* Effect.fail(new ApplicationStillExists());
688 }
689 }).pipe(
690 Effect.retry({
691 while: (e: { _tag: string }) => e._tag === "ApplicationStillExists",
692 schedule: Schedule.max([
693 Schedule.fixed("3 seconds"),
694 Schedule.recurs(50),
695 ]),
696 }),
697 );
698
699const deleteRoleIfExists = Effect.fn(function* (roleName: string) {
700 yield* iam
701 .deleteRolePolicy({ RoleName: roleName, PolicyName: roleName })
702 .pipe(Effect.catchTag("NoSuchEntityException", () => Effect.void));
703 yield* iam
704 .deleteRole({ RoleName: roleName })
705 .pipe(Effect.catchTag("NoSuchEntityException", () => Effect.void));
706});
707
708const toWirePropertyGroups = (
709 groups: ReadonlyArray<PropertyGroupProps>,
710): analytics.PropertyGroup[] =>
711 groups.map((group) => ({
712 PropertyGroupId: group.propertyGroupId,
713 PropertyMap: group.propertyMap,
714 }));
715
716// Canonical string form for observed-vs-desired comparison — order of
717// groups and of keys within a group is not significant.
718const canonicalPropertyGroups = (
719 groups: ReadonlyArray<analytics.PropertyGroup> | undefined,
720) =>
721 JSON.stringify(
722 [...(groups ?? [])]
723 .map((group) => ({
724 id: group.PropertyGroupId,
725 map: Object.fromEntries(
726 Object.entries(group.PropertyMap ?? {})
727 .filter(([, value]) => value !== undefined)
728 .sort(([a], [b]) => a.localeCompare(b)),
729 ),
730 }))
731 .sort((a, b) => a.id.localeCompare(b.id)),
732 );
733
734const sameStringSet = (a: ReadonlyArray<string>, b: ReadonlyArray<string>) =>
735 a.length === b.length && [...a].sort().join(",") === [...b].sort().join(",");
736
737const checkpointDiffers = (
738 desired: CheckpointConfigurationProps,
739 observed: analytics.CheckpointConfigurationDescription | undefined,
740) =>
741 desired.configurationType !== observed?.ConfigurationType ||
742 (desired.checkpointingEnabled !== undefined &&
743 desired.checkpointingEnabled !== observed?.CheckpointingEnabled) ||
744 (desired.checkpointInterval !== undefined &&
745 toWireMillis(desired.checkpointInterval) !==
746 observed?.CheckpointInterval) ||
747 (desired.minPauseBetweenCheckpoints !== undefined &&
748 toWireMillis(desired.minPauseBetweenCheckpoints) !==
749 observed?.MinPauseBetweenCheckpoints);
750
751const monitoringDiffers = (
752 desired: MonitoringConfigurationProps,
753 observed: analytics.MonitoringConfigurationDescription | undefined,
754) =>
755 desired.configurationType !== observed?.ConfigurationType ||
756 (desired.metricsLevel !== undefined &&
757 desired.metricsLevel !== observed?.MetricsLevel) ||
758 (desired.logLevel !== undefined && desired.logLevel !== observed?.LogLevel);
759
760const parallelismDiffers = (
761 desired: ParallelismConfigurationProps,
762 observed: analytics.ParallelismConfigurationDescription | undefined,
763) =>
764 desired.configurationType !== observed?.ConfigurationType ||
765 (desired.parallelism !== undefined &&
766 desired.parallelism !== observed?.Parallelism) ||
767 (desired.parallelismPerKPU !== undefined &&
768 desired.parallelismPerKPU !== observed?.ParallelismPerKPU) ||
769 (desired.autoScalingEnabled !== undefined &&
770 desired.autoScalingEnabled !== observed?.AutoScalingEnabled);
771
772const toFlinkConfiguration = (
773 flink: FlinkConfigurationProps | undefined,
774): analytics.FlinkApplicationConfiguration | undefined => {
775 if (!flink) return undefined;
776 const configuration: analytics.FlinkApplicationConfiguration = {};
777 if (flink.checkpointConfiguration) {
778 configuration.CheckpointConfiguration = {
779 ConfigurationType: flink.checkpointConfiguration.configurationType,
780 CheckpointingEnabled: flink.checkpointConfiguration.checkpointingEnabled,
781 CheckpointInterval: toWireMillis(
782 flink.checkpointConfiguration.checkpointInterval,
783 ),
784 MinPauseBetweenCheckpoints: toWireMillis(
785 flink.checkpointConfiguration.minPauseBetweenCheckpoints,
786 ),
787 };
788 }
789 if (flink.monitoringConfiguration) {
790 configuration.MonitoringConfiguration = {
791 ConfigurationType: flink.monitoringConfiguration.configurationType,
792 MetricsLevel: flink.monitoringConfiguration.metricsLevel,
793 LogLevel: flink.monitoringConfiguration.logLevel,
794 };
795 }
796 if (flink.parallelismConfiguration) {
797 configuration.ParallelismConfiguration = {
798 ConfigurationType: flink.parallelismConfiguration.configurationType,
799 Parallelism: flink.parallelismConfiguration.parallelism,
800 ParallelismPerKPU: flink.parallelismConfiguration.parallelismPerKPU,
801 AutoScalingEnabled: flink.parallelismConfiguration.autoScalingEnabled,
802 };
803 }
804 return Object.keys(configuration).length > 0 ? configuration : undefined;
805};
806
807const toFlinkConfigurationUpdate = (
808 flink: FlinkConfigurationProps,
809 observed: analytics.FlinkApplicationConfigurationDescription | undefined,
810): analytics.FlinkApplicationConfigurationUpdate | undefined => {
811 const update: analytics.FlinkApplicationConfigurationUpdate = {};
812 if (
813 flink.checkpointConfiguration &&
814 checkpointDiffers(
815 flink.checkpointConfiguration,
816 observed?.CheckpointConfigurationDescription,
817 )
818 ) {
819 update.CheckpointConfigurationUpdate = {
820 ConfigurationTypeUpdate: flink.checkpointConfiguration.configurationType,
821 CheckpointingEnabledUpdate:
822 flink.checkpointConfiguration.checkpointingEnabled,
823 CheckpointIntervalUpdate: toWireMillis(
824 flink.checkpointConfiguration.checkpointInterval,
825 ),
826 MinPauseBetweenCheckpointsUpdate: toWireMillis(
827 flink.checkpointConfiguration.minPauseBetweenCheckpoints,
828 ),
829 };
830 }
831 if (
832 flink.monitoringConfiguration &&
833 monitoringDiffers(
834 flink.monitoringConfiguration,
835 observed?.MonitoringConfigurationDescription,
836 )
837 ) {
838 update.MonitoringConfigurationUpdate = {
839 ConfigurationTypeUpdate: flink.monitoringConfiguration.configurationType,
840 MetricsLevelUpdate: flink.monitoringConfiguration.metricsLevel,
841 LogLevelUpdate: flink.monitoringConfiguration.logLevel,
842 };
843 }
844 if (
845 flink.parallelismConfiguration &&
846 parallelismDiffers(
847 flink.parallelismConfiguration,
848 observed?.ParallelismConfigurationDescription,
849 )
850 ) {
851 update.ParallelismConfigurationUpdate = {
852 ConfigurationTypeUpdate: flink.parallelismConfiguration.configurationType,
853 ParallelismUpdate: flink.parallelismConfiguration.parallelism,
854 ParallelismPerKPUUpdate: flink.parallelismConfiguration.parallelismPerKPU,
855 AutoScalingEnabledUpdate:
856 flink.parallelismConfiguration.autoScalingEnabled,
857 };
858 }
859 return Object.keys(update).length > 0 ? update : undefined;
860};
861
862export const ApplicationProvider = () =>
863 Provider.effect(
864 Application,
865 Effect.gen(function* () {
866 const createRoleName = (id: string) =>
867 createPhysicalName({ id, maxLength: 64 });
868
869 // Ensure the synthesized IAM role exists and its inline policy matches
870 // the desired access. `createRole` tolerates the already-exists race;
871 // `putRolePolicy` is an idempotent upsert, so the policy re-converges
872 // on every reconcile (e.g. after the code bucket changes or a VPC
873 // configuration is added).
874 const ensureRole = Effect.fn(function* ({
875 id,
876 roleName,
877 codeBucketArn,
878 vpc,
879 }: {
880 id: string;
881 roleName: string;
882 codeBucketArn: string;
883 vpc: ApplicationVpcProps | undefined;
884 }) {
885 const tags = yield* createInternalTags(id);
886 yield* iam
887 .createRole({
888 RoleName: roleName,
889 AssumeRolePolicyDocument: JSON.stringify({
890 Version: "2012-10-17",
891 Statement: [
892 {
893 Effect: "Allow",
894 Principal: { Service: "kinesisanalytics.amazonaws.com" },
895 Action: "sts:AssumeRole",
896 },
897 ],
898 }),
899 Tags: createTagsList(tags),
900 })
901 .pipe(
902 Effect.catchTag("EntityAlreadyExistsException", () =>
903 iam.getRole({ RoleName: roleName }),
904 ),
905 );
906
907 const statements: PolicyStatement[] = [
908 {
909 Effect: "Allow",
910 Action: ["s3:GetObject", "s3:GetObjectVersion"],
911 Resource: [`${codeBucketArn}/*`],
912 },
913 {
914 Effect: "Allow",
915 Action: ["s3:GetBucketLocation", "s3:ListBucket"],
916 Resource: [codeBucketArn],
917 },
918 {
919 // Log stream ARNs are attached after the application exists (via
920 // ApplicationCloudWatchLoggingOption), so log delivery is granted
921 // broadly rather than per-stream.
922 Effect: "Allow",
923 Action: [
924 "logs:DescribeLogGroups",
925 "logs:DescribeLogStreams",
926 "logs:PutLogEvents",
927 ],
928 Resource: ["*"],
929 },
930 ];
931 if (vpc) {
932 statements.push({
933 // ENI management for VPC-attached applications has no
934 // resource-level IAM support.
935 Effect: "Allow",
936 Action: [
937 "ec2:CreateNetworkInterface",
938 "ec2:CreateNetworkInterfacePermission",
939 "ec2:DeleteNetworkInterface",
940 "ec2:DescribeNetworkInterfaces",
941 "ec2:DescribeVpcs",
942 "ec2:DescribeSubnets",
943 "ec2:DescribeSecurityGroups",
944 "ec2:DescribeDhcpOptions",
945 ],
946 Resource: ["*"],
947 });
948 }
949
950 yield* iam.putRolePolicy({
951 RoleName: roleName,
952 PolicyName: roleName,
953 PolicyDocument: JSON.stringify({
954 Version: "2012-10-17",
955 Statement: statements,
956 }),
957 });
958 });
959
960 return Application.Provider.of({
961 stables: [
962 "applicationName",
963 "applicationArn",
964 "applicationMode",
965 "roleName",
966 ],
967
968 // Enumerate every application in the ambient account/region and
969 // hydrate each summary into the `read` shape with bounded
970 // concurrency, dropping applications that vanish mid-flight.
971 list: () =>
972 Effect.gen(function* () {
973 const summaries = Array.from(
974 yield* analytics.listApplications
975 .items({})
976 .pipe(Stream.runCollect),
977 );
978 const hydrated = yield* Effect.forEach(
979 summaries,
980 (summary) =>
981 readApplication({
982 applicationName: summary.ApplicationName,
983 roleName: undefined,
984 }),
985 { concurrency: 10 },
986 );
987 return hydrated.filter(
988 (attrs): attrs is Application["Attributes"] =>
989 attrs !== undefined,
990 );
991 }),
992
993 read: Effect.fn(function* ({ id, olds, output }) {
994 const applicationName =
995 output?.applicationName ??
996 (yield* createApplicationName(id, olds ?? {}));
997 const state = yield* readApplication({
998 applicationName,
999 roleName: output?.roleName,
1000 });
1001 if (!state) return undefined;
1002 return (yield* hasAlchemyTags(id, state.tags as Tags))
1003 ? state
1004 : Unowned(state);
1005 }),
1006
1007 diff: Effect.fn(function* ({ id, news, olds }) {
1008 if (!isResolved(news)) return undefined;
1009 const oldName = yield* createApplicationName(id, olds ?? {});
1010 const newName = yield* createApplicationName(id, news ?? {});
1011 if (oldName !== newName) {
1012 return { action: "replace" } as const;
1013 }
1014 // The mode is immutable.
1015 if (
1016 (olds?.applicationMode ?? "STREAMING") !==
1017 (news?.applicationMode ?? "STREAMING")
1018 ) {
1019 return { action: "replace" } as const;
1020 }
1021 // The API offers no UpdateApplication field for the description.
1022 if ((olds?.description ?? "") !== (news?.description ?? "")) {
1023 return { action: "replace" } as const;
1024 }
1025 }),
1026
1027 reconcile: Effect.fn(function* ({ id, news, output, session }) {
1028 const { accountId } = yield* AWSEnvironment.current;
1029
1030 if (!news?.code?.bucketArn || !news.code.fileKey) {
1031 return yield* Effect.fail(
1032 new ApplicationValidationError({
1033 message: `Application "${id}" requires code.bucketArn and code.fileKey`,
1034 }),
1035 );
1036 }
1037
1038 const applicationName =
1039 output?.applicationName ?? (yield* createApplicationName(id, news));
1040 const internalTags = yield* createInternalTags(id);
1041 const desiredTags = { ...news.tags, ...internalTags };
1042
1043 // Synthesize the IAM role when the caller didn't supply one.
1044 const roleName = news.serviceExecutionRole
1045 ? undefined
1046 : (output?.roleName ?? (yield* createRoleName(id)));
1047 const serviceExecutionRole =
1048 news.serviceExecutionRole ??
1049 `arn:aws:iam::${accountId}:role/${roleName}`;
1050 if (roleName) {
1051 yield* ensureRole({
1052 id,
1053 roleName,
1054 codeBucketArn: news.code.bucketArn,
1055 vpc: news.vpc,
1056 });
1057 }
1058
1059 // 1. OBSERVE — cloud state is authoritative; `output` is only a
1060 // cache for the physical name.
1061 const observed = yield* describeApplicationDetail(applicationName);
1062
1063 // 2. ENSURE — create the application if it's missing. Tolerate the
1064 // already-exists race (`ResourceInUseException`), retry through the
1065 // fresh role's IAM propagation window, and retry
1066 // `ConcurrentModificationException` — recreating an application
1067 // under a recently-deleted name fails with "Tags are already
1068 // registered for this resource ARN ... please retry later".
1069 if (observed === undefined) {
1070 const createRequest: analytics.CreateApplicationRequest = {
1071 ApplicationName: applicationName,
1072 ApplicationDescription: news.description,
1073 RuntimeEnvironment: news.runtimeEnvironment,
1074 ApplicationMode: news.applicationMode,
1075 ServiceExecutionRole: serviceExecutionRole,
1076 ApplicationConfiguration: {
1077 ApplicationCodeConfiguration: {
1078 CodeContentType: "ZIPFILE",
1079 CodeContent: {
1080 S3ContentLocation: {
1081 BucketARN: news.code.bucketArn,
1082 FileKey: news.code.fileKey,
1083 ObjectVersion: news.code.objectVersion,
1084 },
1085 },
1086 },
1087 EnvironmentProperties: news.environmentProperties
1088 ? {
1089 PropertyGroups: toWirePropertyGroups(
1090 news.environmentProperties,
1091 ),
1092 }
1093 : undefined,
1094 FlinkApplicationConfiguration: toFlinkConfiguration(
1095 news.flinkConfiguration,
1096 ),
1097 ApplicationSnapshotConfiguration:
1098 news.snapshotsEnabled !== undefined
1099 ? { SnapshotsEnabled: news.snapshotsEnabled }
1100 : undefined,
1101 VpcConfigurations: news.vpc
1102 ? [
1103 {
1104 SubnetIds: news.vpc.subnetIds,
1105 SecurityGroupIds: news.vpc.securityGroupIds,
1106 },
1107 ]
1108 : undefined,
1109 },
1110 Tags: createTagsList(desiredTags),
1111 };
1112 const create = (request: analytics.CreateApplicationRequest) =>
1113 retryWhileInUse(
1114 retryThroughRolePropagation(
1115 analytics
1116 .createApplication(request)
1117 .pipe(
1118 Effect.catchTag(
1119 "ResourceInUseException",
1120 () => Effect.void,
1121 ),
1122 ),
1123 ),
1124 );
1125 yield* create(createRequest).pipe(
1126 // The lingering tag registration on a recently-deleted name can
1127 // outlive the bounded retry window. AWS's own guidance in the
1128 // error text: create without tags, then TagResource — the tag
1129 // sync below applies the desired tags right after.
1130 Effect.catchTag("ConcurrentModificationException", () =>
1131 create({ ...createRequest, Tags: undefined }),
1132 ),
1133 );
1134 yield* session.note(`Creating application ${applicationName}...`);
1135 }
1136
1137 // Both a fresh create and a crashed prior run land here — wait for
1138 // a stable status before syncing (bounded).
1139 yield* waitForApplicationStable(applicationName);
1140
1141 // 3. SYNC — read OBSERVED state fresh on every attempt (the
1142 // version id is a compare-and-set token), compute the delta from
1143 // `news`, and apply a single UpdateApplication call. Retried
1144 // through ConcurrentModification/ResourceInUse with a re-read.
1145 const syncConfiguration = Effect.gen(function* () {
1146 const detail = yield* describeApplicationDetail(applicationName);
1147 if (!detail) return;
1148 const config = detail.ApplicationConfigurationDescription;
1149
1150 const configurationUpdate: analytics.ApplicationConfigurationUpdate =
1151 {};
1152
1153 const observedCode =
1154 config?.ApplicationCodeConfigurationDescription
1155 ?.CodeContentDescription?.S3ApplicationCodeLocationDescription;
1156 if (
1157 observedCode?.BucketARN !== news.code.bucketArn ||
1158 observedCode?.FileKey !== news.code.fileKey ||
1159 (news.code.objectVersion !== undefined &&
1160 observedCode?.ObjectVersion !== news.code.objectVersion)
1161 ) {
1162 configurationUpdate.ApplicationCodeConfigurationUpdate = {
1163 CodeContentTypeUpdate: "ZIPFILE",
1164 CodeContentUpdate: {
1165 S3ContentLocationUpdate: {
1166 BucketARNUpdate: news.code.bucketArn,
1167 FileKeyUpdate: news.code.fileKey,
1168 ObjectVersionUpdate: news.code.objectVersion,
1169 },
1170 },
1171 };
1172 }
1173
1174 if (
1175 news.environmentProperties !== undefined &&
1176 canonicalPropertyGroups(
1177 toWirePropertyGroups(news.environmentProperties),
1178 ) !==
1179 canonicalPropertyGroups(
1180 config?.EnvironmentPropertyDescriptions
1181 ?.PropertyGroupDescriptions,
1182 )
1183 ) {
1184 configurationUpdate.EnvironmentPropertyUpdates = {
1185 PropertyGroups: toWirePropertyGroups(
1186 news.environmentProperties,
1187 ),
1188 };
1189 }
1190
1191 if (news.flinkConfiguration) {
1192 const flinkUpdate = toFlinkConfigurationUpdate(
1193 news.flinkConfiguration,
1194 config?.FlinkApplicationConfigurationDescription,
1195 );
1196 if (flinkUpdate) {
1197 configurationUpdate.FlinkApplicationConfigurationUpdate =
1198 flinkUpdate;
1199 }
1200 }
1201
1202 if (
1203 news.snapshotsEnabled !== undefined &&
1204 news.snapshotsEnabled !==
1205 config?.ApplicationSnapshotConfigurationDescription
1206 ?.SnapshotsEnabled
1207 ) {
1208 configurationUpdate.ApplicationSnapshotConfigurationUpdate = {
1209 SnapshotsEnabledUpdate: news.snapshotsEnabled,
1210 };
1211 }
1212
1213 const observedVpc = config?.VpcConfigurationDescriptions?.[0];
1214 if (
1215 news.vpc &&
1216 observedVpc &&
1217 (!sameStringSet(news.vpc.subnetIds, observedVpc.SubnetIds) ||
1218 !sameStringSet(
1219 news.vpc.securityGroupIds,
1220 observedVpc.SecurityGroupIds,
1221 ))
1222 ) {
1223 configurationUpdate.VpcConfigurationUpdates = [
1224 {
1225 VpcConfigurationId: observedVpc.VpcConfigurationId,
1226 SubnetIdUpdates: news.vpc.subnetIds,
1227 SecurityGroupIdUpdates: news.vpc.securityGroupIds,
1228 },
1229 ];
1230 }
1231
1232 const update: analytics.UpdateApplicationRequest = {
1233 ApplicationName: applicationName,
1234 CurrentApplicationVersionId: detail.ApplicationVersionId,
1235 };
1236 let dirty = false;
1237 if (Object.keys(configurationUpdate).length > 0) {
1238 update.ApplicationConfigurationUpdate = configurationUpdate;
1239 dirty = true;
1240 }
1241 if (news.runtimeEnvironment !== detail.RuntimeEnvironment) {
1242 update.RuntimeEnvironmentUpdate = news.runtimeEnvironment;
1243 dirty = true;
1244 }
1245 if (serviceExecutionRole !== detail.ServiceExecutionRole) {
1246 update.ServiceExecutionRoleUpdate = serviceExecutionRole;
1247 dirty = true;
1248 }
1249 if (!dirty) return;
1250
1251 yield* retryThroughRolePropagation(
1252 analytics.updateApplication(update),
1253 );
1254 yield* session.note(
1255 `Updated application configuration for ${applicationName}`,
1256 );
1257 });
1258 yield* retryWhileInUse(syncConfiguration);
1259 yield* waitForApplicationStable(applicationName);
1260
1261 // 3b. SYNC maintenance window — separate API from
1262 // UpdateApplication; only applied when the caller pins a window
1263 // (otherwise the service-assigned window is left as-is).
1264 if (news.maintenanceWindowStartTime !== undefined) {
1265 const syncMaintenanceWindow = Effect.gen(function* () {
1266 const detail = yield* describeApplicationDetail(applicationName);
1267 if (!detail) return;
1268 const observedStart =
1269 detail.ApplicationMaintenanceConfigurationDescription
1270 ?.ApplicationMaintenanceWindowStartTime;
1271 if (observedStart === news.maintenanceWindowStartTime) return;
1272 yield* analytics.updateApplicationMaintenanceConfiguration({
1273 ApplicationName: applicationName,
1274 ApplicationMaintenanceConfigurationUpdate: {
1275 ApplicationMaintenanceWindowStartTimeUpdate:
1276 news.maintenanceWindowStartTime!,
1277 },
1278 });
1279 yield* session.note(
1280 `Updated maintenance window for ${applicationName}`,
1281 );
1282 });
1283 yield* retryWhileInUse(syncMaintenanceWindow);
1284 }
1285
1286 // 3c. SYNC VPC attach/detach — separate APIs from
1287 // UpdateApplication; each re-reads the fresh version id.
1288 const syncVpcAttachment = Effect.gen(function* () {
1289 const detail = yield* describeApplicationDetail(applicationName);
1290 if (!detail) return;
1291 const observedVpcs =
1292 detail.ApplicationConfigurationDescription
1293 ?.VpcConfigurationDescriptions ?? [];
1294 if (news.vpc && observedVpcs.length === 0) {
1295 yield* analytics.addApplicationVpcConfiguration({
1296 ApplicationName: applicationName,
1297 CurrentApplicationVersionId: detail.ApplicationVersionId,
1298 VpcConfiguration: {
1299 SubnetIds: news.vpc.subnetIds,
1300 SecurityGroupIds: news.vpc.securityGroupIds,
1301 },
1302 });
1303 yield* session.note(
1304 `Attached VPC configuration to ${applicationName}`,
1305 );
1306 } else if (!news.vpc && observedVpcs[0]?.VpcConfigurationId) {
1307 yield* analytics.deleteApplicationVpcConfiguration({
1308 ApplicationName: applicationName,
1309 CurrentApplicationVersionId: detail.ApplicationVersionId,
1310 VpcConfigurationId: observedVpcs[0].VpcConfigurationId,
1311 });
1312 yield* session.note(
1313 `Detached VPC configuration from ${applicationName}`,
1314 );
1315 }
1316 });
1317 yield* retryWhileInUse(syncVpcAttachment);
1318 const stable = yield* waitForApplicationStable(applicationName);
1319
1320 // 3d. SYNC run state — start when `start: true`, force-stop a
1321 // running application when `start` is absent/false.
1322 if (news.start === true && stable.ApplicationStatus !== "RUNNING") {
1323 yield* retryWhileInUse(
1324 analytics.startApplication({
1325 ApplicationName: applicationName,
1326 }),
1327 );
1328 yield* session.note(`Starting application ${applicationName}...`);
1329 yield* waitForApplicationRunning(applicationName);
1330 } else if (
1331 news.start !== true &&
1332 (stable.ApplicationStatus === "RUNNING" ||
1333 stable.ApplicationStatus === "AUTOSCALING")
1334 ) {
1335 yield* retryWhileInUse(
1336 analytics.stopApplication({
1337 ApplicationName: applicationName,
1338 Force: true,
1339 }),
1340 );
1341 yield* session.note(`Stopping application ${applicationName}...`);
1342 yield* waitForApplicationStopped(applicationName);
1343 }
1344
1345 // 3e. SYNC TAGS — diff against OBSERVED cloud tags (adoption may
1346 // bring an application with foreign tags), never olds/output.
1347 const applicationArn = stable.ApplicationARN;
1348 const observedTagsResponse = yield* analytics
1349 .listTagsForResource({ ResourceARN: applicationArn })
1350 .pipe(
1351 Effect.catchTag("ResourceNotFoundException", () =>
1352 Effect.succeed(undefined),
1353 ),
1354 );
1355 const observedTags = toTagRecord(observedTagsResponse?.Tags);
1356 const { removed, upsert } = diffTags(observedTags, desiredTags);
1357 if (removed.length > 0) {
1358 yield* retryWhileInUse(
1359 analytics.untagResource({
1360 ResourceARN: applicationArn,
1361 TagKeys: removed,
1362 }),
1363 );
1364 }
1365 if (upsert.length > 0) {
1366 yield* retryWhileInUse(
1367 analytics.tagResource({
1368 ResourceARN: applicationArn,
1369 Tags: upsert,
1370 }),
1371 );
1372 }
1373
1374 yield* session.note(applicationArn);
1375
1376 // Clean up a previously-synthesized role once the application has
1377 // been switched to a user-supplied `serviceExecutionRole` (the
1378 // sync above already re-pointed the app at the new role).
1379 if (!roleName && output?.roleName) {
1380 yield* deleteRoleIfExists(output.roleName);
1381 }
1382
1383 // 4. RETURN fresh attributes reflecting post-sync cloud state.
1384 const final = yield* readApplication({ applicationName, roleName });
1385 if (!final) {
1386 return yield* Effect.fail(
1387 new ApplicationValidationError({
1388 message: `failed to read reconciled application ${applicationName}`,
1389 }),
1390 );
1391 }
1392 return final;
1393 }),
1394
1395 delete: Effect.fn(function* ({ output }) {
1396 // DeleteApplication requires the application's CreateTimestamp —
1397 // observe it fresh rather than trusting persisted state. Missing
1398 // application means the delete already happened.
1399 const detail = yield* describeApplicationDetail(
1400 output.applicationName,
1401 );
1402 if (detail?.CreateTimestamp) {
1403 yield* retryWhileInUse(
1404 analytics
1405 .deleteApplication({
1406 ApplicationName: output.applicationName,
1407 CreateTimestamp: detail.CreateTimestamp,
1408 })
1409 .pipe(
1410 Effect.catchTag(
1411 "ResourceNotFoundException",
1412 () => Effect.void,
1413 ),
1414 ),
1415 );
1416 yield* waitForApplicationDeleted(output.applicationName);
1417 }
1418
1419 // Remove the synthesized IAM role (idempotent — it may already be
1420 // gone from a previous partial delete).
1421 if (output.roleName) {
1422 yield* deleteRoleIfExists(output.roleName);
1423 }
1424 }),
1425 });
1426 }),
1427 );