1import type lambda from "aws-lambda";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as Stream from "effect/Stream";
5import * as Namespace from "../../Namespace.ts";
6import * as Output from "../../Output.ts";
7import {
8 KafkaEventSource as MSKKafkaEventSource,
9 type KafkaEventSourceProps,
10 type KafkaEventSourceService,
11 type MSKRecord,
12} from "../Kafka/ClusterEventSource.ts";
13import type { ServerlessCluster } from "../Kafka/ServerlessCluster.ts";
14import { EventSourceMapping } from "./EventSourceMapping.ts";
15import * as Lambda from "./Function.ts";
16
17export const isMSKEvent = (event: any): event is lambda.MSKEvent =>
18 event?.eventSource === "aws:kafka" &&
19 typeof event?.records === "object" &&
20 event.records !== null;
21
22// The cluster ARN is `arn:...:cluster/name/uuid`. Topic and consumer-group
23// resources live under sibling namespaces (`topic/` and `group/`) scoped to the
24// same cluster path; MSK IAM auth grants data-plane access via these ARNs.
25const topicArnGlob = (clusterArn: string) =>
26 `${clusterArn.replace(":cluster/", ":topic/")}/*`;
27const groupArnGlob = (clusterArn: string) =>
28 `${clusterArn.replace(":cluster/", ":group/")}/*`;
29
30/** @binding */
31export const KafkaEventSource = Layer.effect(
32 MSKKafkaEventSource,
33 Effect.gen(function* () {
34 const host = yield* Lambda.Function;
35 const Mapping = yield* EventSourceMapping;
36
37 return Effect.fn(function* <StreamReq = never, Req = never>(
38 cluster: ServerlessCluster,
39 props: KafkaEventSourceProps,
40 process: (
41 stream: Stream.Stream<MSKRecord, never, StreamReq>,
42 ) => Effect.Effect<void, never, Req>,
43 ) {
44 const ClusterArn = yield* cluster.clusterArn;
45
46 // Deploy-time: grant the IAM actions MSK IAM authentication requires and
47 // create the event-source mapping. Skipped once running inside the
48 // deployed Function (the global guard). Namespaced under the host so the
49 // mapping's logical identity is stable.
50 if (!globalThis.__ALCHEMY_RUNTIME__) {
51 yield* Namespace.push(
52 host.LogicalId,
53 Effect.gen(function* () {
54 yield* host.bind`Allow(${host}, AWS.Lambda.KafkaEventSource(${cluster}))`(
55 {
56 policyStatements: [
57 {
58 // Cluster-level: connect + describe (Lambda poller + IAM auth).
59 Effect: "Allow",
60 Action: [
61 "kafka-cluster:Connect",
62 "kafka-cluster:DescribeCluster",
63 "kafka-cluster:DescribeClusterDynamicConfiguration",
64 "kafka:DescribeClusterV2",
65 "kafka:GetBootstrapBrokers",
66 ],
67 Resource: [cluster.clusterArn],
68 },
69 {
70 // Topic-level: read records + describe topics.
71 Effect: "Allow",
72 Action: [
73 "kafka-cluster:DescribeTopic",
74 "kafka-cluster:ReadData",
75 ],
76 Resource: [
77 cluster.clusterArn.pipe(Output.map(topicArnGlob)),
78 ],
79 },
80 {
81 // Consumer-group-level: join/commit as a consumer group.
82 Effect: "Allow",
83 Action: [
84 "kafka-cluster:DescribeGroup",
85 "kafka-cluster:AlterGroup",
86 ],
87 Resource: [
88 cluster.clusterArn.pipe(Output.map(groupArnGlob)),
89 ],
90 },
91 ],
92 },
93 );
94
95 yield* Mapping(
96 `AWS.Lambda.EventSourceMapping(${host.LogicalId}, ${cluster.LogicalId})`,
97 {
98 functionName: host.functionName,
99 eventSourceArn: cluster.clusterArn,
100 topics: props.topics,
101 amazonManagedKafkaEventSourceConfig: props.consumerGroupId
102 ? { ConsumerGroupId: props.consumerGroupId }
103 : undefined,
104 batchSize: props.batchSize,
105 maximumBatchingWindow: props.maximumBatchingWindow,
106 startingPosition: props.startingPosition ?? "LATEST",
107 enabled: props.enabled ?? true,
108 filterCriteria: props.filterCriteria,
109 provisionedPollerConfig: props.provisionedPollerConfig,
110 },
111 );
112 }),
113 );
114 }
115
116 yield* host.listen(
117 Effect.gen(function* () {
118 return (event: any) => {
119 if (isMSKEvent(event)) {
120 // `event.records` maps `topic-partition` → MSKRecord[]; flatten
121 // to a single stream of records.
122 const records = Object.values(
123 event.records,
124 ).flat() as MSKRecord[];
125 if (records.length > 0) {
126 return process(Stream.fromArray(records)).pipe(Effect.orDie);
127 }
128 }
129 };
130 }),
131 );
132 }) as KafkaEventSourceService;
133 }),
134);