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 { Stream as KinesisStream } from "./Stream.ts";
7
8export type KinesisEventRecord = import("aws-lambda").KinesisStreamRecord;
9
10export interface StreamEventSourceProps {
11 /**
12 * The maximum number of records in each batch that Lambda pulls from the stream.
13 * @default 100
14 */
15 batchSize?: number;
16 /**
17 * The maximum amount of time that Lambda spends gathering records before
18 * invoking the function, e.g. `"5 seconds"` or `Duration.seconds(5)`.
19 * Rounded to whole seconds on the wire.
20 * @default 0
21 */
22 maximumBatchingWindow?: Duration.Input;
23 /**
24 * The position in the stream from which Lambda starts reading.
25 * @default "LATEST"
26 */
27 startingPosition?: "TRIM_HORIZON" | "LATEST" | "AT_TIMESTAMP";
28 /**
29 * The timestamp to start reading from when `startingPosition` is `AT_TIMESTAMP`.
30 */
31 startingPositionTimestamp?: Date;
32 /**
33 * The number of batches to process from each shard concurrently.
34 * @default 1
35 */
36 parallelizationFactor?: number;
37 /**
38 * Split the batch in two and retry if the function returns an error.
39 * @default false
40 */
41 bisectBatchOnFunctionError?: boolean;
42 /**
43 * Discard records older than the specified age, e.g. `"1 hour"` or
44 * `Duration.hours(1)`. Rounded to whole seconds on the wire. Omit to never
45 * discard records by age.
46 */
47 maximumRecordAge?: Duration.Input;
48 /**
49 * Discard records after the specified number of retries.
50 * @default -1
51 */
52 maximumRetryAttempts?: number;
53 /**
54 * The duration of a processing window for tumbling windows, e.g.
55 * `"30 seconds"` or `Duration.seconds(30)`. Rounded to whole seconds on
56 * the wire (0–900 seconds).
57 */
58 tumblingWindow?: Duration.Input;
59 /**
60 * A list of current response type enums applied to the event source mapping.
61 */
62 functionResponseTypes?: "ReportBatchItemFailures"[];
63 /**
64 * A destination for records that failed processing.
65 */
66 destinationConfig?: Lambda.DestinationConfig;
67 /**
68 * Filter criteria to control which records are sent to the function.
69 */
70 filterCriteria?: Lambda.FilterCriteria;
71 /**
72 * The ARN of an AWS KMS key to encrypt the filter criteria.
73 */
74 kmsKeyArn?: string;
75 /**
76 * Metrics configuration for the event source mapping.
77 */
78 metricsConfig?: Lambda.EventSourceMappingMetricsConfig;
79}
80
81/**
82 * Event source connecting a Kinesis `Stream` to the hosting compute.
83 *
84 * The contract is a `Binding.Service`; the Lambda implementation layer
85 * (`AWS.Lambda.StreamEventSource`) creates an event source mapping on the
86 * stream, grants the read IAM actions, and forwards `aws:kinesis` records
87 * into the handler's `Stream`. Use the {@link consumeStreamRecords} helper
88 * rather than calling the service directly.
89 * ### Consuming Records
90 * **Example:** Process Stream Records in a Lambda Function
91 * ```typescript
92 * export default MyFunction.make(
93 * { main: import.meta.url },
94 * Effect.gen(function* () {
95 * const stream = yield* AWS.Kinesis.Stream("OrdersStream");
96 *
97 * // init — registers the event source mapping and the record handler
98 * yield* AWS.Kinesis.consumeStreamRecords(
99 * stream,
100 * { startingPosition: "LATEST", batchSize: 10 },
101 * (records) =>
102 * records.pipe(
103 * Stream.runForEach((record) =>
104 * Effect.log(
105 * Buffer.from(record.kinesis.data, "base64").toString("utf8"),
106 * ),
107 * ),
108 * ),
109 * );
110 *
111 * return {};
112 * }).pipe(Effect.provide(AWS.Lambda.StreamEventSource)),
113 * );
114 * ```
115 *
116 * @binding
117 */
118export interface StreamEventSource extends Binding.Service<
119 StreamEventSource,
120 "AWS.Kinesis.StreamEventSource",
121 StreamEventSourceService
122> {}
123
124export const StreamEventSource = Binding.Service<StreamEventSource>(
125 "AWS.Kinesis.StreamEventSource",
126);
127
128export type StreamEventSourceService = <StreamReq = never, Req = never>(
129 stream: KinesisStream,
130 props: StreamEventSourceProps,
131 process: (
132 stream: Stream.Stream<KinesisEventRecord, never, StreamReq>,
133 ) => Effect.Effect<void, never, Req>,
134) => Effect.Effect<void, never, never>;
135
136/**
137 * Subscribe a runtime to records from a Kinesis stream.
138 *
139 * The Lambda runtime implementation creates an event source mapping and forwards
140 * matching `aws:kinesis` records into the supplied `Stream`.
141 *
142 * @example Forward stream records into an SQS queue
143 * ```typescript
144 * const sink = yield* AWS.SQS.QueueSink(queue);
145 *
146 * yield* AWS.Kinesis.consumeStreamRecords(
147 * stream,
148 * { startingPosition: "LATEST" },
149 * (records) =>
150 * records.pipe(
151 * Stream.map((record) => ({
152 * MessageBody: Buffer.from(record.kinesis.data, "base64").toString("utf8"),
153 * })),
154 * Stream.run(sink),
155 * Effect.orDie,
156 * ),
157 * );
158 * ```
159 */
160export const consumeStreamRecords = <
161 S extends KinesisStream,
162 Req = never,
163 StreamReq = never,
164>(
165 stream: S,
166 props: StreamEventSourceProps = {},
167 process: (
168 stream: Stream.Stream<KinesisEventRecord, never, StreamReq>,
169 ) => Effect.Effect<void, never, Req>,
170) => StreamEventSource.use((source) => source(stream, props, process));