1import * as kafka from "@distilled.cloud/gcp/managedkafka_v1";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import * as Stream from "effect/Stream";
6import { waitForOperation as waitForGcpOperation } from "../Operation.ts";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import { tagRecord } from "../../Tags.ts";
9import {
10 alchemyLabelKeys,
11 createInternalLabels,
12 stripInternalLabels,
13} from "../Labels.ts";
14import { isTransientGcpError } from "../Errors.ts";
15
16export const MAX_NAME_LENGTH = 63;
17/** The API rejects longer cluster and Connect cluster ids. */
18export const MAX_CLUSTER_ID_LENGTH = 53;
19export const DEFAULT_VCPU_COUNT = 3;
20export const DEFAULT_MEMORY_BYTES = 3_221_225_472;
21export const OWNERSHIP_SUBJECT = "alchemy_ownership";
22
23export class ResourceNotResolved extends Data.TaggedError(
24 "GCP.ManagedKafka.ResourceNotResolved",
25)<{
26 name: string;
27}> {}
28
29export class ResourceStillExists extends Data.TaggedError(
30 "GCP.ManagedKafka.ResourceStillExists",
31)<{
32 name: string;
33}> {}
34
35export class ResourceNotReady extends Data.TaggedError(
36 "GCP.ManagedKafka.ResourceNotReady",
37)<{
38 name: string;
39 state: string;
40}> {}
41
42export class ResourceFailed extends Data.TaggedError(
43 "GCP.ManagedKafka.ResourceFailed",
44)<{
45 name: string;
46 state: string;
47}> {}
48
49export const lastSegment = (value: string) => {
50 const trimmed = value.replace(/\/+$/, "");
51 const parts = trimmed.split("/");
52 return parts[parts.length - 1] || trimmed;
53};
54
55export const rfc1035 = (
56 name: string,
57 fallback = "kafka",
58 maxLength = MAX_NAME_LENGTH,
59): string => {
60 let next = name
61 .toLowerCase()
62 .replace(/[^a-z0-9-]/g, "-")
63 .replace(/-+/g, "-")
64 .replace(/^-+|-+$/g, "");
65 if (!/^[a-z]/.test(next)) next = `k${next}`;
66 next = next.slice(0, maxLength).replace(/-+$/g, "");
67 if (next.length === 0) return fallback;
68 if (!/[a-z0-9]$/.test(next)) next = `${next.slice(0, maxLength - 1)}0`;
69 return next.slice(0, maxLength);
70};
71
72export const schemaRegistryIdOf = (
73 name: string,
74 fallback = "schema",
75): string => {
76 let next = name
77 .toLowerCase()
78 .replace(/[^a-z0-9_]/g, "_")
79 .replace(/_+/g, "_")
80 .replace(/^_|_$/g, "");
81 if (!/^[a-z]/.test(next)) next = `s${next}`;
82 next = next.slice(0, MAX_NAME_LENGTH).replace(/_+$/g, "");
83 return next.length > 0 ? next : fallback;
84};
85
86export const normalizeLocation = (
87 location: string | undefined,
88 defaultLocation: string,
89) => lastSegment(location ?? defaultLocation).toLowerCase();
90
91export const locationParent = (project: string, location: string) =>
92 `projects/${project}/locations/${location}`;
93
94export const defaultSubnet = (project: string, location: string) =>
95 `projects/${project}/regions/${location}/subnetworks/default`;
96
97export const toPhysicalId = (
98 id: string,
99 explicit: string | undefined,
100 existing: string | undefined,
101 fallback = "kafka",
102 maxLength = MAX_NAME_LENGTH,
103) =>
104 Effect.gen(function* () {
105 if (explicit !== undefined) return rfc1035(explicit, fallback, maxLength);
106 if (existing !== undefined) return existing;
107 return rfc1035(
108 yield* createPhysicalName({
109 id,
110 maxLength,
111 lowercase: true,
112 }),
113 fallback,
114 maxLength,
115 );
116 });
117
118export const toSchemaRegistryId = (
119 id: string,
120 explicit: string | undefined,
121 existing: string | undefined,
122) =>
123 Effect.gen(function* () {
124 if (explicit !== undefined) return schemaRegistryIdOf(explicit);
125 if (existing !== undefined) return existing;
126 return schemaRegistryIdOf(
127 yield* createPhysicalName({
128 id,
129 maxLength: MAX_NAME_LENGTH,
130 lowercase: true,
131 delimiter: "_",
132 }),
133 );
134 });
135
136export const toSubjectId = (
137 id: string,
138 explicit: string | undefined,
139 existing: string | undefined,
140) =>
141 Effect.gen(function* () {
142 if (explicit !== undefined) return explicit;
143 if (existing !== undefined) return existing;
144 return rfc1035(
145 yield* createPhysicalName({
146 id,
147 maxLength: MAX_NAME_LENGTH,
148 lowercase: true,
149 }),
150 "subject",
151 );
152 });
153
154export const parseName = (name: string, collection: string) => {
155 const parts = name.split("/").filter((part) => part.length > 0);
156 const collectionAt = parts.lastIndexOf(collection);
157 const locationsAt = parts.lastIndexOf("locations");
158 const projectsAt = parts.lastIndexOf("projects");
159 return {
160 project:
161 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
162 // API resource names always carry a `locations/{location}` segment.
163 location:
164 locationsAt >= 0 && parts[locationsAt + 1] ? parts[locationsAt + 1]! : "",
165 id:
166 collectionAt >= 0 && parts[collectionAt + 1]
167 ? parts.slice(collectionAt + 1).join("/")
168 : lastSegment(name),
169 parent:
170 collectionAt > 0
171 ? parts.slice(0, collectionAt).join("/")
172 : parts.slice(0, Math.max(0, parts.length - 1)).join("/"),
173 };
174};
175
176export const expandParent = (
177 value: string,
178 project: string,
179 location: string,
180 collection: string,
181) => {
182 if (value.includes("/")) return value.replace(/\/+$/, "");
183 return `projects/${project}/locations/${location}/${collection}/${value}`;
184};
185
186export const userLabels = (
187 labels: Record<string, string | undefined> | null | undefined,
188): Record<string, string> => stripInternalLabels(tagRecord(labels));
189
190export const hasAlchemyLabelMap = (
191 labels: Record<string, string | undefined> | null | undefined,
192) => Object.keys(labels ?? {}).some((key) => key.startsWith("alchemy-"));
193
194export const stringMapOf = (
195 value: Record<string, string | undefined> | null | undefined,
196): Record<string, string> =>
197 Object.fromEntries(
198 Object.entries(value ?? {}).filter(
199 (entry): entry is [string, string] => entry[1] !== undefined,
200 ),
201 );
202
203export const sameText = (left: string | undefined, right: string | undefined) =>
204 (left ?? "") === (right ?? "");
205
206export const fingerprint = (value: unknown): string =>
207 JSON.stringify(value ?? null);
208
209export const fieldMask = (fields: Array<string | false | undefined>) =>
210 fields
211 .filter((field): field is string => typeof field === "string")
212 .join(",");
213
214export const asCount = (
215 value: number | string | undefined,
216 fallback: number,
217) => {
218 if (value === undefined) return fallback;
219 const parsed = typeof value === "number" ? value : Number(value);
220 return Number.isFinite(parsed) ? parsed : fallback;
221};
222
223export const asCountString = (
224 value: number | string | undefined,
225 fallback: number,
226) => String(asCount(value, fallback));
227
228export const replaceOnIdentity = (input: {
229 previousId?: string;
230 nextId?: string;
231 previousLocation?: string;
232 nextLocation?: string;
233 previousParent?: string;
234 nextParent?: string;
235 extra?: boolean;
236}) => {
237 const parentChanged =
238 (input.previousParent ?? "") !== "" &&
239 (input.nextParent ?? "") !== "" &&
240 (input.previousParent ?? "") !== (input.nextParent ?? "");
241 const locationChanged =
242 (input.previousLocation ?? "") !== "" &&
243 (input.nextLocation ?? "") !== "" &&
244 (input.previousLocation ?? "") !== (input.nextLocation ?? "");
245 const idChanged =
246 input.previousId !== undefined &&
247 input.nextId !== undefined &&
248 input.previousId !== input.nextId;
249 if (
250 !(input.extra === true || parentChanged || locationChanged || idChanged)
251 ) {
252 return undefined;
253 }
254 const samePhysical =
255 !parentChanged &&
256 !locationChanged &&
257 input.previousId !== undefined &&
258 input.nextId === input.previousId;
259 return {
260 action: "replace" as const,
261 deleteFirst: samePhysical,
262 };
263};
264
265const READY_STATES = new Set(["ACTIVE", "RUNNING"]);
266const FAILED_STATES = new Set(["FAILED", "ERROR"]);
267
268export const isReadyState = (state: string | undefined) =>
269 READY_STATES.has((state ?? "").toUpperCase());
270
271export const isFailedState = (state: string | undefined) =>
272 FAILED_STATES.has((state ?? "").toUpperCase());
273
274/**
275 * Wait for a Managed Kafka operation. Cluster creates take 20–30 minutes;
276 * Connect clusters are similar. An operation that finished with
277 * `ALREADY_EXISTS` (6) is a lost create race; `notFoundOk` also accepts
278 * `NOT_FOUND` (5).
279 */
280export const waitForOperation = (
281 operation: kafka.Operation,
282 options?: { notFoundOk?: boolean },
283) =>
284 waitForGcpOperation(
285 operation,
286 (name) =>
287 kafka
288 .getProjectsLocationsOperations({ name })
289 .pipe(
290 Effect.catchTag("NotFound", (error) =>
291 options?.notFoundOk === true
292 ? Effect.succeed<kafka.Operation>({ name, done: true })
293 : Effect.fail(error),
294 ),
295 ),
296 { budget: "45 minutes" },
297 ).pipe(
298 Effect.catchIf(
299 (error) =>
300 error._tag === "GCP.OperationFailed" &&
301 (error.code === 6 ||
302 (options?.notFoundOk === true && error.code === 5)),
303 () => Effect.void,
304 ),
305 Effect.asVoid,
306 );
307
308export const waitUntilExists = <A, E extends { readonly _tag: string }, R>(
309 get: Effect.Effect<A | undefined, E, R>,
310 name: string,
311) =>
312 get.pipe(
313 Effect.filterOrFail(
314 (value): value is A => value !== undefined,
315 () => new ResourceNotResolved({ name }),
316 ),
317 Effect.retry({
318 while: (error) => error._tag === "GCP.ManagedKafka.ResourceNotResolved",
319 times: 8,
320 schedule: Schedule.spaced("1 second"),
321 }),
322 );
323
324export const waitUntilGone = <A, E extends { readonly _tag: string }, R>(
325 get: Effect.Effect<A | undefined, E, R>,
326 name: string,
327) =>
328 get.pipe(
329 Effect.flatMap((value) =>
330 value === undefined
331 ? Effect.void
332 : Effect.fail(new ResourceStillExists({ name })),
333 ),
334 Effect.retry({
335 while: (error) => error._tag === "GCP.ManagedKafka.ResourceStillExists",
336 times: 10,
337 schedule: Schedule.spaced("2 seconds"),
338 }),
339 );
340
341export const waitUntilReady = <A, E extends { readonly _tag: string }, R>(
342 get: Effect.Effect<A | undefined, E, R>,
343 name: string,
344 stateOf: (value: A) => string | undefined,
345) =>
346 get.pipe(
347 Effect.filterOrFail(
348 (value): value is A => value !== undefined,
349 () => new ResourceNotResolved({ name }),
350 ),
351 Effect.filterOrFail(
352 (value) => !isFailedState(stateOf(value) ?? ""),
353 (value) => new ResourceFailed({ name, state: stateOf(value) ?? "" }),
354 ),
355 Effect.filterOrFail(
356 (value) => {
357 const state = stateOf(value) ?? "";
358 return isReadyState(state) || state.length === 0;
359 },
360 (value) => new ResourceNotReady({ name, state: stateOf(value) ?? "" }),
361 ),
362 Effect.retry({
363 while: (error) =>
364 error._tag === "GCP.ManagedKafka.ResourceNotReady" ||
365 error._tag === "GCP.ManagedKafka.ResourceNotResolved",
366 times: 10,
367 schedule: Schedule.spaced("8 seconds"),
368 }),
369 );
370
371export const collectPages = <Page, A, E, R>(
372 pages: Stream.Stream<Page, E, R>,
373 items: (page: Page) => readonly A[] | undefined,
374) =>
375 pages.pipe(
376 Stream.flatMap((page) => Stream.fromIterable(items(page) ?? [])),
377 Stream.runCollect,
378 Effect.map((chunk) => Array.from(chunk)),
379 );
380
381export const listAtLocation = <A, E, R>(
382 project: string,
383 _region: string,
384 list: (parent: string) => Effect.Effect<A[], E, R>,
385) => list(`projects/${project}/locations/-`);
386
387export const listClusters = (project: string, region: string) =>
388 listAtLocation(project, region, (parent) =>
389 collectPages(
390 kafka.listProjectsLocationsClusters.pages({ parent, pageSize: 1000 }),
391 (page) => page.clusters,
392 ).pipe(
393 Effect.catchTag("NotFound", () => Effect.succeed([] as kafka.Cluster[])),
394 ),
395 );
396
397export const listAlchemyClusters = (project: string, region: string) =>
398 listClusters(project, region).pipe(
399 Effect.map((clusters: kafka.Cluster[]) =>
400 clusters.filter((cluster) => hasAlchemyLabelMap(cluster.labels)),
401 ),
402 );
403
404export const listConnectClusters = (project: string, region: string) =>
405 listAtLocation(project, region, (parent) =>
406 collectPages(
407 kafka.listProjectsLocationsConnectClusters.pages({
408 parent,
409 pageSize: 1000,
410 }),
411 (page) => page.connectClusters,
412 ).pipe(
413 Effect.catchTag("NotFound", () =>
414 Effect.succeed([] as kafka.ConnectCluster[]),
415 ),
416 ),
417 );
418
419export const listAlchemyConnectClusters = (project: string, region: string) =>
420 listConnectClusters(project, region).pipe(
421 Effect.map((clusters: kafka.ConnectCluster[]) =>
422 clusters.filter((cluster) => hasAlchemyLabelMap(cluster.labels)),
423 ),
424 );
425
426export const listSchemaRegistries = (project: string, location: string) => {
427 // The `locations/-` wildcard hangs for Schema Registry; list the region.
428 const parents = [`projects/${project}/locations/${location}`];
429 return Effect.forEach(
430 parents,
431 (parent) =>
432 kafka
433 .listProjectsLocationsSchemaRegistries({
434 parent,
435 view: "SCHEMA_REGISTRY_VIEW_FULL",
436 })
437 .pipe(
438 Effect.map((page) => page.schemaRegistries ?? []),
439 Effect.catchTag(["NotFound", "SchemaRegistryRequiresCluster"], () =>
440 Effect.succeed([] as kafka.SchemaRegistry[]),
441 ),
442 ),
443 { concurrency: 1 },
444 ).pipe(
445 Effect.map((groups) => {
446 const seen = new Set<string>();
447 const out: kafka.SchemaRegistry[] = [];
448 for (const registry of groups.flat()) {
449 const name = registry.name ?? "";
450 if (name.length === 0 || seen.has(name)) continue;
451 seen.add(name);
452 out.push(registry);
453 }
454 return out;
455 }),
456 );
457};
458
459export const encodeOwnership = (labels: Record<string, string>) =>
460 `[alchemy ${alchemyLabelKeys.stack}=${labels[alchemyLabelKeys.stack]} ${alchemyLabelKeys.stage}=${labels[alchemyLabelKeys.stage]} ${alchemyLabelKeys.id}=${labels[alchemyLabelKeys.id]}]`;
461
462export const parseOwnership = (text: string | undefined) => {
463 if (!text?.includes("[alchemy ")) return {} as Record<string, string>;
464 const start = text.indexOf("[alchemy ");
465 const end = text.indexOf("]", start);
466 if (end < 0) return {} as Record<string, string>;
467 const labels: Record<string, string> = {};
468 for (const part of text.slice(start + "[alchemy ".length, end).split(/\s+/)) {
469 const eq = part.indexOf("=");
470 if (eq > 0) labels[part.slice(0, eq)] = part.slice(eq + 1);
471 }
472 return labels;
473};
474
475export const hasOwnershipMarker = (text: string | undefined) =>
476 Object.keys(parseOwnership(text)).some((key) => key.startsWith("alchemy-"));
477
478export const ownershipSchema = (labels: Record<string, string>) =>
479 JSON.stringify({
480 type: "record",
481 name: "AlchemyOwnership",
482 namespace: "alchemy",
483 doc: encodeOwnership(labels),
484 fields: [{ name: "ok", type: "string" }],
485 });
486
487export const parseHttpJson = (body: kafka.HttpBody | undefined) =>
488 Effect.sync(() => {
489 const data = body?.data;
490 if (data === undefined || data.length === 0) return undefined;
491 try {
492 return JSON.parse(data) as unknown;
493 } catch {
494 /* base64 */
495 }
496 try {
497 return JSON.parse(
498 Buffer.from(data, "base64").toString("utf8"),
499 ) as unknown;
500 } catch {
501 return undefined;
502 }
503 });
504
505export const subjectParent = (
506 schemaRegistry: string,
507 subject: string,
508 context?: string,
509) =>
510 context !== undefined && context.length > 0
511 ? `${schemaRegistry}/contexts/${context}/subjects/${subject}`
512 : `${schemaRegistry}/subjects/${subject}`;
513
514export const versionName = (parent: string, version: number | string) =>
515 `${parent}/versions/${version}`;
516
517export const retryTransient = <A, E extends { readonly _tag: string }, R>(
518 effect: Effect.Effect<A, E, R>,
519) =>
520 effect.pipe(
521 Effect.retry({
522 while: (error) => isTransientGcpError(error) || error._tag === "NotFound",
523 times: 8,
524 schedule: Schedule.exponential("250 millis"),
525 }),
526 );
527
528/**
529 * Cluster and Connect cluster creates are limited to one request per minute
530 * per region, so a burst of creates is throttled with `TooManyRequests`
531 * until the minute window rolls over. Retry for up to ~4 minutes.
532 */
533export const retryCreateQuota = <A, E extends { readonly _tag: string }, R>(
534 effect: Effect.Effect<A, E, R>,
535) =>
536 effect.pipe(
537 Effect.retry({
538 while: (error) => error._tag === "TooManyRequests",
539 times: 12,
540 schedule: Schedule.spaced("20 seconds"),
541 }),
542 );
543
544export const getCluster = (name: string) =>
545 name.length === 0
546 ? Effect.succeed(undefined)
547 : kafka
548 .getProjectsLocationsClusters({ name })
549 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
550
551export const getConnectCluster = (name: string) =>
552 name.length === 0
553 ? Effect.succeed(undefined)
554 : kafka
555 .getProjectsLocationsConnectClusters({ name })
556 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
557
558export const getAcl = (name: string) =>
559 name.length === 0
560 ? Effect.succeed(undefined)
561 : kafka
562 .getProjectsLocationsClustersAcls({ name })
563 .pipe(
564 Effect.catchTag(["NotFound", "AclClusterNotFound"], () =>
565 Effect.succeed(undefined),
566 ),
567 );
568
569export const getTopic = (name: string) =>
570 name.length === 0
571 ? Effect.succeed(undefined)
572 : kafka
573 .getProjectsLocationsClustersTopics({ name })
574 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
575
576export const getConnector = (name: string) =>
577 name.length === 0
578 ? Effect.succeed(undefined)
579 : kafka
580 .getProjectsLocationsConnectClustersConnectors({ name })
581 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
582
583export const getSchemaRegistry = (name: string) =>
584 name.length === 0
585 ? Effect.succeed(undefined)
586 : kafka
587 .getProjectsLocationsSchemaRegistries({ name })
588 .pipe(
589 Effect.catchTag(["NotFound", "SchemaRegistryRequiresCluster"], () =>
590 Effect.succeed(undefined),
591 ),
592 );
593
594export const getSchemaVersion = (name: string, deleted = false) =>
595 name.length === 0
596 ? Effect.succeed(undefined)
597 : kafka
598 .getProjectsLocationsSchemaRegistriesSubjectsVersions({
599 name,
600 deleted,
601 })
602 .pipe(
603 Effect.catchTag(
604 [
605 "NotFound",
606 "SchemaRegistryRequiresCluster",
607 "SchemaRegistryPathNotFound",
608 ],
609 () => Effect.succeed(undefined),
610 ),
611 );
612
613export const getContextSchemaVersion = (name: string, deleted = false) =>
614 name.length === 0
615 ? Effect.succeed(undefined)
616 : kafka
617 .getProjectsLocationsSchemaRegistriesContextsSubjectsVersions({
618 name,
619 deleted,
620 })
621 .pipe(
622 Effect.catchTag(
623 [
624 "NotFound",
625 "SchemaRegistryRequiresCluster",
626 "SchemaRegistryPathNotFound",
627 ],
628 () => Effect.succeed(undefined),
629 ),
630 );
631
632export const stampSchemaRegistryOwnership = (
633 schemaRegistry: string,
634 labels: Record<string, string>,
635) =>
636 Effect.gen(function* () {
637 const parent = subjectParent(schemaRegistry, OWNERSHIP_SUBJECT);
638 const schema = ownershipSchema(labels);
639 const latest = yield* getSchemaVersion(versionName(parent, "latest"));
640 if (hasOwnershipMarker(latest?.schema) && latest?.schema === schema) {
641 return;
642 }
643 yield* kafka
644 .createProjectsLocationsSchemaRegistriesSubjectsVersions({
645 parent,
646 body: {
647 schemaType: "AVRO",
648 schema,
649 },
650 })
651 .pipe(Effect.catchTag(["Conflict", "BadRequest"], () => Effect.void));
652 });
653
654export const schemaRegistryOwnership = (schemaRegistry: string) =>
655 getSchemaVersion(
656 versionName(subjectParent(schemaRegistry, OWNERSHIP_SUBJECT), "latest"),
657 ).pipe(Effect.map((version) => parseOwnership(version?.schema)));
658
659export const hasSchemaRegistryOwnership = Effect.fn(function* (
660 id: string,
661 schemaRegistry: string,
662) {
663 const expected = yield* createInternalLabels(id);
664 const observed = yield* schemaRegistryOwnership(schemaRegistry);
665 return (
666 observed[alchemyLabelKeys.stack] === expected[alchemyLabelKeys.stack] &&
667 observed[alchemyLabelKeys.stage] === expected[alchemyLabelKeys.stage] &&
668 observed[alchemyLabelKeys.id] === expected[alchemyLabelKeys.id]
669 );
670});