1import * as metastore from "@distilled.cloud/gcp/metastore_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 hasAlchemyLabels,
13 stripInternalLabels,
14} from "../Labels.ts";
15
16export const DEFAULT_HIVE_VERSION = "3.1.2";
17export const MAX_NAME_LENGTH = 63;
18
19export class ResourceNotResolved extends Data.TaggedError(
20 "GCP.Metastore.ResourceNotResolved",
21)<{
22 name: string;
23}> {}
24
25export class ResourceStillExists extends Data.TaggedError(
26 "GCP.Metastore.ResourceStillExists",
27)<{
28 name: string;
29}> {}
30
31export class ResourceNotReady extends Data.TaggedError(
32 "GCP.Metastore.ResourceNotReady",
33)<{
34 name: string;
35 state: string;
36}> {}
37
38export class ResourceFailed extends Data.TaggedError(
39 "GCP.Metastore.ResourceFailed",
40)<{
41 name: string;
42 state: string;
43 details: string | undefined;
44}> {}
45
46export const lastSegment = (value: string) => {
47 const trimmed = value.replace(/\/+$/, "");
48 const parts = trimmed.split("/");
49 return parts[parts.length - 1] || trimmed;
50};
51
52export const rfc1035 = (name: string, fallback = "metastore"): string => {
53 let next = name
54 .toLowerCase()
55 .replace(/[^a-z0-9-]/g, "-")
56 .replace(/-+/g, "-")
57 .replace(/^-+|-+$/g, "");
58 if (!/^[a-z]/.test(next)) next = `m${next}`;
59 next = next.slice(0, MAX_NAME_LENGTH).replace(/-+$/g, "");
60 if (next.length === 0) return fallback;
61 if (!/[a-z0-9]$/.test(next)) next = `${next.slice(0, MAX_NAME_LENGTH - 1)}0`;
62 return next.slice(0, MAX_NAME_LENGTH);
63};
64
65export const normalizeLocation = (
66 location: string | undefined,
67 defaultLocation: string,
68) => lastSegment(location ?? defaultLocation).toLowerCase();
69
70export const locationParent = (project: string, location: string) =>
71 `projects/${project}/locations/${lastSegment(location).toLowerCase()}`;
72
73export const toPhysicalId = (
74 id: string,
75 explicit: string | undefined,
76 existing: string | undefined,
77 fallback = "metastore",
78) =>
79 Effect.gen(function* () {
80 if (explicit !== undefined) return rfc1035(explicit, fallback);
81 if (existing !== undefined) return existing;
82 return rfc1035(
83 yield* createPhysicalName({
84 id,
85 maxLength: MAX_NAME_LENGTH,
86 lowercase: true,
87 }),
88 fallback,
89 );
90 });
91
92export const parseName = (
93 name: string,
94 collection: string,
95 defaultLocation: string,
96) => {
97 const parts = name.split("/").filter((part) => part.length > 0);
98 const collectionAt = parts.lastIndexOf(collection);
99 const locationsAt = parts.lastIndexOf("locations");
100 const projectsAt = parts.lastIndexOf("projects");
101 return {
102 project:
103 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
104 location:
105 locationsAt >= 0 && parts[locationsAt + 1]
106 ? parts[locationsAt + 1]!
107 : defaultLocation,
108 id:
109 collectionAt >= 0 && parts[collectionAt + 1]
110 ? parts[collectionAt + 1]!
111 : lastSegment(name),
112 parent:
113 collectionAt > 0
114 ? parts.slice(0, collectionAt).join("/")
115 : parts.slice(0, Math.max(0, parts.length - 1)).join("/"),
116 };
117};
118
119export const expandParent = (
120 value: string,
121 project: string,
122 location: string,
123 collection: string,
124) => {
125 if (value.includes("/")) return value.replace(/\/+$/, "");
126 return `projects/${project}/locations/${location}/${collection}/${value}`;
127};
128
129export const userLabels = (
130 labels: Record<string, string | undefined> | null | undefined,
131): Record<string, string> => stripInternalLabels(tagRecord(labels));
132
133export const hasAlchemyLabelMap = (
134 labels: Record<string, string | undefined> | null | undefined,
135) => Object.keys(labels ?? {}).some((key) => key.startsWith("alchemy-"));
136
137export const canonical = (value: unknown): unknown => {
138 if (value === undefined || value === null) return undefined;
139 if (typeof value === "boolean" || typeof value === "number") return value;
140 if (typeof value === "string") {
141 return value.length === 0 ? undefined : value;
142 }
143 if (Array.isArray(value)) {
144 const items = value.map(canonical).filter((item) => item !== undefined);
145 return items.length === 0 ? undefined : items;
146 }
147 if (typeof value === "object") {
148 const entries = Object.entries(value as Record<string, unknown>)
149 .map(([key, item]) => [key, canonical(item)] as const)
150 .filter(([, item]) => item !== undefined)
151 .sort(([left], [right]) => left.localeCompare(right));
152 if (entries.length === 0) return undefined;
153 return Object.fromEntries(entries);
154 }
155 return undefined;
156};
157
158export const fingerprint = (value: unknown): string =>
159 JSON.stringify(canonical(value) ?? null);
160
161export const fieldMask = (fields: Array<string | false | undefined>) =>
162 fields
163 .filter((field): field is string => typeof field === "string")
164 .join(",");
165
166export const sameText = (left: string | undefined, right: string | undefined) =>
167 (left ?? "") === (right ?? "");
168
169export const replaceOnIdentity = (input: {
170 previousId: string | undefined;
171 nextId: string | undefined;
172 previousLocation: string;
173 nextLocation: string;
174 extra?: boolean;
175 previousParent?: string;
176 nextParent?: string;
177}) => {
178 const parentChanged =
179 (input.previousParent ?? "") !== "" &&
180 (input.nextParent ?? "") !== "" &&
181 (input.previousParent ?? "") !== (input.nextParent ?? "");
182 const replace =
183 (input.extra ?? false) ||
184 parentChanged ||
185 (input.previousId !== undefined &&
186 input.nextId !== undefined &&
187 input.nextId !== input.previousId) ||
188 input.previousLocation !== input.nextLocation;
189 if (!replace) return undefined;
190 const samePhysical =
191 input.previousLocation === input.nextLocation &&
192 !parentChanged &&
193 input.previousId !== undefined &&
194 input.nextId === input.previousId;
195 return {
196 action: "replace" as const,
197 deleteFirst: samePhysical,
198 };
199};
200
201const READY_STATES = new Set(["ACTIVE", "READY", "SUCCEEDED"]);
202const FAILED_STATES = new Set(["FAILED", "ERROR"]);
203
204export const isReadyState = (state: string | undefined) =>
205 READY_STATES.has((state ?? "").toUpperCase());
206
207export const isFailedState = (state: string | undefined) =>
208 FAILED_STATES.has((state ?? "").toUpperCase());
209
210/**
211 * Wait for a Dataproc Metastore long-running operation. Service creates
212 * and deletes take 20-30 minutes.
213 */
214export const waitForOperation = (operation: metastore.Operation) =>
215 waitForGcpOperation(
216 operation,
217 (name) => metastore.getProjectsLocationsOperations({ name }),
218 { budget: "40 minutes", interval: "10 seconds" },
219 );
220
221export const waitUntilExists = <A, E extends { readonly _tag: string }, R>(
222 get: Effect.Effect<A, E, R>,
223 name: string,
224): Effect.Effect<Exclude<A, undefined>, E | ResourceNotResolved, R> =>
225 get.pipe(
226 Effect.filterOrFail(
227 (value): value is Exclude<A, undefined> => value !== undefined,
228 () => new ResourceNotResolved({ name }),
229 ),
230 Effect.retry({
231 while: (error) => error._tag === "GCP.Metastore.ResourceNotResolved",
232 times: 8,
233 schedule: Schedule.spaced("1 second"),
234 }),
235 );
236
237export const waitUntilGone = <A, E extends { readonly _tag: string }, R>(
238 get: Effect.Effect<A, E, R>,
239 name: string,
240): Effect.Effect<void, E | ResourceStillExists, R> =>
241 get.pipe(
242 Effect.filterOrFail(
243 (value) => value === undefined,
244 () => new ResourceStillExists({ name }),
245 ),
246 Effect.asVoid,
247 Effect.retry({
248 while: (error) => error._tag === "GCP.Metastore.ResourceStillExists",
249 times: 10,
250 schedule: Schedule.spaced("5 seconds"),
251 }),
252 );
253
254export const waitUntilReady = <A, E extends { readonly _tag: string }, R>(
255 get: Effect.Effect<A, E, R>,
256 name: string,
257 stateOf: (value: Exclude<A, undefined>) => string | undefined,
258 detailsOf?: (value: Exclude<A, undefined>) => string | undefined,
259 options?: {
260 times?: number;
261 interval?: `${number} seconds`;
262 },
263): Effect.Effect<
264 Exclude<A, undefined>,
265 E | ResourceNotResolved | ResourceFailed | ResourceNotReady,
266 R
267> =>
268 get.pipe(
269 Effect.filterOrFail(
270 (value): value is Exclude<A, undefined> => value !== undefined,
271 () => new ResourceNotResolved({ name }),
272 ),
273 Effect.filterOrFail(
274 (value) => !isFailedState(stateOf(value)),
275 (value) =>
276 new ResourceFailed({
277 name,
278 state: stateOf(value) ?? "",
279 details: detailsOf?.(value),
280 }),
281 ),
282 Effect.filterOrFail(
283 (value) => {
284 const state = stateOf(value) ?? "";
285 return isReadyState(state) || state.length === 0;
286 },
287 (value) => new ResourceNotReady({ name, state: stateOf(value) ?? "" }),
288 ),
289 Effect.retry({
290 while: (error) =>
291 error._tag === "GCP.Metastore.ResourceNotReady" ||
292 error._tag === "GCP.Metastore.ResourceNotResolved",
293 times: options?.times ?? 8,
294 schedule: Schedule.spaced(options?.interval ?? "3 seconds"),
295 }),
296 );
297
298// Metastore accepts the `locations/-` wildcard, so one call covers every region.
299export const listAtLocation = <A, E, R>(
300 project: string,
301 list: (parent: string) => Effect.Effect<A[], E, R>,
302): Effect.Effect<A[], E, R> => list(`projects/${project}/locations/-`);
303
304export const listAtNested = <A, E, R>(
305 project: string,
306 nested: string,
307 list: (parent: string) => Effect.Effect<A[], E, R>,
308): Effect.Effect<A[], E, R> =>
309 list(`projects/${project}/locations/-/${nested}`);
310
311export const listLabeledPages = <Page, A, E, R>(
312 pages: Stream.Stream<Page, E, R>,
313 items: (page: Page) => readonly A[] | undefined,
314 labelsOf: (item: A) => Record<string, string | undefined> | null | undefined,
315): Effect.Effect<A[], E, R> =>
316 listOwnedPages(pages, items, (item) => hasAlchemyLabelMap(labelsOf(item)));
317
318export const listOwnedPages = <Page, A, E, R>(
319 pages: Stream.Stream<Page, E, R>,
320 items: (page: Page) => readonly A[] | undefined,
321 owned: (item: A) => boolean,
322): Effect.Effect<A[], E, R> =>
323 pages.pipe(
324 Stream.flatMap((page) => Stream.fromIterable(items(page) ?? [])),
325 Stream.filter(owned),
326 Stream.runCollect,
327 Effect.map((chunk) => Array.from(chunk)),
328 );
329
330const markerOf = (
331 labels: Record<string, string>,
332 stack: string,
333 stage: string,
334 id: string,
335) =>
336 `[alchemy ${alchemyLabelKeys.stack}=${stack} ${alchemyLabelKeys.stage}=${stage} ${alchemyLabelKeys.id}=${id}]`;
337
338const fitMarker = (labels: Record<string, string>, maxLength: number) => {
339 let stack = labels[alchemyLabelKeys.stack] ?? "x";
340 let stage = labels[alchemyLabelKeys.stage] ?? "x";
341 let id = labels[alchemyLabelKeys.id] ?? "x";
342 let marker = markerOf(labels, stack, stage, id);
343 while (
344 marker.length > maxLength &&
345 (stack.length > 1 || stage.length > 1 || id.length > 1)
346 ) {
347 if (stack.length >= stage.length && stack.length >= id.length) {
348 stack = stack.slice(0, -1);
349 } else if (stage.length >= id.length) {
350 stage = stage.slice(0, -1);
351 } else {
352 id = id.slice(0, -1);
353 }
354 marker = markerOf(labels, stack, stage, id);
355 }
356 return marker.length <= maxLength
357 ? marker
358 : `${marker.slice(0, maxLength - 1)}]`;
359};
360
361export const encodeOwnership = (
362 labels: Record<string, string>,
363 text: string | undefined,
364): string => {
365 const marker = fitMarker(labels, 8000);
366 const trimmed = text?.trim();
367 return trimmed && trimmed.length > 0 ? `${marker}\n${trimmed}` : marker;
368};
369
370export const parseOwnership = (
371 text: string | undefined,
372): {
373 labels: Record<string, string>;
374 text: string | undefined;
375} => {
376 if (!text?.startsWith("[alchemy ")) {
377 return { labels: {}, text };
378 }
379 const end = text.indexOf("]");
380 if (end < 0) return { labels: {}, text };
381 const labels: Record<string, string> = {};
382 for (const part of text.slice("[alchemy ".length, end).split(/\s+/)) {
383 const eq = part.indexOf("=");
384 if (eq > 0) {
385 labels[part.slice(0, eq)] = part.slice(eq + 1);
386 }
387 }
388 const rest = text.slice(end + 1).replace(/^[\s\n]+/, "");
389 return { labels, text: rest.length > 0 ? rest : undefined };
390};
391
392export const hasOwnershipMarker = (text: string | undefined) =>
393 Object.keys(parseOwnership(text).labels).some((key) =>
394 key.startsWith("alchemy-"),
395 );
396
397const prefixMatch = (expected: string, observed: string) =>
398 expected === observed ||
399 expected.startsWith(observed) ||
400 observed.startsWith(expected);
401
402export const ownedByAlchemy = (id: string, text: string | undefined) =>
403 Effect.gen(function* () {
404 const expected = yield* createInternalLabels(id);
405 const { labels } = parseOwnership(text);
406 if (!hasOwnershipMarker(text)) return false;
407 const exact = yield* hasAlchemyLabels(id, labels);
408 if (exact) return true;
409 return (
410 prefixMatch(
411 expected[alchemyLabelKeys.stack] ?? "",
412 labels[alchemyLabelKeys.stack] ?? "",
413 ) &&
414 prefixMatch(
415 expected[alchemyLabelKeys.stage] ?? "",
416 labels[alchemyLabelKeys.stage] ?? "",
417 ) &&
418 prefixMatch(
419 expected[alchemyLabelKeys.id] ?? "",
420 labels[alchemyLabelKeys.id] ?? "",
421 )
422 );
423 });