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 the EventBridge schema registry delivers when a
11 * schema is created or a new version is published. Fields are optional
12 * (the schema grows over time).
13 */
14export interface SchemaEventDetail {
15 /** The name of the schema the event is about. */
16 SchemaName?: string;
17 /** The type of the schema document, e.g. `OpenApi3`. */
18 SchemaType?: string;
19 /** The registry the schema belongs to, e.g. `discovered-schemas`. */
20 RegistryName?: string;
21 /** The version that was created. */
22 SchemaVersion?: string;
23 /** When the schema version was created. */
24 CreationDate?: string;
25 /** Additional event fields (the schema grows over time). */
26 [key: string]: unknown;
27}
28
29/** A schema-registry EventBridge event delivered to the handler. */
30export type SchemaEvent = EventRecord<SchemaEventDetail>;
31
32/** Which schema-registry notifications to subscribe to. */
33export type SchemaEventKind = "schema-created" | "schema-version-created";
34
35const DETAIL_TYPES: Record<SchemaEventKind, string> = {
36 "schema-created": "Schema Created",
37 "schema-version-created": "Schema Version Created",
38};
39
40export interface SchemaEventSourceProps extends EventRouteProps {
41 /**
42 * Logical id for the backing EventBridge rule.
43 * @default "SchemasEvents"
44 */
45 id?: string;
46 /**
47 * Which notifications to subscribe to: brand-new schemas and/or new
48 * versions of existing schemas (a schema's first version emits both).
49 * @default ["schema-created", "schema-version-created"]
50 */
51 kinds?: readonly SchemaEventKind[];
52 /**
53 * Restrict to events about schemas in specific registries (matched
54 * against the event's `detail.RegistryName`, e.g. `discovered-schemas`).
55 */
56 registryNames?: readonly string[];
57 /**
58 * Restrict to events about specific schemas (matched against the event's
59 * top-level `resources`, which includes the schema's ARN).
60 */
61 schemaArns?: readonly string[];
62}
63
64/**
65 * Event source connecting EventBridge schema-registry notifications to the
66 * hosting compute. The registry publishes `Schema Created` and
67 * `Schema Version Created` events to the account's default bus (source
68 * `aws.schemas`) whenever a schema is registered — including schemas the
69 * discoverer infers from live traffic — so a function can react to new or
70 * drifting event contracts (regenerate types, notify consumers, run
71 * compatibility checks).
72 *
73 * The registry publishes to EventBridge automatically — no additional
74 * resource is created besides the EventBridge rule targeting the host.
75 * Provide the host-specific implementation layer (e.g.
76 * `AWS.Lambda.EventSource`) on the Function effect.
77 *
78 * ### Consuming Schema Registry Events
79 * **Example:** React To New Schema Versions
80 * ```typescript
81 * import * as AWS from "alchemy/AWS";
82 *
83 * export default ContractWatcher.make(
84 * { main: import.meta.url },
85 * Effect.gen(function* () {
86 * yield* AWS.Schemas.consumeSchemaEvents(
87 * { kinds: ["schema-version-created"] },
88 * (events) =>
89 * Stream.runForEach(events, (event) =>
90 * Effect.log(
91 * `schema ${event.detail.RegistryName}/${event.detail.SchemaName} ` +
92 * `published version ${event.detail.SchemaVersion}`,
93 * ),
94 * ),
95 * );
96 * return {};
97 * }).pipe(Effect.provide(AWS.Lambda.EventSource)),
98 * );
99 * ```
100 */
101export const consumeSchemaEvents = <StreamReq = never, Req = never>(
102 props: SchemaEventSourceProps,
103 process: (
104 events: Stream.Stream<SchemaEvent, never, StreamReq>,
105 ) => Effect.Effect<void, never, Req>,
106) =>
107 consumeBusEvents(
108 props.id ?? "SchemasEvents",
109 {
110 source: ["aws.schemas"],
111 "detail-type": (
112 props.kinds ?? (["schema-created", "schema-version-created"] as const)
113 ).map((kind) => DETAIL_TYPES[kind]),
114 ...(props.registryNames !== undefined
115 ? { detail: { RegistryName: [...props.registryNames] } }
116 : {}),
117 ...(props.schemaArns !== undefined
118 ? { resources: [...props.schemaArns] }
119 : {}),
120 },
121 { description: props.description, state: props.state },
122 process,
123 );