1import * as Effect from "effect/Effect";
2import * as Layer from "effect/Layer";
3import * as Stream from "effect/Stream";
4import * as Namespace from "../../Namespace.ts";
5import {
6 ClusterEventSource as DocDBClusterEventSourceContract,
7 type ClusterEventSourceProps,
8 type ClusterEventSourceService,
9 type DocumentDBRecord,
10} from "../DocDB/ClusterEventSource.ts";
11import type { DBCluster } from "../DocDB/DBCluster.ts";
12import { EventSourceMapping } from "./EventSourceMapping.ts";
13import * as Lambda from "./Function.ts";
14
15export const isDocumentDBEvent = (
16 event: any,
17): event is { eventSourceArn?: string; events: unknown[] } =>
18 typeof event?.eventSourceArn === "string" && Array.isArray(event?.events);
19
20/** @binding */
21export const DocDBClusterEventSource = Layer.effect(
22 DocDBClusterEventSourceContract,
23 Effect.gen(function* () {
24 const host = yield* Lambda.Function;
25 const Mapping = yield* EventSourceMapping;
26
27 return Effect.fn(function* <TDoc = unknown, Req = never>(
28 cluster: DBCluster,
29 props: ClusterEventSourceProps,
30 process: (
31 stream: Stream.Stream<DocumentDBRecord<TDoc>>,
32 ) => Effect.Effect<void, never, Req>,
33 ) {
34 const ClusterArn = yield* cluster.dbClusterArn;
35
36 // Deploy-time: grant IAM (secret access) and create the DocumentDB
37 // event-source mapping. Skipped once running inside the deployed Function
38 // (the global guard), where the only work is registering the runtime
39 // handler below. Namespaced under the host so the mapping's logical
40 // identity is stable.
41 if (!globalThis.__ALCHEMY_RUNTIME__) {
42 yield* Namespace.push(
43 host.LogicalId,
44 Effect.gen(function* () {
45 yield* host.bind`Allow(${host}, AWS.DocDB.ClusterEventSource(${cluster}))`(
46 {
47 policyStatements: [
48 {
49 Effect: "Allow",
50 Action: ["secretsmanager:GetSecretValue"],
51 Resource: [props.secretArn],
52 },
53 ],
54 },
55 );
56
57 yield* Mapping(
58 `AWS.Lambda.EventSourceMapping(${host.LogicalId}, ${cluster.LogicalId})`,
59 {
60 functionName: host.functionName,
61 eventSourceArn: cluster.dbClusterArn,
62 startingPosition: props.startingPosition ?? "LATEST",
63 batchSize: props.batchSize,
64 maximumBatchingWindow: props.maximumBatchingWindow,
65 enabled: props.enabled ?? true,
66 documentDBEventSourceConfig: {
67 DatabaseName: props.databaseName,
68 CollectionName: props.collectionName,
69 FullDocument: props.fullDocument ?? "Default",
70 },
71 sourceAccessConfigurations: [
72 { Type: "BASIC_AUTH", URI: props.secretArn },
73 ],
74 },
75 );
76 }),
77 );
78 }
79
80 yield* host.listen(
81 Effect.gen(function* () {
82 const clusterArn = yield* ClusterArn;
83 return (event: any) => {
84 if (
85 isDocumentDBEvent(event) &&
86 event.eventSourceArn === clusterArn
87 ) {
88 return process(
89 Stream.fromArray(event.events as DocumentDBRecord<TDoc>[]),
90 ).pipe(Effect.orDie);
91 }
92 };
93 }),
94 );
95 }) as ClusterEventSourceService;
96 }),
97);