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 MWAA Serverless delivers to EventBridge for
11 * workflow-run and task state-change events. The schema is not pinned by
12 * the service docs, so fields are optional and unknown fields are allowed
13 * (the schema grows over time).
14 */
15export interface WorkflowRunEventDetail {
16 /** ARN of the workflow the event is about. */
17 workflowArn?: string;
18 /** The run the event is about. */
19 runId?: string;
20 /** Task events: the task instance the event is about. */
21 taskInstanceId?: string;
22 /** The state transitioned to, e.g. `RUNNING`, `SUCCESS`, `FAILED`. */
23 status?: string;
24 /** Additional event fields (the schema grows over time). */
25 [key: string]: unknown;
26}
27
28/** An MWAA Serverless EventBridge event delivered to the handler. */
29export type WorkflowRunEvent = EventRecord<WorkflowRunEventDetail>;
30
31/**
32 * Workflow-run states MWAA Serverless publishes as
33 * `MWAA Serverless Workflow Run {State}` detail-types.
34 */
35export type WorkflowRunEventState =
36 | "Started"
37 | "Queued"
38 | "Running"
39 | "Succeeded"
40 | "Failed"
41 | "Stopped"
42 | "Timeout";
43
44/**
45 * Task-instance states MWAA Serverless publishes as
46 * `MWAA Serverless Task {State}` detail-types.
47 */
48export type TaskEventState =
49 | "Queued"
50 | "Scheduled"
51 | "Upstream Failed"
52 | "Running"
53 | "Succeeded"
54 | "Failed"
55 | "Up For Retry"
56 | "Timeout";
57
58const ALL_RUN_STATES: readonly WorkflowRunEventState[] = [
59 "Started",
60 "Queued",
61 "Running",
62 "Succeeded",
63 "Failed",
64 "Stopped",
65 "Timeout",
66];
67
68export interface WorkflowRunEventSourceProps extends EventRouteProps {
69 /**
70 * Logical id for the backing EventBridge rule.
71 * @default "MWAAServerlessWorkflowRunEvents"
72 */
73 id?: string;
74 /**
75 * Which workflow-run state transitions to subscribe to, e.g. `["Failed"]`
76 * to alert only on failures.
77 * @default all run states
78 */
79 runStates?: readonly WorkflowRunEventState[];
80 /**
81 * Which task-instance state transitions to additionally subscribe to.
82 * @default none
83 */
84 taskStates?: readonly TaskEventState[];
85}
86
87/**
88 * Event source connecting Amazon MWAA Serverless notifications to the
89 * hosting compute. MWAA Serverless publishes workflow-run state changes
90 * (started, queued, running, succeeded, failed, stopped, timeout) and task
91 * state changes to the account's default EventBridge bus (source
92 * `aws.airflow-serverless`); this subscribes the host Function to those
93 * events so it can alert on failed runs or chain follow-up work when a run
94 * succeeds.
95 *
96 * MWAA Serverless publishes to EventBridge automatically — no additional
97 * resource is created besides the EventBridge rule targeting the host.
98 * Provide the host-specific implementation layer (e.g.
99 * `AWS.Lambda.EventSource`) on the Function effect.
100 *
101 * ### Consuming Workflow Run Events
102 * **Example:** Alert On Failed Runs
103 * ```typescript
104 * import * as AWS from "alchemy/AWS";
105 *
106 * export default AlertFunction.make(
107 * { main: import.meta.url },
108 * Effect.gen(function* () {
109 * yield* AWS.MWAAServerless.consumeWorkflowRunEvents(
110 * { runStates: ["Failed", "Timeout"] },
111 * (events) =>
112 * Stream.runForEach(events, (event) =>
113 * Effect.logError(
114 * `workflow run failed: ${event["detail-type"]}`,
115 * event.detail,
116 * ),
117 * ),
118 * );
119 * return {};
120 * }).pipe(Effect.provide(AWS.Lambda.EventSource)),
121 * );
122 * ```
123 */
124export const consumeWorkflowRunEvents = <StreamReq = never, Req = never>(
125 props: WorkflowRunEventSourceProps,
126 process: (
127 events: Stream.Stream<WorkflowRunEvent, never, StreamReq>,
128 ) => Effect.Effect<void, never, Req>,
129) =>
130 consumeBusEvents(
131 props.id ?? "MWAAServerlessWorkflowRunEvents",
132 {
133 source: ["aws.airflow-serverless"],
134 "detail-type": [
135 ...(props.runStates ?? ALL_RUN_STATES).map(
136 (state) => `MWAA Serverless Workflow Run ${state}`,
137 ),
138 ...(props.taskStates ?? []).map(
139 (state) => `MWAA Serverless Task ${state}`,
140 ),
141 ],
142 },
143 { description: props.description, state: props.state },
144 process,
145 );