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 Data Exchange delivers to EventBridge. Revision
11 * events carry the published/revoked `RevisionIds`; product events carry
12 * `DataSetIds`; provider-generated notifications carry the notification
13 * comment and scope. Fields not shared by every event kind are optional
14 * (the schema grows over time).
15 */
16export interface DataSetEventDetail {
17 /** Revision events: the revisions that were published or revoked. */
18 RevisionIds?: string[];
19 /** Product events: the data sets that were added to or removed from a product. */
20 DataSetIds?: string[];
21 /** Revocation events: the provider's revocation comment. */
22 RevocationComment?: string;
23 /** Additional event fields (the schema grows over time). */
24 [key: string]: unknown;
25}
26
27/** An AWS Data Exchange EventBridge event delivered to the handler. */
28export type DataSetEvent = EventRecord<DataSetEventDetail>;
29
30/**
31 * Which AWS Data Exchange notifications to subscribe to. Each kind maps to
32 * one documented EventBridge detail-type (source `aws.dataexchange`).
33 */
34export type DataSetEventKind =
35 | "revision-published"
36 | "revision-revoked"
37 | "data-sets-published-to-product"
38 | "data-set-removed-from-product"
39 | "update-delayed"
40 | "data-updated"
41 | "deprecation-planned"
42 | "schema-change-planned"
43 | "auto-export-completed"
44 | "auto-export-failed"
45 | "data-grant-accepted"
46 | "data-grant-extended"
47 | "data-grant-revoked";
48
49const DETAIL_TYPES: Record<DataSetEventKind, string> = {
50 "revision-published": "Revision Published To Data Set",
51 "revision-revoked": "Revision Revoked",
52 "data-sets-published-to-product": "Data Sets Published To Product",
53 "data-set-removed-from-product": "Data Set Removed From Product",
54 "update-delayed": "Data Set Update Delayed",
55 "data-updated": "Data Updated in Data Set",
56 "deprecation-planned": "Deprecation Planned for Data Set",
57 "schema-change-planned": "Schema Change Planned for Data Set",
58 "auto-export-completed": "Auto-export Job Completed",
59 "auto-export-failed": "Auto-export Job Failed",
60 "data-grant-accepted": "Data Grant Accepted",
61 "data-grant-extended": "Data Grant Extended",
62 "data-grant-revoked": "Data Grant Revoked",
63};
64
65export interface DataSetEventSourceProps extends EventRouteProps {
66 /**
67 * Logical id for the backing EventBridge rule.
68 * @default "DataExchangeEvents"
69 */
70 id?: string;
71 /**
72 * Which notifications to subscribe to. Data Exchange publishes provider
73 * events (new revisions, revocations, product changes), provider-generated
74 * notifications (data updated/delayed, schema change, deprecation),
75 * auto-export job outcomes, and data grant lifecycle events.
76 * @default ["revision-published"]
77 */
78 kinds?: readonly DataSetEventKind[];
79 /**
80 * Raw EventBridge detail-types to subscribe to in addition to `kinds` —
81 * the escape hatch for the long tail of asset-type-specific detail-types
82 * (Redshift datashare, API Gateway API, S3 data access, Lake Formation).
83 */
84 detailTypes?: readonly string[];
85 /**
86 * Restrict to events about specific data sets (matched against the
87 * event's top-level `resources`, which carries the data set id).
88 */
89 dataSetIds?: readonly string[];
90}
91
92/**
93 * Event source connecting AWS Data Exchange notifications to the hosting
94 * compute. Data Exchange publishes subscriber-facing events to the
95 * account's default EventBridge bus (source `aws.dataexchange`) — most
96 * importantly `Revision Published To Data Set` when a provider publishes
97 * fresh data to an entitled data set — and this subscribes the host
98 * Function to those events so it can trigger export jobs or downstream
99 * pipelines the moment new data arrives.
100 *
101 * Data Exchange publishes to EventBridge automatically — no additional
102 * resource is created besides the EventBridge rule targeting the host.
103 * Provide the host-specific implementation layer (e.g.
104 * `AWS.Lambda.EventSource`) on the Function effect.
105 *
106 * ### Consuming Data Exchange Events
107 * **Example:** Process Newly Published Revisions
108 * ```typescript
109 * import * as AWS from "alchemy/AWS";
110 *
111 * export default IngestFunction.make(
112 * { main: import.meta.url },
113 * Effect.gen(function* () {
114 * yield* AWS.DataExchange.consumeDataSetEvents(
115 * { kinds: ["revision-published", "revision-revoked"] },
116 * (events) =>
117 * Stream.runForEach(events, (event) =>
118 * Effect.log(
119 * `${event["detail-type"]}: revisions ${event.detail.RevisionIds}`,
120 * ),
121 * ),
122 * );
123 * return {};
124 * }).pipe(Effect.provide(AWS.Lambda.EventSource)),
125 * );
126 * ```
127 */
128export const consumeDataSetEvents = <StreamReq = never, Req = never>(
129 props: DataSetEventSourceProps,
130 process: (
131 events: Stream.Stream<DataSetEvent, never, StreamReq>,
132 ) => Effect.Effect<void, never, Req>,
133) =>
134 consumeBusEvents(
135 props.id ?? "DataExchangeEvents",
136 {
137 source: ["aws.dataexchange"],
138 "detail-type": [
139 ...(props.kinds ?? (["revision-published"] as const)).map(
140 (kind) => DETAIL_TYPES[kind],
141 ),
142 ...(props.detailTypes ?? []),
143 ],
144 ...(props.dataSetIds !== undefined
145 ? { resources: [...props.dataSetIds] }
146 : {}),
147 },
148 { description: props.description, state: props.state },
149 process,
150 );