1import * as metastore from "@distilled.cloud/gcp/metastore_v1";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import * as Provider from "../../Provider.ts";
7import { Resource } from "../../Resource.ts";
8import { tagRecord } from "../../Tags.ts";
9import { GcpEnvironment } from "../Environment.ts";
10import {
11 createInternalLabels,
12 diffLabels,
13 hasAlchemyLabels,
14 toLabels,
15} from "../Labels.ts";
16import type { Providers } from "../Providers.ts";
17import {
18 DEFAULT_HIVE_VERSION,
19 fieldMask,
20 fingerprint,
21 listAtLocation,
22 listLabeledPages,
23 locationParent,
24 normalizeLocation,
25 parseName,
26 replaceOnIdentity,
27 ResourceNotResolved,
28 toPhysicalId,
29 userLabels,
30 waitForOperation,
31 waitUntilExists,
32 waitUntilGone,
33 waitUntilReady,
34} from "./internal.ts";
35
36export type BackendMetastore = {
37 /**
38 * Relative resource name of the backend. BigQuery uses
39 * `projects/{project}`. Dataproc Metastore uses
40 * `projects/{project}/locations/{location}/services/{service}`.
41 */
42 name: string;
43 /**
44 * Backend type (`BIGQUERY` or `DATAPROC_METASTORE`).
45 */
46 metastoreType?: metastore.BackendMetastoreMetastoreTypeEnum | (string & {});
47};
48
49export type FederationProps = {
50 /**
51 * Federation id (the `{federation}` segment of
52 * `projects/{project}/locations/{location}/federations/{federation}`).
53 * If omitted, a unique RFC1035 name is generated. Must be 2-63
54 * characters, start with a letter, and end with a letter or number.
55 * Immutable — changing it replaces the federation.
56 */
57 federationId?: string;
58 /**
59 * Region (`us-central1`, …). Immutable — changing it replaces the
60 * federation. `US-CENTRAL1` is accepted and normalized to `us-central1`.
61 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
62 */
63 location?: string;
64 /**
65 * Apache Hive metastore version of the federation. All backends must
66 * be compatible. Immutable — changing it replaces the federation.
67 * @default "3.1.2"
68 */
69 version?: string;
70 /**
71 * Ranked backends used to resolve database names at query time. Keys
72 * are ranks (`0`, `1`, …); lower ranks are evaluated first. When
73 * omitted, a BigQuery backend for the current project is used.
74 */
75 backendMetastores?: Record<string, BackendMetastore>;
76 /**
77 * User labels. Alchemy ownership labels are merged in automatically.
78 */
79 labels?: Record<string, string>;
80};
81
82export type Federation = Resource<
83 "GCP.Metastore.Federation",
84 FederationProps,
85 {
86 /** Full resource name. */
87 name: string;
88 /** Federation id (last path segment). */
89 federationId: string;
90 /** Project id. */
91 project: string;
92 /** Location id. */
93 location: string;
94 /** Hive metastore version. */
95 version: string | undefined;
96 /** Ranked backend metastores. */
97 backendMetastores: Record<string, BackendMetastore>;
98 /** User labels (Alchemy ownership labels stripped). */
99 labels: Record<string, string>;
100 /** Federation endpoint URI. */
101 endpointUri: string | undefined;
102 /** Server-reported state. */
103 state: string | undefined;
104 /** Extra status text, if any. */
105 stateMessage: string | undefined;
106 /** Server-generated resource uid. */
107 uid: string | undefined;
108 /** RFC3339 creation timestamp. */
109 createTime: string | undefined;
110 /** RFC3339 last-update timestamp. */
111 updateTime: string | undefined;
112 },
113 never,
114 Providers
115>;
116
117/**
118 * A Dataproc Metastore federation that serves metadata from ranked
119 * BigQuery and Dataproc Metastore backends.
120 *
121 * Changing `federationId`, `location`, or `version` replaces the
122 * federation. Labels and `backendMetastores` update in place.
123 *
124 * ### Creating a Federation
125 * **Example:** BigQuery backend
126 * ```typescript
127 * const federation = yield* GCP.Metastore.Federation("Lakehouse", {
128 * version: "3.1.2",
129 * backendMetastores: {
130 * "1": {
131 * name: "projects/my-project",
132 * metastoreType: "BIGQUERY",
133 * },
134 * },
135 * labels: { env: "prod" },
136 * });
137 * ```
138 *
139 * **Example:** Generated name
140 * ```typescript
141 * const federation = yield* GCP.Metastore.Federation("Lakehouse", {
142 * labels: { env: "test" },
143 * });
144 * ```
145 *
146 * ### Updating a Federation
147 * **Example:** Labels and backends
148 * ```typescript
149 * // Same logical id, changed props: the engine updates it in place.
150 * const federation = yield* GCP.Metastore.Federation("Lakehouse", {
151 * version: "3.1.2",
152 * backendMetastores: {
153 * "1": {
154 * name: "projects/my-project",
155 * metastoreType: "BIGQUERY",
156 * },
157 * },
158 * labels: { env: "prod", team: "data" },
159 * });
160 * ```
161 *
162 * @resource
163 * @category Metastore
164 */
165export const Federation = Resource<Federation>("GCP.Metastore.Federation");
166
167const resourceName = (
168 project: string,
169 location: string,
170 federationId: string,
171) => `${locationParent(project, location)}/federations/${federationId}`;
172
173const toBackends = (
174 value: Record<string, metastore.BackendMetastore | undefined> | undefined,
175): Record<string, BackendMetastore> => {
176 const next: Record<string, BackendMetastore> = {};
177 for (const [rank, item] of Object.entries(value ?? {})) {
178 if (item === undefined) continue;
179 next[rank] = {
180 name: item.name ?? "",
181 metastoreType: item.metastoreType,
182 };
183 }
184 return next;
185};
186
187const desiredBackends = (
188 project: string,
189 value: Record<string, BackendMetastore> | undefined,
190): Record<string, BackendMetastore> => {
191 if (value !== undefined) return toBackends(value);
192 return {
193 "1": {
194 name: `projects/${project}`,
195 metastoreType: "BIGQUERY",
196 },
197 };
198};
199
200const toAttrs = (
201 item: metastore.Federation,
202 project: string,
203 region: string,
204) => {
205 const name = item.name ?? "";
206 const parsed = parseName(name, "federations", region);
207 return {
208 name,
209 federationId: parsed.id,
210 project: parsed.project || project,
211 location: parsed.location,
212 version: item.version,
213 backendMetastores: toBackends(item.backendMetastores),
214 labels: userLabels(item.labels),
215 endpointUri: item.endpointUri,
216 state: item.state,
217 stateMessage: item.stateMessage,
218 uid: item.uid,
219 createTime: item.createTime,
220 updateTime: item.updateTime,
221 };
222};
223
224const getByName = (name: string) =>
225 name.length === 0
226 ? Effect.succeed(undefined)
227 : metastore
228 .getProjectsLocationsFederations({ name })
229 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
230
231const listOwned = (project: string, region: string) =>
232 listAtLocation(project, (parent) =>
233 listLabeledPages(
234 metastore.listProjectsLocationsFederations.pages({
235 parent,
236 pageSize: 1000,
237 }),
238 (page) => page.federations,
239 (item) => item.labels,
240 ),
241 );
242
243export const FederationProvider = () =>
244 Provider.succeed(Federation, {
245 stables: [
246 "name",
247 "federationId",
248 "project",
249 "location",
250 "version",
251 "uid",
252 "createTime",
253 ],
254
255 diff: Effect.fn(function* ({ news, olds, output }) {
256 if (!isResolved(news)) return undefined;
257 const env = yield* GcpEnvironment.current;
258 const previousVersion = olds?.version ?? output?.version;
259 const nextVersion =
260 news.version ??
261 olds?.version ??
262 output?.version ??
263 DEFAULT_HIVE_VERSION;
264 return replaceOnIdentity({
265 previousId: olds?.federationId ?? output?.federationId,
266 nextId: news.federationId ?? olds?.federationId ?? output?.federationId,
267 previousLocation: normalizeLocation(
268 olds?.location ?? output?.location,
269 env.region,
270 ),
271 nextLocation: normalizeLocation(
272 news.location ?? olds?.location ?? output?.location,
273 env.region,
274 ),
275 extra: previousVersion !== undefined && nextVersion !== previousVersion,
276 });
277 }),
278
279 read: Effect.fn(function* ({ id, olds, output }) {
280 const env = yield* GcpEnvironment.current;
281 const federationId = yield* toPhysicalId(
282 id,
283 olds?.federationId,
284 output?.federationId,
285 "federation",
286 );
287 const location = normalizeLocation(
288 olds?.location ?? output?.location,
289 env.region,
290 );
291 const name =
292 output?.name ?? resourceName(env.project, location, federationId);
293 const existing = yield* getByName(name);
294 if (existing === undefined) return undefined;
295 const attrs = toAttrs(existing, env.project, env.region);
296 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
297 ? attrs
298 : Unowned(attrs);
299 }),
300
301 list: () =>
302 Effect.gen(function* () {
303 const env = yield* GcpEnvironment.current;
304 const items = yield* listOwned(env.project, env.region);
305 return items.map((item) => toAttrs(item, env.project, env.region));
306 }),
307
308 reconcile: Effect.fn(function* ({ id, news, output }) {
309 const env = yield* GcpEnvironment.current;
310 const federationId = yield* toPhysicalId(
311 id,
312 news.federationId,
313 output?.federationId,
314 "federation",
315 );
316 const location = normalizeLocation(
317 news.location ?? output?.location,
318 env.region,
319 );
320 const name = resourceName(env.project, location, federationId);
321 const version = news.version ?? output?.version ?? DEFAULT_HIVE_VERSION;
322 const backends = desiredBackends(env.project, news.backendMetastores);
323 const desiredLabels = {
324 ...toLabels(news.labels),
325 ...(yield* createInternalLabels(id)),
326 };
327
328 let current = yield* getByName(output?.name ?? name);
329
330 if (current === undefined) {
331 const created = yield* metastore
332 .createProjectsLocationsFederations({
333 parent: locationParent(env.project, location),
334 federationId,
335 body: {
336 version,
337 backendMetastores: backends,
338 labels: desiredLabels,
339 },
340 })
341 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
342 if (created !== undefined) {
343 // ALREADY_EXISTS (6): a concurrent create won the race.
344 yield* waitForOperation(created).pipe(
345 Effect.catchTag("GCP.OperationFailed", (error) =>
346 error.code === 6 ? Effect.void : Effect.fail(error),
347 ),
348 );
349 }
350 current = yield* waitUntilExists(getByName(name), name);
351 current = yield* waitUntilReady(
352 getByName(name),
353 name,
354 (item) => item.state,
355 (item) => item.stateMessage,
356 );
357 }
358
359 if (current === undefined) {
360 return yield* new ResourceNotResolved({ name });
361 }
362
363 const observedLabels = tagRecord(current.labels);
364 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
365 const labelsChanged = upsert.length > 0 || removed.length > 0;
366 const backendsChanged =
367 fingerprint(toBackends(current.backendMetastores)) !==
368 fingerprint(backends);
369 const mask = fieldMask([
370 labelsChanged && "labels",
371 backendsChanged && "backendMetastores",
372 ]);
373
374 if (mask.length > 0) {
375 const operation = yield* metastore.patchProjectsLocationsFederations({
376 name: current.name ?? name,
377 updateMask: mask,
378 body: {
379 labels: desiredLabels,
380 backendMetastores: backends,
381 },
382 });
383 yield* waitForOperation(operation);
384 current = yield* waitUntilReady(
385 getByName(current.name ?? name),
386 current.name ?? name,
387 (item) => item.state,
388 (item) => item.stateMessage,
389 );
390 }
391
392 if (current === undefined) {
393 return yield* new ResourceNotResolved({ name });
394 }
395
396 return toAttrs(current, env.project, env.region);
397 }),
398
399 delete: Effect.fn(function* ({ output }) {
400 const operation = yield* metastore
401 .deleteProjectsLocationsFederations({ name: output.name })
402 .pipe(
403 Effect.retry({
404 while: (error) => error._tag === "Conflict",
405 times: 8,
406 schedule: Schedule.spaced("2 seconds"),
407 }),
408 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
409 );
410 if (operation !== undefined) {
411 // NOT_FOUND (5): already gone.
412 yield* waitForOperation(operation).pipe(
413 Effect.catchTag("GCP.OperationFailed", (error) =>
414 error.code === 5 ? Effect.void : Effect.fail(error),
415 ),
416 );
417 }
418 yield* waitUntilGone(getByName(output.name), output.name);
419 }),
420 });