1import * as kafka from "@distilled.cloud/gcp/managedkafka_v1";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import * as Provider from "../../Provider.ts";
7import { Resource } from "../../Resource.ts";
8import { GcpEnvironment } from "../Environment.ts";
9import type { Providers } from "../Providers.ts";
10import {
11 expandParent,
12 getContextSchemaVersion,
13 hasAlchemyLabelMap,
14 listSchemaRegistries,
15 normalizeLocation,
16 OWNERSHIP_SUBJECT,
17 parseHttpJson,
18 parseName,
19 replaceOnIdentity,
20 retryTransient,
21 schemaRegistryOwnership,
22 subjectParent,
23 toSubjectId,
24 versionName,
25} from "./internal.ts";
26
27export type ContextSchemaReference = {
28 /** Name of the referenced schema. */
29 name?: string;
30 /** Subject of the referenced schema. */
31 subject?: string;
32 /** Version of the referenced schema. */
33 version?: number;
34};
35
36export type SchemaRegistriesContextsSubjectsVersionProps = {
37 /**
38 * Parent schema registry. Full name
39 * `projects/{project}/locations/{location}/schemaRegistries/{id}` or
40 * the registry id (combined with `location`). Immutable — changing it
41 * replaces the version.
42 */
43 schemaRegistry: string;
44 /**
45 * Schema context. Immutable — changing it replaces the version.
46 */
47 context: string;
48 /**
49 * Region used when `schemaRegistry` is a bare id.
50 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
51 */
52 location?: string;
53 /**
54 * Subject name. If omitted, a unique name is generated. Immutable —
55 * changing it replaces the version.
56 */
57 subject?: string;
58 /**
59 * Version number. If omitted, the server assigns the next id.
60 * Immutable — changing it replaces the version.
61 */
62 version?: number;
63 /**
64 * Schema payload. Immutable — changing it replaces the version.
65 */
66 schema: string;
67 /**
68 * Schema type.
69 * @default "AVRO"
70 */
71 schemaType?: kafka.CreateVersionRequestSchemaTypeEnum | (string & {});
72 /**
73 * Normalize the schema before storing.
74 * @default false
75 */
76 normalize?: boolean;
77 /**
78 * Explicit schema id. Must not collide with a different schema.
79 */
80 id?: number;
81 /**
82 * Schema references.
83 */
84 references?: ContextSchemaReference[];
85};
86
87export type SchemaRegistriesContextsSubjectsVersion = Resource<
88 "GCP.ManagedKafka.SchemaRegistriesContextsSubjectsVersion",
89 SchemaRegistriesContextsSubjectsVersionProps,
90 {
91 /** Full resource name `.../contexts/{context}/subjects/{subject}/versions/{version}`. */
92 name: string;
93 /** Parent schema registry resource name. */
94 schemaRegistry: string;
95 /** Context id. */
96 context: string;
97 /** Project id. */
98 project: string;
99 /** Location id. */
100 location: string;
101 /** Subject name. */
102 subject: string;
103 /** Version number. */
104 version: number;
105 /** Schema id. */
106 schemaId: number | undefined;
107 /** Schema payload. */
108 schema: string | undefined;
109 /** Schema type. */
110 schemaType: string | undefined;
111 /** Schema references. */
112 references: ContextSchemaReference[];
113 },
114 never,
115 Providers
116>;
117
118/**
119 * A schema version under a subject in a named context of a Managed Kafka
120 * schema registry.
121 *
122 * Versions are immutable. Changing context, subject, registry, schema,
123 * type, or version replaces the resource.
124 *
125 * ### Creating a Version
126 * **Example:** Avro record in a context
127 * ```typescript
128 * const version =
129 * yield* GCP.ManagedKafka.SchemaRegistriesContextsSubjectsVersion(
130 * "OrderSchema",
131 * {
132 * schemaRegistry: registry.name,
133 * context: "prod",
134 * subject: "orders",
135 * schemaType: "AVRO",
136 * schema: JSON.stringify({
137 * type: "record",
138 * name: "Order",
139 * fields: [{ name: "id", type: "string" }],
140 * }),
141 * },
142 * );
143 * ```
144 *
145 * @resource
146 * @category ManagedKafka
147 */
148export const SchemaRegistriesContextsSubjectsVersion =
149 Resource<SchemaRegistriesContextsSubjectsVersion>(
150 "GCP.ManagedKafka.SchemaRegistriesContextsSubjectsVersion",
151 );
152
153export class SchemaRegistriesContextsSubjectsVersionNotResolved extends Data.TaggedError(
154 "GCP.ManagedKafka.SchemaRegistriesContextsSubjectsVersionNotResolved",
155)<{
156 name: string;
157}> {}
158
159const registryOf = (value: string, project: string, location: string) =>
160 expandParent(value, project, location, "schemaRegistries");
161
162/**
163 * Context-scoped subjects come back qualified as `:.{context}:{subject}`;
164 * resource paths and attributes use the bare subject.
165 */
166const unqualifySubject = (subject: string | undefined, context: string) =>
167 subject?.replace(`:.${context}:`, "");
168
169const toAttrs = (
170 version: kafka.SchemaVersion,
171 schemaRegistry: string,
172 context: string,
173 project: string,
174 location: string,
175 name: string,
176) => {
177 const parsed = parseName(schemaRegistry, "schemaRegistries");
178 return {
179 name,
180 schemaRegistry,
181 context,
182 project: parsed.project || project,
183 location: parsed.location || location,
184 subject: unqualifySubject(version.subject, context) ?? "",
185 version: version.version ?? 0,
186 schemaId: version.id,
187 schema: version.schema,
188 schemaType: version.schemaType,
189 references: version.references ?? [],
190 };
191};
192
193const parseVersionList = (value: unknown): number[] => {
194 if (!Array.isArray(value)) return [];
195 return value
196 .map((item) => {
197 if (typeof item === "number") return item;
198 if (typeof item === "string") return Number(item);
199 if (item && typeof item === "object" && "version" in item) {
200 return Number((item as { version?: unknown }).version);
201 }
202 return Number.NaN;
203 })
204 .filter((item) => Number.isFinite(item));
205};
206
207export const SchemaRegistriesContextsSubjectsVersionProvider = () =>
208 Provider.succeed(SchemaRegistriesContextsSubjectsVersion, {
209 stables: [
210 "name",
211 "schemaRegistry",
212 "context",
213 "project",
214 "location",
215 "subject",
216 "version",
217 "schemaId",
218 ],
219
220 diff: Effect.fn(function* ({ news, olds, output }) {
221 if (!isResolved(news)) return undefined;
222 const env = yield* GcpEnvironment.current;
223 const location = normalizeLocation(
224 news.location ?? output?.location,
225 env.region,
226 );
227 const schemaChanged =
228 (olds?.schema ?? output?.schema ?? "") !== news.schema ||
229 (olds?.schemaType ?? output?.schemaType ?? "") !==
230 (news.schemaType ?? olds?.schemaType ?? output?.schemaType ?? "") ||
231 (olds?.context ?? output?.context ?? "") !== news.context;
232 return replaceOnIdentity({
233 previousId:
234 olds?.subject !== undefined && olds.version !== undefined
235 ? `${olds.context}/${olds.subject}/${olds.version}`
236 : output !== undefined
237 ? `${output.context}/${output.subject}/${output.version}`
238 : undefined,
239 nextId:
240 news.subject !== undefined && news.version !== undefined
241 ? `${news.context}/${news.subject}/${news.version}`
242 : undefined,
243 previousParent: olds?.schemaRegistry ?? output?.schemaRegistry,
244 nextParent: registryOf(news.schemaRegistry, env.project, location),
245 extra: schemaChanged,
246 });
247 }),
248
249 read: Effect.fn(function* ({ id, olds, output }) {
250 const env = yield* GcpEnvironment.current;
251 const location = normalizeLocation(
252 olds?.location ?? output?.location,
253 env.region,
254 );
255 const subject = yield* toSubjectId(id, olds?.subject, output?.subject);
256 const context = olds?.context ?? output?.context ?? "";
257 const schemaRegistry =
258 olds?.schemaRegistry !== undefined
259 ? registryOf(olds.schemaRegistry, env.project, location)
260 : (output?.schemaRegistry ?? "");
261 const version = olds?.version ?? output?.version ?? "latest";
262 const name =
263 output?.name ??
264 (schemaRegistry.length > 0 && context.length > 0
265 ? versionName(
266 subjectParent(schemaRegistry, subject, context),
267 version,
268 )
269 : "");
270 const existing = yield* getContextSchemaVersion(name);
271 if (existing === undefined) return undefined;
272 const resolvedName = versionName(
273 subjectParent(schemaRegistry, existing.subject ?? subject, context),
274 existing.version ?? version,
275 );
276 const attrs = toAttrs(
277 existing,
278 schemaRegistry,
279 context,
280 env.project,
281 location,
282 resolvedName,
283 );
284 return output !== undefined || olds !== undefined
285 ? attrs
286 : Unowned(attrs);
287 }),
288
289 list: () =>
290 Effect.gen(function* () {
291 const env = yield* GcpEnvironment.current;
292 const registries = yield* listSchemaRegistries(env.project, env.region);
293 const owned = yield* Effect.forEach(
294 registries.filter((registry) => (registry.name ?? "").length > 0),
295 (registry) =>
296 schemaRegistryOwnership(registry.name!).pipe(
297 Effect.map((labels) =>
298 hasAlchemyLabelMap(labels) ? registry : undefined,
299 ),
300 ),
301 { concurrency: 4 },
302 );
303 const versions = yield* Effect.forEach(
304 owned.filter(
305 (registry): registry is kafka.SchemaRegistry =>
306 registry !== undefined,
307 ),
308 (registry) =>
309 Effect.gen(function* () {
310 const contexts = (registry.contexts ?? []).filter(
311 (context) => context.length > 0 && context !== ".",
312 );
313 return yield* Effect.forEach(
314 contexts,
315 (context) =>
316 Effect.gen(function* () {
317 const contextName = context.includes("/")
318 ? context
319 : `${registry.name}/contexts/${context}`;
320 const body = yield* kafka
321 .listProjectsLocationsSchemaRegistriesContextsSubjects({
322 parent: contextName,
323 })
324 .pipe(
325 Effect.catchTag(
326 ["NotFound", "SchemaRegistryRequiresCluster"],
327 () => Effect.succeed<kafka.HttpBody>({}),
328 ),
329 );
330 const subjects = (yield* parseHttpJson(body)) as unknown;
331 const names = Array.isArray(subjects)
332 ? subjects.filter(
333 (item): item is string =>
334 typeof item === "string" &&
335 item !== OWNERSHIP_SUBJECT,
336 )
337 : [];
338 const contextId = parseName(contextName, "contexts").id;
339 return yield* Effect.forEach(
340 names,
341 (subject) =>
342 Effect.gen(function* () {
343 const listed = yield* kafka
344 .listProjectsLocationsSchemaRegistriesContextsSubjectsVersions(
345 {
346 parent: subjectParent(
347 registry.name!,
348 subject,
349 contextId,
350 ),
351 },
352 )
353 .pipe(
354 Effect.catchTag(
355 ["NotFound", "SchemaRegistryRequiresCluster"],
356 () => Effect.succeed<kafka.HttpBody>({}),
357 ),
358 );
359 const ids = parseVersionList(
360 yield* parseHttpJson(listed),
361 );
362 return yield* Effect.forEach(
363 ids,
364 (version) =>
365 getContextSchemaVersion(
366 versionName(
367 subjectParent(
368 registry.name!,
369 subject,
370 contextId,
371 ),
372 version,
373 ),
374 ).pipe(
375 Effect.map((item) =>
376 item
377 ? toAttrs(
378 item,
379 registry.name!,
380 contextId,
381 env.project,
382 parseName(
383 registry.name!,
384 "schemaRegistries",
385 ).location,
386 versionName(
387 subjectParent(
388 registry.name!,
389 subject,
390 contextId,
391 ),
392 version,
393 ),
394 )
395 : undefined,
396 ),
397 ),
398 { concurrency: 4 },
399 );
400 }),
401 { concurrency: 4 },
402 );
403 }),
404 { concurrency: 2 },
405 );
406 }),
407 { concurrency: 2 },
408 );
409 return versions
410 .flat(3)
411 .filter(
412 (item): item is NonNullable<typeof item> => item !== undefined,
413 );
414 }),
415
416 reconcile: Effect.fn(function* ({ id, news, output }) {
417 const env = yield* GcpEnvironment.current;
418 const location = normalizeLocation(
419 news.location ?? output?.location,
420 env.region,
421 );
422 const schemaRegistry = registryOf(
423 news.schemaRegistry,
424 env.project,
425 location,
426 );
427 const subject = yield* toSubjectId(id, news.subject, output?.subject);
428 const context = news.context;
429 const parent = subjectParent(schemaRegistry, subject, context);
430 const versionHint = news.version ?? output?.version;
431 const name = output?.name ?? versionName(parent, versionHint ?? "latest");
432
433 let current = yield* getContextSchemaVersion(name);
434
435 if (current === undefined) {
436 const created = yield* retryTransient(
437 kafka
438 .createProjectsLocationsSchemaRegistriesContextsSubjectsVersions({
439 parent,
440 body: {
441 schemaType: news.schemaType,
442 version: news.version,
443 normalize: news.normalize,
444 id: news.id,
445 schema: news.schema,
446 references: news.references,
447 },
448 })
449 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined))),
450 );
451 const lookedUp = yield* kafka
452 .lookupVersionProjectsLocationsSchemaRegistriesContextsSubjects({
453 parent,
454 body: {
455 schema: news.schema,
456 schemaType: news.schemaType,
457 normalize: news.normalize,
458 references: news.references,
459 },
460 })
461 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
462 current =
463 lookedUp ??
464 (yield* getContextSchemaVersion(
465 versionName(parent, news.version ?? "latest"),
466 ));
467 void created;
468 }
469
470 if (current === undefined) {
471 return yield* new SchemaRegistriesContextsSubjectsVersionNotResolved({
472 name,
473 });
474 }
475
476 const resolved = versionName(
477 subjectParent(
478 schemaRegistry,
479 unqualifySubject(current.subject, context) ?? subject,
480 context,
481 ),
482 current.version ?? versionHint ?? "latest",
483 );
484 return toAttrs(
485 current,
486 schemaRegistry,
487 context,
488 env.project,
489 location,
490 resolved,
491 );
492 }),
493
494 delete: Effect.fn(function* ({ output }) {
495 yield* kafka
496 .deleteProjectsLocationsSchemaRegistriesContextsSubjectsVersions({
497 name: output.name,
498 permanent: false,
499 })
500 .pipe(
501 Effect.catchTag(
502 [
503 "NotFound",
504 "SchemaRegistryRequiresCluster",
505 "SchemaRegistryPathNotFound",
506 ],
507 () => Effect.void,
508 ),
509 );
510 yield* kafka
511 .deleteProjectsLocationsSchemaRegistriesContextsSubjectsVersions({
512 name: output.name,
513 permanent: true,
514 })
515 .pipe(
516 Effect.catchTag(
517 [
518 "NotFound",
519 "SchemaRegistryRequiresCluster",
520 "SchemaRegistryPathNotFound",
521 ],
522 () => Effect.void,
523 ),
524 );
525 }),
526 });