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 getSchemaVersion,
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 SchemaReference = {
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 SchemaRegistriesSubjectsVersionProps = {
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 * Region used when `schemaRegistry` is a bare id.
46 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
47 */
48 location?: string;
49 /**
50 * Subject name. If omitted, a unique name is generated. Immutable —
51 * changing it replaces the version.
52 */
53 subject?: string;
54 /**
55 * Version number. If omitted, the server assigns the next id.
56 * Immutable — changing it replaces the version.
57 */
58 version?: number;
59 /**
60 * Schema payload. Immutable — changing it replaces the version.
61 */
62 schema: string;
63 /**
64 * Schema type.
65 * @default "AVRO"
66 */
67 schemaType?: kafka.CreateVersionRequestSchemaTypeEnum | (string & {});
68 /**
69 * Normalize the schema before storing.
70 * @default false
71 */
72 normalize?: boolean;
73 /**
74 * Explicit schema id. Must not collide with a different schema.
75 */
76 id?: number;
77 /**
78 * Schema references.
79 */
80 references?: SchemaReference[];
81};
82
83export type SchemaRegistriesSubjectsVersion = Resource<
84 "GCP.ManagedKafka.SchemaRegistriesSubjectsVersion",
85 SchemaRegistriesSubjectsVersionProps,
86 {
87 /** Full resource name `.../subjects/{subject}/versions/{version}`. */
88 name: string;
89 /** Parent schema registry resource name. */
90 schemaRegistry: string;
91 /** Project id. */
92 project: string;
93 /** Location id. */
94 location: string;
95 /** Subject name. */
96 subject: string;
97 /** Version number. */
98 version: number;
99 /** Schema id. */
100 schemaId: number | undefined;
101 /** Schema payload. */
102 schema: string | undefined;
103 /** Schema type. */
104 schemaType: string | undefined;
105 /** Schema references. */
106 references: SchemaReference[];
107 },
108 never,
109 Providers
110>;
111
112/**
113 * A schema version under a subject in a Managed Kafka schema registry
114 * (default context).
115 *
116 * Versions are immutable. Changing subject, registry, schema, type, or
117 * version replaces the resource.
118 *
119 * ### Creating a Version
120 * **Example:** Avro record
121 * ```typescript
122 * const version = yield* GCP.ManagedKafka.SchemaRegistriesSubjectsVersion(
123 * "OrderSchema",
124 * {
125 * schemaRegistry: registry.name,
126 * subject: "orders",
127 * schemaType: "AVRO",
128 * schema: JSON.stringify({
129 * type: "record",
130 * name: "Order",
131 * fields: [{ name: "id", type: "string" }],
132 * }),
133 * },
134 * );
135 * ```
136 *
137 * @resource
138 * @category ManagedKafka
139 */
140export const SchemaRegistriesSubjectsVersion =
141 Resource<SchemaRegistriesSubjectsVersion>(
142 "GCP.ManagedKafka.SchemaRegistriesSubjectsVersion",
143 );
144
145export class SchemaRegistriesSubjectsVersionNotResolved extends Data.TaggedError(
146 "GCP.ManagedKafka.SchemaRegistriesSubjectsVersionNotResolved",
147)<{
148 name: string;
149}> {}
150
151const registryOf = (value: string, project: string, location: string) =>
152 expandParent(value, project, location, "schemaRegistries");
153
154const toAttrs = (
155 version: kafka.SchemaVersion,
156 schemaRegistry: string,
157 project: string,
158 location: string,
159 name: string,
160) => {
161 const parsed = parseName(schemaRegistry, "schemaRegistries");
162 return {
163 name,
164 schemaRegistry,
165 project: parsed.project || project,
166 location: parsed.location || location,
167 subject: version.subject ?? "",
168 version: version.version ?? 0,
169 schemaId: version.id,
170 schema: version.schema,
171 schemaType: version.schemaType,
172 references: version.references ?? [],
173 };
174};
175
176const parseVersionList = (value: unknown): number[] => {
177 if (!Array.isArray(value)) return [];
178 return value
179 .map((item) => {
180 if (typeof item === "number") return item;
181 if (typeof item === "string") return Number(item);
182 if (item && typeof item === "object" && "version" in item) {
183 return Number((item as { version?: unknown }).version);
184 }
185 return Number.NaN;
186 })
187 .filter((item) => Number.isFinite(item));
188};
189
190export const SchemaRegistriesSubjectsVersionProvider = () =>
191 Provider.succeed(SchemaRegistriesSubjectsVersion, {
192 stables: [
193 "name",
194 "schemaRegistry",
195 "project",
196 "location",
197 "subject",
198 "version",
199 "schemaId",
200 ],
201
202 diff: Effect.fn(function* ({ news, olds, output }) {
203 if (!isResolved(news)) return undefined;
204 const env = yield* GcpEnvironment.current;
205 const location = normalizeLocation(
206 news.location ?? output?.location,
207 env.region,
208 );
209 const schemaChanged =
210 (olds?.schema ?? output?.schema ?? "") !== news.schema ||
211 (olds?.schemaType ?? output?.schemaType ?? "") !==
212 (news.schemaType ?? olds?.schemaType ?? output?.schemaType ?? "");
213 return replaceOnIdentity({
214 previousId:
215 olds?.subject !== undefined && olds.version !== undefined
216 ? `${olds.subject}/${olds.version}`
217 : output !== undefined
218 ? `${output.subject}/${output.version}`
219 : undefined,
220 nextId:
221 news.subject !== undefined && news.version !== undefined
222 ? `${news.subject}/${news.version}`
223 : undefined,
224 previousParent: olds?.schemaRegistry ?? output?.schemaRegistry,
225 nextParent: registryOf(news.schemaRegistry, env.project, location),
226 extra: schemaChanged,
227 });
228 }),
229
230 read: Effect.fn(function* ({ id, olds, output }) {
231 const env = yield* GcpEnvironment.current;
232 const location = normalizeLocation(
233 olds?.location ?? output?.location,
234 env.region,
235 );
236 const subject = yield* toSubjectId(id, olds?.subject, output?.subject);
237 const schemaRegistry =
238 olds?.schemaRegistry !== undefined
239 ? registryOf(olds.schemaRegistry, env.project, location)
240 : (output?.schemaRegistry ?? "");
241 const version = olds?.version ?? output?.version ?? "latest";
242 const name =
243 output?.name ??
244 (schemaRegistry.length > 0
245 ? versionName(subjectParent(schemaRegistry, subject), version)
246 : "");
247 const existing = yield* getSchemaVersion(name);
248 if (existing === undefined) return undefined;
249 const resolvedName = versionName(
250 subjectParent(schemaRegistry, existing.subject ?? subject),
251 existing.version ?? version,
252 );
253 const attrs = toAttrs(
254 existing,
255 schemaRegistry,
256 env.project,
257 location,
258 resolvedName,
259 );
260 return output !== undefined || olds !== undefined
261 ? attrs
262 : Unowned(attrs);
263 }),
264
265 list: () =>
266 Effect.gen(function* () {
267 const env = yield* GcpEnvironment.current;
268 const registries = yield* listSchemaRegistries(env.project, env.region);
269 const owned = yield* Effect.forEach(
270 registries.filter((registry) => (registry.name ?? "").length > 0),
271 (registry) =>
272 schemaRegistryOwnership(registry.name!).pipe(
273 Effect.map((labels) =>
274 hasAlchemyLabelMap(labels) ? registry : undefined,
275 ),
276 ),
277 { concurrency: 4 },
278 );
279 const versions = yield* Effect.forEach(
280 owned.filter(
281 (registry): registry is kafka.SchemaRegistry =>
282 registry !== undefined,
283 ),
284 (registry) =>
285 Effect.gen(function* () {
286 const body = yield* kafka
287 .listProjectsLocationsSchemaRegistriesSubjects({
288 parent: registry.name!,
289 })
290 .pipe(
291 Effect.catchTag(
292 ["NotFound", "SchemaRegistryRequiresCluster"],
293 () => Effect.succeed<kafka.HttpBody>({}),
294 ),
295 );
296 const subjects = (yield* parseHttpJson(body)) as unknown;
297 const names = Array.isArray(subjects)
298 ? subjects.filter(
299 (item): item is string =>
300 typeof item === "string" && item !== OWNERSHIP_SUBJECT,
301 )
302 : [];
303 return yield* Effect.forEach(
304 names,
305 (subject) =>
306 Effect.gen(function* () {
307 const listed = yield* kafka
308 .listProjectsLocationsSchemaRegistriesSubjectsVersions({
309 parent: subjectParent(registry.name!, subject),
310 })
311 .pipe(
312 Effect.catchTag(
313 ["NotFound", "SchemaRegistryRequiresCluster"],
314 () => Effect.succeed<kafka.HttpBody>({}),
315 ),
316 );
317 const ids = parseVersionList(yield* parseHttpJson(listed));
318 return yield* Effect.forEach(
319 ids,
320 (version) =>
321 getSchemaVersion(
322 versionName(
323 subjectParent(registry.name!, subject),
324 version,
325 ),
326 ).pipe(
327 Effect.map((item) =>
328 item
329 ? toAttrs(
330 item,
331 registry.name!,
332 env.project,
333 parseName(registry.name!, "schemaRegistries")
334 .location,
335 versionName(
336 subjectParent(registry.name!, subject),
337 version,
338 ),
339 )
340 : undefined,
341 ),
342 ),
343 { concurrency: 4 },
344 );
345 }),
346 { concurrency: 4 },
347 );
348 }),
349 { concurrency: 2 },
350 );
351 return versions
352 .flat(2)
353 .filter(
354 (item): item is NonNullable<typeof item> => item !== undefined,
355 );
356 }),
357
358 reconcile: Effect.fn(function* ({ id, news, output }) {
359 const env = yield* GcpEnvironment.current;
360 const location = normalizeLocation(
361 news.location ?? output?.location,
362 env.region,
363 );
364 const schemaRegistry = registryOf(
365 news.schemaRegistry,
366 env.project,
367 location,
368 );
369 const subject = yield* toSubjectId(id, news.subject, output?.subject);
370 const parent = subjectParent(schemaRegistry, subject);
371 const versionHint = news.version ?? output?.version;
372 const name = output?.name ?? versionName(parent, versionHint ?? "latest");
373
374 let current = yield* getSchemaVersion(name);
375
376 if (current === undefined) {
377 const created = yield* retryTransient(
378 kafka
379 .createProjectsLocationsSchemaRegistriesSubjectsVersions({
380 parent,
381 body: {
382 schemaType: news.schemaType,
383 version: news.version,
384 normalize: news.normalize,
385 id: news.id,
386 schema: news.schema,
387 references: news.references,
388 },
389 })
390 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined))),
391 );
392 const lookedUp = yield* kafka
393 .lookupVersionProjectsLocationsSchemaRegistriesSubjects({
394 parent,
395 body: {
396 schema: news.schema,
397 schemaType: news.schemaType,
398 normalize: news.normalize,
399 references: news.references,
400 },
401 })
402 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
403 current =
404 lookedUp ??
405 (yield* getSchemaVersion(
406 versionName(parent, news.version ?? "latest"),
407 ));
408 void created;
409 }
410
411 if (current === undefined) {
412 return yield* new SchemaRegistriesSubjectsVersionNotResolved({
413 name,
414 });
415 }
416
417 const resolved = versionName(
418 subjectParent(schemaRegistry, current.subject ?? subject),
419 current.version ?? versionHint ?? "latest",
420 );
421 return toAttrs(current, schemaRegistry, env.project, location, resolved);
422 }),
423
424 delete: Effect.fn(function* ({ output }) {
425 yield* kafka
426 .deleteProjectsLocationsSchemaRegistriesSubjectsVersions({
427 name: output.name,
428 permanent: false,
429 })
430 .pipe(
431 Effect.catchTag(
432 [
433 "NotFound",
434 "SchemaRegistryRequiresCluster",
435 "SchemaRegistryPathNotFound",
436 ],
437 () => Effect.void,
438 ),
439 );
440 yield* kafka
441 .deleteProjectsLocationsSchemaRegistriesSubjectsVersions({
442 name: output.name,
443 permanent: true,
444 })
445 .pipe(
446 Effect.catchTag(
447 [
448 "NotFound",
449 "SchemaRegistryRequiresCluster",
450 "SchemaRegistryPathNotFound",
451 ],
452 () => Effect.void,
453 ),
454 );
455 }),
456 });