1import type * as Effect from "effect/Effect";
2import type * as Stream from "effect/Stream";
3import {
4 consumeBusEvents,
5 type EventRecord,
6 type EventRouteProps,
7} from "../EventBridge/EventSource.ts";
8
9/**
10 * The `detail` payload DMS delivers to EventBridge when a replication
11 * instance or replication task changes state (source `aws.dms`). Fields not
12 * shared by every event kind are optional (the schema grows over time).
13 */
14export interface DmsReplicationEventDetail {
15 /** The DMS event categories, e.g. `["failure"]`, `["state change"]`. */
16 category?: string;
17 /** The DMS event id, e.g. `DMS-EVENT-0079`. */
18 eventId?: string;
19 /** The event message, e.g. `Replication task has stopped.`. */
20 eventMessage?: string;
21 /** Human-readable event type, e.g. `REPLICATION_TASK_STOPPED`. */
22 eventType?: string;
23 /** The source identifier (instance or task identifier). */
24 sourceId?: string;
25 /** Additional event fields (the schema grows over time). */
26 [key: string]: unknown;
27}
28
29/** A DMS EventBridge event delivered to the handler. */
30export type DmsReplicationEvent = EventRecord<DmsReplicationEventDetail>;
31
32/** Which DMS state-change events to subscribe to. */
33export type DmsReplicationEventKind = "instance-state" | "task-state";
34
35const DETAIL_TYPES: Record<DmsReplicationEventKind, string> = {
36 "instance-state": "DMS Replication Instance State Change",
37 "task-state": "DMS Replication Task State Change",
38};
39
40export interface ReplicationEventSourceProps extends EventRouteProps {
41 /**
42 * Logical id for the backing EventBridge rule.
43 * @default "DmsReplicationEvents"
44 */
45 id?: string;
46 /**
47 * Which state-change events to subscribe to.
48 * @default all kinds
49 */
50 kinds?: readonly DmsReplicationEventKind[];
51}
52
53/**
54 * Event source connecting DMS replication state changes to the hosting
55 * compute. DMS publishes replication instance and replication task state
56 * changes (creation, failover, failure, task stop/start) to the account's
57 * default EventBridge bus (source `aws.dms`); this subscribes the host
58 * Function to those events so it can alert on failures or chain
59 * post-migration automation.
60 *
61 * DMS publishes to EventBridge automatically — no additional resource is
62 * created besides the EventBridge rule targeting the host. Provide the
63 * host-specific implementation layer (e.g. `AWS.Lambda.EventSource`) on the
64 * Function effect.
65 *
66 * ### Consuming Replication Events
67 * **Example:** Alert On Task Failures
68 * ```typescript
69 * import * as AWS from "alchemy/AWS";
70 *
71 * export default AlertFunction.make(
72 * { main: import.meta.url },
73 * Effect.gen(function* () {
74 * yield* AWS.DMS.consumeReplicationEvents(
75 * { kinds: ["task-state"] },
76 * (events) =>
77 * Stream.runForEach(events, (event) =>
78 * event.detail.category === "failure"
79 * ? Effect.log(`DMS task ${event.detail.sourceId} failed`)
80 * : Effect.void,
81 * ),
82 * );
83 * return {};
84 * }).pipe(Effect.provide(AWS.Lambda.EventSource)),
85 * );
86 * ```
87 */
88export const consumeReplicationEvents = <StreamReq = never, Req = never>(
89 props: ReplicationEventSourceProps,
90 process: (
91 events: Stream.Stream<DmsReplicationEvent, never, StreamReq>,
92 ) => Effect.Effect<void, never, Req>,
93) =>
94 consumeBusEvents(
95 props.id ?? "DmsReplicationEvents",
96 {
97 source: ["aws.dms"],
98 "detail-type": (
99 props.kinds ?? (Object.keys(DETAIL_TYPES) as DmsReplicationEventKind[])
100 ).map((kind) => DETAIL_TYPES[kind]),
101 },
102 { description: props.description, state: props.state },
103 process,
104 );