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 on EKS delivers to EventBridge for job
11 * run state changes. Fields not shared by every event are optional (the
12 * schema grows over time).
13 */
14export interface JobRunEventDetail {
15 /** The ID of the job run. */
16 id?: string;
17 /** The name of the job run. */
18 name?: string;
19 /** The ID of the virtual cluster the job run belongs to. */
20 virtualClusterId?: string;
21 /** The ARN of the job run. */
22 arn?: string;
23 /** The state the job run transitioned to (e.g. `COMPLETED`, `FAILED`). */
24 state?: string;
25 /** For failed runs: the failure reason (e.g. `USER_ERROR`). */
26 failureReason?: string;
27 /** Additional detail about the state transition. */
28 stateDetails?: string;
29 /** Additional event fields (the schema grows over time). */
30 [key: string]: unknown;
31}
32
33/** An EMR on EKS job run EventBridge event delivered to the handler. */
34export type JobRunEvent = EventRecord<JobRunEventDetail>;
35
36export interface JobRunEventSourceProps extends EventRouteProps {
37 /**
38 * Logical id for the backing EventBridge rule.
39 * @default "EMRContainersJobRunEvents"
40 */
41 id?: string;
42 /**
43 * Restrict to job runs on specific virtual clusters (matched against the
44 * event detail's `virtualClusterId`).
45 */
46 virtualClusterIds?: readonly string[];
47 /**
48 * Restrict to specific job run states (e.g. `["FAILED", "COMPLETED"]`).
49 * @default all states
50 */
51 states?: readonly string[];
52}
53
54/**
55 * Event source connecting Amazon EMR on EKS job run state changes to the
56 * hosting compute. EMR on EKS publishes every job run state transition
57 * (`PENDING` → `SUBMITTED` → `RUNNING` → `COMPLETED`/`FAILED`/`CANCELLED`)
58 * to the account's default EventBridge bus (source `aws.emr-containers`,
59 * detail-type `EMR Job Run State Change`); this subscribes the host
60 * Function to those events so it can react when Spark jobs finish or fail.
61 *
62 * EMR on EKS publishes to EventBridge automatically — no additional
63 * resource is created besides the EventBridge rule targeting the host.
64 * Provide the host-specific implementation layer (e.g.
65 * `AWS.Lambda.EventSource`) on the Function effect.
66 *
67 * ### Consuming Job Run Events
68 * **Example:** Alert On Failed Spark Jobs
69 * ```typescript
70 * import * as AWS from "alchemy/AWS";
71 *
72 * export default AlertFunction.make(
73 * { main: import.meta.url },
74 * Effect.gen(function* () {
75 * yield* AWS.EMRContainers.consumeJobRunEvents(
76 * { states: ["FAILED"] },
77 * (events) =>
78 * Stream.runForEach(events, (event) =>
79 * Effect.logError(
80 * `job run ${event.detail.id} failed: ${event.detail.failureReason}`,
81 * ),
82 * ),
83 * );
84 * return {};
85 * }).pipe(Effect.provide(AWS.Lambda.EventSource)),
86 * );
87 * ```
88 */
89export const consumeJobRunEvents = <StreamReq = never, Req = never>(
90 props: JobRunEventSourceProps,
91 process: (
92 events: Stream.Stream<JobRunEvent, never, StreamReq>,
93 ) => Effect.Effect<void, never, Req>,
94) =>
95 consumeBusEvents(
96 props.id ?? "EMRContainersJobRunEvents",
97 {
98 source: ["aws.emr-containers"],
99 "detail-type": ["EMR Job Run State Change"],
100 ...(props.virtualClusterIds !== undefined || props.states !== undefined
101 ? {
102 detail: {
103 ...(props.virtualClusterIds !== undefined
104 ? { virtualClusterId: [...props.virtualClusterIds] }
105 : {}),
106 ...(props.states !== undefined
107 ? { state: [...props.states] }
108 : {}),
109 },
110 }
111 : {}),
112 },
113 { description: props.description, state: props.state },
114 process,
115 );