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 Serverless delivers to EventBridge.
11 * Job-run state-change events describe the run's transition; application
12 * state-change events describe the application's lifecycle. Fields not
13 * shared by every event kind are optional (the schema grows over time).
14 */
15export interface JobRunEventDetail {
16 /** The id of the EMR Serverless application the event is about. */
17 applicationId?: string;
18 /** Job-run events: the id of the job run. */
19 jobRunId?: string;
20 /** The state transitioned to, e.g. `RUNNING`, `SUCCESS`, `FAILED`. */
21 state?: string;
22 /** Why the state changed (e.g. the failure reason). */
23 stateDetails?: string;
24 /** Additional event fields (the schema grows over time). */
25 [key: string]: unknown;
26}
27
28/** An EMR Serverless EventBridge event delivered to the handler. */
29export type JobRunEvent = EventRecord<JobRunEventDetail>;
30
31/** Which EMR Serverless notifications to subscribe to. */
32export type JobRunEventKind =
33 | "job-run-state-change"
34 | "application-state-change";
35
36const DETAIL_TYPES: Record<JobRunEventKind, string> = {
37 "job-run-state-change": "EMR Serverless Job Run State Change",
38 "application-state-change": "EMR Serverless Application State Change",
39};
40
41export interface JobRunEventSourceProps extends EventRouteProps {
42 /**
43 * Logical id for the backing EventBridge rule.
44 * @default "EMRServerlessJobRunEvents"
45 */
46 id?: string;
47 /**
48 * Which notifications to subscribe to: job-run state changes and/or
49 * application state changes.
50 * @default ["job-run-state-change"]
51 */
52 kinds?: readonly JobRunEventKind[];
53 /**
54 * Restrict to events about specific applications (matched against the
55 * event detail's `applicationId`).
56 */
57 applicationIds?: readonly string[];
58 /**
59 * Restrict to specific states (matched against the event detail's
60 * `state`), e.g. `["FAILED"]` to alert only on failures.
61 */
62 states?: readonly string[];
63}
64
65/**
66 * Event source connecting Amazon EMR Serverless notifications to the
67 * hosting compute. EMR Serverless publishes job-run state changes (most
68 * importantly a job dropping into `FAILED`) and application state changes
69 * to the account's default EventBridge bus (source `aws.emr-serverless`);
70 * this subscribes the host Function to those events so it can alert on
71 * failed jobs or chain follow-up work when a job succeeds.
72 *
73 * EMR Serverless publishes to EventBridge automatically — no additional
74 * resource is created besides the EventBridge rule targeting the host.
75 * Provide the host-specific implementation layer (e.g.
76 * `AWS.Lambda.EventSource`) on the Function effect.
77 *
78 * ### Consuming Job Run Events
79 * **Example:** Alert On Failed Jobs
80 * ```typescript
81 * import * as AWS from "alchemy/AWS";
82 *
83 * export default AlertFunction.make(
84 * { main: import.meta.url },
85 * Effect.gen(function* () {
86 * yield* AWS.EMRServerless.consumeJobRunEvents(
87 * { states: ["FAILED"] },
88 * (events) =>
89 * Stream.runForEach(events, (event) =>
90 * Effect.logError(
91 * `job ${event.detail.jobRunId} failed: ${event.detail.stateDetails}`,
92 * ),
93 * ),
94 * );
95 * return {};
96 * }).pipe(Effect.provide(AWS.Lambda.EventSource)),
97 * );
98 * ```
99 */
100export const consumeJobRunEvents = <StreamReq = never, Req = never>(
101 props: JobRunEventSourceProps,
102 process: (
103 events: Stream.Stream<JobRunEvent, never, StreamReq>,
104 ) => Effect.Effect<void, never, Req>,
105) =>
106 consumeBusEvents(
107 props.id ?? "EMRServerlessJobRunEvents",
108 {
109 source: ["aws.emr-serverless"],
110 "detail-type": (props.kinds ?? (["job-run-state-change"] as const)).map(
111 (kind) => DETAIL_TYPES[kind],
112 ),
113 ...(props.applicationIds !== undefined || props.states !== undefined
114 ? {
115 detail: {
116 ...(props.applicationIds !== undefined
117 ? { applicationId: [...props.applicationIds] }
118 : {}),
119 ...(props.states !== undefined
120 ? { state: [...props.states] }
121 : {}),
122 },
123 }
124 : {}),
125 },
126 { description: props.description, state: props.state },
127 process,
128 );