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 DataBrew delivers to EventBridge when a job run
11 * changes state. Fields not shared by every event kind are optional (the
12 * schema grows over time).
13 */
14export interface JobEventDetail {
15 /** Name of the job the run belongs to. */
16 jobName?: string;
17 /** The run's id (`db_…`). */
18 jobRunId?: string;
19 /**
20 * The new run state — `RUNNING`, `SUCCEEDED`, `FAILED`, `STOPPED`, or
21 * `TIMEOUT`.
22 */
23 state?: string;
24 /** Event severity, e.g. `INFO`. */
25 severity?: string;
26 /** Human-readable state-change message. */
27 message?: string;
28 /** The account the job belongs to. */
29 accountId?: string;
30 /** Additional event fields (the schema grows over time). */
31 [key: string]: unknown;
32}
33
34/** A DataBrew EventBridge event delivered to the handler. */
35export type JobEvent = EventRecord<JobEventDetail>;
36
37export interface JobEventSourceProps extends EventRouteProps {
38 /**
39 * Logical id for the backing EventBridge rule.
40 * @default "DataBrewJobEvents"
41 */
42 id?: string;
43 /**
44 * Restrict to events about specific jobs (matched against the event's
45 * `jobName`).
46 */
47 jobNames?: readonly string[];
48 /**
49 * Restrict to specific run states, e.g. `["FAILED", "TIMEOUT"]`.
50 */
51 states?: readonly string[];
52}
53
54/**
55 * Event source connecting DataBrew job-run notifications to the hosting
56 * compute. DataBrew publishes every job-run state change to the account's
57 * default EventBridge bus (source `aws.databrew`, detail-type
58 * `DataBrew Job State Change`); this subscribes the host Function to those
59 * events so it can alert on failed runs or chain post-run automation.
60 *
61 * DataBrew publishes to EventBridge automatically — no additional resource
62 * is created besides the EventBridge rule targeting the host. Provide the
63 * host-specific implementation layer (e.g. `AWS.Lambda.EventSource`) on the
64 * Function effect.
65 *
66 * ### Consuming Job Events
67 * **Example:** Alert On Failed Runs
68 * ```typescript
69 * import * as AWS from "alchemy/AWS";
70 *
71 * export default AlertFunction.make(
72 * { main: import.meta.url },
73 * Effect.gen(function* () {
74 * yield* AWS.DataBrew.consumeJobEvents(
75 * { states: ["FAILED", "TIMEOUT"] },
76 * (events) =>
77 * Stream.runForEach(events, (event) =>
78 * Effect.log(
79 * `run ${event.detail.jobRunId} of ${event.detail.jobName} failed`,
80 * ),
81 * ),
82 * );
83 * return {};
84 * }).pipe(Effect.provide(AWS.Lambda.EventSource)),
85 * );
86 * ```
87 */
88export const consumeJobEvents = <StreamReq = never, Req = never>(
89 props: JobEventSourceProps,
90 process: (
91 events: Stream.Stream<JobEvent, never, StreamReq>,
92 ) => Effect.Effect<void, never, Req>,
93) =>
94 consumeBusEvents(
95 props.id ?? "DataBrewJobEvents",
96 {
97 source: ["aws.databrew"],
98 "detail-type": ["DataBrew Job State Change"],
99 ...(props.jobNames !== undefined || props.states !== undefined
100 ? {
101 detail: {
102 ...(props.jobNames !== undefined
103 ? { jobName: [...props.jobNames] }
104 : {}),
105 ...(props.states !== undefined
106 ? { state: [...props.states] }
107 : {}),
108 },
109 }
110 : {}),
111 },
112 { description: props.description, state: props.state },
113 process,
114 );