1import * as datalineage from "@distilled.cloud/gcp/datalineage_v1";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import * as Provider from "../../Provider.ts";
7import { Resource } from "../../Resource.ts";
8import { GcpEnvironment } from "../Environment.ts";
9import type { Providers } from "../Providers.ts";
10import {
11 fingerprint,
12 hasAlchemyAttributeMap,
13 listLineageEvents,
14 listOwnedProcesses,
15 listRuns,
16 normalizeLocation,
17 parseName,
18 processOf,
19 replaceOnIdentity,
20 toPhysicalId,
21 waitUntilGone,
22} from "./internal.ts";
23
24export type LineageEntityReference = {
25 /**
26 * Fully qualified name of the entity, for example
27 * `bigquery:project.dataset.table` or `custom:project.system.table`.
28 */
29 fullyQualifiedName: string;
30 /**
31 * Optional nested field path within the entity. Empty for asset-level
32 * lineage.
33 */
34 field?: string[];
35};
36
37export type LineageEventLink = {
38 /** Source entity. */
39 source: LineageEntityReference;
40 /** Target entity. */
41 target: LineageEntityReference;
42 /**
43 * How the target depends on the source (`EXACT_COPY` or `OTHER`).
44 */
45 dependencyType?:
46 | datalineage.GoogleCloudDatacatalogLineageV1DependencyInfoDependencyTypeEnum
47 | (string & {});
48};
49
50export type ProcessesRunsLineageEventProps = {
51 /**
52 * Parent run. Full name
53 * `projects/{project}/locations/{location}/processes/{process}/runs/{run}`
54 * or a run id combined with `process`. Immutable — changing it replaces
55 * the event.
56 */
57 run: string;
58 /**
59 * Parent process used when `run` is a bare id. Full process name or
60 * process id combined with `location`.
61 */
62 process?: string;
63 /**
64 * Event id (the `{lineage_event}` segment). If omitted, a unique name
65 * is generated. Max 200 characters: letters, digits, underscores,
66 * hyphens, colons, periods. Immutable — changing it replaces the event.
67 */
68 lineageEventId?: string;
69 /**
70 * Region used when `run` / `process` is a bare id.
71 * @default the stack's GCP region (`GCP.Region`, profile region, `us-central1`)
72 */
73 location?: string;
74 /**
75 * RFC3339 start of the transformation. Required by the API; Alchemy
76 * reuses the observed value on subsequent reconciles when omitted.
77 */
78 startTime?: string;
79 /**
80 * RFC3339 end of the transformation.
81 */
82 endTime?: string;
83 /**
84 * Source-target links (max 100). Lineage events have no labels or
85 * attributes field, so Alchemy lists events under alchemy-owned
86 * parent processes for nuke.
87 */
88 links?: LineageEventLink[];
89};
90
91export type ProcessesRunsLineageEvent = Resource<
92 "GCP.DataLineage.ProcessesRunsLineageEvent",
93 ProcessesRunsLineageEventProps,
94 {
95 /** Full resource name `.../runs/{run}/lineageEvents/{lineage_event}`. */
96 name: string;
97 /** Lineage event id (last path segment). */
98 lineageEventId: string;
99 /** Parent run resource name. */
100 run: string;
101 /** Parent process resource name. */
102 process: string;
103 /** Project id. */
104 project: string;
105 /** Location id. */
106 location: string;
107 /** RFC3339 start time. */
108 startTime: string | undefined;
109 /** RFC3339 end time. */
110 endTime: string | undefined;
111 /** Source-target links. */
112 links: LineageEventLink[];
113 },
114 never,
115 Providers
116>;
117
118/**
119 * A Data Lineage event — one operation that moved data from source
120 * assets to target assets.
121 *
122 * Events are immutable after create (no patch API). Identity is the
123 * parent run plus `lineageEventId`. Changing links or timestamps
124 * replaces the event. Events have no labels, so `list` / nuke walks
125 * alchemy-owned parent processes.
126 *
127 * ### Creating a Lineage Event
128 * **Example:** One source-to-target link
129 * ```typescript
130 * const event = yield* GCP.DataLineage.ProcessesRunsLineageEvent("Load", {
131 * run: run.name,
132 * startTime: "2024-01-01T00:00:00Z",
133 * links: [
134 * {
135 * source: { fullyQualifiedName: "custom:proj.raw.orders" },
136 * target: { fullyQualifiedName: "custom:proj.dw.orders" },
137 * },
138 * ],
139 * });
140 * ```
141 *
142 * @resource
143 * @category DataLineage
144 */
145export const ProcessesRunsLineageEvent = Resource<ProcessesRunsLineageEvent>(
146 "GCP.DataLineage.ProcessesRunsLineageEvent",
147);
148
149/** Alias matching the Data Lineage resource name. */
150export const LineageEvent = ProcessesRunsLineageEvent;
151export type LineageEvent = ProcessesRunsLineageEvent;
152
153export class ProcessesRunsLineageEventNotResolved extends Data.TaggedError(
154 "GCP.DataLineage.ProcessesRunsLineageEventNotResolved",
155)<{
156 name: string;
157}> {}
158
159const resourceName = (run: string, lineageEventId: string) =>
160 `${run}/lineageEvents/${lineageEventId}`;
161
162const nowIso = () => Effect.sync(() => new Date().toISOString());
163
164const runOf = (
165 run: string,
166 process: string | undefined,
167 project: string,
168 location: string,
169) => {
170 if (run.includes("/")) return run.replace(/\/+$/, "");
171 const parent = processOf(process ?? "", project, location);
172 return `${parent}/runs/${run}`;
173};
174
175const linksOf = (
176 links: readonly LineageEventLink[] | undefined,
177): datalineage.GoogleCloudDatacatalogLineageV1EventLink[] =>
178 (links ?? []).map((link) => ({
179 source: {
180 fullyQualifiedName: link.source.fullyQualifiedName,
181 field: link.source.field,
182 },
183 target: {
184 fullyQualifiedName: link.target.fullyQualifiedName,
185 field: link.target.field,
186 },
187 dependencyInfo:
188 link.dependencyType !== undefined
189 ? { dependencyType: link.dependencyType }
190 : undefined,
191 }));
192
193const linksFrom = (
194 links:
195 | readonly datalineage.GoogleCloudDatacatalogLineageV1EventLink[]
196 | undefined,
197): LineageEventLink[] =>
198 (links ?? [])
199 .filter(
200 (link) =>
201 (link.source?.fullyQualifiedName ?? "").length > 0 &&
202 (link.target?.fullyQualifiedName ?? "").length > 0,
203 )
204 .map((link) => ({
205 source: {
206 fullyQualifiedName: link.source?.fullyQualifiedName ?? "",
207 field: link.source?.field,
208 },
209 target: {
210 fullyQualifiedName: link.target?.fullyQualifiedName ?? "",
211 field: link.target?.field,
212 },
213 dependencyType: link.dependencyInfo?.dependencyType,
214 }));
215
216const toAttrs = (
217 event: datalineage.GoogleCloudDatacatalogLineageV1LineageEvent,
218 project: string,
219 region: string,
220) => {
221 const name = event.name ?? "";
222 const parsed = parseName(name, "lineageEvents", region);
223 const process = parseName(parsed.parent, "runs", region).parent;
224 return {
225 name,
226 lineageEventId: parsed.id,
227 run: parsed.parent,
228 process,
229 project: parsed.project || project,
230 location: parsed.location,
231 startTime: event.startTime,
232 endTime: event.endTime,
233 links: linksFrom(event.links),
234 };
235};
236
237const getByName = (name: string) =>
238 name.length === 0
239 ? Effect.succeed(undefined)
240 : datalineage
241 .getProjectsLocationsProcessesRunsLineageEvents({ name })
242 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
243
244const getProcess = (name: string) =>
245 name.length === 0
246 ? Effect.succeed(undefined)
247 : datalineage
248 .getProjectsLocationsProcesses({ name })
249 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
250
251const identityFingerprint = (news: {
252 startTime?: string;
253 endTime?: string;
254 links?: LineageEventLink[];
255}) =>
256 fingerprint({
257 startTime: news.startTime,
258 endTime: news.endTime,
259 links: linksOf(news.links),
260 });
261
262export const ProcessesRunsLineageEventProvider = () =>
263 Provider.succeed(ProcessesRunsLineageEvent, {
264 stables: [
265 "name",
266 "lineageEventId",
267 "run",
268 "process",
269 "project",
270 "location",
271 ],
272
273 diff: Effect.fn(function* ({ news, olds, output }) {
274 if (!isResolved(news)) return undefined;
275 const env = yield* GcpEnvironment.current;
276 const previousLocation = normalizeLocation(
277 olds?.location ?? output?.location,
278 env.region,
279 );
280 const nextLocation = normalizeLocation(
281 news.location ?? olds?.location ?? output?.location,
282 env.region,
283 );
284 const previous = identityFingerprint({
285 startTime: olds?.startTime ?? output?.startTime,
286 endTime: olds?.endTime ?? output?.endTime,
287 links: olds?.links ?? output?.links,
288 });
289 const next = identityFingerprint({
290 startTime: news.startTime ?? olds?.startTime ?? output?.startTime,
291 endTime: news.endTime ?? olds?.endTime ?? output?.endTime,
292 links: news.links ?? olds?.links ?? output?.links,
293 });
294 return replaceOnIdentity({
295 previousId: olds?.lineageEventId ?? output?.lineageEventId,
296 nextId: news.lineageEventId,
297 previousParent: olds?.run ?? output?.run,
298 nextParent: news.run,
299 extra:
300 (output !== undefined && previousLocation !== nextLocation) ||
301 (output !== undefined && previous !== next),
302 });
303 }),
304
305 read: Effect.fn(function* ({ id, olds, output }) {
306 const env = yield* GcpEnvironment.current;
307 const location = normalizeLocation(
308 olds?.location ?? output?.location,
309 env.region,
310 );
311 const run = runOf(
312 olds?.run ?? output?.run ?? "",
313 olds?.process ?? output?.process,
314 env.project,
315 location,
316 );
317 const lineageEventId = yield* toPhysicalId(
318 id,
319 olds?.lineageEventId,
320 output?.lineageEventId,
321 );
322 const name = output?.name ?? resourceName(run, lineageEventId);
323 const existing = yield* getByName(name);
324 if (existing === undefined) return undefined;
325 const attrs = toAttrs(existing, env.project, env.region);
326 const parent = yield* getProcess(attrs.process);
327 return parent === undefined || hasAlchemyAttributeMap(parent.attributes)
328 ? attrs
329 : Unowned(attrs);
330 }),
331
332 list: () =>
333 Effect.gen(function* () {
334 const env = yield* GcpEnvironment.current;
335 const processes = yield* listOwnedProcesses(env.project, env.region);
336 const runs = (yield* Effect.forEach(
337 processes,
338 (process) => listRuns(process.name ?? ""),
339 { concurrency: 4 },
340 )).flat();
341 const events = (yield* Effect.forEach(
342 runs,
343 (run) => listLineageEvents(run.name ?? ""),
344 { concurrency: 4 },
345 )).flat();
346 return events.map((event) => toAttrs(event, env.project, env.region));
347 }),
348
349 reconcile: Effect.fn(function* ({ id, news, output }) {
350 const env = yield* GcpEnvironment.current;
351 const location = normalizeLocation(
352 news.location ?? output?.location,
353 env.region,
354 );
355 const run = runOf(news.run, news.process, env.project, location);
356 const lineageEventId = yield* toPhysicalId(
357 id,
358 news.lineageEventId,
359 output?.lineageEventId,
360 );
361 const name = output?.name ?? resourceName(run, lineageEventId);
362
363 let current = yield* getByName(name);
364
365 if (current === undefined) {
366 const startTime = news.startTime ?? (yield* nowIso());
367 const created = yield* datalineage
368 .createProjectsLocationsProcessesRunsLineageEvents({
369 parent: run,
370 body: {
371 name,
372 startTime,
373 endTime: news.endTime,
374 links: linksOf(news.links),
375 },
376 })
377 .pipe(Effect.catchTag("Conflict", () => getByName(name)));
378 current = created ?? undefined;
379 }
380
381 if (current === undefined) {
382 return yield* new ProcessesRunsLineageEventNotResolved({ name });
383 }
384
385 return toAttrs(current, env.project, env.region);
386 }),
387
388 delete: Effect.fn(function* ({ output }) {
389 if (!output.name) return;
390 yield* datalineage
391 .deleteProjectsLocationsProcessesRunsLineageEvents({
392 name: output.name,
393 allowMissing: true,
394 })
395 .pipe(Effect.catchTag("NotFound", () => Effect.void));
396 yield* waitUntilGone(getByName(output.name), output.name);
397 }),
398 });