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 { createInternalLabels } from "../Labels.ts";
10import type { Providers } from "../Providers.ts";
11import {
12 getSchemaRegistry,
13 hasAlchemyLabelMap,
14 hasSchemaRegistryOwnership,
15 listSchemaRegistries,
16 locationParent,
17 normalizeLocation,
18 parseName,
19 replaceOnIdentity,
20 retryTransient,
21 schemaRegistryIdOf,
22 schemaRegistryOwnership,
23 stampSchemaRegistryOwnership,
24 toSchemaRegistryId,
25} from "./internal.ts";
26
27export type SchemaRegistryProps = {
28 /**
29 * Schema registry id. Letters, numbers, and underscores; max 63
30 * characters; must not start with a number. If omitted, a unique id is
31 * generated. Immutable — changing it replaces the registry.
32 */
33 schemaRegistryId?: string;
34 /**
35 * Region (`us-central1`, …). Immutable.
36 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
37 */
38 location?: string;
39};
40
41export type SchemaRegistry = Resource<
42 "GCP.ManagedKafka.SchemaRegistry",
43 SchemaRegistryProps,
44 {
45 /** Full resource name `.../schemaRegistries/{schemaRegistry}`. */
46 name: string;
47 /** Schema registry id. */
48 schemaRegistryId: string;
49 /** Project id. */
50 project: string;
51 /** Location id. */
52 location: string;
53 /** Contexts currently present. */
54 contexts: string[];
55 },
56 never,
57 Providers
58>;
59
60/**
61 * A Managed Kafka schema registry instance.
62 *
63 * Schema registries have no labels field, so Alchemy stamps ownership on
64 * a reserved `alchemy_ownership` subject. Name and location are identity
65 * — changing either replaces the registry.
66 *
67 * ### Creating a Schema Registry
68 * **Example:** Generated name
69 * ```typescript
70 * const registry = yield* GCP.ManagedKafka.SchemaRegistry("Schemas", {});
71 * ```
72 *
73 * **Example:** Explicit id
74 * ```typescript
75 * const registry = yield* GCP.ManagedKafka.SchemaRegistry("Schemas", {
76 * schemaRegistryId: "app_schemas",
77 * location: "us-central1",
78 * });
79 * ```
80 *
81 * @resource
82 * @category ManagedKafka
83 */
84export const SchemaRegistry = Resource<SchemaRegistry>(
85 "GCP.ManagedKafka.SchemaRegistry",
86);
87
88export class SchemaRegistryNotResolved extends Data.TaggedError(
89 "GCP.ManagedKafka.SchemaRegistryNotResolved",
90)<{
91 name: string;
92}> {}
93
94const resourceName = (
95 project: string,
96 location: string,
97 schemaRegistryId: string,
98) =>
99 `projects/${project}/locations/${location}/schemaRegistries/${schemaRegistryId}`;
100
101const toAttrs = (registry: kafka.SchemaRegistry, project: string) => {
102 const name = registry.name ?? "";
103 const parsed = parseName(name, "schemaRegistries");
104 return {
105 name,
106 schemaRegistryId: parsed.id,
107 project: parsed.project || project,
108 location: parsed.location,
109 contexts: registry.contexts ?? [],
110 };
111};
112
113export const SchemaRegistryProvider = () =>
114 Provider.succeed(SchemaRegistry, {
115 stables: ["name", "schemaRegistryId", "project", "location"],
116
117 diff: Effect.fn(function* ({ news, olds, output }) {
118 if (!isResolved(news)) return undefined;
119 const env = yield* GcpEnvironment.current;
120 return replaceOnIdentity({
121 previousId: olds?.schemaRegistryId ?? output?.schemaRegistryId,
122 nextId: news.schemaRegistryId
123 ? schemaRegistryIdOf(news.schemaRegistryId)
124 : (olds?.schemaRegistryId ?? output?.schemaRegistryId),
125 previousLocation: normalizeLocation(
126 olds?.location ?? output?.location,
127 env.region,
128 ),
129 nextLocation: normalizeLocation(
130 news.location ?? output?.location,
131 env.region,
132 ),
133 });
134 }),
135
136 read: Effect.fn(function* ({ id, olds, output }) {
137 const env = yield* GcpEnvironment.current;
138 const schemaRegistryId = yield* toSchemaRegistryId(
139 id,
140 olds?.schemaRegistryId,
141 output?.schemaRegistryId,
142 );
143 const location = normalizeLocation(
144 olds?.location ?? output?.location,
145 env.region,
146 );
147 const name =
148 output?.name ?? resourceName(env.project, location, schemaRegistryId);
149 const existing = yield* getSchemaRegistry(name);
150 if (existing === undefined) return undefined;
151 const attrs = toAttrs(existing, env.project);
152 return (yield* hasSchemaRegistryOwnership(id, name))
153 ? attrs
154 : Unowned(attrs);
155 }),
156
157 list: () =>
158 Effect.gen(function* () {
159 const env = yield* GcpEnvironment.current;
160 const registries = yield* listSchemaRegistries(env.project, env.region);
161 const owned = yield* Effect.forEach(
162 registries.filter((registry) => (registry.name ?? "").length > 0),
163 (registry) =>
164 schemaRegistryOwnership(registry.name!).pipe(
165 Effect.map((labels) =>
166 hasAlchemyLabelMap(labels)
167 ? toAttrs(registry, env.project)
168 : undefined,
169 ),
170 ),
171 { concurrency: 4 },
172 );
173 return owned.filter(
174 (attrs): attrs is NonNullable<typeof attrs> => attrs !== undefined,
175 );
176 }),
177
178 reconcile: Effect.fn(function* ({ id, news, output }) {
179 const env = yield* GcpEnvironment.current;
180 const schemaRegistryId = yield* toSchemaRegistryId(
181 id,
182 news.schemaRegistryId,
183 output?.schemaRegistryId,
184 );
185 const location = normalizeLocation(
186 news.location ?? output?.location,
187 env.region,
188 );
189 const name =
190 output?.name ?? resourceName(env.project, location, schemaRegistryId);
191 const labels = yield* createInternalLabels(id);
192
193 let current = yield* getSchemaRegistry(name);
194
195 if (current === undefined) {
196 const created = yield* retryTransient(
197 kafka
198 .createProjectsLocationsSchemaRegistries({
199 parent: locationParent(env.project, location),
200 body: {
201 schemaRegistryId,
202 schemaRegistry: {},
203 },
204 })
205 .pipe(Effect.catchTag("Conflict", () => getSchemaRegistry(name))),
206 );
207 current = created ?? undefined;
208 }
209
210 if (current === undefined) {
211 return yield* new SchemaRegistryNotResolved({ name });
212 }
213
214 yield* stampSchemaRegistryOwnership(current.name ?? name, labels);
215 return toAttrs(current, env.project);
216 }),
217
218 delete: Effect.fn(function* ({ output }) {
219 yield* kafka
220 .deleteProjectsLocationsSchemaRegistries({ name: output.name })
221 .pipe(
222 Effect.catchTag(
223 ["NotFound", "SchemaRegistryRequiresCluster"],
224 () => Effect.void,
225 ),
226 );
227 }),
228 });