1import * as schemas from "@distilled.cloud/aws/schemas";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import * as Stream from "effect/Stream";
5import { Unowned } from "../../AdoptPolicy.ts";
6import { isResolved } from "../../Diff.ts";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import { hasAlchemyTags } from "../../Tags.ts";
11import {
12 normalizePolicyDocument,
13 stringifyPolicyDocument,
14 type PolicyDocument,
15} from "../IAM/Policy.ts";
16import type { Providers } from "../Providers.ts";
17import { syncSchemasTags } from "./internal.ts";
18
19export interface RegistryProps {
20 /**
21 * Name of the registry. Must match `[a-zA-Z0-9-_.@]+` and be at most 64
22 * characters. If omitted, a unique name is generated. Changing it replaces
23 * the registry.
24 */
25 registryName?: string;
26
27 /**
28 * A description of the registry.
29 */
30 description?: string;
31
32 /**
33 * User tags to attach to the registry.
34 */
35 tags?: Record<string, string>;
36
37 /**
38 * Resource-based policy attached to the registry, granting other AWS
39 * accounts or principals access to the registry and its schemas. Provided
40 * as a structured {@link PolicyDocument} or a raw JSON string. Omitting it
41 * removes any existing resource policy.
42 */
43 policy?: PolicyDocument | string;
44}
45
46export interface Registry extends Resource<
47 "AWS.Schemas.Registry",
48 RegistryProps,
49 {
50 /** The name of the registry. */
51 registryName: string;
52 /** The ARN of the registry. */
53 registryArn: string;
54 },
55 never,
56 Providers
57> {}
58
59/**
60 * An EventBridge Schema Registry — a named collection of event schemas.
61 * Custom registries hold your own OpenAPI 3 / JSONSchema Draft 4 schema
62 * documents; AWS also maintains the built-in `aws.events` and
63 * `discovered-schemas` registries.
64 *
65 * ### Creating a Registry
66 * **Example:** Basic Registry
67 * ```typescript
68 * const registry = yield* AWS.Schemas.Registry("app-events", {
69 * description: "Schemas for application events",
70 * });
71 * ```
72 *
73 * **Example:** Registry with Tags
74 * ```typescript
75 * const registry = yield* AWS.Schemas.Registry("orders", {
76 * description: "Order lifecycle events",
77 * tags: { team: "payments" },
78 * });
79 * ```
80 *
81 * ### Sharing a Registry
82 * **Example:** Registry with a Resource Policy
83 * ```typescript
84 * const registry = yield* AWS.Schemas.Registry("shared-events", {
85 * policy: {
86 * Version: "2012-10-17",
87 * Statement: [
88 * {
89 * Effect: "Allow",
90 * Principal: { AWS: `arn:aws:iam::${otherAccountId}:root` },
91 * Action: ["schemas:DescribeRegistry", "schemas:SearchSchemas"],
92 * Resource: registryArn,
93 * },
94 * ],
95 * },
96 * });
97 * ```
98 *
99 * ### Adding Schemas
100 * **Example:** Registry with a Schema
101 * ```typescript
102 * const registry = yield* AWS.Schemas.Registry("app-events", {});
103 * const schema = yield* AWS.Schemas.Schema("OrderCreated", {
104 * registryName: registry.registryName,
105 * type: "OpenApi3",
106 * content: JSON.stringify(openApiDocument),
107 * });
108 * ```
109 *
110 * @resource
111 */
112export const Registry = Resource<Registry>("AWS.Schemas.Registry");
113
114export const RegistryProvider = () =>
115 Provider.effect(
116 Registry,
117 Effect.gen(function* () {
118 const createName = Effect.fn(function* (
119 id: string,
120 props: RegistryProps,
121 ) {
122 return (
123 props.registryName ??
124 (yield* createPhysicalName({ id, maxLength: 64 }))
125 );
126 });
127
128 const describe = (registryName: string) =>
129 schemas
130 .describeRegistry({ RegistryName: registryName })
131 .pipe(
132 Effect.catchTag("NotFoundException", () =>
133 Effect.succeed(undefined),
134 ),
135 );
136
137 return Registry.Provider.of({
138 stables: ["registryName", "registryArn"],
139
140 // Top-level, account/region-scoped: enumerate every customer-owned
141 // registry. Scope LOCAL excludes the AWS-managed registries
142 // (aws.events, discovered-schemas).
143 list: () =>
144 schemas.listRegistries.pages({ Scope: "LOCAL" }).pipe(
145 Stream.runCollect,
146 Effect.map((chunk) =>
147 Array.from(chunk).flatMap((page) =>
148 (page.Registries ?? [])
149 .filter(
150 (r) => r.RegistryName != null && r.RegistryArn != null,
151 )
152 .map((r) => ({
153 registryName: r.RegistryName!,
154 registryArn: r.RegistryArn!,
155 })),
156 ),
157 ),
158 ),
159
160 read: Effect.fn(function* ({ id, olds, output }) {
161 const registryName =
162 output?.registryName ?? (yield* createName(id, olds ?? {}));
163 const found = yield* describe(registryName);
164 if (!found) return undefined;
165 const attrs = {
166 registryName,
167 registryArn: found.RegistryArn!,
168 };
169 return (yield* hasAlchemyTags(id, found.Tags))
170 ? attrs
171 : Unowned(attrs);
172 }),
173
174 diff: Effect.fn(function* ({ id, news, olds }) {
175 if (!isResolved(news)) return undefined;
176 const oldName = yield* createName(id, olds);
177 const newName = yield* createName(id, news);
178 if (oldName !== newName) return { action: "replace" } as const;
179 return undefined;
180 }),
181
182 reconcile: Effect.fn(function* ({ id, news, output, session }) {
183 const registryName =
184 output?.registryName ?? (yield* createName(id, news));
185
186 // OBSERVE
187 let live = yield* describe(registryName);
188
189 // ENSURE — tolerate the create race as "already exists".
190 if (live === undefined) {
191 yield* schemas
192 .createRegistry({
193 RegistryName: registryName,
194 Description: news.description,
195 })
196 .pipe(Effect.catchTag("ConflictException", () => Effect.void));
197 live = yield* schemas.describeRegistry({
198 RegistryName: registryName,
199 });
200 }
201
202 // SYNC description — diff observed against desired.
203 if ((news.description ?? "") !== (live.Description ?? "")) {
204 yield* schemas.updateRegistry({
205 RegistryName: registryName,
206 Description: news.description ?? "",
207 });
208 }
209
210 // SYNC resource policy — diff observed against desired, comparing
211 // canonicalized documents so a re-deploy of an equivalent policy
212 // (key order, whitespace) is a no-op.
213 const desiredPolicy =
214 news.policy === undefined
215 ? undefined
216 : typeof news.policy === "string"
217 ? news.policy
218 : stringifyPolicyDocument(news.policy);
219 const currentPolicy = yield* schemas
220 .getResourcePolicy({ RegistryName: registryName })
221 .pipe(
222 Effect.map((r) => (r.Policy ? r.Policy : undefined)),
223 Effect.catchTag("NotFoundException", () =>
224 Effect.succeed(undefined),
225 ),
226 );
227 const policyDrifted =
228 desiredPolicy === undefined || currentPolicy === undefined
229 ? desiredPolicy !== currentPolicy
230 : normalizePolicyDocument(currentPolicy) !==
231 normalizePolicyDocument(desiredPolicy);
232 if (policyDrifted) {
233 if (desiredPolicy === undefined) {
234 yield* schemas
235 .deleteResourcePolicy({ RegistryName: registryName })
236 .pipe(Effect.catchTag("NotFoundException", () => Effect.void));
237 } else {
238 yield* schemas.putResourcePolicy({
239 RegistryName: registryName,
240 Policy: desiredPolicy,
241 });
242 }
243 }
244
245 // SYNC tags — diff against observed cloud tags.
246 yield* syncSchemasTags(live.RegistryArn!, id, news.tags);
247
248 yield* session.note(registryName);
249 return {
250 registryName,
251 registryArn: live.RegistryArn!,
252 };
253 }),
254
255 delete: Effect.fn(function* ({ output, force }) {
256 const RegistryName = output.registryName;
257 if (force !== true) {
258 yield* schemas
259 .deleteRegistry({ RegistryName })
260 .pipe(Effect.catchTag("NotFoundException", () => Effect.void));
261 return;
262 }
263
264 const purgeAndDelete = Effect.gen(function* () {
265 // Schemas are registry-scoped and globally invisible to nuke.
266 // Only its explicit operator-confirmed force path may remove them
267 // or an attached registry policy.
268 const childSchemas = yield* schemas.listSchemas
269 .items({ RegistryName })
270 .pipe(
271 Stream.runCollect,
272 Effect.map((chunk) => Array.from(chunk)),
273 );
274 yield* Effect.forEach(
275 childSchemas,
276 (schema) =>
277 schema.SchemaName === undefined
278 ? Effect.void
279 : schemas
280 .deleteSchema({
281 RegistryName,
282 SchemaName: schema.SchemaName,
283 })
284 .pipe(
285 Effect.catchTag("NotFoundException", () => Effect.void),
286 ),
287 { concurrency: 4, discard: true },
288 );
289 yield* schemas
290 .deleteResourcePolicy({ RegistryName })
291 .pipe(Effect.catchTag("NotFoundException", () => Effect.void));
292 yield* schemas.deleteRegistry({ RegistryName });
293 });
294
295 yield* purgeAndDelete.pipe(
296 // Registry deletion can briefly observe schemas that have already
297 // accepted deletion. Re-list and retry only bounded transient or
298 // dependency-shaped failures.
299 Effect.retry({
300 while: (e) =>
301 e._tag === "BadRequestException" ||
302 e._tag === "InternalServerErrorException" ||
303 e._tag === "ServiceUnavailableException",
304 schedule: Schedule.max([
305 Schedule.exponential(500),
306 Schedule.recurs(8),
307 ]),
308 }),
309 Effect.catchTag("NotFoundException", () => Effect.void),
310 );
311 }),
312 });
313 }),
314 );