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 createOwnership,
12 desiredAttributes,
13 findOwnedRun,
14 hasAlchemyAttributeMap,
15 listOwnedProcesses,
16 listRuns,
17 normalizeLocation,
18 ownedByAlchemy,
19 parseName,
20 processOf,
21 replaceOnIdentity,
22 sameJson,
23 sameText,
24 toPhysicalId,
25 userAttributes,
26 waitForOperation,
27 waitUntilGone,
28} from "./internal.ts";
29
30export type ProcessesRunProps = {
31 /**
32 * Parent process. Full name
33 * `projects/{project}/locations/{location}/processes/{process}` or the
34 * process id (combined with `location`). Immutable — changing it
35 * replaces the run.
36 */
37 process: string;
38 /**
39 * Run id (the `{run}` segment of
40 * `.../processes/{process}/runs/{run}`). If omitted, a unique name is
41 * generated. Max 200 characters: letters, digits, underscores, hyphens,
42 * colons, periods. Immutable — changing it replaces the run.
43 */
44 runId?: string;
45 /**
46 * Region used when `process` is a bare id.
47 * @default the stack's GCP region (`GCP.Region`, profile region, `us-central1`)
48 */
49 location?: string;
50 /**
51 * RFC3339 start time. Required by the API; Alchemy reuses the observed
52 * value on updates when omitted.
53 */
54 startTime?: string;
55 /**
56 * RFC3339 end time.
57 */
58 endTime?: string;
59 /**
60 * Run state.
61 * @default "STARTED"
62 */
63 state?:
64 | datalineage.GoogleCloudDatacatalogLineageV1RunStateEnum
65 | (string & {});
66 /**
67 * Human-readable name shown in UIs. Max 200 characters.
68 */
69 displayName?: string;
70 /**
71 * Non-semantic attributes. Runs have no labels field, so Alchemy stamps
72 * `alchemy-stack` / `alchemy-stage` / `alchemy-id` here for `list` /
73 * nuke and strips them from attributes.
74 */
75 attributes?: Record<string, unknown>;
76};
77
78export type ProcessesRun = Resource<
79 "GCP.DataLineage.ProcessesRun",
80 ProcessesRunProps,
81 {
82 /** Full resource name `.../processes/{process}/runs/{run}`. */
83 name: string;
84 /** Run id (last path segment). */
85 runId: string;
86 /** Parent process resource name. */
87 process: string;
88 /** Project id. */
89 project: string;
90 /** Location id. */
91 location: string;
92 /** RFC3339 start time. */
93 startTime: string | undefined;
94 /** RFC3339 end time. */
95 endTime: string | undefined;
96 /** Run state (`STARTED`, `COMPLETED`, …). */
97 state: string | undefined;
98 /** Human-readable display name. */
99 displayName: string | undefined;
100 /** User attributes (Alchemy ownership keys stripped). */
101 attributes: Record<string, unknown>;
102 },
103 never,
104 Providers
105>;
106
107/**
108 * A Data Lineage run — one execution of a process that produces lineage
109 * events.
110 *
111 * Runs have no labels field, so Alchemy stamps ownership into
112 * `attributes` for `list` / nuke. Parent process, `runId`, and location
113 * are identity. State, timestamps, display name, and attributes update
114 * in place.
115 *
116 * ### Creating a Run
117 * **Example:** Started run under a process
118 * ```typescript
119 * const run = yield* GCP.DataLineage.ProcessesRun("Nightly", {
120 * process: process.name,
121 * startTime: "2024-01-01T00:00:00Z",
122 * state: "STARTED",
123 * attributes: { env: "dev" },
124 * });
125 * ```
126 *
127 * **Example:** Completed run
128 * ```typescript
129 * const run = yield* GCP.DataLineage.ProcessesRun("Nightly", {
130 * process: process.name,
131 * startTime: "2024-01-01T00:00:00Z",
132 * endTime: "2024-01-01T01:00:00Z",
133 * state: "COMPLETED",
134 * displayName: "nightly 2024-01-01",
135 * });
136 * ```
137 *
138 * @resource
139 * @category DataLineage
140 */
141export const ProcessesRun = Resource<ProcessesRun>(
142 "GCP.DataLineage.ProcessesRun",
143);
144
145/** Alias matching the Data Lineage resource name. */
146export const Run = ProcessesRun;
147export type Run = ProcessesRun;
148
149export class ProcessesRunNotResolved extends Data.TaggedError(
150 "GCP.DataLineage.ProcessesRunNotResolved",
151)<{
152 name: string;
153}> {}
154
155const resourceName = (process: string, runId: string) =>
156 `${process}/runs/${runId}`;
157
158const nowIso = () => Effect.sync(() => new Date().toISOString());
159
160const toAttrs = (
161 run: datalineage.GoogleCloudDatacatalogLineageV1Run,
162 project: string,
163 region: string,
164) => {
165 const name = run.name ?? "";
166 const parsed = parseName(name, "runs", region);
167 return {
168 name,
169 runId: parsed.id,
170 process: parsed.parent,
171 project: parsed.project || project,
172 location: parsed.location,
173 startTime: run.startTime,
174 endTime: run.endTime,
175 state: run.state,
176 displayName: run.displayName,
177 attributes: userAttributes(run.attributes),
178 };
179};
180
181const getByName = (name: string) =>
182 name.length === 0
183 ? Effect.succeed(undefined)
184 : datalineage
185 .getProjectsLocationsProcessesRuns({ name })
186 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
187
188export const ProcessesRunProvider = () =>
189 Provider.succeed(ProcessesRun, {
190 stables: ["name", "runId", "process", "project", "location"],
191
192 diff: Effect.fn(function* ({ news, olds, output }) {
193 if (!isResolved(news)) return undefined;
194 const env = yield* GcpEnvironment.current;
195 const previousLocation = normalizeLocation(
196 olds?.location ?? output?.location,
197 env.region,
198 );
199 const nextLocation = normalizeLocation(
200 news.location ?? olds?.location ?? output?.location,
201 env.region,
202 );
203 const previousProcess = olds?.process ?? output?.process;
204 return replaceOnIdentity({
205 previousId: olds?.runId ?? output?.runId,
206 nextId: news.runId,
207 previousParent: previousProcess,
208 nextParent: news.process,
209 extra: output !== undefined && previousLocation !== nextLocation,
210 });
211 }),
212
213 read: Effect.fn(function* ({ id, olds, output }) {
214 const env = yield* GcpEnvironment.current;
215 const location = normalizeLocation(
216 olds?.location ?? output?.location,
217 env.region,
218 );
219 const process = processOf(
220 olds?.process ?? output?.process ?? "",
221 env.project,
222 location,
223 );
224 const runId = yield* toPhysicalId(id, olds?.runId, output?.runId);
225 const name = output?.name ?? resourceName(process, runId);
226 const existing =
227 (yield* getByName(name)) ?? (yield* findOwnedRun(id, process));
228 if (existing === undefined) return undefined;
229 const attrs = toAttrs(existing, env.project, env.region);
230 return (yield* ownedByAlchemy(id, existing.attributes))
231 ? attrs
232 : Unowned(attrs);
233 }),
234
235 list: () =>
236 Effect.gen(function* () {
237 const env = yield* GcpEnvironment.current;
238 const processes = yield* listOwnedProcesses(env.project, env.region);
239 const groups = yield* Effect.forEach(
240 processes,
241 (process) => listRuns(process.name ?? ""),
242 { concurrency: 4 },
243 );
244 return groups
245 .flat()
246 .filter((run) => hasAlchemyAttributeMap(run.attributes))
247 .map((run) => toAttrs(run, env.project, env.region));
248 }),
249
250 reconcile: Effect.fn(function* ({ id, news, output }) {
251 const env = yield* GcpEnvironment.current;
252 const location = normalizeLocation(
253 news.location ?? output?.location,
254 env.region,
255 );
256 const process = processOf(news.process, env.project, location);
257 const runId = yield* toPhysicalId(id, news.runId, output?.runId);
258 const name = output?.name ?? resourceName(process, runId);
259 const ownership = yield* createOwnership(id);
260 const attributes = desiredAttributes(news.attributes, ownership);
261
262 let current =
263 (yield* getByName(name)) ?? (yield* findOwnedRun(id, process));
264
265 if (current === undefined) {
266 const startTime = news.startTime ?? (yield* nowIso());
267 const created = yield* datalineage
268 .createProjectsLocationsProcessesRuns({
269 parent: process,
270 body: {
271 name,
272 startTime,
273 endTime: news.endTime,
274 state: news.state ?? "STARTED",
275 displayName: news.displayName,
276 attributes,
277 },
278 })
279 .pipe(
280 Effect.catchTag("Conflict", () =>
281 getByName(name).pipe(
282 Effect.flatMap((row) =>
283 row ? Effect.succeed(row) : findOwnedRun(id, process),
284 ),
285 ),
286 ),
287 );
288 current = created ?? undefined;
289 }
290
291 if (current === undefined) {
292 return yield* new ProcessesRunNotResolved({ name });
293 }
294
295 const resolvedName = current.name ?? name;
296 const startTime =
297 news.startTime ?? current.startTime ?? (yield* nowIso());
298 const endTime = news.endTime ?? current.endTime;
299 const state = news.state ?? current.state ?? "STARTED";
300 const displayName = news.displayName ?? current.displayName;
301 const startChanged = !sameText(current.startTime, startTime);
302 const endChanged = !sameText(current.endTime, endTime);
303 const stateChanged = !sameText(current.state, state);
304 const displayNameChanged = !sameText(current.displayName, displayName);
305 const attributesChanged = !sameJson(current.attributes, attributes);
306
307 if (
308 startChanged ||
309 endChanged ||
310 stateChanged ||
311 displayNameChanged ||
312 attributesChanged
313 ) {
314 current = yield* datalineage.patchProjectsLocationsProcessesRuns({
315 name: resolvedName,
316 body: {
317 name: resolvedName,
318 startTime,
319 endTime,
320 state,
321 displayName,
322 attributes,
323 },
324 });
325 }
326
327 return toAttrs(current, env.project, env.region);
328 }),
329
330 delete: Effect.fn(function* ({ output }) {
331 if (!output.name) return;
332 const operation = yield* datalineage
333 .deleteProjectsLocationsProcessesRuns({
334 name: output.name,
335 allowMissing: true,
336 })
337 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
338 if (operation !== undefined) {
339 yield* waitForOperation(operation, { notFoundOk: true });
340 }
341 yield* waitUntilGone(getByName(output.name), output.name);
342 }),
343 });