1import * as AWS from "@/AWS";
2import { Application } from "@/AWS/KinesisAnalyticsV2";
3import * as Test from "@/Test/Alchemy";
4import * as iam from "@distilled.cloud/aws/iam";
5import * as analytics from "@distilled.cloud/aws/kinesis-analytics-v2";
6import { describe, expect } from "alchemy-test";
7import * as Data from "effect/Data";
8import * as Effect from "effect/Effect";
9import * as Result from "effect/Result";
10import * as Schedule from "effect/Schedule";
11import {
12 codeKey,
13 deleteCodeBucketIdempotent,
14 provisionCodeBucket,
15} from "./code-bucket.ts";
16
17const { test } = Test.make({ providers: AWS.providers() });
18
19// Deterministic out-of-band code buckets (see code-bucket.ts) — the Flink
20// code object must exist before the application resource, and an in-stack
21// bucket orphans on a crashed run because its generated name is lost with
22// the state.
23const appCodeBucket = "alchemy-test-kav2-app-code";
24const renameCodeBucket = "alchemy-test-kav2-rename-code";
25
26class ApplicationStillExists extends Data.TaggedError(
27 "ApplicationStillExists",
28) {}
29
30const assertApplicationDeleted = Effect.fn(function* (applicationName: string) {
31 yield* analytics
32 .describeApplication({ ApplicationName: applicationName })
33 .pipe(
34 Effect.flatMap(() => Effect.fail(new ApplicationStillExists())),
35 Effect.retry({
36 while: (e: { _tag: string }) => e._tag === "ApplicationStillExists",
37 schedule: Schedule.max([
38 Schedule.fixed("3 seconds"),
39 Schedule.recurs(40),
40 ]),
41 }),
42 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
43 );
44});
45
46describe.skipIf(!!process.env.FAST)(
47 "AWS.KinesisAnalyticsV2.Application",
48 {
49 tags: [
50 "provider:aws",
51 "provider:aws:kinesisanalyticsv2",
52 "provider:aws:s3",
53 "live",
54 ],
55 },
56 () => {
57 test.provider(
58 "create Flink application in READY, update configuration in place, destroy",
59 (stack) =>
60 Effect.gen(function* () {
61 yield* stack.destroy();
62
63 // Stage the code object BEFORE the application exists — the service
64 // reads (and hashes) the object at create/update time.
65 const codeBucketArn = yield* provisionCodeBucket(appCodeBucket);
66
67 const deployed = yield* stack.deploy(
68 Effect.gen(function* () {
69 const app = yield* Application("FlinkApp", {
70 runtimeEnvironment: "FLINK-1_20",
71 code: { bucketArn: codeBucketArn, fileKey: codeKey },
72 environmentProperties: [
73 {
74 propertyGroupId: "AppProperties",
75 propertyMap: { mode: "test" },
76 },
77 ],
78 tags: { Environment: "test" },
79 });
80 return { app };
81 }),
82 );
83
84 expect(deployed.app.applicationName).toBeDefined();
85 expect(deployed.app.applicationStatus).toEqual("READY");
86 expect(deployed.app.applicationVersionId).toEqual(1);
87 expect(deployed.app.applicationArn).toContain(":kinesisanalytics:");
88 expect(deployed.app.runtimeEnvironment).toEqual("FLINK-1_20");
89 expect(deployed.app.roleName).toBeDefined();
90 expect(deployed.app.serviceExecutionRole).toContain(":role/");
91 expect(deployed.app.codeBucketArn).toEqual(codeBucketArn);
92 expect(deployed.app.codeFileKey).toEqual(codeKey);
93 expect(deployed.app.vpcConfigurationId).toBeUndefined();
94
95 // Out-of-band verification via distilled.
96 const described = yield* analytics.describeApplication({
97 ApplicationName: deployed.app.applicationName,
98 });
99 expect(described.ApplicationDetail.ApplicationStatus).toEqual(
100 "READY",
101 );
102 expect(described.ApplicationDetail.RuntimeEnvironment).toEqual(
103 "FLINK-1_20",
104 );
105 const observedGroups =
106 described.ApplicationDetail.ApplicationConfigurationDescription
107 ?.EnvironmentPropertyDescriptions?.PropertyGroupDescriptions;
108 expect(observedGroups?.[0]?.PropertyGroupId).toEqual("AppProperties");
109 expect(observedGroups?.[0]?.PropertyMap?.mode).toEqual("test");
110
111 // Ownership + user tags.
112 const tags = yield* analytics.listTagsForResource({
113 ResourceARN: deployed.app.applicationArn,
114 });
115 const tagKeys = (tags.Tags ?? []).map((tag) => tag.Key);
116 expect(tagKeys).toContain("alchemy::stack");
117 expect(tagKeys).toContain("alchemy::stage");
118 expect(tagKeys).toContain("alchemy::id");
119 expect(tags.Tags).toContainEqual({
120 Key: "Environment",
121 Value: "test",
122 });
123
124 // Typed error probe: describing a nonexistent snapshot of a live
125 // application surfaces the typed ResourceNotFoundException.
126 const snapshotError = yield* analytics
127 .describeApplicationSnapshot({
128 ApplicationName: deployed.app.applicationName,
129 SnapshotName: "does-not-exist",
130 })
131 .pipe(Effect.flip);
132 expect(snapshotError._tag).toEqual("ResourceNotFoundException");
133
134 // Update runtime properties + tags in place (UpdateApplication).
135 const updated = yield* stack.deploy(
136 Effect.gen(function* () {
137 const app = yield* Application("FlinkApp", {
138 runtimeEnvironment: "FLINK-1_20",
139 code: { bucketArn: codeBucketArn, fileKey: codeKey },
140 environmentProperties: [
141 {
142 propertyGroupId: "AppProperties",
143 propertyMap: { mode: "production", region: "us-west-2" },
144 },
145 ],
146 flinkConfiguration: {
147 parallelismConfiguration: {
148 configurationType: "CUSTOM",
149 parallelism: 1,
150 parallelismPerKPU: 1,
151 autoScalingEnabled: false,
152 },
153 },
154 maintenanceWindowStartTime: "02:00",
155 tags: { Environment: "production", Team: "platform" },
156 });
157 return { app };
158 }),
159 );
160
161 // Same physical application — updated in place, not replaced.
162 expect(updated.app.applicationName).toEqual(
163 deployed.app.applicationName,
164 );
165 expect(updated.app.applicationVersionId).toBeGreaterThan(1);
166 expect(updated.app.applicationStatus).toEqual("READY");
167 // Maintenance window applied via UpdateApplicationMaintenanceConfiguration.
168 expect(updated.app.maintenanceWindowStartTime).toEqual("02:00");
169 expect(updated.app.maintenanceWindowEndTime).toBeDefined();
170
171 const updatedDescribe = yield* analytics.describeApplication({
172 ApplicationName: deployed.app.applicationName,
173 });
174 const updatedConfig =
175 updatedDescribe.ApplicationDetail
176 .ApplicationConfigurationDescription;
177 expect(
178 updatedConfig?.EnvironmentPropertyDescriptions
179 ?.PropertyGroupDescriptions?.[0]?.PropertyMap?.mode,
180 ).toEqual("production");
181 expect(
182 updatedConfig?.FlinkApplicationConfigurationDescription
183 ?.ParallelismConfigurationDescription?.ConfigurationType,
184 ).toEqual("CUSTOM");
185
186 const updatedTags = yield* analytics.listTagsForResource({
187 ResourceARN: deployed.app.applicationArn,
188 });
189 expect(updatedTags.Tags).toContainEqual({
190 Key: "Environment",
191 Value: "production",
192 });
193 expect(updatedTags.Tags).toContainEqual({
194 Key: "Team",
195 Value: "platform",
196 });
197
198 // A no-change redeploy must not bump the application version
199 // (observed-vs-desired diffing skips the UpdateApplication call).
200 const steady = yield* stack.deploy(
201 Effect.gen(function* () {
202 const app = yield* Application("FlinkApp", {
203 runtimeEnvironment: "FLINK-1_20",
204 code: { bucketArn: codeBucketArn, fileKey: codeKey },
205 environmentProperties: [
206 {
207 propertyGroupId: "AppProperties",
208 propertyMap: { mode: "production", region: "us-west-2" },
209 },
210 ],
211 flinkConfiguration: {
212 parallelismConfiguration: {
213 configurationType: "CUSTOM",
214 parallelism: 1,
215 parallelismPerKPU: 1,
216 autoScalingEnabled: false,
217 },
218 },
219 maintenanceWindowStartTime: "02:00",
220 tags: { Environment: "production", Team: "platform" },
221 });
222 return { app };
223 }),
224 );
225 expect(steady.app.applicationVersionId).toEqual(
226 updated.app.applicationVersionId,
227 );
228
229 yield* stack.destroy();
230
231 yield* assertApplicationDeleted(deployed.app.applicationName);
232
233 // The synthesized IAM role is cleaned up with the application.
234 const roleResult = yield* iam
235 .getRole({ RoleName: deployed.app.roleName! })
236 .pipe(Effect.result);
237 expect(Result.isFailure(roleResult)).toBe(true);
238 }).pipe(Effect.ensuring(deleteCodeBucketIdempotent(appCodeBucket))),
239 { tags: ["provider:aws:iam"], timeout: 300_000 },
240 );
241
242 test.provider(
243 "replaces the application when the name changes",
244 (stack) =>
245 Effect.gen(function* () {
246 yield* stack.destroy();
247
248 const codeBucketArn = yield* provisionCodeBucket(renameCodeBucket);
249
250 const initial = yield* stack.deploy(
251 Effect.gen(function* () {
252 const app = yield* Application("RenamedApp", {
253 runtimeEnvironment: "FLINK-1_20",
254 code: { bucketArn: codeBucketArn, fileKey: codeKey },
255 });
256 return { app };
257 }),
258 );
259
260 const replaced = yield* stack.deploy(
261 Effect.gen(function* () {
262 const app = yield* Application("RenamedApp", {
263 applicationName: "alchemy-test-kav2-renamed",
264 runtimeEnvironment: "FLINK-1_20",
265 code: { bucketArn: codeBucketArn, fileKey: codeKey },
266 });
267 return { app };
268 }),
269 );
270
271 // Name change is a replacement — new physical application.
272 expect(replaced.app.applicationName).toEqual(
273 "alchemy-test-kav2-renamed",
274 );
275 expect(replaced.app.applicationName).not.toEqual(
276 initial.app.applicationName,
277 );
278
279 // The replaced (old) application is deleted by the engine.
280 yield* assertApplicationDeleted(initial.app.applicationName);
281
282 yield* stack.destroy();
283
284 yield* assertApplicationDeleted(replaced.app.applicationName);
285 }).pipe(Effect.ensuring(deleteCodeBucketIdempotent(renameCodeBucket))),
286 { timeout: 300_000 },
287 );
288 },
289);