1import * as AWS from "@/AWS";
2import * as Core from "@/Test/Core";
3import * as Test from "@/Test/Alchemy";
4import * as analytics from "@distilled.cloud/aws/kinesis-analytics-v2";
5import { describe, expect } from "alchemy-test";
6import * as Data from "effect/Data";
7import * as Effect from "effect/Effect";
8import * as Schedule from "effect/Schedule";
9import * as HttpClient from "effect/http/HttpClient";
10import * as HttpClientRequest from "effect/http/HttpClientRequest";
11import {
12 deleteCodeBucketIdempotent,
13 provisionCodeBucket,
14} from "./code-bucket.ts";
15import KinesisAnalyticsV2TestFunctionLive, {
16 FIXTURE_CODE_BUCKET,
17 KinesisAnalyticsV2TestFunction,
18} from "./handler";
19
20const testOptions = { providers: AWS.providers() };
21const { test, beforeAll, afterAll } = Test.make(testOptions);
22const sharedStack = Core.scratchStack(
23 testOptions,
24 "KinesisAnalyticsV2Bindings",
25);
26
27// Lambda function URL cold-start (DNS, IAM propagation, init) can take well
28// over 60s on a fresh deploy.
29const readinessPolicy = Schedule.max([
30 Schedule.fixed("2 seconds"),
31 Schedule.recurs(75),
32]);
33
34let baseUrl: string;
35// Physical name of the Application the fixture creates — captured in
36// beforeAll so afterAll can assert it is GONE out-of-band after destroy.
37let applicationName: string | undefined;
38
39class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
40 readonly status: number;
41 readonly body: string;
42}> {}
43
44// The shared Lambda fixture occasionally answers a transient 5xx under load
45// (cold re-init, IAM propagation on the freshly attached policy). Retry only
46// 5xx; a genuine 4xx/assertion failure surfaces immediately.
47const send = (request: HttpClientRequest.HttpClientRequest) =>
48 HttpClient.execute(request).pipe(
49 Effect.flatMap((response) =>
50 response.status >= 500
51 ? response.text.pipe(
52 Effect.flatMap((body) =>
53 Effect.fail(
54 new TransientUpstream({ status: response.status, body }),
55 ),
56 ),
57 )
58 : Effect.succeed(response),
59 ),
60 Effect.retry({
61 while: (e) => e._tag === "TransientUpstream",
62 schedule: Schedule.max([
63 Schedule.exponential("500 millis"),
64 Schedule.recurs(6),
65 ]),
66 }),
67 );
68
69const getJson = (path: string) =>
70 send(HttpClientRequest.get(`${baseUrl}${path}`)).pipe(
71 Effect.flatMap((r) => r.json),
72 );
73
74const postJson = (path: string) =>
75 send(HttpClientRequest.post(`${baseUrl}${path}`)).pipe(
76 Effect.flatMap((r) => r.json),
77 );
78
79describe.sequential(
80 "KinesisAnalyticsV2 Bindings",
81 {
82 tags: [
83 "provider:aws",
84 "provider:aws:kinesisanalyticsv2",
85 "provider:aws:lambda",
86 "provider:aws:s3",
87 "live",
88 ],
89 },
90 () => {
91 beforeAll(
92 Effect.gen(function* () {
93 yield* Effect.logInfo(
94 "KinesisAnalyticsV2 test setup: destroying previous resources",
95 );
96 yield* sharedStack.destroy();
97
98 // Stage the code object BEFORE the Application resource deploys — the
99 // service reads (and hashes) the object at create time. Direct
100 // distilled calls need the AWS provider context (credentials, region)
101 // that `test.provider` bodies get implicitly.
102 yield* Core.withProviders(
103 provisionCodeBucket(FIXTURE_CODE_BUCKET),
104 testOptions,
105 "KinesisAnalyticsV2Bindings",
106 );
107
108 yield* Effect.logInfo(
109 "KinesisAnalyticsV2 test setup: deploying fixture",
110 );
111 const { functionUrl } = yield* sharedStack.deploy(
112 Effect.gen(function* () {
113 return yield* KinesisAnalyticsV2TestFunction;
114 }).pipe(Effect.provide(KinesisAnalyticsV2TestFunctionLive)),
115 );
116
117 expect(functionUrl).toBeTruthy();
118 baseUrl = functionUrl!.replace(/\/+$/, "");
119
120 const readinessUrl = `${baseUrl}/bindings`;
121 yield* Effect.logInfo(
122 `KinesisAnalyticsV2 test setup: probing readiness at ${readinessUrl}`,
123 );
124 yield* HttpClient.get(readinessUrl).pipe(
125 Effect.flatMap((response) =>
126 response.status === 200
127 ? Effect.succeed(response)
128 : Effect.fail(
129 new Error(`Function not ready: ${response.status}`),
130 ),
131 ),
132 Effect.tapError((error) =>
133 Effect.logWarning(
134 `KinesisAnalyticsV2 test setup: fixture not ready yet (${String(error)})`,
135 ),
136 ),
137 Effect.retry({ schedule: readinessPolicy }),
138 );
139
140 // `/bindings` only proves the function URL is up. The first real
141 // SDK call (`DescribeApplication`) can still 200 with `{ errorTag }`
142 // while IAM on the freshly attached policy is propagating — poll
143 // until the application name is present.
144 const described = yield* getJson("/describe").pipe(
145 Effect.flatMap((body) => {
146 const name = (body as { name?: string }).name;
147 return typeof name === "string" && name.length > 0
148 ? Effect.succeed(body as { name: string })
149 : Effect.fail(
150 new Error(
151 `KinesisAnalyticsV2 describe not ready yet (${JSON.stringify(body)})`,
152 ),
153 );
154 }),
155 Effect.retry({ schedule: readinessPolicy }),
156 );
157 applicationName = described.name;
158 }),
159 { timeout: 300_000 },
160 );
161
162 afterAll(
163 Effect.gen(function* () {
164 yield* sharedStack.destroy();
165
166 // Assert the fixture Application is GONE out-of-band — the trailing
167 // destroy must leave zero orphaned cloud resources.
168 if (applicationName !== undefined) {
169 const gone = yield* Core.withProviders(
170 analytics
171 .describeApplication({ ApplicationName: applicationName })
172 .pipe(Effect.flip),
173 testOptions,
174 "KinesisAnalyticsV2Bindings",
175 );
176 expect(gone._tag).toEqual("ResourceNotFoundException");
177 }
178 }).pipe(
179 Effect.ensuring(
180 Core.withProviders(
181 deleteCodeBucketIdempotent(FIXTURE_CODE_BUCKET),
182 testOptions,
183 "KinesisAnalyticsV2Bindings",
184 ),
185 ),
186 ),
187 { timeout: 180_000 },
188 );
189
190 describe("binding registration", () => {
191 test.provider("all 14 capabilities initialize in the runtime", (_stack) =>
192 Effect.gen(function* () {
193 const response = yield* getJson("/bindings");
194 expect((response as any).bound).toHaveLength(14);
195 }),
196 );
197 });
198
199 describe("DescribeApplication", () => {
200 test.provider("reads the READY application detail", (_stack) =>
201 Effect.gen(function* () {
202 const response = (yield* getJson("/describe")) as {
203 status: string;
204 runtime: string;
205 };
206 expect(response.status).toBe("READY");
207 expect(response.runtime).toBe("FLINK-1_20");
208 }),
209 );
210 });
211
212 describe("ListApplications", () => {
213 test.provider("account list contains the bound application", (_stack) =>
214 Effect.gen(function* () {
215 const response = (yield* getJson("/applications")) as {
216 count?: number;
217 containsSelf?: boolean;
218 errorTag?: string;
219 };
220 expect(response.errorTag).toBeUndefined();
221 expect(response.containsSelf).toBe(true);
222 }),
223 );
224 });
225
226 describe("ListApplicationVersions", () => {
227 test.provider("lists at least the initial version", (_stack) =>
228 Effect.gen(function* () {
229 const response = (yield* getJson("/versions")) as { count: number };
230 expect(response.count).toBeGreaterThanOrEqual(1);
231 }),
232 );
233 });
234
235 describe("DescribeApplicationVersion", () => {
236 test.provider("reads version 1", (_stack) =>
237 Effect.gen(function* () {
238 const response = (yield* getJson("/version")) as {
239 versionId: number;
240 };
241 expect(response.versionId).toBe(1);
242 }),
243 );
244 });
245
246 describe("ListApplicationOperations", () => {
247 test.provider("lists the (possibly empty) operation history", (_stack) =>
248 Effect.gen(function* () {
249 const response = (yield* getJson("/operations")) as { count: number };
250 expect(response.count).toBeGreaterThanOrEqual(0);
251 }),
252 );
253 });
254
255 describe("DescribeApplicationOperation", () => {
256 test.provider(
257 "surfaces a typed tag for a nonexistent operation (proving the grant)",
258 (_stack) =>
259 Effect.gen(function* () {
260 const response = (yield* getJson("/operation")) as {
261 errorTag: string;
262 };
263 expect([
264 "ResourceNotFoundException",
265 "InvalidArgumentException",
266 "UnsupportedOperationException",
267 ]).toContain(response.errorTag);
268 }),
269 );
270 });
271
272 describe("ListApplicationSnapshots", () => {
273 test.provider("lists zero snapshots on the fresh application", (_stack) =>
274 Effect.gen(function* () {
275 const response = (yield* getJson("/snapshots")) as { count: number };
276 expect(response.count).toBe(0);
277 }),
278 );
279 });
280
281 describe("DescribeApplicationSnapshot", () => {
282 test.provider(
283 "surfaces the typed not-found tag for a nonexistent snapshot",
284 (_stack) =>
285 Effect.gen(function* () {
286 const response = (yield* getJson("/snapshot")) as {
287 errorTag: string;
288 };
289 expect(response.errorTag).toBe("ResourceNotFoundException");
290 }),
291 );
292 });
293
294 describe("CreateApplicationSnapshot", () => {
295 test.provider(
296 "reaches the typed not-running rejection (proving the grant)",
297 (_stack) =>
298 Effect.gen(function* () {
299 const response = (yield* postJson("/snapshot/create")) as {
300 errorTag: string;
301 };
302 expect([
303 "InvalidApplicationConfigurationException",
304 "InvalidRequestException",
305 "ResourceInUseException",
306 "UnsupportedOperationException",
307 ]).toContain(response.errorTag);
308 }),
309 );
310 });
311
312 describe("DeleteApplicationSnapshot", () => {
313 test.provider(
314 "surfaces a typed tag for a nonexistent snapshot (proving the grant)",
315 (_stack) =>
316 Effect.gen(function* () {
317 const response = (yield* postJson("/snapshot/delete")) as {
318 errorTag: string;
319 };
320 expect([
321 "ResourceNotFoundException",
322 "InvalidArgumentException",
323 "InvalidRequestException",
324 ]).toContain(response.errorTag);
325 }),
326 );
327 });
328
329 describe("CreateApplicationPresignedUrl", () => {
330 test.provider(
331 "mints a URL or reaches a typed rejection (proving the grant)",
332 (_stack) =>
333 Effect.gen(function* () {
334 const response = (yield* getJson("/presigned-url")) as {
335 hasUrl?: boolean;
336 errorTag?: string;
337 };
338 if (response.errorTag !== undefined) {
339 expect([
340 "ResourceNotFoundException",
341 "ResourceInUseException",
342 "InvalidArgumentException",
343 ]).toContain(response.errorTag);
344 } else {
345 expect(response.hasUrl).toBe(true);
346 }
347 }),
348 );
349 });
350
351 describe("RollbackApplication", () => {
352 test.provider(
353 "reaches the typed nothing-to-roll-back rejection (proving the grant)",
354 (_stack) =>
355 Effect.gen(function* () {
356 const response = (yield* postJson("/rollback")) as {
357 errorTag: string;
358 };
359 expect([
360 "InvalidRequestException",
361 "InvalidArgumentException",
362 "ResourceInUseException",
363 "ConcurrentModificationException",
364 "UnsupportedOperationException",
365 ]).toContain(response.errorTag);
366 }),
367 );
368 });
369
370 describe("StopApplication", () => {
371 test.provider(
372 "reaches the typed not-running rejection (proving the grant)",
373 (_stack) =>
374 Effect.gen(function* () {
375 const response = (yield* postJson("/stop")) as {
376 errorTag?: string;
377 ok?: boolean;
378 };
379 if (response.errorTag !== undefined) {
380 expect([
381 "InvalidApplicationConfigurationException",
382 "InvalidRequestException",
383 "ResourceInUseException",
384 "ConcurrentModificationException",
385 ]).toContain(response.errorTag);
386 } else {
387 expect(response.ok).toBe(true);
388 }
389 }),
390 );
391 });
392
393 // Last on purpose: if the service unexpectedly accepts the start, the
394 // route force-stops immediately and no other test observes the transition.
395 describe("StartApplication", () => {
396 test.provider(
397 "rejects a restore from a nonexistent snapshot (proving the grant)",
398 (_stack) =>
399 Effect.gen(function* () {
400 const response = (yield* postJson("/start")) as {
401 errorTag?: string;
402 started?: boolean;
403 };
404 if (response.errorTag !== undefined) {
405 expect([
406 "ResourceNotFoundException",
407 "InvalidArgumentException",
408 "InvalidApplicationConfigurationException",
409 "InvalidRequestException",
410 "ResourceInUseException",
411 ]).toContain(response.errorTag);
412 } else {
413 expect(response.started).toBe(true);
414 }
415 }),
416 { timeout: 120_000 },
417 );
418 });
419 },
420);