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 AWS Batch delivers to EventBridge when a job changes
11 * state (`Batch Job State Change`) or a job queue becomes blocked
12 * (`Batch Job Queue Blocked`). Fields not shared by every event kind are
13 * optional (the schema grows over time).
14 */
15export interface BatchJobEventDetail {
16 /** The job's ARN (`arn:…:job/{jobId}`). */
17 jobArn?: string;
18 /** The job id. */
19 jobId?: string;
20 /** The job name given at submission. */
21 jobName?: string;
22 /** ARN of the job queue the job was submitted to. */
23 jobQueue?: string;
24 /** The new job status, e.g. `RUNNABLE`, `STARTING`, `SUCCEEDED`, `FAILED`. */
25 status?: string;
26 /** Human-readable reason for the status, when present. */
27 statusReason?: string;
28 /** Revision-qualified job definition ARN the job runs. */
29 jobDefinition?: string;
30 /** Container detail (image, exit code, log stream, …) on state changes. */
31 container?: Record<string, unknown>;
32 /** `Ref::` parameter substitutions supplied at submission. */
33 parameters?: Record<string, string>;
34 /** Additional event fields (the schema grows over time). */
35 [key: string]: unknown;
36}
37
38/** An AWS Batch EventBridge event delivered to the handler. */
39export type BatchJobEvent = EventRecord<BatchJobEventDetail>;
40
41/** Which AWS Batch state-change events to subscribe to. */
42export type BatchJobEventKind = "job-state" | "job-queue-blocked";
43
44const DETAIL_TYPES: Record<BatchJobEventKind, string> = {
45 "job-state": "Batch Job State Change",
46 "job-queue-blocked": "Batch Job Queue Blocked",
47};
48
49export interface JobEventSourceProps extends EventRouteProps {
50 /**
51 * Logical id for the backing EventBridge rule.
52 * @default "BatchJobEvents"
53 */
54 id?: string;
55 /**
56 * Which state-change events to subscribe to.
57 * @default all kinds
58 */
59 kinds?: readonly BatchJobEventKind[];
60}
61
62/**
63 * Event source connecting AWS Batch job state changes to the hosting compute.
64 * AWS Batch publishes every job state transition (and job-queue-blocked
65 * notifications) to the account's default EventBridge bus (source
66 * `aws.batch`); this subscribes the host Function to those events so it can
67 * alert on `FAILED` jobs or chain post-completion automation.
68 *
69 * AWS Batch publishes to EventBridge automatically — no additional resource
70 * is 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:** Alert On Failed Jobs
76 * ```typescript
77 * import * as AWS from "alchemy/AWS";
78 *
79 * export default AlertFunction.make(
80 * { main: import.meta.url },
81 * Effect.gen(function* () {
82 * yield* AWS.Batch.consumeJobEvents(
83 * { kinds: ["job-state"] },
84 * (events) =>
85 * Stream.runForEach(events, (event) =>
86 * event.detail.status === "FAILED"
87 * ? Effect.log(`batch job ${event.detail.jobId} failed`)
88 * : Effect.void,
89 * ),
90 * );
91 * return {};
92 * }).pipe(Effect.provide(AWS.Lambda.EventSource)),
93 * );
94 * ```
95 */
96export const consumeJobEvents = <StreamReq = never, Req = never>(
97 props: JobEventSourceProps,
98 process: (
99 events: Stream.Stream<BatchJobEvent, never, StreamReq>,
100 ) => Effect.Effect<void, never, Req>,
101) =>
102 consumeBusEvents(
103 props.id ?? "BatchJobEvents",
104 {
105 source: ["aws.batch"],
106 "detail-type": (
107 props.kinds ?? (Object.keys(DETAIL_TYPES) as BatchJobEventKind[])
108 ).map((kind) => DETAIL_TYPES[kind]),
109 },
110 { description: props.description, state: props.state },
111 process,
112 );