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";
8import type { StateMachine } from "./StateMachine.ts";
9
10/**
11 * The `detail` payload Step Functions delivers to EventBridge when an
12 * execution changes status (`Step Functions Execution Status Change`).
13 * Fields not shared by every status are optional (the schema grows over
14 * time).
15 */
16export interface ExecutionEventDetail {
17 /** The execution's ARN (`arn:…:execution:{machine}:{name}`). */
18 executionArn?: string;
19 /** ARN of the state machine the execution belongs to. */
20 stateMachineArn?: string;
21 /** The execution name. */
22 name?: string;
23 /** The new status: `RUNNING`, `SUCCEEDED`, `FAILED`, `TIMED_OUT`, `ABORTED`. */
24 status?: string;
25 /** Epoch millis the execution started. */
26 startDate?: number;
27 /** Epoch millis the execution stopped (terminal statuses only). */
28 stopDate?: number | null;
29 /** The execution input (JSON string), when execution data is included. */
30 input?: string | null;
31 /** The execution output (JSON string) on `SUCCEEDED`, else `null`. */
32 output?: string | null;
33 /** The `Error` name on `FAILED`/`TIMED_OUT`, when present. */
34 error?: string | null;
35 /** The `Cause` string on `FAILED`/`TIMED_OUT`, when present. */
36 cause?: string | null;
37 /** Additional event fields (the schema grows over time). */
38 [key: string]: unknown;
39}
40
41/** A Step Functions EventBridge event delivered to the handler. */
42export type ExecutionEvent = EventRecord<ExecutionEventDetail>;
43
44/** An execution status a subscription can filter on. */
45export type ExecutionEventStatus =
46 | "RUNNING"
47 | "SUCCEEDED"
48 | "FAILED"
49 | "TIMED_OUT"
50 | "ABORTED";
51
52export interface ExecutionEventSourceProps extends EventRouteProps {
53 /**
54 * Logical id for the backing EventBridge rule.
55 * @default "SfnExecutionEvents"
56 */
57 id?: string;
58 /**
59 * Restrict the subscription to executions of specific state machines.
60 * Omit to receive status changes for every state machine in the
61 * account/region.
62 */
63 stateMachines?: readonly StateMachine[];
64 /**
65 * Which execution statuses to subscribe to.
66 * @default all statuses
67 */
68 statuses?: readonly ExecutionEventStatus[];
69}
70
71/**
72 * Event source connecting Step Functions execution status changes to the
73 * hosting compute. Step Functions publishes every STANDARD execution status
74 * transition to the account's default EventBridge bus (source `aws.states`,
75 * detail-type `Step Functions Execution Status Change`); this subscribes
76 * the host Function to those events so it can alert on `FAILED` executions
77 * or chain post-completion automation.
78 *
79 * Step Functions publishes to EventBridge automatically — no additional
80 * resource is created besides the EventBridge rule targeting the host.
81 * Provide the host-specific implementation layer (e.g.
82 * `AWS.Lambda.EventSource`) on the Function effect.
83 *
84 * `EXPRESS` workflows do not emit execution status change events.
85 *
86 * ### Consuming Execution Events
87 * **Example:** Alert On Failed Executions
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.StepFunctions.consumeExecutionEvents(
95 * { stateMachines: [orderWorkflow], statuses: ["FAILED", "TIMED_OUT"] },
96 * (events) =>
97 * Stream.runForEach(events, (event) =>
98 * Effect.log(`execution ${event.detail.executionArn} failed`),
99 * ),
100 * );
101 * return {};
102 * }).pipe(Effect.provide(AWS.Lambda.EventSource)),
103 * );
104 * ```
105 */
106export const consumeExecutionEvents = <StreamReq = never, Req = never>(
107 props: ExecutionEventSourceProps,
108 process: (
109 events: Stream.Stream<ExecutionEvent, never, StreamReq>,
110 ) => Effect.Effect<void, never, Req>,
111) =>
112 consumeBusEvents(
113 props.id ?? "SfnExecutionEvents",
114 {
115 source: ["aws.states"],
116 "detail-type": ["Step Functions Execution Status Change"],
117 ...(props.stateMachines !== undefined && props.stateMachines.length > 0
118 ? props.statuses !== undefined && props.statuses.length > 0
119 ? {
120 detail: {
121 stateMachineArn: props.stateMachines.map(
122 (machine) => machine.stateMachineArn,
123 ),
124 status: [...props.statuses],
125 },
126 }
127 : {
128 detail: {
129 stateMachineArn: props.stateMachines.map(
130 (machine) => machine.stateMachineArn,
131 ),
132 },
133 }
134 : props.statuses !== undefined && props.statuses.length > 0
135 ? { detail: { status: [...props.statuses] } }
136 : {}),
137 },
138 { description: props.description, state: props.state },
139 process,
140 );