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 findOwnedProcess,
14 listOwnedProcesses,
15 locationParent,
16 normalizeLocation,
17 ownedByAlchemy,
18 parseName,
19 replaceOnIdentity,
20 sameJson,
21 sameText,
22 toPhysicalId,
23 userAttributes,
24 waitForOperation,
25 waitUntilGone,
26} from "./internal.ts";
27
28export type ProcessOrigin = {
29 /**
30 * Origin system. Use `CUSTOM` unless reporting lineage on behalf of a
31 * Google system. Non-`CUSTOM` sources may be restricted and billed.
32 * @default "CUSTOM"
33 */
34 sourceType?:
35 | datalineage.GoogleCloudDatacatalogLineageV1OriginSourceTypeEnum
36 | (string & {});
37 /**
38 * Origin name. For `CUSTOM`, any identifier. For Google systems, a
39 * resource name in the same project and location.
40 */
41 name?: string;
42};
43
44export type ProcessProps = {
45 /**
46 * Process id (the `{process}` segment of
47 * `projects/{project}/locations/{location}/processes/{process}`). If
48 * omitted, a unique name is generated from the stack, stage, and logical
49 * id. Max 200 characters: letters, digits, underscores, hyphens, colons,
50 * periods. Immutable — changing it replaces the process.
51 */
52 processId?: string;
53 /**
54 * Region (`us-central1`, …). Immutable — changing it replaces the
55 * process. `US-CENTRAL1` is accepted and normalized to `us-central1`.
56 * @default the stack's GCP region (`GCP.Region`, profile region, `us-central1`)
57 */
58 location?: string;
59 /**
60 * Human-readable name shown in UIs. Max 200 characters.
61 */
62 displayName?: string;
63 /**
64 * Origin of this process and its runs. Defaults to `{ sourceType:
65 * "CUSTOM", name: "alchemy" }`.
66 */
67 origin?: ProcessOrigin;
68 /**
69 * Non-semantic attributes (classifying or labeling the process). Max
70 * 100 entries. Processes have no labels field, so Alchemy stamps
71 * `alchemy-stack` / `alchemy-stage` / `alchemy-id` here for `list` /
72 * nuke and strips them from attributes.
73 */
74 attributes?: Record<string, unknown>;
75};
76
77export type Process = Resource<
78 "GCP.DataLineage.Process",
79 ProcessProps,
80 {
81 /** Full resource name `projects/{project}/locations/{location}/processes/{process}`. */
82 name: string;
83 /** Process id (last path segment). */
84 processId: string;
85 /** Project id. */
86 project: string;
87 /** Location id. */
88 location: string;
89 /** Human-readable display name. */
90 displayName: string | undefined;
91 /** Origin of the process. */
92 origin: ProcessOrigin | undefined;
93 /** User attributes (Alchemy ownership keys stripped). */
94 attributes: Record<string, unknown>;
95 },
96 never,
97 Providers
98>;
99
100/**
101 * A Data Lineage process — the definition of a data transformation.
102 *
103 * Processes have no labels field, so Alchemy stamps ownership into
104 * `attributes` for `list` / nuke. `processId` and `location` are identity
105 * — changing either replaces the process. Display name, origin, and
106 * attributes update in place.
107 *
108 * ### Creating a Process
109 * **Example:** Generated name
110 * ```typescript
111 * const process = yield* GCP.DataLineage.Process("Etl", {
112 * displayName: "nightly etl",
113 * attributes: { env: "dev" },
114 * });
115 * ```
116 *
117 * **Example:** Named process with a custom origin
118 * ```typescript
119 * const process = yield* GCP.DataLineage.Process("Etl", {
120 * processId: "orders-etl",
121 * location: "us-central1",
122 * displayName: "orders etl",
123 * origin: { sourceType: "CUSTOM", name: "dbt" },
124 * attributes: { env: "prod" },
125 * });
126 * ```
127 *
128 * @resource
129 * @category DataLineage
130 */
131export const Process = Resource<Process>("GCP.DataLineage.Process");
132
133export class ProcessNotResolved extends Data.TaggedError(
134 "GCP.DataLineage.ProcessNotResolved",
135)<{
136 name: string;
137}> {}
138
139const resourceName = (project: string, location: string, processId: string) =>
140 `${locationParent(project, location)}/processes/${processId}`;
141
142const defaultOrigin = (
143 origin: ProcessOrigin | undefined,
144): datalineage.GoogleCloudDatacatalogLineageV1Origin => ({
145 sourceType: origin?.sourceType ?? "CUSTOM",
146 name: origin?.name ?? "alchemy",
147});
148
149const toAttrs = (
150 process: datalineage.GoogleCloudDatacatalogLineageV1Process,
151 project: string,
152 region: string,
153) => {
154 const name = process.name ?? "";
155 const parsed = parseName(name, "processes", region);
156 return {
157 name,
158 processId: parsed.id,
159 project: parsed.project || project,
160 location: parsed.location,
161 displayName: process.displayName,
162 origin: process.origin
163 ? {
164 sourceType: process.origin.sourceType,
165 name: process.origin.name,
166 }
167 : undefined,
168 attributes: userAttributes(process.attributes),
169 };
170};
171
172const getByName = (name: string) =>
173 name.length === 0
174 ? Effect.succeed(undefined)
175 : datalineage
176 .getProjectsLocationsProcesses({ name })
177 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
178
179export const ProcessProvider = () =>
180 Provider.succeed(Process, {
181 stables: ["name", "processId", "project", "location"],
182
183 diff: Effect.fn(function* ({ news, olds, output }) {
184 if (!isResolved(news)) return undefined;
185 const env = yield* GcpEnvironment.current;
186 const previousLocation = normalizeLocation(
187 olds?.location ?? output?.location,
188 env.region,
189 );
190 const nextLocation = normalizeLocation(
191 news.location ?? olds?.location ?? output?.location,
192 env.region,
193 );
194 return replaceOnIdentity({
195 previousId: olds?.processId ?? output?.processId,
196 nextId: news.processId,
197 extra: output !== undefined && previousLocation !== nextLocation,
198 });
199 }),
200
201 read: Effect.fn(function* ({ id, olds, output }) {
202 const env = yield* GcpEnvironment.current;
203 const location = normalizeLocation(
204 olds?.location ?? output?.location,
205 env.region,
206 );
207 const processId = yield* toPhysicalId(
208 id,
209 olds?.processId,
210 output?.processId,
211 );
212 const name =
213 output?.name ?? resourceName(env.project, location, processId);
214 const existing =
215 (yield* getByName(name)) ??
216 (yield* findOwnedProcess(id, env.project, location));
217 if (existing === undefined) return undefined;
218 const attrs = toAttrs(existing, env.project, env.region);
219 return (yield* ownedByAlchemy(id, existing.attributes))
220 ? attrs
221 : Unowned(attrs);
222 }),
223
224 list: () =>
225 Effect.gen(function* () {
226 const env = yield* GcpEnvironment.current;
227 const rows = yield* listOwnedProcesses(env.project, env.region);
228 return rows.map((row) => toAttrs(row, env.project, env.region));
229 }),
230
231 reconcile: Effect.fn(function* ({ id, news, output }) {
232 const env = yield* GcpEnvironment.current;
233 const location = normalizeLocation(
234 news.location ?? output?.location,
235 env.region,
236 );
237 const processId = yield* toPhysicalId(
238 id,
239 news.processId,
240 output?.processId,
241 );
242 const name =
243 output?.name ?? resourceName(env.project, location, processId);
244 const ownership = yield* createOwnership(id);
245 const attributes = desiredAttributes(news.attributes, ownership);
246
247 let current =
248 (yield* getByName(name)) ??
249 (yield* findOwnedProcess(id, env.project, location));
250
251 if (current === undefined) {
252 const created = yield* datalineage
253 .createProjectsLocationsProcesses({
254 parent: locationParent(env.project, location),
255 body: {
256 name,
257 displayName: news.displayName,
258 origin: defaultOrigin(news.origin),
259 attributes,
260 },
261 })
262 .pipe(
263 Effect.catchTag("Conflict", () =>
264 getByName(name).pipe(
265 Effect.flatMap((row) =>
266 row
267 ? Effect.succeed(row)
268 : findOwnedProcess(id, env.project, location),
269 ),
270 ),
271 ),
272 );
273 current = created ?? undefined;
274 }
275
276 if (current === undefined) {
277 return yield* new ProcessNotResolved({ name });
278 }
279
280 const resolvedName = current.name ?? name;
281 const desiredOrigin = defaultOrigin(news.origin ?? current.origin);
282 const displayName = news.displayName ?? current.displayName;
283 const originChanged = !sameJson(current.origin, desiredOrigin);
284 const displayNameChanged = !sameText(current.displayName, displayName);
285 const attributesChanged = !sameJson(current.attributes, attributes);
286
287 if (originChanged || displayNameChanged || attributesChanged) {
288 current = yield* datalineage.patchProjectsLocationsProcesses({
289 name: resolvedName,
290 body: {
291 name: resolvedName,
292 displayName,
293 origin: desiredOrigin,
294 attributes,
295 },
296 });
297 }
298
299 return toAttrs(current, env.project, env.region);
300 }),
301
302 delete: Effect.fn(function* ({ output }) {
303 if (!output.name) return;
304 const operation = yield* datalineage
305 .deleteProjectsLocationsProcesses({
306 name: output.name,
307 allowMissing: true,
308 })
309 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
310 if (operation !== undefined) {
311 yield* waitForOperation(operation, { notFoundOk: true });
312 }
313 yield* waitUntilGone(getByName(output.name), output.name);
314 }),
315 });