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 Deadline Cloud delivers to EventBridge. Every
11 * event kind identifies the farm and the affected resource (queue, job,
12 * step, task, fleet, worker, or budget); status-change events carry the
13 * previous and new status. Fields not shared by every event kind are
14 * optional (the schema grows over time).
15 */
16export interface FarmEventDetail {
17 /** The farm the event is about. */
18 farmId?: string;
19 /** Job/step/task and budget events: the queue the resource belongs to. */
20 queueId?: string;
21 /** Job/step/task events: the job the event is about. */
22 jobId?: string;
23 /** Step/task events: the step the event is about. */
24 stepId?: string;
25 /** Task events: the task the event is about. */
26 taskId?: string;
27 /** Fleet and worker events: the fleet the event is about. */
28 fleetId?: string;
29 /** Worker events: the worker the event is about. */
30 workerId?: string;
31 /** Budget Threshold Reached: the budget that crossed a threshold. */
32 budgetId?: string;
33 /** Status-change events: the status before the transition. */
34 previousStatus?: string;
35 /** Status-change events: the status after the transition. */
36 status?: string;
37 /** Additional event fields (the schema grows over time). */
38 [key: string]: unknown;
39}
40
41/** A Deadline Cloud EventBridge event delivered to the handler. */
42export type FarmEvent = EventRecord<FarmEventDetail>;
43
44/** Which Deadline Cloud notifications to subscribe to. */
45export type FarmEventKind =
46 | "job-lifecycle"
47 | "job-run"
48 | "step-lifecycle"
49 | "step-run"
50 | "task-run"
51 | "budget-threshold"
52 | "fleet-size-recommendation"
53 | "worker-unhealthy";
54
55const DETAIL_TYPES: Record<FarmEventKind, string> = {
56 "job-lifecycle": "Job Lifecycle Status Change",
57 "job-run": "Job Run Status Change",
58 "step-lifecycle": "Step Lifecycle Status Change",
59 "step-run": "Step Run Status Change",
60 "task-run": "Task Run Status Change",
61 "budget-threshold": "Budget Threshold Reached",
62 "fleet-size-recommendation": "Fleet Size Recommendation Change",
63 "worker-unhealthy": "Worker Status Unhealthy",
64};
65
66export interface FarmEventSourceProps extends EventRouteProps {
67 /**
68 * Logical id for the backing EventBridge rule.
69 * @default "DeadlineFarmEvents"
70 */
71 id?: string;
72 /**
73 * Which notifications to subscribe to: job/step lifecycle and run status
74 * changes, task run status changes, budget threshold crossings, fleet
75 * size recommendations, and unhealthy-worker reports.
76 * @default ["job-run"]
77 */
78 kinds?: readonly FarmEventKind[];
79 /**
80 * Restrict to events about specific farms (matched against the event
81 * detail's `farmId`).
82 */
83 farmIds?: readonly string[];
84}
85
86/**
87 * Event source connecting AWS Deadline Cloud notifications to the hosting
88 * compute. Deadline Cloud publishes job/step/task status changes, budget
89 * threshold crossings, fleet size recommendations, and unhealthy-worker
90 * reports to the account's default EventBridge bus (source `aws.deadline`);
91 * this subscribes the host Function to those events so it can react to
92 * finished renders or runaway spend.
93 *
94 * Deadline Cloud publishes to EventBridge automatically — no additional
95 * resource is created besides the EventBridge rule targeting the host.
96 * Provide the host-specific implementation layer (e.g.
97 * `AWS.Lambda.EventSource`) on the Function effect.
98 *
99 * ### Consuming Farm Events
100 * **Example:** Alert When A Job Finishes
101 * ```typescript
102 * import * as AWS from "alchemy/AWS";
103 *
104 * export default RenderAlertFunction.make(
105 * { main: import.meta.url },
106 * Effect.gen(function* () {
107 * yield* AWS.Deadline.consumeFarmEvents(
108 * { kinds: ["job-run"] },
109 * (events) =>
110 * Stream.runForEach(events, (event) =>
111 * event.detail.status === "SUCCEEDED"
112 * ? Effect.log(`render ${event.detail.jobId} finished`)
113 * : Effect.void,
114 * ),
115 * );
116 * return {};
117 * }).pipe(Effect.provide(AWS.Lambda.EventSource)),
118 * );
119 * ```
120 *
121 * **Example:** Stop The Farm On A Budget Threshold
122 * ```typescript
123 * yield* AWS.Deadline.consumeFarmEvents(
124 * { kinds: ["budget-threshold"], farmIds: [farm.farmId] },
125 * (events) =>
126 * Stream.runForEach(events, (event) =>
127 * Effect.logWarning(`budget ${event.detail.budgetId} threshold hit`),
128 * ),
129 * );
130 * ```
131 */
132export const consumeFarmEvents = <StreamReq = never, Req = never>(
133 props: FarmEventSourceProps,
134 process: (
135 events: Stream.Stream<FarmEvent, never, StreamReq>,
136 ) => Effect.Effect<void, never, Req>,
137) =>
138 consumeBusEvents(
139 props.id ?? "DeadlineFarmEvents",
140 {
141 source: ["aws.deadline"],
142 "detail-type": (props.kinds ?? (["job-run"] as const)).map(
143 (kind) => DETAIL_TYPES[kind],
144 ),
145 ...(props.farmIds !== undefined
146 ? { detail: { farmId: [...props.farmIds] } }
147 : {}),
148 },
149 { description: props.description, state: props.state },
150 process,
151 );