1import * as kafka from "@distilled.cloud/aws/kafka";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import * as Stream from "effect/Stream";
5import { Unowned } from "../../AdoptPolicy.ts";
6import { isResolved } from "../../Diff.ts";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
11import type { Providers } from "../Providers.ts";
12
13export interface ServerlessClusterProps {
14 /**
15 * Name of the MSK Serverless cluster. Must be 1-64 characters. If omitted, a
16 * deterministic physical name is generated. Changing the name replaces the
17 * cluster.
18 */
19 clusterName?: string;
20 /**
21 * VPC subnets the cluster's brokers are reachable from. At least two subnets
22 * in different Availability Zones are required. Changing subnets replaces the
23 * cluster.
24 */
25 subnetIds: string[];
26 /**
27 * VPC security groups that control network access to the cluster. If omitted,
28 * the VPC's default security group is used. Changing security groups replaces
29 * the cluster.
30 */
31 securityGroupIds?: string[];
32 /**
33 * User-defined tags for the cluster.
34 */
35 tags?: Record<string, string>;
36}
37
38export interface ServerlessCluster extends Resource<
39 "AWS.Kafka.ServerlessCluster",
40 ServerlessClusterProps,
41 {
42 /** Name of the cluster. */
43 clusterName: string;
44 /** ARN of the cluster. */
45 clusterArn: string;
46 /** Cluster type — always `SERVERLESS` for this resource. */
47 clusterType: string;
48 /** Current state (`CREATING`, `ACTIVE`, `DELETING`, ...). */
49 state: string;
50 /**
51 * SASL/IAM bootstrap broker connection string — the endpoint kafkajs-style
52 * clients connect to using IAM authentication.
53 */
54 bootstrapBrokerStringSaslIam: string | undefined;
55 /** Subnets the cluster is reachable from. */
56 subnetIds: string[];
57 /** Security groups attached to the cluster. */
58 securityGroupIds: string[];
59 /** Tags observed on the cluster. */
60 tags: Record<string, string>;
61 },
62 never,
63 Providers
64> {}
65
66/**
67 * An Amazon MSK (Managed Streaming for Apache Kafka) **Serverless** cluster.
68 *
69 * MSK Serverless clusters use IAM authentication exclusively and scale broker
70 * capacity automatically — there is no broker count, instance type, or storage
71 * to configure. They are reachable only from inside a VPC. Creation takes
72 * roughly 5-10 minutes.
73 *
74 * The provisioned (broker-count) MSK cluster is a separate, much slower
75 * (~20-40 minute) resource and is intentionally not modeled here.
76 *
77 * ### Creating a Serverless Cluster
78 * **Example:** Serverless Cluster in a VPC
79 * ```typescript
80 * const cluster = yield* ServerlessCluster("Events", {
81 * subnetIds: [subnetA.subnetId, subnetB.subnetId],
82 * securityGroupIds: [kafkaSecurityGroup.securityGroupId],
83 * });
84 * ```
85 *
86 * ### Consuming from a Lambda Function
87 * **Example:** Wire a topic to a Lambda
88 * ```typescript
89 * yield* Kafka.consumeKafkaTopic(cluster, { topics: ["orders"] }, (records) =>
90 * records.pipe(Stream.runForEach((r) => Effect.log(r.value))),
91 * );
92 * ```
93 *
94 * @resource
95 */
96export const ServerlessCluster = Resource<ServerlessCluster>(
97 "AWS.Kafka.ServerlessCluster",
98);
99
100const sameStringSet = (
101 a: readonly string[] | undefined,
102 b: readonly string[] | undefined,
103): boolean => {
104 const left = [...(a ?? [])].sort();
105 const right = [...(b ?? [])].sort();
106 return left.length === right.length && left.every((v, i) => v === right[i]);
107};
108
109const toTagRecord = (
110 tags: { [key: string]: string | undefined } | undefined,
111): Record<string, string> =>
112 Object.fromEntries(
113 Object.entries(tags ?? {}).filter(
114 (entry): entry is [string, string] => typeof entry[1] === "string",
115 ),
116 );
117
118export const ServerlessClusterProvider = () =>
119 Provider.effect(
120 ServerlessCluster,
121 Effect.gen(function* () {
122 const toName = (id: string, props: Partial<ServerlessClusterProps>) =>
123 props.clusterName
124 ? Effect.succeed(props.clusterName)
125 : createPhysicalName({ id, maxLength: 64 });
126
127 // Serverless clusters can only be looked up by ARN via describeClusterV2.
128 // Before we have an ARN (or after a state-persistence failure) we scan
129 // listClustersV2 filtered by name.
130 const findByName = Effect.fn(function* (name: string) {
131 return yield* kafka.listClustersV2
132 .pages({ ClusterNameFilter: name, ClusterTypeFilter: "SERVERLESS" })
133 .pipe(
134 Stream.runCollect,
135 Effect.map((chunk) =>
136 Array.from(chunk)
137 .flatMap((page) => page.ClusterInfoList ?? [])
138 .find((c) => c.ClusterName === name),
139 ),
140 );
141 });
142
143 const describeByArn = Effect.fn(function* (arn: string) {
144 const response = yield* kafka
145 .describeClusterV2({ ClusterArn: arn })
146 .pipe(
147 Effect.catchTag("NotFoundException", () =>
148 Effect.succeed(undefined),
149 ),
150 );
151 return response?.ClusterInfo;
152 });
153
154 const readBrokers = Effect.fn(function* (arn: string) {
155 return yield* kafka.getBootstrapBrokers({ ClusterArn: arn }).pipe(
156 Effect.map((r) => r.BootstrapBrokerStringSaslIam),
157 // Brokers are only resolvable once the cluster is ACTIVE.
158 Effect.catchTag("BadRequestException", () =>
159 Effect.succeed(undefined),
160 ),
161 Effect.catchTag("ConflictException", () => Effect.succeed(undefined)),
162 );
163 });
164
165 // Bounded readiness wait. MSK Serverless create typically completes in
166 // 5-10 minutes; budget ~15 min (60 * 15s).
167 const waitForActive = Effect.fn(function* (arn: string) {
168 const policy = Schedule.max([
169 Schedule.fixed("15 seconds"),
170 Schedule.recurs(60),
171 ]);
172 return yield* describeByArn(arn).pipe(
173 Effect.flatMap((cluster) => {
174 if (!cluster) {
175 return Effect.fail(new Error(`MSK cluster '${arn}' not found`));
176 }
177 if (cluster.State === "FAILED") {
178 return Effect.die(
179 new Error(
180 `MSK cluster '${arn}' entered FAILED state: ${cluster.StateInfo?.Message ?? ""}`,
181 ),
182 );
183 }
184 if (cluster.State !== "ACTIVE") {
185 return Effect.fail(
186 new Error(
187 `MSK cluster '${arn}' not active (state: ${cluster.State})`,
188 ),
189 );
190 }
191 return Effect.succeed(cluster);
192 }),
193 Effect.retry({ schedule: policy }),
194 );
195 });
196
197 const toAttrs = Effect.fn(function* (cluster: kafka.Cluster) {
198 if (!cluster.ClusterName || !cluster.ClusterArn) {
199 return yield* Effect.fail(
200 new Error(
201 `MSK cluster is missing its name or ARN (state: ${cluster.State})`,
202 ),
203 );
204 }
205 const bootstrap =
206 cluster.State === "ACTIVE"
207 ? yield* readBrokers(cluster.ClusterArn)
208 : undefined;
209 const vpcConfigs = cluster.Serverless?.VpcConfigs ?? [];
210 return {
211 clusterName: cluster.ClusterName,
212 clusterArn: cluster.ClusterArn,
213 clusterType: cluster.ClusterType ?? "SERVERLESS",
214 state: cluster.State ?? "ACTIVE",
215 bootstrapBrokerStringSaslIam: bootstrap,
216 subnetIds: vpcConfigs.flatMap((v) => v.SubnetIds ?? []),
217 securityGroupIds: vpcConfigs.flatMap((v) => v.SecurityGroupIds ?? []),
218 tags: toTagRecord(cluster.Tags),
219 };
220 });
221
222 return {
223 stables: ["clusterName", "clusterArn", "clusterType"],
224
225 diff: Effect.fn(function* ({ id, olds, news }) {
226 if (!isResolved(news)) return undefined;
227 if (
228 (yield* toName(id, olds ?? {})) !== (yield* toName(id, news ?? {}))
229 ) {
230 return { action: "replace" } as const;
231 }
232 // VPC configuration is create-only for serverless clusters.
233 if (
234 olds?.subnetIds !== undefined &&
235 !sameStringSet(news?.subnetIds, olds.subnetIds)
236 ) {
237 return { action: "replace" } as const;
238 }
239 if (
240 olds?.securityGroupIds !== undefined &&
241 !sameStringSet(news?.securityGroupIds, olds.securityGroupIds)
242 ) {
243 return { action: "replace" } as const;
244 }
245 }),
246
247 read: Effect.fn(function* ({ id, olds, output }) {
248 const cluster = output?.clusterArn
249 ? yield* describeByArn(output.clusterArn)
250 : yield* findByName(yield* toName(id, olds ?? {}));
251 if (!cluster?.ClusterArn) return undefined;
252 const attrs = yield* toAttrs(cluster);
253 return (yield* hasAlchemyTags(id, attrs.tags))
254 ? attrs
255 : Unowned(attrs);
256 }),
257
258 reconcile: Effect.fn(function* ({ id, news, output, session }) {
259 const name = output?.clusterName ?? (yield* toName(id, news));
260 const internalTags = yield* createInternalTags(id);
261 const desiredTags = { ...internalTags, ...news.tags };
262
263 // 1. Observe — cloud state is authoritative. Prefer the cached ARN,
264 // fall back to a name scan (recovery / adoption).
265 let observed = output?.clusterArn
266 ? yield* describeByArn(output.clusterArn)
267 : yield* findByName(name);
268
269 // 2. Ensure — create if missing; tolerate a concurrent create
270 // (ConflictException) by re-scanning for the cluster by name.
271 if (!observed?.ClusterArn) {
272 const created = yield* kafka
273 .createClusterV2({
274 ClusterName: name,
275 Tags: desiredTags,
276 Serverless: {
277 VpcConfigs: [
278 {
279 SubnetIds: news.subnetIds,
280 SecurityGroupIds: news.securityGroupIds,
281 },
282 ],
283 ClientAuthentication: { Sasl: { Iam: { Enabled: true } } },
284 },
285 })
286 .pipe(
287 Effect.map((r) => r.ClusterArn),
288 Effect.catchTag("ConflictException", () =>
289 findByName(name).pipe(Effect.map((c) => c?.ClusterArn)),
290 ),
291 );
292 if (!created) {
293 return yield* Effect.die(
294 new Error(`MSK ServerlessCluster(${id}) could not be created`),
295 );
296 }
297 observed = yield* describeByArn(created);
298 }
299
300 if (!observed?.ClusterArn) {
301 return yield* Effect.die(
302 new Error(`MSK ServerlessCluster(${id}) could not be reconciled`),
303 );
304 }
305 const arn = observed.ClusterArn;
306
307 // Wait for ACTIVE (bounded) so tag sync and broker resolution work.
308 observed = yield* waitForActive(arn);
309
310 // 3. Sync tags — diff against OBSERVED cloud tags so adoption
311 // rewrites ownership tags correctly.
312 const { removed, upsert } = diffTags(
313 toTagRecord(observed.Tags),
314 desiredTags,
315 );
316 if (removed.length > 0) {
317 yield* kafka.untagResource({ ResourceArn: arn, TagKeys: removed });
318 }
319 if (upsert.length > 0) {
320 yield* kafka.tagResource({
321 ResourceArn: arn,
322 Tags: Object.fromEntries(
323 upsert.map(({ Key, Value }) => [Key, Value]),
324 ),
325 });
326 observed = (yield* describeByArn(arn)) ?? observed;
327 }
328
329 // 4. Return fresh attributes.
330 yield* session.note(name);
331 return yield* toAttrs(observed);
332 }),
333
334 delete: Effect.fn(function* ({ output }) {
335 yield* kafka.deleteCluster({ ClusterArn: output.clusterArn }).pipe(
336 Effect.catchTag("NotFoundException", () => Effect.void),
337 // A cluster still CREATING rejects deletion; retry until it can
338 // be deleted (or is gone). Bounded.
339 Effect.retry({
340 while: (e) => e._tag === "BadRequestException",
341 schedule: Schedule.max([
342 Schedule.fixed("15 seconds"),
343 Schedule.recurs(40),
344 ]),
345 }),
346 Effect.catchTag("BadRequestException", () => Effect.void),
347 );
348 }),
349
350 list: () =>
351 kafka.listClustersV2.pages({ ClusterTypeFilter: "SERVERLESS" }).pipe(
352 Stream.runCollect,
353 Effect.map((chunk) =>
354 Array.from(chunk).flatMap((page) =>
355 (page.ClusterInfoList ?? []).filter(
356 (c) =>
357 c.ClusterName !== undefined && c.ClusterArn !== undefined,
358 ),
359 ),
360 ),
361 Effect.flatMap(
362 Effect.forEach((cluster) => toAttrs(cluster), { concurrency: 4 }),
363 ),
364 ),
365 };
366 }),
367 );