1import * as AWS from "@/AWS";
2import * as Test from "@/Test/Alchemy";
3import * as Lambda from "@distilled.cloud/aws/lambda";
4import { describe, expect } from "alchemy-test";
5import * as Data from "effect/Data";
6import * as Effect from "effect/Effect";
7import * as Schedule from "effect/Schedule";
8import * as HttpClient from "effect/http/HttpClient";
9import KafkaTestFunctionLive, {
10 FixtureCluster,
11 KafkaTestFunction,
12} from "./kafka-handler.ts";
13
14const { test } = Test.make({ providers: AWS.providers() });
15
16// The MSK Serverless cluster this test provisions takes ~5-10 minutes to
17// create and is metered while it exists, so the whole event-source e2e is
18// gated behind AWS_TEST_SLOW=1. It deploys a cluster + a Lambda that binds
19// `consumeKafkaTopic`, then verifies the event source mapping was created and
20// points at the cluster with the configured topic. (A full produce/consume
21// roundtrip additionally requires a Kafka admin/producer with MSK IAM auth to
22// create and write the topic, which is out of scope here.)
23describe.sequential(
24 "AWS.Kafka.KafkaEventSource",
25 {
26 tags: [
27 "provider:aws",
28 "provider:aws:ec2",
29 "provider:aws:kafka",
30 "provider:aws:lambda",
31 "live",
32 ],
33 },
34 () => {
35 test.provider.skipIf(!process.env.AWS_TEST_SLOW)(
36 "creates an event source mapping pointing a Lambda at the cluster topic",
37 (stack) =>
38 Effect.gen(function* () {
39 yield* stack.destroy();
40
41 const { cluster, fn } = yield* stack.deploy(
42 Effect.gen(function* () {
43 const { cluster } = yield* FixtureCluster;
44 const fn = yield* KafkaTestFunction;
45 return { cluster, fn };
46 }).pipe(Effect.provide(KafkaTestFunctionLive)),
47 );
48
49 expect(cluster.clusterArn).toContain(":cluster/");
50
51 // Verify the event source mapping exists, targets the cluster, and
52 // carries the configured topic. MSK ESMs progress
53 // CREATING -> ENABLING -> ENABLED; assert it reached at least CREATING
54 // and never FAILED.
55 const mapping = yield* waitForMapping(
56 fn.functionName,
57 cluster.clusterArn,
58 );
59 expect(mapping.Topics).toEqual(["orders"]);
60 expect(mapping.State).not.toBe("Failed");
61
62 // Drive the control-plane bindings end-to-end through the deployed
63 // function URL: GetBootstrapBrokers, ConnectReadWrite (env-published
64 // endpoint), ListTopics, and the CreateTopic → DescribeTopic →
65 // DeleteTopic roundtrip.
66 const baseUrl = fn.functionUrl!.replace(/\/+$/, "");
67 const get = (path: string) =>
68 HttpClient.get(`${baseUrl}${path}`).pipe(
69 Effect.retry({
70 schedule: Schedule.exponential("500 millis"),
71 times: 10,
72 }),
73 Effect.flatMap((res) => res.json),
74 );
75
76 const brokers = (yield* get("/brokers")) as { saslIam?: string };
77 expect(brokers.saslIam).toContain(":9098");
78
79 const connect = (yield* get("/connect")) as {
80 bootstrapServers: string;
81 brokers: string[];
82 clusterArn: string;
83 authentication: string;
84 };
85 expect(connect.authentication).toBe("iam");
86 expect(connect.bootstrapServers).toContain(":9098");
87 expect(connect.clusterArn).toBe(cluster.clusterArn);
88
89 const topics = (yield* get("/topics")) as { topics: string[] };
90 expect(Array.isArray(topics.topics)).toBe(true);
91
92 const roundtrip = (yield* get("/topics/roundtrip")) as {
93 created: boolean;
94 partitions?: number;
95 error?: string;
96 };
97 // Serverless clusters that don't support the topic control plane
98 // surface a typed BadRequestException — either branch proves the
99 // grant + ClusterArn injection (an IAM gap would be a 500).
100 if (roundtrip.created) {
101 expect(roundtrip.partitions).toBe(1);
102 } else {
103 expect(roundtrip.error).toBeDefined();
104 }
105
106 yield* stack.destroy();
107 }),
108 { timeout: 1_200_000 },
109 );
110 },
111);
112
113class MappingNotReady extends Data.TaggedError("MappingNotReady")<{}> {}
114
115const waitForMapping = Effect.fn(function* (
116 functionName: string,
117 eventSourceArn: string,
118) {
119 return yield* Lambda.listEventSourceMappings({
120 FunctionName: functionName,
121 EventSourceArn: eventSourceArn,
122 }).pipe(
123 Effect.flatMap((result) => {
124 const mapping = result.EventSourceMappings?.[0];
125 return mapping?.UUID
126 ? Effect.succeed(mapping)
127 : Effect.fail(new MappingNotReady());
128 }),
129 Effect.retry({
130 while: (e) => e._tag === "MappingNotReady",
131 schedule: Schedule.max([
132 Schedule.fixed("5 seconds"),
133 Schedule.recurs(24),
134 ]),
135 }),
136 );
137});