1import * as Context from "effect/Context";
2import type * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import * as Stream from "effect/Stream";
5import type { DBCluster } from "./DBCluster.ts";
6
7/**
8 * The `fullDocument` delivery mode for a DocumentDB change-stream event.
9 * `UpdateLookup` includes the current full document on updates; `Default`
10 * delivers only the change description.
11 */
12export type FullDocument = "UpdateLookup" | "Default";
13
14/**
15 * A single DocumentDB change-stream record delivered to the Lambda function.
16 * `TDoc` is the shape of the watched collection's documents.
17 */
18export interface DocumentDBRecord<TDoc = unknown> {
19 /**
20 * The change-stream event payload (MongoDB change event).
21 */
22 event: {
23 /** The operation type, e.g. `insert`, `update`, `delete`, `replace`. */
24 operationType?: string;
25 /** The namespace (database + collection) the change occurred in. */
26 ns?: { db?: string; coll?: string };
27 /** The full document (present when `fullDocument` is `UpdateLookup`). */
28 fullDocument?: TDoc;
29 /** The `_id` of the changed document. */
30 documentKey?: { _id?: unknown };
31 /** The change-stream resume token. */
32 _id?: unknown;
33 };
34}
35
36export interface ClusterEventSourceProps {
37 /**
38 * The DocumentDB database to watch. Required.
39 */
40 databaseName: string;
41 /**
42 * The specific collection to watch. Omit to watch every collection in the
43 * database.
44 */
45 collectionName?: string;
46 /**
47 * Whether Lambda delivers the full document on update events.
48 * @default "Default"
49 */
50 fullDocument?: FullDocument;
51 /**
52 * ARN of a Secrets Manager secret holding the DocumentDB connection
53 * credentials (`{ username, password }`). Used for `BASIC_AUTH` access.
54 * Required — a managed cluster exposes this as `cluster.masterUserSecretArn`.
55 */
56 secretArn: string;
57 /**
58 * Maximum number of records per batch delivered to the function.
59 */
60 batchSize?: number;
61 /**
62 * Maximum time Lambda spends gathering records before invoking, e.g.
63 * `"5 seconds"` or `Duration.seconds(5)`. Rounded to whole seconds on the
64 * wire.
65 */
66 maximumBatchingWindow?: Duration.Input;
67 /**
68 * Where in the change stream to begin reading.
69 * @default "LATEST"
70 */
71 startingPosition?: "LATEST" | "TRIM_HORIZON";
72 /**
73 * Whether the mapping is active on create.
74 * @default true
75 */
76 enabled?: boolean;
77}
78
79type ChangesHandler<TDoc, Req> = (
80 stream: Stream.Stream<DocumentDBRecord<TDoc>>,
81) => Effect.Effect<void, never, Req>;
82
83/**
84 * Subscribe an Effect handler to change-stream events produced by an Amazon
85 * DocumentDB {@link DBCluster}.
86 *
87 * The underlying Lambda event-source mapping polls the cluster's change stream
88 * and delivers batches of change events. Note that DocumentDB change streams
89 * must be enabled on the target database/collection (a data-plane operation run
90 * against the cluster) before events flow.
91 *
92 * @param cluster The DocumentDB cluster to consume change events from.
93 * @param props Event-source configuration (database, collection, credentials).
94 * @param process The handler invoked with a stream of change records.
95 *
96 * @example Consume change-stream events in a Lambda function
97 * ```typescript
98 * // inside the Function's Effect.gen, with
99 * // Effect.provide(AWS.Lambda.DocDBClusterEventSource)
100 * const secretArn = yield* cluster.masterUserSecretArn;
101 * yield* AWS.DocDB.consumeClusterChanges(
102 * cluster,
103 * {
104 * databaseName: "app",
105 * collectionName: "orders",
106 * secretArn,
107 * fullDocument: "UpdateLookup",
108 * startingPosition: "LATEST",
109 * },
110 * (stream) =>
111 * stream.pipe(
112 * Stream.runForEach((record) =>
113 * Effect.log(
114 * `${record.event.operationType}: ${JSON.stringify(record.event.documentKey)}`,
115 * ),
116 * ),
117 * ),
118 * );
119 * ```
120 */
121export function consumeClusterChanges<TDoc = unknown, Req = never>(
122 cluster: DBCluster,
123 props: ClusterEventSourceProps,
124 process: ChangesHandler<TDoc, Req>,
125): Effect.Effect<void, never, ClusterEventSource> {
126 return ClusterEventSource.use((source) => source(cluster, props, process));
127}
128
129export class ClusterEventSource extends Context.Service<
130 ClusterEventSource,
131 ClusterEventSourceService
132>()("AWS.DocDB.ClusterEventSource") {}
133
134export type ClusterEventSourceService = <TDoc = unknown, Req = never>(
135 cluster: DBCluster,
136 props: ClusterEventSourceProps,
137 process: (
138 stream: Stream.Stream<DocumentDBRecord<TDoc>>,
139 ) => Effect.Effect<void, never, Req>,
140) => Effect.Effect<void, never, never>;