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 Bedrock Data Automation delivers to EventBridge when
11 * an asynchronous job changes state. Emitted only for jobs invoked with
12 * `notificationConfiguration.eventBridgeConfiguration.eventBridgeEnabled`.
13 */
14export interface DataAutomationJobEventDetail {
15 /** The job id (the trailing segment of the invocation ARN). */
16 job_id?: string;
17 /** The job status, e.g. `SUCCESS`, `CLIENT_ERROR`, `SERVICE_ERROR`. */
18 job_status?: string;
19 /** The detected modality, e.g. `Document`, `Image`, `Audio`, `Video`. */
20 semantic_modality?: string;
21 /** The S3 input object the job processed. */
22 input_s3_object?: { s3_bucket?: string; name?: string };
23 /** The S3 location the job wrote results to. */
24 output_s3_location?: { s3_bucket?: string; name?: string };
25 /** The error message for failed jobs (empty on success). */
26 error_message?: string;
27 /** Additional event fields (the schema grows over time). */
28 [key: string]: unknown;
29}
30
31/** A Bedrock Data Automation EventBridge event delivered to the handler. */
32export type DataAutomationJobEvent = EventRecord<DataAutomationJobEventDetail>;
33
34/** Which Bedrock Data Automation job state-change events to subscribe to. */
35export type DataAutomationJobEventKind =
36 | "created"
37 | "succeeded"
38 | "client-error"
39 | "service-error";
40
41const DETAIL_TYPES: Record<DataAutomationJobEventKind, string> = {
42 created: "Bedrock Data Automation Job Created",
43 succeeded: "Bedrock Data Automation Job Succeeded",
44 "client-error": "Bedrock Data Automation Job Failed With Client Error",
45 "service-error": "Bedrock Data Automation Job Failed With Service Error",
46};
47
48export interface DataAutomationJobEventSourceProps extends EventRouteProps {
49 /**
50 * Logical id for the backing EventBridge rule.
51 * @default "DataAutomationJobEvents"
52 */
53 id?: string;
54 /**
55 * Which job state-change events to subscribe to.
56 * @default all kinds
57 */
58 kinds?: readonly DataAutomationJobEventKind[];
59}
60
61/**
62 * Event source connecting Bedrock Data Automation job state changes to the
63 * hosting compute. Jobs started via `InvokeDataAutomationAsync` with
64 * `notificationConfiguration.eventBridgeConfiguration.eventBridgeEnabled`
65 * publish created / succeeded / failed events to the account's default
66 * EventBridge bus (source `aws.bedrock`); this subscribes the host Function
67 * to those events so it can pick up results the moment a job settles.
68 *
69 * Bedrock publishes to EventBridge directly — no additional resource is
70 * created besides the EventBridge rule targeting the host. Provide the
71 * host-specific implementation layer (e.g. `AWS.Lambda.EventSource`) on the
72 * Function effect.
73 *
74 * ### Consuming Job Events
75 * **Example:** Process Results When A Job Succeeds
76 * ```typescript
77 * import * as AWS from "alchemy/AWS";
78 *
79 * export default ResultsFunction.make(
80 * { main: import.meta.url },
81 * Effect.gen(function* () {
82 * yield* AWS.BedrockDataAutomation.consumeDataAutomationJobEvents(
83 * { kinds: ["succeeded", "client-error", "service-error"] },
84 * (events) =>
85 * Stream.runForEach(events, (event) =>
86 * event.detail.job_status === "SUCCESS"
87 * ? Effect.log(
88 * `job ${event.detail.job_id} wrote ${event.detail.output_s3_location?.name}`,
89 * )
90 * : Effect.log(`job failed: ${event.detail.error_message}`),
91 * ),
92 * );
93 * return {};
94 * }).pipe(Effect.provide(AWS.Lambda.EventSource)),
95 * );
96 * ```
97 */
98export const consumeDataAutomationJobEvents = <StreamReq = never, Req = never>(
99 props: DataAutomationJobEventSourceProps,
100 process: (
101 events: Stream.Stream<DataAutomationJobEvent, never, StreamReq>,
102 ) => Effect.Effect<void, never, Req>,
103) =>
104 consumeBusEvents(
105 props.id ?? "DataAutomationJobEvents",
106 {
107 source: ["aws.bedrock"],
108 "detail-type": (
109 props.kinds ??
110 (Object.keys(DETAIL_TYPES) as DataAutomationJobEventKind[])
111 ).map((kind) => DETAIL_TYPES[kind]),
112 },
113 { description: props.description, state: props.state },
114 process,
115 );