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 OpenSearch Service delivers to EventBridge. The schema
11 * is loosely specified and differs slightly per event kind, so every field is
12 * optional (the schema grows over time).
13 */
14export interface DomainEventDetail {
15 /** The event that occurred, e.g. `"Service Software Update"`. */
16 event?: string;
17 /** Status of the event, e.g. `"Available"`, `"Completed"`, `"Failed"`. */
18 status?: string;
19 /** Severity of the event: `"Informational"`, `"Low"`, `"Medium"`, `"High"`. */
20 severity?: string;
21 /** Human-readable description of what happened. */
22 description?: string;
23 /** Additional event fields (the schema grows over time). */
24 [key: string]: unknown;
25}
26
27/** An OpenSearch Service EventBridge event delivered to the handler. */
28export type DomainEvent = EventRecord<DomainEventDetail>;
29
30/**
31 * Which OpenSearch Service notifications to subscribe to. Each kind maps to
32 * one of the detail-types OpenSearch publishes to the account's default bus.
33 */
34export type DomainEventKind =
35 | "software-update"
36 | "auto-tune"
37 | "cluster-status"
38 | "domain-update"
39 | "vpc-endpoint"
40 | "node-retirement"
41 | "dry-run-progress";
42
43const DETAIL_TYPES: Record<DomainEventKind, string> = {
44 "software-update": "Amazon OpenSearch Service Software Update Notification",
45 "auto-tune": "Amazon OpenSearch Service Auto-Tune Notification",
46 "cluster-status": "Amazon OpenSearch Service Cluster Status Notification",
47 "domain-update": "Amazon OpenSearch Service Domain Update Notification",
48 "vpc-endpoint": "Amazon OpenSearch Service VPC Endpoint Notification",
49 "node-retirement": "Amazon OpenSearch Service Node Retirement Notification",
50 "dry-run-progress": "Amazon OpenSearch Service Dry Run Progress Notification",
51};
52
53export interface DomainEventSourceProps extends EventRouteProps {
54 /**
55 * Logical id for the backing EventBridge rule.
56 * @default "OpenSearchDomainEvents"
57 */
58 id?: string;
59 /**
60 * Which notifications to subscribe to.
61 * @default all OpenSearch Service detail-types
62 */
63 kinds?: readonly DomainEventKind[];
64 /**
65 * Restrict to events about specific domains (matched against the event's
66 * top-level `resources`, which carries the domain ARNs).
67 */
68 domainArns?: readonly string[];
69}
70
71/**
72 * Event source connecting OpenSearch Service notifications to the hosting
73 * compute. OpenSearch publishes domain lifecycle events — available and
74 * completed service software updates, Auto-Tune optimizations, cluster
75 * health transitions (e.g. a red cluster or blocked writes), VPC endpoint
76 * changes, node retirements, and blue/green update progress — to the
77 * account's default EventBridge bus (source `aws.es`); this subscribes the
78 * host Function to those events so it can alert on degraded clusters or
79 * auto-approve pending software updates.
80 *
81 * OpenSearch publishes to EventBridge automatically — no additional resource
82 * is created besides the EventBridge rule targeting the host. Provide the
83 * host-specific implementation layer (e.g. `AWS.Lambda.EventSource`) on the
84 * Function effect.
85 *
86 * ### Consuming Domain Events
87 * **Example:** Alert When a Cluster Degrades or an Update Fails
88 * ```typescript
89 * import * as AWS from "alchemy/AWS";
90 *
91 * export default AlertFunction.make(
92 * { main: import.meta.url },
93 * Effect.gen(function* () {
94 * yield* AWS.OpenSearch.consumeDomainEvents(
95 * { kinds: ["cluster-status", "software-update"] },
96 * (events) =>
97 * Stream.runForEach(events, (event) =>
98 * Effect.logError(
99 * `${event.detail.event}: ${event.detail.description}`,
100 * ),
101 * ),
102 * );
103 * return {};
104 * }).pipe(Effect.provide(AWS.Lambda.EventSource)),
105 * );
106 * ```
107 */
108export const consumeDomainEvents = <StreamReq = never, Req = never>(
109 props: DomainEventSourceProps,
110 process: (
111 events: Stream.Stream<DomainEvent, never, StreamReq>,
112 ) => Effect.Effect<void, never, Req>,
113) =>
114 consumeBusEvents(
115 props.id ?? "OpenSearchDomainEvents",
116 {
117 source: ["aws.es"],
118 ...(props.kinds !== undefined
119 ? { "detail-type": props.kinds.map((kind) => DETAIL_TYPES[kind]) }
120 : {}),
121 ...(props.domainArns !== undefined
122 ? { resources: [...props.domainArns] }
123 : {}),
124 },
125 { description: props.description, state: props.state },
126 process,
127 );