1import type * as DynamoDB from "@distilled.cloud/aws/dynamodb";
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 { Table, TableEvent, TableRecord } from "./Table.ts";
7
8export type StreamRecord<Data> = TableRecord<Data>;
9
10export type StreamEvent<Data> = TableEvent<Data>;
11
12/**
13 * Event source binding that subscribes a Lambda function to a DynamoDB
14 * table's change stream. Enables the stream on the table (via the binding
15 * contract) and creates the Lambda event source mapping.
16 *
17 * Prefer the {@link consumeTableChanges} helper for ergonomic use; provide
18 * the runtime-specific implementation layer (e.g. `Lambda.TableEventSource`)
19 * on the Function.
20 *
21 * @binding
22 */
23export interface TableEventSource extends Binding.Service<
24 TableEventSource,
25 "AWS.DynamoDB.TableEventSource",
26 TableEventSourceService
27> {}
28export const TableEventSource = Binding.Service<TableEventSource>(
29 "AWS.DynamoDB.TableEventSource",
30);
31
32export type TableEventSourceService = <
33 Data = unknown,
34 StreamReq = never,
35 Req = never,
36>(
37 table: Table,
38 props: StreamsProps,
39 process: (
40 stream: Stream.Stream<StreamRecord<Data>, never, StreamReq>,
41 ) => Effect.Effect<void, never, Req>,
42) => Effect.Effect<void, never, never>;
43
44export interface TableEventSourceProps {
45 /**
46 * The maximum number of records in each batch that Lambda pulls from the stream.
47 * @default 100
48 */
49 batchSize?: number;
50 /**
51 * The maximum amount of time that Lambda spends gathering records before
52 * invoking the function, e.g. `"5 seconds"` or `Duration.seconds(5)`.
53 * Rounded to whole seconds on the wire.
54 * @default 0
55 */
56 maximumBatchingWindow?: Duration.Input;
57 /**
58 * The position in the stream from which to start reading.
59 * @default "LATEST"
60 */
61 startingPosition?: "TRIM_HORIZON" | "LATEST" | "AT_TIMESTAMP";
62 /**
63 * The timestamp to start reading from when `startingPosition` is `AT_TIMESTAMP`.
64 */
65 startingPositionTimestamp?: Date;
66 /**
67 * The number of batches to process from each shard concurrently.
68 * @default 1
69 */
70 parallelizationFactor?: number;
71 /**
72 * Split the batch in two and retry if the function returns an error.
73 * @default false
74 */
75 bisectBatchOnFunctionError?: boolean;
76 /**
77 * Discard records older than the specified age, e.g. `"1 hour"` or
78 * `Duration.hours(1)`. Rounded to whole seconds on the wire. Omit to never
79 * discard records by age.
80 */
81 maximumRecordAge?: Duration.Input;
82 /**
83 * Discard records after the specified number of retries.
84 * @default -1
85 */
86 maximumRetryAttempts?: number;
87 /**
88 * The duration of a processing window for tumbling windows, e.g.
89 * `"30 seconds"` or `Duration.seconds(30)`. Rounded to whole seconds on
90 * the wire (0–900 seconds).
91 */
92 tumblingWindow?: Duration.Input;
93}
94
95export interface StreamsProps extends TableEventSourceProps {
96 /**
97 * The DynamoDB stream view type to enable on the table.
98 * @default "NEW_AND_OLD_IMAGES"
99 */
100 streamViewType?: DynamoDB.StreamViewType;
101}
102
103/**
104 * Consume change data capture events from a DynamoDB table via a Lambda
105 * event source mapping. The stream is enabled automatically through the
106 * binding contract. The handler receives each delivered batch as an Effect
107 * `Stream` of change records.
108 *
109 * Provide `Lambda.TableEventSource` on the Function to satisfy the binding.
110 *
111 * @example Log every table change
112 * ```typescript
113 * yield* DynamoDB.consumeTableChanges(
114 * table,
115 * { streamViewType: "NEW_AND_OLD_IMAGES", startingPosition: "TRIM_HORIZON" },
116 * (stream) =>
117 * stream.pipe(
118 * Stream.runForEach((record) =>
119 * Effect.log(`${record.eventName}: ${JSON.stringify(record.dynamodb.Keys)}`),
120 * ),
121 * ),
122 * );
123 * ```
124 *
125 * @example Forward changes to an SQS queue via QueueSink
126 * ```typescript
127 * const sink = yield* AWS.SQS.QueueSink(queue);
128 *
129 * yield* DynamoDB.consumeTableChanges(
130 * table,
131 * { streamViewType: "NEW_AND_OLD_IMAGES", batchSize: 10 },
132 * (stream) =>
133 * stream.pipe(
134 * Stream.map((record) => ({
135 * MessageBody: JSON.stringify({
136 * eventName: record.eventName,
137 * keys: record.dynamodb.Keys,
138 * newImage: record.dynamodb.NewImage,
139 * }),
140 * })),
141 * Stream.run(sink),
142 * Effect.orDie,
143 * ),
144 * );
145 * ```
146 */
147export const consumeTableChanges = <
148 Data = unknown,
149 Req = never,
150 StreamReq = never,
151>(
152 table: Table,
153 props: StreamsProps = {},
154 handler: (
155 stream: Stream.Stream<StreamRecord<Data>, never, StreamReq>,
156 ) => Effect.Effect<void, never, Req>,
157) => TableEventSource.use((source) => source(table, props, handler));