1/**
2 * Out-of-band S3 code-bucket helpers for the KinesisAnalyticsV2 tests.
3 *
4 * Flink applications read their code zip from S3 out-of-band (the object
5 * must exist BEFORE the application resource is created), so the bucket is
6 * deliberately NOT an in-stack `AWS.S3.Bucket` resource: an in-stack bucket
7 * gets an engine-generated (per-instance-unique) physical name, and a
8 * crashed/failed run orphans it where the next run's `stack.destroy()`
9 * cannot find it. Instead each test uses a DETERMINISTIC bucket name,
10 * provisions it with a delete-if-exists pre-clean, and guarantees teardown
11 * via an idempotent `Effect.ensuring` finalizer (empty + delete, tolerating
12 * NoSuchBucket) — the same pattern as `test/AWS/SNS/Topic.test.ts` and
13 * `test/AWS/RDSData/reap.ts`.
14 */
15import { AWSEnvironment } from "@/AWS/Environment.ts";
16import type { BucketLocationConstraint } from "@distilled.cloud/aws/s3";
17import * as s3 from "@distilled.cloud/aws/s3";
18import * as Effect from "effect/Effect";
19import * as Schedule from "effect/Schedule";
20import { makeDummyFlinkCodeZip } from "./dummy-zip.ts";
21
22/** Key of the staged code object inside the code bucket. */
23export const codeKey = "code/app.zip";
24
25/**
26 * Empty the bucket (bounded pagination) and delete it. Idempotent —
27 * `NoSuchBucket` at any step is success. Unexpected errors are defects
28 * (`Effect.orDie`), so the error channel is `never` and this is a valid
29 * `Effect.ensuring` finalizer.
30 */
31export const deleteCodeBucketIdempotent = (bucketName: string) =>
32 Effect.gen(function* () {
33 // Bounded pagination — test buckets hold at most a handful of objects.
34 for (let page = 0; page < 10; page++) {
35 const listed = yield* s3.listObjectsV2({ Bucket: bucketName });
36 const objects = (listed.Contents ?? []).flatMap((object) =>
37 object.Key ? [{ Key: object.Key }] : [],
38 );
39 if (objects.length === 0) break;
40 yield* s3.deleteObjects({
41 Bucket: bucketName,
42 Delete: { Objects: objects, Quiet: true },
43 });
44 if (!listed.IsTruncated) break;
45 }
46 yield* s3.deleteBucket({ Bucket: bucketName });
47 }).pipe(
48 Effect.catchTag("NoSuchBucket", () => Effect.void),
49 Effect.orDie,
50 );
51
52/**
53 * Delete-if-exists pre-clean, create the bucket, stage the dummy Flink code
54 * zip at {@link codeKey}, and return the bucket ARN. Recreating a
55 * just-deleted bucket name can race S3's conflicting-operation window, so
56 * creation retries `OperationAborted` boundedly and tolerates
57 * `BucketAlreadyOwnedByYou`.
58 */
59export const provisionCodeBucket = Effect.fn(function* (bucketName: string) {
60 yield* deleteCodeBucketIdempotent(bucketName);
61 const { region } = yield* AWSEnvironment.current;
62 yield* s3
63 .createBucket({
64 Bucket: bucketName,
65 ...(region === "us-east-1"
66 ? {}
67 : {
68 CreateBucketConfiguration: {
69 LocationConstraint: region as BucketLocationConstraint,
70 },
71 }),
72 })
73 .pipe(
74 Effect.catchTag("BucketAlreadyOwnedByYou", () => Effect.void),
75 Effect.retry({
76 while: (e): boolean =>
77 e._tag === "OperationAborted" || e._tag === "ServiceUnavailable",
78 schedule: Schedule.fixed("5 seconds"),
79 times: 8,
80 }),
81 );
82 const zip = yield* Effect.sync(() => makeDummyFlinkCodeZip());
83 yield* s3.putObject({ Bucket: bucketName, Key: codeKey, Body: zip });
84 return `arn:aws:s3:::${bucketName}`;
85});