1import * as AWS from "@/AWS";
2import * as EC2 from "@distilled.cloud/aws/ec2";
3import * as Context from "effect/Context";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Stream from "effect/Stream";
7import { HttpServerRequest } from "effect/http/HttpServerRequest";
8import * as HttpServerResponse from "effect/http/HttpServerResponse";
9import { getDefaultVpc } from "../DefaultVpc.ts";
10
11export class KafkaTestFunction extends AWS.Lambda.Function<AWS.Lambda.Function>()(
12 "KafkaTestFunction",
13) {}
14
15export class FixtureCluster extends Context.Service<
16 FixtureCluster,
17 { cluster: AWS.Kafka.ServerlessCluster }
18>()("FixtureCluster") {}
19
20/**
21 * Resolve two default-for-AZ subnets from the account's default VPC — MSK
22 * requires at least two subnets in distinct AZs. Runtime-guarded: the Lambda
23 * runtime re-executes this props effect on cold start with no ec2:Describe*
24 * permission, so return an empty list there (VPC config is deploy-time only).
25 */
26const resolveSubnets = Effect.gen(function* () {
27 if (globalThis.__ALCHEMY_RUNTIME__) return [] as string[];
28 const vpc = yield* getDefaultVpc;
29 const subnets = yield* EC2.describeSubnets({
30 Filters: [
31 { Name: "vpc-id", Values: [vpc.vpcId] },
32 { Name: "default-for-az", Values: ["true"] },
33 ],
34 });
35 const byAz = new Map<string, string>();
36 for (const s of subnets.Subnets ?? []) {
37 if (s.AvailabilityZone && s.SubnetId && !byAz.has(s.AvailabilityZone)) {
38 byAz.set(s.AvailabilityZone, s.SubnetId);
39 }
40 }
41 return [...byAz.values()].slice(0, 3);
42}).pipe(
43 // Deploy-time lookup only; a failure here is a fixture defect, not a typed
44 // error the Function impl contract can carry.
45 Effect.orDie,
46);
47
48export const FixtureClusterLive = Layer.effect(
49 FixtureCluster,
50 Effect.gen(function* () {
51 const subnetIds = yield* resolveSubnets;
52 const cluster = yield* AWS.Kafka.ServerlessCluster("FixtureCluster", {
53 subnetIds,
54 tags: { fixture: "kafka-serverless" },
55 });
56 return { cluster };
57 }),
58);
59
60export default KafkaTestFunction.make(
61 { main: import.meta.url, functionUrl: true },
62 Effect.gen(function* () {
63 const { cluster } = yield* FixtureCluster;
64
65 // Wire the "orders" topic to this function. At deploy time this grants MSK
66 // IAM-auth actions and creates the event source mapping; at runtime it
67 // registers the record handler below.
68 yield* AWS.Kafka.consumeKafkaTopic(
69 cluster,
70 { topics: ["orders"], consumerGroupId: "alchemy-fixture" },
71 (records) => records.pipe(Stream.runForEach((r) => Effect.log(r.value))),
72 );
73
74 // Cluster-scoped control-plane bindings (grant + ClusterArn injection).
75 const getBootstrapBrokers = yield* AWS.Kafka.GetBootstrapBrokers(cluster);
76 const listTopics = yield* AWS.Kafka.ListTopics(cluster);
77 const createTopic = yield* AWS.Kafka.CreateTopic(cluster);
78 const describeTopic = yield* AWS.Kafka.DescribeTopic(cluster);
79 const deleteTopic = yield* AWS.Kafka.DeleteTopic(cluster);
80 // Data-plane connection descriptor (env-published bootstrap endpoint).
81 const connect = yield* AWS.Kafka.ConnectReadWrite(cluster);
82
83 const ClusterArn = cluster.clusterArn;
84
85 return {
86 fetch: Effect.gen(function* () {
87 const request = yield* HttpServerRequest;
88 const url = new URL(request.originalUrl ?? request.url, "http://x");
89 if (url.pathname === "/cluster") {
90 const clusterArn = yield* ClusterArn;
91 return yield* HttpServerResponse.json({ clusterArn });
92 }
93 if (url.pathname === "/brokers") {
94 const brokers = yield* getBootstrapBrokers();
95 return yield* HttpServerResponse.json({
96 saslIam: brokers.BootstrapBrokerStringSaslIam,
97 });
98 }
99 if (url.pathname === "/connect") {
100 const info = yield* connect;
101 return yield* HttpServerResponse.json(info);
102 }
103 if (url.pathname === "/topics") {
104 const page = yield* listTopics();
105 return yield* HttpServerResponse.json({
106 topics: (page.Topics ?? []).map((t) => t.TopicName),
107 });
108 }
109 if (url.pathname === "/topics/roundtrip") {
110 // Create → describe → delete a probe topic through the MSK topic
111 // control plane. Typed 400s are surfaced (not thrown) so the test
112 // can distinguish "grant works, cluster type unsupported" from an
113 // IAM failure (which dies as a 500).
114 const outcome = yield* createTopic({
115 TopicName: "alchemy-probe",
116 PartitionCount: 1,
117 }).pipe(
118 Effect.flatMap(() =>
119 describeTopic({ TopicName: "alchemy-probe" }).pipe(
120 Effect.map((d) => ({
121 created: true,
122 partitions: d.PartitionCount,
123 })),
124 ),
125 ),
126 Effect.tap(() =>
127 deleteTopic({ TopicName: "alchemy-probe" }).pipe(
128 Effect.catchTag("NotFoundException", () => Effect.void),
129 ),
130 ),
131 Effect.catchTag(
132 ["BadRequestException", "TopicExistsException"],
133 (e) => Effect.succeed({ created: false, error: e._tag }),
134 ),
135 );
136 return yield* HttpServerResponse.json(outcome);
137 }
138 return HttpServerResponse.text("ok");
139 }).pipe(Effect.orDie),
140 };
141 }).pipe(
142 Effect.provide(
143 Layer.provideMerge(
144 Layer.mergeAll(
145 AWS.Lambda.KafkaEventSource,
146 AWS.Kafka.GetBootstrapBrokersHttp,
147 AWS.Kafka.ListTopicsHttp,
148 AWS.Kafka.CreateTopicHttp,
149 AWS.Kafka.DescribeTopicHttp,
150 AWS.Kafka.DeleteTopicHttp,
151 AWS.Kafka.ConnectReadWriteHttp,
152 ),
153 FixtureClusterLive,
154 ),
155 ),
156 ),
157);