1import type * as Lambda from "@distilled.cloud/aws/lambda";
2import type * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import * as Stream from "effect/Stream";
5import * as Binding from "../../Binding.ts";
6import type { ServerlessCluster } from "./ServerlessCluster.ts";
7
8export type MSKRecord = import("aws-lambda").MSKRecord;
9
10export interface KafkaEventSourceProps {
11 /**
12 * The Kafka topic name(s) the function consumes. At least one is required.
13 */
14 topics: string[];
15 /**
16 * The Kafka consumer group id used by the event source poller.
17 * @default a Lambda-generated id
18 */
19 consumerGroupId?: string;
20 /**
21 * The position in the topic from which the poller starts reading.
22 * @default "LATEST"
23 */
24 startingPosition?: "TRIM_HORIZON" | "LATEST";
25 /**
26 * The maximum number of records in each batch that Lambda pulls from the topic.
27 * @default 100
28 */
29 batchSize?: number;
30 /**
31 * The maximum time Lambda spends gathering records before invoking the
32 * function, e.g. `"5 seconds"` or `Duration.seconds(5)`. Rounded to whole
33 * seconds on the wire.
34 * @default 500ms (0)
35 */
36 maximumBatchingWindow?: Duration.Input;
37 /**
38 * Whether the event source mapping actively polls.
39 * @default true
40 */
41 enabled?: boolean;
42 /**
43 * Filter criteria to control which records are sent to the function.
44 */
45 filterCriteria?: Lambda.FilterCriteria;
46 /**
47 * Provisioned poller configuration for predictable throughput.
48 */
49 provisionedPollerConfig?: Lambda.ProvisionedPollerConfig;
50}
51
52/**
53 * Event source connecting topics on an MSK `ServerlessCluster` to the
54 * hosting compute.
55 *
56 * The contract is a `Binding.Service`; the Lambda implementation layer
57 * (`AWS.Lambda.KafkaEventSource`) grants the IAM actions MSK IAM
58 * authentication requires, creates an event source mapping on the cluster,
59 * and forwards `aws:kafka` records into the handler's `Stream`. Use the
60 * {@link consumeKafkaTopic} helper rather than calling the service directly.
61 * ### Consuming Topics
62 * **Example:** Consume a Topic in a Lambda Function
63 * ```typescript
64 * export default MyFunction.make(
65 * { main: import.meta.url },
66 * Effect.gen(function* () {
67 * const cluster = yield* AWS.Kafka.ServerlessCluster("Events", {
68 * subnetIds,
69 * });
70 *
71 * // init — registers the event source mapping and the record handler
72 * yield* AWS.Kafka.consumeKafkaTopic(
73 * cluster,
74 * { topics: ["orders"], consumerGroupId: "my-service" },
75 * (records) =>
76 * records.pipe(Stream.runForEach((r) => Effect.log(r.value))),
77 * );
78 *
79 * return {};
80 * }).pipe(Effect.provide(AWS.Lambda.KafkaEventSource)),
81 * );
82 * ```
83 *
84 * @binding
85 */
86export interface KafkaEventSource extends Binding.Service<
87 KafkaEventSource,
88 "AWS.Kafka.KafkaEventSource",
89 KafkaEventSourceService
90> {}
91
92export const KafkaEventSource = Binding.Service<KafkaEventSource>(
93 "AWS.Kafka.KafkaEventSource",
94);
95
96export type KafkaEventSourceService = <StreamReq = never, Req = never>(
97 cluster: ServerlessCluster,
98 props: KafkaEventSourceProps,
99 process: (
100 stream: Stream.Stream<MSKRecord, never, StreamReq>,
101 ) => Effect.Effect<void, never, Req>,
102) => Effect.Effect<void, never, never>;
103
104/**
105 * Subscribe a runtime to records from one or more topics on an MSK
106 * {@link ServerlessCluster}.
107 *
108 * The Lambda runtime implementation grants the IAM actions MSK IAM
109 * authentication requires, creates an event source mapping pointing at the
110 * cluster, and forwards matching `aws:kafka` records into the supplied
111 * `Stream`.
112 *
113 * @example Consume the "orders" topic
114 * ```typescript
115 * yield* AWS.Kafka.consumeKafkaTopic(
116 * cluster,
117 * { topics: ["orders"], startingPosition: "TRIM_HORIZON" },
118 * (records) =>
119 * records.pipe(
120 * Stream.runForEach((record) =>
121 * Effect.log(Buffer.from(record.value, "base64").toString("utf8")),
122 * ),
123 * ),
124 * );
125 * ```
126 */
127export const consumeKafkaTopic = <
128 C extends ServerlessCluster,
129 Req = never,
130 StreamReq = never,
131>(
132 cluster: C,
133 props: KafkaEventSourceProps,
134 process: (
135 stream: Stream.Stream<MSKRecord, never, StreamReq>,
136 ) => Effect.Effect<void, never, Req>,
137) => KafkaEventSource.use((source) => source(cluster, props, process));