1import * as dataplex from "@distilled.cloud/gcp/dataplex_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 { waitForOperation as waitForGcpOperation } from "../Operation.ts";
8import { tagRecord } from "../../Tags.ts";
9import {
10 createInternalLabels,
11 hasAlchemyLabels,
12 stripInternalLabels,
13} from "../Labels.ts";
14
15export const MAX_NAME_LENGTH = 63;
16
17export const GENERIC_ENTRY_TYPE =
18 "projects/dataplex-types/locations/global/entryTypes/generic";
19
20export const RELATED_ENTRY_LINK_TYPE =
21 "projects/dataplex-types/locations/global/entryLinkTypes/related";
22
23export class DataplexNotResolved extends Data.TaggedError(
24 "GCP.Dataplex.ResourceNotResolved",
25)<{
26 name: string;
27}> {}
28
29export class DataplexStillExists extends Data.TaggedError(
30 "GCP.Dataplex.ResourceStillExists",
31)<{
32 name: string;
33}> {}
34
35export const lastSegment = (value: string) => {
36 const trimmed = value.replace(/\/+$/, "");
37 const parts = trimmed.split("/");
38 return parts[parts.length - 1] || trimmed;
39};
40
41export const normalizeLocation = (
42 location: string | undefined,
43 defaultLocation: string,
44) => lastSegment(location ?? defaultLocation).toLowerCase();
45
46export const rfc1035 = (
47 name: string,
48 fallbackOrMax: string | number = "dataplex",
49): string => {
50 const maxLength =
51 typeof fallbackOrMax === "number" ? fallbackOrMax : MAX_NAME_LENGTH;
52 const fallback =
53 typeof fallbackOrMax === "string" ? fallbackOrMax : "dataplex";
54 let next = name
55 .toLowerCase()
56 .replace(/[^a-z0-9-]/g, "-")
57 .replace(/-+/g, "-")
58 .replace(/^-+|-+$/g, "");
59 if (!/^[a-z]/.test(next)) next = `d${next}`;
60 next = next.slice(0, maxLength).replace(/-+$/g, "");
61 if (next.length === 0) return fallback;
62 if (!/[a-z0-9]$/.test(next)) next = `${next.slice(0, maxLength - 1)}0`;
63 return next.slice(0, maxLength);
64};
65
66export const toPhysicalId = (
67 id: string,
68 explicit: string | undefined,
69 existing: string | undefined,
70 fallback: string,
71) =>
72 Effect.gen(function* () {
73 if (explicit !== undefined) return rfc1035(explicit, fallback);
74 if (existing !== undefined) return existing;
75 return rfc1035(
76 yield* createPhysicalName({
77 id,
78 maxLength: MAX_NAME_LENGTH,
79 lowercase: true,
80 }),
81 fallback,
82 );
83 });
84
85export const parseName = (name: string, collection: string) => {
86 const parts = name.split("/").filter((part) => part.length > 0);
87 const collectionAt = parts.lastIndexOf(collection);
88 const locationsAt = parts.lastIndexOf("locations");
89 const projectsAt = parts.lastIndexOf("projects");
90 const orgsAt = parts.lastIndexOf("organizations");
91 return {
92 project:
93 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
94 organization: orgsAt >= 0 && parts[orgsAt + 1] ? parts[orgsAt + 1]! : "",
95 // API resource names always carry a `locations/{location}` segment.
96 location:
97 locationsAt >= 0 && parts[locationsAt + 1] ? parts[locationsAt + 1]! : "",
98 id:
99 collectionAt >= 0
100 ? parts.slice(collectionAt + 1).join("/")
101 : lastSegment(name),
102 parent:
103 collectionAt > 0
104 ? parts.slice(0, collectionAt).join("/")
105 : parts.slice(0, Math.max(0, parts.length - 1)).join("/"),
106 };
107};
108
109export const expandParent = (
110 value: string,
111 project: string,
112 location: string,
113 collection: string,
114) => {
115 if (value.includes("/")) return value.replace(/\/+$/, "");
116 return `projects/${project}/locations/${location}/${collection}/${value}`;
117};
118
119export const parentOf = (project: string, location: string) =>
120 `projects/${project}/locations/${lastSegment(location).toLowerCase()}`;
121
122export const userLabels = (
123 labels: Record<string, string | undefined> | null | undefined,
124): Record<string, string> => stripInternalLabels(tagRecord(labels));
125
126export const hasAlchemyLabelMap = (
127 labels: Record<string, string | undefined> | null | undefined,
128) => Object.keys(labels ?? {}).some((key) => key.startsWith("alchemy-"));
129
130export const canonical = (value: unknown): unknown => {
131 if (value === undefined || value === null) return undefined;
132 if (typeof value === "boolean" || typeof value === "number") return value;
133 if (typeof value === "string") {
134 return value.length === 0 ? undefined : value;
135 }
136 if (Array.isArray(value)) {
137 const items = value.map(canonical).filter((item) => item !== undefined);
138 return items.length === 0 ? undefined : items;
139 }
140 if (typeof value === "object") {
141 const entries = Object.entries(value as Record<string, unknown>)
142 .map(([key, item]) => [key, canonical(item)] as const)
143 .filter(([, item]) => item !== undefined)
144 .sort(([left], [right]) => left.localeCompare(right));
145 if (entries.length === 0) return undefined;
146 return Object.fromEntries(entries);
147 }
148 return undefined;
149};
150
151export const fingerprint = (value: unknown): string =>
152 JSON.stringify(canonical(value) ?? null);
153
154export const retryQuota = <A, E extends { _tag: string }, R>(
155 effect: Effect.Effect<A, E, R>,
156) =>
157 effect.pipe(
158 Effect.retry({
159 while: (error) => error._tag === "TooManyRequests",
160 times: 8,
161 schedule: Schedule.exponential("2 seconds"),
162 }),
163 );
164
165/**
166 * Wait for a Dataplex operation; lakes, zones, and assets provision for
167 * several minutes. ALREADY_EXISTS (code 6) counts as success (create
168 * race); with `notFoundOk`, so does NOT_FOUND (code 5, delete race) and an
169 * operation that is already gone.
170 */
171export const waitForOperation = (
172 operation: dataplex.GoogleLongrunningOperation,
173 options?: { notFoundOk?: boolean },
174) =>
175 waitForGcpOperation(
176 operation,
177 (name) =>
178 retryQuota(
179 name.startsWith("organizations/")
180 ? dataplex.getOrganizationsLocationsOperations({ name })
181 : dataplex.getProjectsLocationsOperations({ name }),
182 ),
183 { budget: "20 minutes", interval: "2 seconds" },
184 ).pipe(
185 Effect.catchIf(
186 (error) =>
187 (error._tag === "GCP.OperationFailed" &&
188 (error.code === 6 ||
189 (options?.notFoundOk === true && error.code === 5))) ||
190 (options?.notFoundOk === true && error._tag === "NotFound"),
191 () => Effect.succeed(operation),
192 ),
193 );
194
195export const waitUntilExists = <A, E extends { readonly _tag: string }, R>(
196 get: Effect.Effect<A | undefined, E, R>,
197 name: string,
198) =>
199 get.pipe(
200 Effect.flatMap((value) =>
201 value
202 ? Effect.succeed(value)
203 : Effect.fail(new DataplexNotResolved({ name })),
204 ),
205 Effect.retry({
206 while: (error) => error._tag === "GCP.Dataplex.ResourceNotResolved",
207 times: 8,
208 schedule: Schedule.spaced("1 second"),
209 }),
210 );
211
212export const waitUntilGone = <A, E extends { readonly _tag: string }, R>(
213 get: Effect.Effect<A | undefined, E, R>,
214 name: string,
215) =>
216 get.pipe(
217 Effect.flatMap((value) =>
218 value === undefined
219 ? Effect.void
220 : Effect.fail(new DataplexStillExists({ name })),
221 ),
222 Effect.retry({
223 while: (error) => error._tag === "GCP.Dataplex.ResourceStillExists",
224 times: 10,
225 schedule: Schedule.spaced("1 second"),
226 }),
227 );
228
229export const collectPages = <Page, A, E, R>(
230 pages: Stream.Stream<Page, E, R>,
231 items: (page: Page) => readonly A[] | undefined,
232) =>
233 pages.pipe(
234 Stream.flatMap((page) => Stream.fromIterable(items(page) ?? [])),
235 Stream.runCollect,
236 Effect.map((chunk) => Array.from(chunk)),
237 );
238
239export const listAtLocation = <A, E, R>(
240 project: string,
241 region: string,
242 list: (parent: string) => Effect.Effect<A[], E, R>,
243) =>
244 // Prefer the all-locations wildcard; fall back to the default region
245 // (whose error, if any, propagates).
246 list(`projects/${project}/locations/-`).pipe(
247 Effect.catch(() => list(`projects/${project}/locations/${region}`)),
248 );
249
250export const replaceOnIdentity = (input: {
251 previousId: string | undefined;
252 nextId: string | undefined;
253 previousLocation: string;
254 nextLocation: string;
255 extra?: boolean;
256 previousParent?: string;
257 nextParent?: string;
258}) => {
259 const parentChanged =
260 (input.previousParent ?? "") !== "" &&
261 (input.nextParent ?? "") !== "" &&
262 lastSegment(input.previousParent ?? "") !==
263 lastSegment(input.nextParent ?? "") &&
264 (input.previousParent ?? "") !== (input.nextParent ?? "");
265 const replace =
266 (input.extra ?? false) ||
267 parentChanged ||
268 (input.previousId !== undefined &&
269 input.nextId !== undefined &&
270 input.nextId !== input.previousId) ||
271 input.previousLocation !== input.nextLocation;
272 if (!replace) return undefined;
273 const samePhysical =
274 input.previousLocation === input.nextLocation &&
275 !parentChanged &&
276 input.previousId !== undefined &&
277 input.nextId === input.previousId;
278 return {
279 action: "replace" as const,
280 deleteFirst: samePhysical,
281 };
282};
283
284export const isPendingState = (state: string | undefined) =>
285 state === "CREATING" ||
286 state === "DELETING" ||
287 state === "UPDATING" ||
288 state === "STATE_UNSPECIFIED" ||
289 state === undefined ||
290 state === "";
291
292export const snakeId = (name: string, maxLength = 256): string => {
293 let next = name
294 .toLowerCase()
295 .replace(/[^a-z0-9_]/g, "_")
296 .replace(/_+/g, "_")
297 .replace(/^_+|_+$/g, "");
298 if (!/^[a-z]/.test(next)) next = `e${next}`;
299 next = next.slice(0, maxLength).replace(/_+$/g, "");
300 if (next.length === 0) return "entity";
301 if (!/[a-z0-9]$/.test(next)) next = `${next.slice(0, maxLength - 1)}0`;
302 return next.slice(0, maxLength);
303};
304
305export const toPhysicalSnake = (
306 id: string,
307 explicit: string | undefined,
308 existing: string | undefined,
309 maxLength = 256,
310) =>
311 Effect.gen(function* () {
312 if (explicit !== undefined) return snakeId(explicit, maxLength);
313 if (existing !== undefined) return existing;
314 return snakeId(
315 yield* createPhysicalName({
316 id,
317 maxLength,
318 lowercase: true,
319 delimiter: "_",
320 }),
321 maxLength,
322 );
323 });
324
325export const createOwnership = (id: string) => createInternalLabels(id);
326
327export const ownedLabels = (id: string, labels: Record<string, string>) =>
328 hasAlchemyLabels(id, labels);
329
330export const parseResourceName = parseName;
331export const locationParent = parentOf;
332export const hasAlchemyLabelKeys = hasAlchemyLabelMap;
333export const LABELS_FILTER = "labels.alchemy-id:*";
334export const sameJson = (left: unknown, right: unknown) =>
335 fingerprint(left) === fingerprint(right);
336export const sameStringList = (
337 left: readonly string[] | undefined,
338 right: readonly string[] | undefined,
339) =>
340 JSON.stringify([...(left ?? [])].sort()) ===
341 JSON.stringify([...(right ?? [])].sort());
342export const toResourceId = (
343 id: string,
344 explicit: string | undefined,
345 existing: string | undefined,
346) => toPhysicalId(id, explicit, existing, "dataplex");
347export const updateMaskOf = (
348 fields: ReadonlyArray<readonly [boolean, string]>,
349): string =>
350 fields
351 .filter(([changed]) => changed)
352 .map(([, field]) => field)
353 .join(",");
354export const emptyOnMissing = <A, E extends { readonly _tag: string }, R>(
355 effect: Effect.Effect<A[], E, R>,
356) =>
357 effect.pipe(
358 Effect.catchIf(
359 (error) => error._tag === "NotFound",
360 () => Effect.succeed([] as A[]),
361 ),
362 );