1import * as schemas from "@distilled.cloud/aws/schemas";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import * as Provider from "../../Provider.ts";
7import { Resource } from "../../Resource.ts";
8import { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
9import type { Providers } from "../Providers.ts";
10import { syncSchemasTags } from "./internal.ts";
11
12export type DiscovererState = "STARTED" | "STOPPED";
13
14export interface DiscovererProps {
15 /**
16 * The ARN of the event bus to discover schemas on. Changing it replaces
17 * the discoverer.
18 */
19 sourceArn: string;
20
21 /**
22 * A description of the discoverer.
23 */
24 description?: string;
25
26 /**
27 * Whether the discoverer also discovers schemas from events sent by other
28 * accounts.
29 * @default true
30 */
31 crossAccount?: boolean;
32
33 /**
34 * The desired state of the discoverer. A discoverer starts automatically
35 * on creation.
36 * @default "STARTED"
37 */
38 state?: DiscovererState;
39
40 /**
41 * User tags to attach to the discoverer.
42 */
43 tags?: Record<string, string>;
44}
45
46export interface Discoverer extends Resource<
47 "AWS.Schemas.Discoverer",
48 DiscovererProps,
49 {
50 /** The ID of the discoverer. */
51 discovererId: string;
52 /** The ARN of the discoverer. */
53 discovererArn: string;
54 /** The ARN of the event bus being discovered. */
55 sourceArn: string;
56 /** The current state of the discoverer. */
57 state: DiscovererState;
58 },
59 never,
60 Providers
61> {}
62
63/**
64 * An EventBridge schema discoverer — automatically infers schemas from the
65 * events flowing through an event bus and publishes them (versioned) to the
66 * AWS-managed `discovered-schemas` registry.
67 *
68 * ### Creating a Discoverer
69 * **Example:** Discover Schemas on an Event Bus
70 * ```typescript
71 * const bus = yield* AWS.EventBridge.EventBus("AppBus", {});
72 *
73 * const discoverer = yield* AWS.Schemas.Discoverer("AppDiscoverer", {
74 * sourceArn: bus.eventBusArn,
75 * description: "Discovers schemas for application events",
76 * });
77 * ```
78 *
79 * **Example:** Stopped Discoverer
80 * ```typescript
81 * const discoverer = yield* AWS.Schemas.Discoverer("PausedDiscoverer", {
82 * sourceArn: bus.eventBusArn,
83 * state: "STOPPED",
84 * });
85 * ```
86 * The discoverer is provisioned but paused; set `state: "STARTED"` (or omit
87 * it) to resume discovery.
88 *
89 * @resource
90 */
91export const Discoverer = Resource<Discoverer>("AWS.Schemas.Discoverer");
92
93export const DiscovererProvider = () =>
94 Provider.effect(
95 Discoverer,
96 Effect.gen(function* () {
97 const describe = (discovererId: string) =>
98 schemas
99 .describeDiscoverer({ DiscovererId: discovererId })
100 .pipe(
101 Effect.catchTag("NotFoundException", () =>
102 Effect.succeed(undefined),
103 ),
104 );
105
106 const toAttrs = (d: {
107 DiscovererId?: string;
108 DiscovererArn?: string;
109 SourceArn?: string;
110 State?: string;
111 }) => ({
112 discovererId: d.DiscovererId!,
113 discovererArn: d.DiscovererArn!,
114 sourceArn: d.SourceArn!,
115 state: (d.State ?? "STARTED") as DiscovererState,
116 });
117
118 return Discoverer.Provider.of({
119 stables: ["discovererId", "discovererArn", "sourceArn"],
120
121 list: () =>
122 schemas.listDiscoverers.pages({}).pipe(
123 Stream.runCollect,
124 Effect.map((chunk) =>
125 Array.from(chunk).flatMap((page) =>
126 (page.Discoverers ?? [])
127 .filter(
128 (d) =>
129 d.DiscovererId != null &&
130 d.DiscovererArn != null &&
131 d.SourceArn != null,
132 )
133 .map(toAttrs),
134 ),
135 ),
136 ),
137
138 read: Effect.fn(function* ({ id, output }) {
139 if (output?.discovererId) {
140 const found = yield* describe(output.discovererId);
141 if (!found) return undefined;
142 const attrs = toAttrs(found);
143 return (yield* hasAlchemyTags(id, found.Tags))
144 ? attrs
145 : Unowned(attrs);
146 }
147 // No cached id (the discoverer id is auto-assigned) — recover by
148 // scanning for a discoverer branded with our ownership tags.
149 const all = yield* schemas.listDiscoverers.items({}).pipe(
150 Stream.runCollect,
151 Effect.map((c) => Array.from(c)),
152 );
153 for (const d of all) {
154 if (
155 d.DiscovererId != null &&
156 d.DiscovererArn != null &&
157 d.SourceArn != null &&
158 (yield* hasAlchemyTags(id, d.Tags))
159 ) {
160 return toAttrs(d);
161 }
162 }
163 return undefined;
164 }),
165
166 diff: Effect.fn(function* ({ news, olds }) {
167 if (!isResolved(news)) return undefined;
168 if (news.sourceArn !== olds.sourceArn) {
169 return { action: "replace" } as const;
170 }
171 return undefined;
172 }),
173
174 reconcile: Effect.fn(function* ({ id, news, output, session }) {
175 const internalTags = yield* createInternalTags(id);
176
177 // OBSERVE — output.discovererId is only an id cache.
178 let live:
179 | {
180 Description?: string;
181 DiscovererArn?: string;
182 DiscovererId?: string;
183 SourceArn?: string;
184 State?: string;
185 CrossAccount?: boolean;
186 }
187 | undefined;
188 if (output?.discovererId) {
189 live = yield* describe(output.discovererId);
190 }
191
192 // ENSURE — create if missing. The id is auto-assigned, so there is
193 // no name race to tolerate.
194 if (live === undefined) {
195 live = yield* schemas.createDiscoverer({
196 SourceArn: news.sourceArn,
197 Description: news.description,
198 CrossAccount: news.crossAccount,
199 Tags: { ...news.tags, ...internalTags },
200 });
201 }
202 const discovererId = live.DiscovererId!;
203 const discovererArn = live.DiscovererArn!;
204
205 // SYNC description / crossAccount — diff observed against desired.
206 const desiredCrossAccount = news.crossAccount ?? true;
207 if (
208 (news.description ?? "") !== (live.Description ?? "") ||
209 desiredCrossAccount !== (live.CrossAccount ?? true)
210 ) {
211 live = yield* schemas.updateDiscoverer({
212 DiscovererId: discovererId,
213 Description: news.description ?? "",
214 CrossAccount: desiredCrossAccount,
215 });
216 }
217
218 // SYNC state — start/stop only on delta.
219 const desiredState: DiscovererState = news.state ?? "STARTED";
220 let state = (live.State ?? "STARTED") as DiscovererState;
221 if (state !== desiredState) {
222 if (desiredState === "STARTED") {
223 const r = yield* schemas.startDiscoverer({
224 DiscovererId: discovererId,
225 });
226 state = (r.State ?? "STARTED") as DiscovererState;
227 } else {
228 const r = yield* schemas.stopDiscoverer({
229 DiscovererId: discovererId,
230 });
231 state = (r.State ?? "STOPPED") as DiscovererState;
232 }
233 }
234
235 // SYNC tags — diff against observed cloud tags.
236 yield* syncSchemasTags(discovererArn, id, news.tags);
237
238 yield* session.note(discovererId);
239 return {
240 discovererId,
241 discovererArn,
242 sourceArn: live.SourceArn ?? news.sourceArn,
243 state,
244 };
245 }),
246
247 delete: Effect.fn(function* ({ output }) {
248 yield* schemas
249 .deleteDiscoverer({ DiscovererId: output.discovererId })
250 .pipe(Effect.catchTag("NotFoundException", () => Effect.void));
251 }),
252 });
253 }),
254 );