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 of a `Glue Crawler State Change` EventBridge event.
11 * Emitted on the account's default bus (source `aws.glue`) when a crawl
12 * starts, succeeds, or fails.
13 */
14export interface CrawlerEventDetail {
15 /** The name of the crawler. */
16 crawlerName?: string;
17 /** The state the crawl transitioned to. */
18 state?: "Started" | "Succeeded" | "Failed" | (string & {});
19 /** A human-readable message (e.g. tables created/updated counts). */
20 message?: string;
21 /** Additional event fields (the schema grows over time). */
22 [key: string]: unknown;
23}
24
25/** A Glue crawler state-change EventBridge event delivered to the handler. */
26export type CrawlerEvent = EventRecord<CrawlerEventDetail>;
27
28export interface CrawlerEventSourceProps extends EventRouteProps {
29 /**
30 * Logical id for the backing EventBridge rule.
31 * @default "GlueCrawlerEvents"
32 */
33 id?: string;
34 /**
35 * Restrict to specific crawlers (matched against `detail.crawlerName`).
36 * Omit to receive events for every crawler in the account.
37 */
38 crawlerNames?: readonly string[];
39 /**
40 * Restrict to specific states (matched against `detail.state`).
41 * @default all states
42 */
43 states?: readonly ("Started" | "Succeeded" | "Failed")[];
44}
45
46/**
47 * Event source connecting Glue crawler state changes to the hosting
48 * compute. Glue publishes `Glue Crawler State Change` events to the
49 * account's default EventBridge bus (source `aws.glue`) when a crawl
50 * starts, succeeds, or fails; this subscribes the host Function to those
51 * events so it can kick off downstream processing once the Data Catalog is
52 * refreshed.
53 *
54 * Glue publishes to EventBridge automatically — no additional resource is
55 * created besides the EventBridge rule targeting the host. Provide the
56 * host-specific implementation layer (e.g. `AWS.Lambda.EventSource`) on the
57 * Function effect.
58 *
59 * ### Consuming Crawler Events
60 * **Example:** Run Downstream Work After a Crawl
61 * ```typescript
62 * import * as AWS from "alchemy/AWS";
63 *
64 * export default PipelineFunction.make(
65 * { main: import.meta.url },
66 * Effect.gen(function* () {
67 * yield* AWS.Glue.consumeCrawlerEvents(
68 * { states: ["Succeeded"] },
69 * (events) =>
70 * Stream.runForEach(events, (event) =>
71 * Effect.log(`catalog refreshed by ${event.detail.crawlerName}`),
72 * ),
73 * );
74 * return {};
75 * }).pipe(Effect.provide(AWS.Lambda.EventSource)),
76 * );
77 * ```
78 */
79export const consumeCrawlerEvents = <StreamReq = never, Req = never>(
80 props: CrawlerEventSourceProps,
81 process: (
82 events: Stream.Stream<CrawlerEvent, never, StreamReq>,
83 ) => Effect.Effect<void, never, Req>,
84) =>
85 consumeBusEvents(
86 props.id ?? "GlueCrawlerEvents",
87 {
88 source: ["aws.glue"],
89 "detail-type": ["Glue Crawler State Change"],
90 ...(props.crawlerNames !== undefined || props.states !== undefined
91 ? {
92 detail: {
93 ...(props.crawlerNames !== undefined
94 ? { crawlerName: [...props.crawlerNames] }
95 : {}),
96 ...(props.states !== undefined
97 ? { state: [...props.states] }
98 : {}),
99 },
100 }
101 : {}),
102 },
103 { description: props.description, state: props.state },
104 process,
105 );