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 TableEventSource as DynamoDBTableEventSource,
9 type StreamRecord,
10 type StreamsProps,
11 type TableEventSourceService,
12} from "../DynamoDB/Stream.ts";
13
14import type { Table } from "../DynamoDB/Table.ts";
15import { EventSourceMapping } from "./EventSourceMapping.ts";
16import * as Lambda from "./Function.ts";
17
18export const isDynamoDBStreamEvent = (
19 event: any,
20): event is lambda.DynamoDBStreamEvent =>
21 Array.isArray(event?.Records) &&
22 event.Records.length > 0 &&
23 event.Records[0].eventSource === "aws:dynamodb";
24
25/** @binding */
26export const TableEventSource = Layer.effect(
27 DynamoDBTableEventSource,
28 Effect.gen(function* () {
29 const host = yield* Lambda.Function;
30 const Mapping = yield* EventSourceMapping;
31
32 return Effect.fn(function* <Data = unknown, StreamReq = never, Req = never>(
33 table: Table,
34 props: StreamsProps,
35 process: (
36 stream: Stream.Stream<StreamRecord<Data>, never, StreamReq>,
37 ) => Effect.Effect<void, never, Req>,
38 ) {
39 const TableArn = yield* table.tableArn;
40
41 // Deploy-time: enable the table stream, grant IAM, and create the
42 // event-source mapping. Skipped once running inside the deployed Function
43 // (the global guard), where the only work is registering the runtime
44 // handler below. Namespaced under the host so the mapping's logical
45 // identity matches the previous Binding.Policy.
46 if (!globalThis.__ALCHEMY_RUNTIME__) {
47 yield* Namespace.push(
48 host.LogicalId,
49 Effect.gen(function* () {
50 const latestStreamArn = table.latestStreamArn.pipe(
51 Output.mapEffect((arn) =>
52 typeof arn === "string"
53 ? Effect.succeed(arn)
54 : Effect.die(`latestStreamArn is not a string: ${arn}`),
55 ),
56 );
57
58 const streamViewType = props.streamViewType ?? "NEW_AND_OLD_IMAGES";
59
60 yield* Effect.logInfo(
61 `Lambda TableEventSource: binding stream ${streamViewType} for ${table.LogicalId}`,
62 );
63 yield* table.bind`AWS.DynamoDB.Stream(${host}, ${table}, ${streamViewType})`(
64 {
65 streamSpecification: {
66 StreamEnabled: true,
67 StreamViewType: streamViewType,
68 },
69 },
70 );
71
72 yield* Effect.logInfo(
73 `Lambda TableEventSourcePolicy: creating mapping for ${host.LogicalId} <- ${table.LogicalId}`,
74 );
75
76 yield* host.bind`Allow(${host}, AWS.DynamoDB.Table.ReadStream(${table}))`(
77 {
78 policyStatements: [
79 {
80 Effect: "Allow",
81 Action: [
82 "dynamodb:DescribeStream",
83 "dynamodb:GetRecords",
84 "dynamodb:GetShardIterator",
85 ],
86 Resource: [latestStreamArn],
87 },
88 ],
89 },
90 );
91
92 yield* host.bind`Allow(${host}, AWS.DynamoDB.ListStreams(${table}))`(
93 {
94 policyStatements: [
95 {
96 Effect: "Allow",
97 Action: ["dynamodb:ListStreams"],
98 Resource: [table.tableArn],
99 },
100 ],
101 },
102 );
103
104 yield* Mapping(
105 `AWS.Lambda.EventSourceMapping(${host.LogicalId}, ${table.LogicalId})`,
106 {
107 functionName: host.functionName,
108 eventSourceArn: latestStreamArn,
109 batchSize: props.batchSize,
110 maximumBatchingWindow: props.maximumBatchingWindow,
111 enabled: true,
112 startingPosition: props.startingPosition ?? "LATEST",
113 startingPositionTimestamp: props.startingPositionTimestamp,
114 parallelizationFactor: props.parallelizationFactor,
115 bisectBatchOnFunctionError: props.bisectBatchOnFunctionError,
116 maximumRecordAge: props.maximumRecordAge,
117 maximumRetryAttempts: props.maximumRetryAttempts,
118 tumblingWindow: props.tumblingWindow,
119 },
120 );
121 }),
122 );
123 }
124
125 yield* host.listen(
126 Effect.gen(function* () {
127 const tableArn = yield* TableArn;
128 const streamArnPrefix = `${tableArn}/stream/`;
129
130 return (event: any) => {
131 if (isDynamoDBStreamEvent(event)) {
132 const records = event.Records.filter(
133 (record) =>
134 record.eventSourceARN?.startsWith(streamArnPrefix) === true,
135 );
136 if (records.length > 0) {
137 return process(
138 Stream.fromArray(records as StreamRecord<Data>[]),
139 ).pipe(Effect.orDie);
140 }
141 }
142 };
143 }),
144 );
145 }) as TableEventSourceService;
146 }),
147);