1import * as AWS from "@/AWS";
2import { AWSEnvironment } from "@/AWS/Environment";
3import * as Test from "@/Test/Alchemy";
4import * as EC2 from "@distilled.cloud/aws/ec2";
5import * as Kafka from "@distilled.cloud/aws/kafka";
6import { expect } from "alchemy-test";
7import * as Effect from "effect/Effect";
8import * as Schedule from "effect/Schedule";
9import { getDefaultVpc } from "../DefaultVpc.ts";
10
11const { test } = Test.make({ providers: AWS.providers() });
12
13// Ungated typed-error probe: proves the distilled kafka error union carries the
14// NotFoundException tag this provider's read/delete paths depend on. A
15// well-formed but nonexistent cluster ARN returns NotFoundException.
16test.provider(
17 "describeClusterV2 on a nonexistent cluster fails with NotFoundException",
18 () =>
19 Effect.gen(function* () {
20 const { accountId, region } = yield* AWSEnvironment.current;
21 const error = yield* Effect.flip(
22 Kafka.describeClusterV2({
23 ClusterArn: `arn:aws:kafka:${region}:${accountId}:cluster/alchemy-nonexistent-probe/00000000-0000-0000-0000-000000000000-1`,
24 }),
25 );
26 expect(error._tag).toBe("NotFoundException");
27 }),
28 { tags: ["provider:aws", "provider:aws:kafka", "live"] },
29);
30
31const nonexistentClusterArn = Effect.gen(function* () {
32 const { accountId, region } = yield* AWSEnvironment.current;
33 return `arn:aws:kafka:${region}:${accountId}:cluster/alchemy-nonexistent-probe/00000000-0000-0000-0000-000000000000-1`;
34});
35
36// Ungated typed-error probes for the operations behind the runtime bindings
37// (GetBootstrapBrokers + the topic control plane). Each proves the distilled
38// error union carries the tag a caller must handle for a missing cluster.
39test.provider(
40 "getBootstrapBrokers on a nonexistent cluster fails with a typed tag",
41 () =>
42 Effect.gen(function* () {
43 const ClusterArn = yield* nonexistentClusterArn;
44 const error = yield* Effect.flip(
45 Kafka.getBootstrapBrokers({ ClusterArn }),
46 );
47 expect(error._tag).toBe("NotFoundException");
48 }),
49 { tags: ["provider:aws", "provider:aws:kafka", "live"] },
50);
51
52test.provider(
53 "listTopics on a nonexistent cluster fails with a typed tag",
54 () =>
55 Effect.gen(function* () {
56 const ClusterArn = yield* nonexistentClusterArn;
57 const error = yield* Effect.flip(Kafka.listTopics({ ClusterArn }));
58 expect(error._tag).toBe("NotFoundException");
59 }),
60 { tags: ["provider:aws", "provider:aws:kafka", "live"] },
61);
62
63test.provider(
64 "describeTopic on a nonexistent cluster fails with a typed tag",
65 () =>
66 Effect.gen(function* () {
67 const ClusterArn = yield* nonexistentClusterArn;
68 const error = yield* Effect.flip(
69 Kafka.describeTopic({ ClusterArn, TopicName: "alchemy-probe" }),
70 );
71 expect(error._tag).toBe("NotFoundException");
72 }),
73 { tags: ["provider:aws", "provider:aws:kafka", "live"] },
74);
75
76test.provider(
77 "createTopic on a nonexistent cluster fails with a typed tag",
78 () =>
79 Effect.gen(function* () {
80 const ClusterArn = yield* nonexistentClusterArn;
81 const error = yield* Effect.flip(
82 Kafka.createTopic({
83 ClusterArn,
84 TopicName: "alchemy-probe",
85 PartitionCount: 1,
86 }),
87 );
88 // The topic control plane validates the request body before resolving
89 // the cluster, so a nonexistent cluster surfaces the typed 400.
90 expect(error._tag).toBe("BadRequestException");
91 }),
92 { tags: ["provider:aws", "provider:aws:kafka", "live"] },
93);
94
95// Resolve two default-for-AZ subnets from the account's default VPC — MSK
96// requires at least two subnets in distinct Availability Zones.
97const resolveSubnets = Effect.gen(function* () {
98 const vpc = yield* getDefaultVpc;
99 const subnets = yield* EC2.describeSubnets({
100 Filters: [
101 { Name: "vpc-id", Values: [vpc.vpcId] },
102 { Name: "default-for-az", Values: ["true"] },
103 ],
104 });
105 const byAz = new Map<string, string>();
106 for (const s of subnets.Subnets ?? []) {
107 if (s.AvailabilityZone && s.SubnetId && !byAz.has(s.AvailabilityZone)) {
108 byAz.set(s.AvailabilityZone, s.SubnetId);
109 }
110 }
111 return [...byAz.values()].slice(0, 3);
112});
113
114// MSK Serverless clusters take ~5-10 minutes to create and are metered while
115// they exist. The full lifecycle is gated behind AWS_TEST_SLOW=1 and always
116// destroys what it created.
117test.provider.skipIf(!process.env.AWS_TEST_SLOW)(
118 "create serverless cluster, verify IAM-auth brokers, destroy, verify gone",
119 (stack) =>
120 Effect.gen(function* () {
121 yield* stack.destroy();
122
123 const subnetIds = yield* resolveSubnets;
124 expect(subnetIds.length).toBeGreaterThanOrEqual(2);
125
126 const { cluster } = yield* stack.deploy(
127 Effect.gen(function* () {
128 const cluster = yield* AWS.Kafka.ServerlessCluster("Events", {
129 subnetIds,
130 tags: { fixture: "kafka-serverless" },
131 });
132 return { cluster };
133 }),
134 );
135
136 expect(cluster.clusterName).toBeDefined();
137 expect(cluster.clusterArn).toContain(":cluster/");
138 expect(cluster.clusterType).toBe("SERVERLESS");
139 expect(cluster.state).toBe("ACTIVE");
140 // Serverless clusters only expose the SASL/IAM bootstrap endpoint.
141 expect(cluster.bootstrapBrokerStringSaslIam).toBeDefined();
142 expect(cluster.bootstrapBrokerStringSaslIam).toContain(":9098");
143
144 // Out-of-band verification via distilled.
145 const described = yield* Kafka.describeClusterV2({
146 ClusterArn: cluster.clusterArn,
147 });
148 const info = described.ClusterInfo;
149 expect(info?.State).toBe("ACTIVE");
150 expect(info?.ClusterType).toBe("SERVERLESS");
151 expect(info?.Serverless?.ClientAuthentication?.Sasl?.Iam?.Enabled).toBe(
152 true,
153 );
154
155 yield* stack.destroy();
156 yield* assertClusterDeleted(cluster.clusterArn);
157 }),
158 // create (~5-10 min) + destroy initiation, one test.
159 {
160 tags: ["provider:aws", "provider:aws:ec2", "provider:aws:kafka", "live"],
161 timeout: 1_200_000,
162 },
163);
164
165// Deletion is verified as INITIATED (state DELETING, irreversible) or fully
166// gone. Full disappearance takes several more minutes server-side.
167const assertClusterDeleted = (arn: string) =>
168 Effect.gen(function* () {
169 const state = yield* Kafka.describeClusterV2({ ClusterArn: arn }).pipe(
170 Effect.map((r) => r.ClusterInfo?.State ?? "gone"),
171 Effect.catchTag("NotFoundException", () =>
172 Effect.succeed("gone" as const),
173 ),
174 );
175 if (state !== "gone" && state !== "DELETING") {
176 return yield* Effect.fail(
177 new Error(`MSK cluster '${arn}' still exists (state: ${state})`),
178 );
179 }
180 }).pipe(
181 Effect.retry({
182 schedule: Schedule.max([
183 Schedule.fixed("10 seconds"),
184 Schedule.recurs(18),
185 ]),
186 }),
187 );