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 type { Stream as KinesisStream } from "../Kinesis/Stream.ts";
7import {
8 StreamEventSource as KinesisStreamEventSource,
9 type KinesisEventRecord,
10 type StreamEventSourceProps,
11 type StreamEventSourceService,
12} from "../Kinesis/StreamEventSource.ts";
13import { EventSourceMapping } from "./EventSourceMapping.ts";
14import * as Lambda from "./Function.ts";
15
16export const isKinesisStreamEvent = (
17 event: any,
18): event is lambda.KinesisStreamEvent =>
19 Array.isArray(event?.Records) &&
20 event.Records.length > 0 &&
21 event.Records[0].eventSource === "aws:kinesis";
22
23/** @binding */
24export const StreamEventSource = Layer.effect(
25 KinesisStreamEventSource,
26 Effect.gen(function* () {
27 const host = yield* Lambda.Function;
28 const Mapping = yield* EventSourceMapping;
29
30 return Effect.fn(function* <StreamReq = never, Req = never>(
31 stream: KinesisStream,
32 props: StreamEventSourceProps,
33 process: (
34 stream: Stream.Stream<KinesisEventRecord, never, StreamReq>,
35 ) => Effect.Effect<void, never, Req>,
36 ) {
37 const StreamArn = yield* stream.streamArn;
38
39 // Deploy-time: grant IAM and create the event-source mapping. Skipped once
40 // running inside the deployed Function (the global guard), where the only
41 // work is registering the runtime handler below. Namespaced under the host
42 // so the mapping's logical identity matches the previous Binding.Policy.
43 if (!globalThis.__ALCHEMY_RUNTIME__) {
44 yield* Namespace.push(
45 host.LogicalId,
46 Effect.gen(function* () {
47 yield* host.bind`Allow(${host}, AWS.Lambda.StreamEventSource(${stream}))`(
48 {
49 policyStatements: [
50 {
51 Effect: "Allow",
52 Action: [
53 "kinesis:DescribeStream",
54 "kinesis:GetRecords",
55 "kinesis:GetShardIterator",
56 "kinesis:ListShards",
57 ],
58 Resource: [stream.streamArn],
59 },
60 ],
61 },
62 );
63
64 yield* Mapping(
65 `AWS.Lambda.EventSourceMapping(${host.LogicalId}, ${stream.LogicalId})`,
66 {
67 functionName: host.functionName,
68 eventSourceArn: stream.streamArn,
69 batchSize: props.batchSize,
70 maximumBatchingWindow: props.maximumBatchingWindow,
71 enabled: true,
72 startingPosition: props.startingPosition ?? "LATEST",
73 startingPositionTimestamp: props.startingPositionTimestamp,
74 parallelizationFactor: props.parallelizationFactor,
75 bisectBatchOnFunctionError: props.bisectBatchOnFunctionError,
76 maximumRecordAge: props.maximumRecordAge,
77 maximumRetryAttempts: props.maximumRetryAttempts,
78 tumblingWindow: props.tumblingWindow,
79 functionResponseTypes: props.functionResponseTypes,
80 destinationConfig: props.destinationConfig,
81 filterCriteria: props.filterCriteria,
82 kmsKeyArn: props.kmsKeyArn,
83 metricsConfig: props.metricsConfig,
84 },
85 );
86 }),
87 );
88 }
89
90 yield* host.listen(
91 Effect.gen(function* () {
92 const streamArn = yield* StreamArn;
93
94 return (event: any) => {
95 if (isKinesisStreamEvent(event)) {
96 const records = event.Records.filter(
97 (record) =>
98 record.eventSourceARN?.startsWith(streamArn) === true,
99 );
100 if (records.length > 0) {
101 return process(
102 Stream.fromArray(records as KinesisEventRecord[]),
103 ).pipe(Effect.orDie);
104 }
105 }
106 };
107 }),
108 );
109 }) as StreamEventSourceService;
110 }),
111);