1import * as biglake from "@distilled.cloud/gcp/biglake_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 { createPhysicalName } from "../../PhysicalName.ts";
7import { tagRecord } from "../../Tags.ts";
8import {
9 alchemyLabelKeys,
10 createInternalLabels,
11 hasAlchemyLabels,
12 stripInternalLabels,
13 toLabels,
14} from "../Labels.ts";
15import { isTransientGcpError } from "../Errors.ts";
16
17/** Locations `list` scans: the pre-`GCP.Region` default plus the stack region. */
18export const listLocations = (region: string): string[] => [
19 ...new Set(["us-central1", region]),
20];
21export const MAX_ID_LENGTH = 63;
22
23export class BiglakeNotResolved extends Data.TaggedError(
24 "GCP.BigLake.ResourceNotResolved",
25)<{
26 name: string;
27}> {}
28
29export const lastSegment = (value: string) => {
30 const trimmed = value.replace(/\/+$/, "");
31 const parts = trimmed.split("/");
32 return parts[parts.length - 1] || trimmed;
33};
34
35export const locationParent = (project: string, location: string) =>
36 `projects/${project}/locations/${location}`;
37
38export const normalizeLocation = (
39 location: string | undefined,
40 fallback: string,
41) => {
42 const raw = lastSegment(location ?? fallback);
43 const upper = raw.toUpperCase();
44 if (upper === "US" || upper === "EU") return upper;
45 return raw.toLowerCase();
46};
47
48export const parseResourceName = (name: string, collection: string) => {
49 const parts = name.split("/").filter((part) => part.length > 0);
50 const collectionAt = parts.lastIndexOf(collection);
51 const locationsAt = parts.lastIndexOf("locations");
52 const projectsAt = parts.lastIndexOf("projects");
53 const catalogsAt = parts.lastIndexOf("catalogs");
54 const databasesAt = parts.lastIndexOf("databases");
55 return {
56 project:
57 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
58 location:
59 locationsAt >= 0 && parts[locationsAt + 1] ? parts[locationsAt + 1]! : "",
60 catalogId:
61 catalogsAt >= 0 && parts[catalogsAt + 1] ? parts[catalogsAt + 1]! : "",
62 catalog: catalogsAt >= 0 ? parts.slice(0, catalogsAt + 2).join("/") : "",
63 databaseId:
64 databasesAt >= 0 && parts[databasesAt + 1] ? parts[databasesAt + 1]! : "",
65 database: databasesAt >= 0 ? parts.slice(0, databasesAt + 2).join("/") : "",
66 id:
67 collectionAt >= 0 && parts[collectionAt + 1]
68 ? parts[collectionAt + 1]!
69 : lastSegment(name),
70 parent:
71 collectionAt > 0
72 ? parts.slice(0, collectionAt).join("/")
73 : parts.slice(0, Math.max(0, parts.length - 1)).join("/"),
74 };
75};
76
77export const expandCatalog = (
78 value: string,
79 project: string,
80 location: string,
81) =>
82 value.includes("/catalogs/")
83 ? value.replace(/\/+$/, "")
84 : `${locationParent(project, location)}/catalogs/${value}`;
85
86export const expandDatabase = (
87 value: string,
88 project: string,
89 location: string,
90 catalog?: string,
91) => {
92 if (value.includes("/databases/")) return value.replace(/\/+$/, "");
93 const parent =
94 catalog !== undefined && catalog.length > 0
95 ? expandCatalog(catalog, project, location)
96 : locationParent(project, location);
97 return `${parent}/databases/${value}`;
98};
99
100export const rfc1035 = (name: string, maxLength = MAX_ID_LENGTH): string => {
101 let next = name
102 .toLowerCase()
103 .replace(/[^a-z0-9_]/g, "_")
104 .replace(/_+/g, "_")
105 .replace(/^_+|_+$/g, "");
106 if (!/^[a-z]/.test(next)) next = `a${next}`;
107 next = next.slice(0, maxLength).replace(/_+$/g, "");
108 if (next.length === 0) return "catalog";
109 if (next.length < 2) next = `${next}x`.slice(0, maxLength);
110 if (!/[a-z0-9]$/.test(next)) next = `${next.slice(0, maxLength - 1)}0`;
111 return next.slice(0, maxLength);
112};
113
114export const toPhysicalId = (
115 id: string,
116 explicit: string | undefined,
117 existing: string | undefined,
118 maxLength = MAX_ID_LENGTH,
119) =>
120 Effect.gen(function* () {
121 if (explicit !== undefined) return explicit;
122 if (existing !== undefined) return existing;
123 return rfc1035(
124 yield* createPhysicalName({
125 id,
126 maxLength,
127 lowercase: true,
128 delimiter: "_",
129 }),
130 maxLength,
131 );
132 });
133
134export const userLabels = (
135 labels: Record<string, string | undefined> | null | undefined,
136): Record<string, string> => stripInternalLabels(tagRecord(labels));
137
138export const hasAlchemyLabelMap = (
139 labels: Record<string, string | undefined> | null | undefined,
140) => Object.keys(labels ?? {}).some((key) => key.startsWith("alchemy-"));
141
142export const ownershipMarker = (labels: Record<string, string>) =>
143 `[alchemy ${alchemyLabelKeys.stack}=${labels[alchemyLabelKeys.stack]} ${alchemyLabelKeys.stage}=${labels[alchemyLabelKeys.stage]} ${alchemyLabelKeys.id}=${labels[alchemyLabelKeys.id]}]`;
144
145export const parseOwnershipMarker = (text: string | undefined) => {
146 if (!text?.includes("[alchemy ")) return {} as Record<string, string>;
147 const start = text.indexOf("[alchemy ");
148 const end = text.indexOf("]", start);
149 if (end < 0) return {} as Record<string, string>;
150 const labels: Record<string, string> = {};
151 for (const part of text.slice(start + "[alchemy ".length, end).split(/\s+/)) {
152 const eq = part.indexOf("=");
153 if (eq > 0) labels[part.slice(0, eq)] = part.slice(eq + 1);
154 }
155 return labels;
156};
157
158export const hasOwnershipMarker = (text: string | undefined) =>
159 Object.keys(parseOwnershipMarker(text)).some((key) =>
160 key.startsWith("alchemy-"),
161 );
162
163export const sameText = (left: string | undefined, right: string | undefined) =>
164 (left ?? "") === (right ?? "");
165
166export const canonical = (value: unknown): unknown => {
167 if (value === undefined || value === null) return undefined;
168 if (typeof value === "boolean" || typeof value === "number") return value;
169 if (typeof value === "string") {
170 return value.length === 0 ? undefined : value;
171 }
172 if (Array.isArray(value)) {
173 const items = value.map(canonical).filter((item) => item !== undefined);
174 return items.length === 0 ? undefined : items;
175 }
176 if (typeof value === "object") {
177 const entries = Object.entries(value as Record<string, unknown>)
178 .map(([key, item]) => [key, canonical(item)] as const)
179 .filter(([, item]) => item !== undefined)
180 .sort(([left], [right]) => left.localeCompare(right));
181 if (entries.length === 0) return undefined;
182 return Object.fromEntries(entries);
183 }
184 return undefined;
185};
186
187export const fingerprint = (value: unknown): string =>
188 JSON.stringify(canonical(value) ?? null);
189
190export const sameJson = (left: unknown, right: unknown) =>
191 fingerprint(left) === fingerprint(right);
192
193export const updateMaskOf = (...fields: Array<string | undefined>) =>
194 fields.filter((field): field is string => field !== undefined).join(",");
195
196export const replaceOnIdentity = (input: {
197 previousId?: string;
198 nextId?: string;
199 previousParent?: string;
200 nextParent?: string;
201 previousLocation?: string;
202 nextLocation?: string;
203}) => {
204 const idChanged =
205 input.previousId !== undefined &&
206 input.nextId !== undefined &&
207 input.previousId !== input.nextId;
208 const parentChanged =
209 (input.previousParent ?? "") !== "" &&
210 (input.nextParent ?? "") !== "" &&
211 (input.previousParent ?? "") !== input.nextParent;
212 const locationChanged =
213 (input.previousLocation ?? "") !== "" &&
214 (input.nextLocation ?? "") !== "" &&
215 (input.previousLocation ?? "") !== input.nextLocation;
216 if (!idChanged && !parentChanged && !locationChanged) return undefined;
217 const samePhysical =
218 !parentChanged &&
219 !locationChanged &&
220 input.previousId !== undefined &&
221 input.nextId === input.previousId;
222 return {
223 action: "replace" as const,
224 deleteFirst: samePhysical,
225 };
226};
227
228export const retryTransient = <A, E extends { readonly _tag: string }, R>(
229 effect: Effect.Effect<A, E, R>,
230) =>
231 effect.pipe(
232 Effect.retry({
233 while: isTransientGcpError,
234 times: 8,
235 schedule: Schedule.exponential("250 millis"),
236 }),
237 );
238
239/**
240 * BigLake answers a missing catalog, database or table with 403 "... (or it
241 * may not exist)", typed `BigLakeResourceNotFound`.
242 */
243const isMissing = (error: { readonly _tag: string }) =>
244 error._tag === "NotFound" || error._tag === "BigLakeResourceNotFound";
245
246export const ignoreGone = <A, E extends { readonly _tag: string }, R>(
247 effect: Effect.Effect<A, E, R>,
248) =>
249 effect.pipe(
250 Effect.catchIf(
251 (
252 error,
253 ): error is Extract<
254 E,
255 { readonly _tag: "NotFound" | "BigLakeResourceNotFound" }
256 > => isMissing(error),
257 () => Effect.void,
258 ),
259 );
260
261export const missingGet =
262 <A, E extends { readonly _tag: string }, R>(
263 effect: (input: { name: string }) => Effect.Effect<A, E, R>,
264 ) =>
265 (name: string) =>
266 name.length === 0
267 ? Effect.succeed(undefined)
268 : effect({ name }).pipe(
269 Effect.catchIf(
270 (
271 error,
272 ): error is Extract<
273 E,
274 { readonly _tag: "NotFound" | "BigLakeResourceNotFound" }
275 > => isMissing(error),
276 () => Effect.succeed(undefined),
277 ),
278 );
279
280const emptyList = <A>() => Effect.succeed<A[]>([]);
281
282export const collectPages = <
283 Page,
284 Item,
285 E extends { readonly _tag: string },
286 R,
287>(
288 pages: Stream.Stream<Page, E, R>,
289 items: (page: Page) => readonly Item[] | null | undefined,
290) =>
291 pages.pipe(
292 Stream.flatMap((page) => Stream.fromIterable(items(page) ?? [])),
293 Stream.runCollect,
294 Effect.map((chunk) => Array.from(chunk) as Item[]),
295 Effect.catchIf(
296 (
297 error,
298 ): error is Extract<
299 E,
300 { readonly _tag: "NotFound" | "BigLakeResourceNotFound" }
301 > => isMissing(error),
302 () => emptyList<Item>(),
303 ),
304 );
305
306export const listCatalogs = (parent: string) =>
307 parent.length === 0
308 ? emptyList<biglake.Catalog>()
309 : collectPages(
310 biglake.listProjectsLocationsCatalogs.pages({
311 parent,
312 pageSize: 1000,
313 }),
314 (page) => page.catalogs,
315 );
316
317export const listDatabases = (parent: string) =>
318 parent.length === 0
319 ? emptyList<biglake.Database>()
320 : collectPages(
321 biglake.listProjectsLocationsCatalogsDatabases.pages({
322 parent,
323 pageSize: 1000,
324 }),
325 (page) => page.databases,
326 );
327
328export const listTables = (parent: string) =>
329 parent.length === 0
330 ? emptyList<biglake.Table>()
331 : collectPages(
332 biglake.listProjectsLocationsCatalogsDatabasesTables.pages({
333 parent,
334 pageSize: 1000,
335 view: "FULL",
336 }),
337 (page) => page.tables,
338 );
339
340export const namedOf = <T extends { name?: string }>(items: readonly T[]) =>
341 items.filter((item) => (item.name ?? "").length > 0);
342
343export const listChildResources = <A, E, R>(
344 parents: readonly { name?: string }[],
345 list: (name: string) => Effect.Effect<A[], E, R>,
346) =>
347 Effect.forEach(namedOf(parents), (parent) => list(parent.name!), {
348 concurrency: 4,
349 }).pipe(Effect.map((groups) => groups.flat()));
350
351export class DeleteNotConfirmed extends Data.TaggedError(
352 "GCP.BigLake.DeleteNotConfirmed",
353)<{}> {}
354
355/** Poll until the resource is gone; fails if it is still readable after ~60s. */
356export const waitUntilGone = <A, E, R>(
357 get: Effect.Effect<A | undefined, E, R>,
358) =>
359 get.pipe(
360 Effect.repeat({
361 schedule: Schedule.spaced("2 seconds"),
362 until: (value) => value === undefined,
363 times: 30,
364 }),
365 Effect.flatMap((value): Effect.Effect<void, DeleteNotConfirmed> =>
366 value === undefined ? Effect.void : Effect.fail(new DeleteNotConfirmed()),
367 ),
368 );
369
370export const mergeParameters = (
371 parameters: Record<string, string> | undefined,
372 ownership: Record<string, string>,
373): Record<string, string> => ({
374 ...tagRecord(parameters),
375 ...ownership,
376});
377
378export { createInternalLabels, hasAlchemyLabels, toLabels };