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 Amazon EMR delivers to EventBridge when a cluster,
11 * step, instance group/fleet, or auto-scaling policy changes state. EMR uses
12 * camelCase keys; fields not shared by every event kind are optional (the
13 * schema grows over time).
14 */
15export interface ClusterEventDetail {
16 /** The id of the cluster the event is about (`j-…`). */
17 clusterId?: string;
18 /** The new state, e.g. `RUNNING`, `WAITING`, `TERMINATED`, `COMPLETED`. */
19 state?: string;
20 /** Event severity: `CRITICAL`, `ERROR`, or `INFO`. */
21 severity?: string;
22 /** Human-readable description of the state change. */
23 message?: string;
24 /** Cluster events: the cluster's name. Step events: the step's name. */
25 name?: string;
26 /** Cluster events: why the cluster changed state. */
27 stateChangeReason?: string;
28 /** Step events: the step's id (`s-…`). */
29 stepId?: string;
30 /** Step events: the step's configured action on failure. */
31 actionOnFailure?: string;
32 /** Instance-group events: the group's id (`ig-…`). */
33 instanceGroupId?: string;
34 /** Instance-fleet events: the fleet's id (`if-…`). */
35 instanceFleetId?: string;
36 /** Additional event fields (the schema grows over time). */
37 [key: string]: unknown;
38}
39
40/** An EMR EventBridge event delivered to the handler. */
41export type ClusterEvent = EventRecord<ClusterEventDetail>;
42
43/** Which EMR notifications to subscribe to. */
44export type ClusterEventKind =
45 | "cluster"
46 | "step"
47 | "instance-group"
48 | "instance-fleet"
49 | "auto-scaling";
50
51const DETAIL_TYPES: Record<ClusterEventKind, string> = {
52 cluster: "EMR Cluster State Change",
53 step: "EMR Step Status Change",
54 "instance-group": "EMR Instance Group State Change",
55 "instance-fleet": "EMR Instance Fleet State Change",
56 "auto-scaling": "EMR Auto Scaling Policy State Change",
57};
58
59export interface ClusterEventSourceProps extends EventRouteProps {
60 /**
61 * Logical id for the backing EventBridge rule.
62 * @default "EMRClusterEvents"
63 */
64 id?: string;
65 /**
66 * Which notifications to subscribe to: cluster state changes, step status
67 * changes, instance group/fleet state changes, auto-scaling policy state
68 * changes, or any combination.
69 * @default ["cluster"]
70 */
71 kinds?: readonly ClusterEventKind[];
72 /**
73 * Restrict to events about specific clusters (matched against the event's
74 * `clusterId`).
75 */
76 clusterIds?: readonly string[];
77}
78
79/**
80 * Event source connecting Amazon EMR notifications to the hosting compute.
81 * EMR publishes every cluster state change, step status change, instance
82 * group/fleet state change, and auto-scaling policy state change to the
83 * account's default EventBridge bus (source `aws.emr`); this subscribes the
84 * host Function to those events so it can alert on failed steps or chain
85 * post-cluster automation.
86 *
87 * EMR publishes to EventBridge automatically — no additional resource is
88 * created besides the EventBridge rule targeting the host. Provide the
89 * host-specific implementation layer (e.g. `AWS.Lambda.EventSource`) on the
90 * Function effect.
91 *
92 * ### Consuming Cluster Events
93 * **Example:** Alert On Failed Steps
94 * ```typescript
95 * import * as AWS from "alchemy/AWS";
96 *
97 * export default AlertFunction.make(
98 * { main: import.meta.url },
99 * Effect.gen(function* () {
100 * yield* AWS.EMR.consumeClusterEvents(
101 * { kinds: ["step"] },
102 * (events) =>
103 * Stream.runForEach(events, (event) =>
104 * event.detail.state === "FAILED"
105 * ? Effect.log(`step ${event.detail.stepId} failed`)
106 * : Effect.void,
107 * ),
108 * );
109 * return {};
110 * }).pipe(Effect.provide(AWS.Lambda.EventSource)),
111 * );
112 * ```
113 */
114export const consumeClusterEvents = <StreamReq = never, Req = never>(
115 props: ClusterEventSourceProps,
116 process: (
117 events: Stream.Stream<ClusterEvent, never, StreamReq>,
118 ) => Effect.Effect<void, never, Req>,
119) =>
120 consumeBusEvents(
121 props.id ?? "EMRClusterEvents",
122 {
123 source: ["aws.emr"],
124 "detail-type": (props.kinds ?? (["cluster"] as const)).map(
125 (kind) => DETAIL_TYPES[kind],
126 ),
127 ...(props.clusterIds !== undefined
128 ? { detail: { clusterId: [...props.clusterIds] } }
129 : {}),
130 },
131 { description: props.description, state: props.state },
132 process,
133 );