1import * as datalineage from "@distilled.cloud/gcp/datalineage_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 {
8 createInternalLabels,
9 hasAlchemyLabels,
10 stripInternalLabels,
11} from "../Labels.ts";
12import { waitForOperation as waitForGcpOperation } from "../Operation.ts";
13
14export const MAX_ID_LENGTH = 200;
15export const MAX_DISPLAY_NAME_LENGTH = 200;
16
17export class DatalineageNotResolved extends Data.TaggedError(
18 "GCP.DataLineage.ResourceNotResolved",
19)<{
20 name: string;
21}> {}
22
23export class DatalineageStillExists extends Data.TaggedError(
24 "GCP.DataLineage.ResourceStillExists",
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 normalizeLocation = (
36 location: string | undefined,
37 fallback: string,
38) => lastSegment(location ?? fallback).toLowerCase();
39
40export const locationParent = (project: string, location: string) =>
41 `projects/${project}/locations/${lastSegment(location).toLowerCase()}`;
42
43export const parentOf = (name: string) => {
44 const parts = name.split("/").filter((part) => part.length > 0);
45 return parts.slice(0, Math.max(0, parts.length - 2)).join("/");
46};
47
48export const parseName = (
49 name: string,
50 collection: string,
51 fallbackLocation: string,
52) => {
53 const parts = name.split("/").filter((part) => part.length > 0);
54 const collectionAt = parts.lastIndexOf(collection);
55 const locationsAt = parts.lastIndexOf("locations");
56 const projectsAt = parts.lastIndexOf("projects");
57 return {
58 project:
59 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
60 location:
61 locationsAt >= 0 && parts[locationsAt + 1]
62 ? parts[locationsAt + 1]!
63 : fallbackLocation,
64 id:
65 collectionAt >= 0 && parts[collectionAt + 1]
66 ? parts[collectionAt + 1]!
67 : lastSegment(name),
68 parent:
69 collectionAt > 0
70 ? parts.slice(0, collectionAt).join("/")
71 : parentOf(name),
72 };
73};
74
75export const expandParent = (
76 value: string,
77 project: string,
78 location: string,
79 collection: string,
80) => {
81 if (value.includes("/")) return value.replace(/\/+$/, "");
82 return `projects/${project}/locations/${location}/${collection}/${value}`;
83};
84
85export const processOf = (value: string, project: string, location: string) =>
86 expandParent(value, project, location, "processes");
87
88export const runOf = (value: string, project: string, location: string) =>
89 expandParent(value, project, location, "runs");
90
91export const sanitizeId = (name: string, fallback = "lineage"): string => {
92 let next = name
93 .toLowerCase()
94 .replace(/[^a-z0-9_\-.]/g, "-")
95 .replace(/-+/g, "-")
96 .replace(/^-+|-+$/g, "");
97 if (next.length === 0) return fallback;
98 if (!/^[a-z0-9]/.test(next)) next = `p${next}`;
99 next = next.slice(0, MAX_ID_LENGTH).replace(/[-.]+$/g, "");
100 return next.length > 0 ? next : fallback;
101};
102
103export const toPhysicalId = (
104 id: string,
105 explicit: string | undefined,
106 existing: string | undefined,
107 fallback = "lineage",
108) =>
109 Effect.gen(function* () {
110 if (explicit !== undefined) return sanitizeId(explicit, fallback);
111 if (existing !== undefined) return existing;
112 return sanitizeId(
113 yield* createPhysicalName({
114 id,
115 maxLength: 63,
116 lowercase: true,
117 }),
118 fallback,
119 );
120 });
121
122export const sameText = (left: string | undefined, right: string | undefined) =>
123 (left ?? "") === (right ?? "");
124
125export const canonical = (value: unknown): unknown => {
126 if (value === undefined || value === null) return undefined;
127 if (typeof value === "boolean" || typeof value === "number") return value;
128 if (typeof value === "string") {
129 return value.length === 0 ? undefined : value;
130 }
131 if (Array.isArray(value)) {
132 const items = value.map(canonical).filter((item) => item !== undefined);
133 return items.length === 0 ? undefined : items;
134 }
135 if (typeof value === "object") {
136 const entries = Object.entries(value as Record<string, unknown>)
137 .map(([key, item]) => [key, canonical(item)] as const)
138 .filter(([, item]) => item !== undefined)
139 .sort(([left], [right]) => left.localeCompare(right));
140 if (entries.length === 0) return undefined;
141 return Object.fromEntries(entries);
142 }
143 return undefined;
144};
145
146export const fingerprint = (value: unknown): string =>
147 JSON.stringify(canonical(value) ?? null);
148
149export const sameJson = (left: unknown, right: unknown) =>
150 fingerprint(left) === fingerprint(right);
151
152export const replaceOnIdentity = (input: {
153 previousId?: string;
154 nextId?: string;
155 previousParent?: string;
156 nextParent?: string;
157 extra?: boolean;
158}) => {
159 if (input.extra === true) {
160 return { action: "replace" as const, deleteFirst: false };
161 }
162 if (
163 input.previousId !== undefined &&
164 input.nextId !== undefined &&
165 input.previousId !== input.nextId
166 ) {
167 return { action: "replace" as const, deleteFirst: true };
168 }
169 if (
170 input.previousParent !== undefined &&
171 input.nextParent !== undefined &&
172 input.previousParent !== input.nextParent
173 ) {
174 return { action: "replace" as const, deleteFirst: false };
175 }
176 return undefined;
177};
178
179export const attributeTags = (
180 attributes: datalineage.DocumentMap | null | undefined,
181): Record<string, string> => {
182 const tags: Record<string, string> = {};
183 for (const [key, value] of Object.entries(attributes ?? {})) {
184 if (value === undefined) continue;
185 tags[key] = typeof value === "string" ? value : JSON.stringify(value);
186 }
187 return tags;
188};
189
190export const userAttributes = (
191 attributes: datalineage.DocumentMap | null | undefined,
192): Record<string, unknown> => {
193 const next: Record<string, unknown> = {};
194 for (const [key, value] of Object.entries(
195 stripInternalLabels(attributeTags(attributes)),
196 )) {
197 const raw = attributes?.[key];
198 next[key] = raw !== undefined ? raw : value;
199 }
200 return next;
201};
202
203export const desiredAttributes = (
204 attributes: Record<string, unknown> | undefined,
205 ownership: Record<string, string>,
206): datalineage.DocumentMap => ({
207 ...(attributes ?? {}),
208 ...ownership,
209});
210
211export const hasAlchemyAttributeMap = (
212 attributes: datalineage.DocumentMap | null | undefined,
213) => Object.keys(attributes ?? {}).some((key) => key.startsWith("alchemy-"));
214
215export const ownedByAlchemy = (
216 id: string,
217 attributes: datalineage.DocumentMap | null | undefined,
218) => hasAlchemyLabels(id, attributeTags(attributes));
219
220export const collectPages = <Page, Item, E, R>(
221 pages: Stream.Stream<Page, E, R>,
222 items: (page: Page) => readonly Item[] | null | undefined,
223) =>
224 pages.pipe(
225 Stream.flatMap((page) => Stream.fromIterable(items(page) ?? [])),
226 Stream.runCollect,
227 Effect.map((chunk) => Array.from(chunk) as Item[]),
228 );
229
230const emptyList = <A>() => Effect.succeed<A[]>([]);
231
232const emptyOnMissing = <A, E extends { readonly _tag: string }, R>(
233 effect: Effect.Effect<A[], E, R>,
234) =>
235 effect.pipe(
236 Effect.catchIf(
237 (error) => error._tag === "NotFound",
238 () => emptyList<A>(),
239 ),
240 );
241
242export const listProcesses = (parent: string) =>
243 parent.length === 0
244 ? emptyList<datalineage.GoogleCloudDatacatalogLineageV1Process>()
245 : emptyOnMissing(
246 collectPages(
247 datalineage.listProjectsLocationsProcesses.pages({
248 parent,
249 pageSize: 100,
250 }),
251 (page) => page.processes,
252 ),
253 );
254
255export const listRuns = (parent: string) =>
256 parent.length === 0
257 ? emptyList<datalineage.GoogleCloudDatacatalogLineageV1Run>()
258 : emptyOnMissing(
259 collectPages(
260 datalineage.listProjectsLocationsProcessesRuns.pages({
261 parent,
262 pageSize: 100,
263 }),
264 (page) => page.runs,
265 ),
266 );
267
268export const listLineageEvents = (parent: string) =>
269 parent.length === 0
270 ? emptyList<datalineage.GoogleCloudDatacatalogLineageV1LineageEvent>()
271 : emptyOnMissing(
272 collectPages(
273 datalineage.listProjectsLocationsProcessesRunsLineageEvents.pages({
274 parent,
275 pageSize: 100,
276 }),
277 (page) => page.lineageEvents,
278 ),
279 );
280
281export const listOwnedProcesses = (project: string, location: string) =>
282 listProcesses(locationParent(project, location)).pipe(
283 Effect.map((rows) =>
284 rows.filter((row) => hasAlchemyAttributeMap(row.attributes)),
285 ),
286 );
287
288export const findOwnedProcess = (
289 id: string,
290 project: string,
291 location: string,
292) =>
293 Effect.gen(function* () {
294 const rows = yield* listProcesses(locationParent(project, location));
295 for (const row of rows) {
296 if (yield* ownedByAlchemy(id, row.attributes)) return row;
297 }
298 return undefined;
299 });
300
301export const findOwnedRun = (id: string, process: string) =>
302 Effect.gen(function* () {
303 const rows = yield* listRuns(process);
304 for (const row of rows) {
305 if (yield* ownedByAlchemy(id, row.attributes)) return row;
306 }
307 return undefined;
308 });
309
310/** Wait for an operation through the shared GCP waiter. */
311export const waitForOperation = (
312 operation: datalineage.GoogleLongrunningOperation,
313 options?: { notFoundOk?: boolean; alreadyExistsOk?: boolean },
314) =>
315 waitForGcpOperation(
316 operation,
317 (name) =>
318 datalineage.getProjectsLocationsOperations({ name }).pipe(
319 Effect.catchTag("NotFound", (error) =>
320 options?.notFoundOk === true
321 ? Effect.succeed<datalineage.GoogleLongrunningOperation>({
322 name,
323 done: true,
324 })
325 : Effect.fail(error),
326 ),
327 ),
328 { budget: "10 minutes" },
329 ).pipe(
330 // ALREADY_EXISTS (6): a concurrent create won the race. NOT_FOUND (5)
331 // is success for a delete.
332 Effect.catchIf(
333 (error) =>
334 error._tag === "GCP.OperationFailed" &&
335 ((options?.alreadyExistsOk !== false && error.code === 6) ||
336 (options?.notFoundOk === true && error.code === 5)),
337 () => Effect.void,
338 ),
339 // Re-read the finished operation for its typed response and metadata.
340 Effect.flatMap(() =>
341 operation.name === undefined || operation.name.length === 0
342 ? Effect.succeed(operation)
343 : datalineage
344 .getProjectsLocationsOperations({ name: operation.name })
345 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(operation))),
346 ),
347 );
348
349export const waitUntilGone = <A, E, R>(
350 get: Effect.Effect<A | undefined, E, R>,
351 name: string,
352): Effect.Effect<void, E | DatalineageStillExists, R> =>
353 get.pipe(
354 Effect.filterOrFail(
355 (value) => value === undefined,
356 () => new DatalineageStillExists({ name }),
357 ),
358 Effect.retry({
359 while: (error) => error instanceof DatalineageStillExists,
360 times: 10,
361 schedule: Schedule.spaced("1 second"),
362 }),
363 Effect.asVoid,
364 );
365
366export const createOwnership = (id: string) => createInternalLabels(id);