1import * as datapipelines from "@distilled.cloud/gcp/datapipelines_v1";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import { createPhysicalName } from "../../PhysicalName.ts";
5import { tagRecord } from "../../Tags.ts";
6import {
7 alchemyLabelKeys,
8 createInternalLabels,
9 hasAlchemyLabels,
10 stripInternalLabels,
11} from "../Labels.ts";
12
13export const DEFAULT_TYPE = "PIPELINE_TYPE_BATCH";
14export const DEFAULT_STATE = "STATE_ACTIVE";
15export const MAX_ID_LENGTH = 63;
16export const MAX_DISPLAY_NAME_LENGTH = 64;
17// Data Pipelines rejects the `-` location wildcard on list with a bare
18// INVALID_ARGUMENT, so enumerate regions explicitly.
19export const LIST_LOCATIONS = [
20 "us-central1",
21 "us-east1",
22 "us-west1",
23 "europe-west1",
24 "asia-east1",
25] as const;
26
27export const lastSegment = (value: string) => {
28 const trimmed = value.replace(/\/+$/, "");
29 const parts = trimmed.split("/");
30 return parts[parts.length - 1] || trimmed;
31};
32
33export const sameText = (left: string | undefined, right: string | undefined) =>
34 (left ?? "") === (right ?? "");
35
36// Key-order-insensitive: the API echoes maps (e.g. launch `parameters`) in
37// its own order, which would otherwise read as a workload change.
38const canonical = (value: unknown): unknown =>
39 Array.isArray(value)
40 ? value.map(canonical)
41 : value !== null && typeof value === "object"
42 ? Object.fromEntries(
43 Object.entries(value)
44 .filter(([, entry]) => entry !== undefined)
45 .sort(([a], [b]) => a.localeCompare(b))
46 .map(([key, entry]) => [key, canonical(entry)]),
47 )
48 : value;
49
50export const jsonEqual = (left: unknown, right: unknown) =>
51 JSON.stringify(canonical(left ?? null)) ===
52 JSON.stringify(canonical(right ?? null));
53
54export const updateMaskOf = (...fields: Array<string | undefined>) =>
55 fields.filter((field): field is string => field !== undefined).join(",");
56
57export const normalizeLocation = (
58 location: string | undefined,
59 fallback: string,
60) => lastSegment(location ?? fallback).toLowerCase();
61
62export const locationParent = (project: string, location: string) =>
63 `projects/${project}/locations/${location}`;
64
65export const resourceName = (
66 project: string,
67 location: string,
68 pipelineId: string,
69) => `${locationParent(project, location)}/pipelines/${pipelineId}`;
70
71export const parseName = (name: string, fallbackLocation: string) => {
72 const parts = name.split("/").filter((part) => part.length > 0);
73 const projectsAt = parts.lastIndexOf("projects");
74 const locationsAt = parts.lastIndexOf("locations");
75 const pipelinesAt = parts.lastIndexOf("pipelines");
76 return {
77 project:
78 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
79 location:
80 locationsAt >= 0 && parts[locationsAt + 1]
81 ? parts[locationsAt + 1]!
82 : fallbackLocation,
83 pipelineId:
84 pipelinesAt >= 0 && parts[pipelinesAt + 1]
85 ? parts[pipelinesAt + 1]!
86 : lastSegment(name),
87 parent:
88 pipelinesAt > 0
89 ? parts.slice(0, pipelinesAt).join("/")
90 : parts.slice(0, Math.max(0, parts.length - 1)).join("/"),
91 };
92};
93
94export const rfc1035 = (
95 name: string,
96 maxLength = MAX_ID_LENGTH,
97 fallback = "pipeline",
98): string => {
99 let next = name
100 .toLowerCase()
101 .replace(/[^a-z0-9-]/g, "-")
102 .replace(/-+/g, "-")
103 .replace(/^-+|-+$/g, "");
104 if (!/^[a-z]/.test(next)) next = `p${next}`;
105 next = next.slice(0, maxLength).replace(/-+$/g, "");
106 if (next.length === 0) return fallback.slice(0, maxLength);
107 if (!/[a-z0-9]$/.test(next)) {
108 next = `${next.slice(0, maxLength - 1)}0`;
109 }
110 return next.slice(0, maxLength);
111};
112
113export const sanitizeDisplayName = (value: string) => {
114 let next = value
115 .replace(/[^A-Za-z0-9_-]/g, "_")
116 .replace(/_+/g, "_")
117 .replace(/^_+|_+$/g, "")
118 .slice(0, MAX_DISPLAY_NAME_LENGTH);
119 if (next.length === 0) return "pipeline";
120 if (!/^[A-Za-z]/.test(next)) {
121 next = `p${next}`.slice(0, MAX_DISPLAY_NAME_LENGTH);
122 }
123 return next;
124};
125
126export const toPipelineId = (
127 id: string,
128 requested: string | undefined,
129 existing: string | undefined,
130) =>
131 Effect.gen(function* () {
132 if (requested !== undefined && requested.length > 0) {
133 return rfc1035(requested);
134 }
135 if (existing !== undefined && existing.length > 0) return existing;
136 return rfc1035(
137 yield* createPhysicalName({
138 id,
139 maxLength: MAX_ID_LENGTH,
140 lowercase: true,
141 }),
142 );
143 });
144
145export const toDisplayName = (
146 requested: string | undefined,
147 fallback: string,
148) =>
149 requested !== undefined && requested.length > 0
150 ? sanitizeDisplayName(requested)
151 : sanitizeDisplayName(fallback);
152
153export const stringMapOf = (
154 value: Record<string, string | undefined> | null | undefined,
155): Record<string, string> => tagRecord(value);
156
157export const userSources = (
158 sources: Record<string, string | undefined> | null | undefined,
159): Record<string, string> => stripInternalLabels(stringMapOf(sources));
160
161export const ownershipSources = (
162 labels: Record<string, string>,
163 user?: Record<string, string>,
164): Record<string, string> => ({
165 ...(user ?? {}),
166 [alchemyLabelKeys.stack]: labels[alchemyLabelKeys.stack] ?? "x",
167 [alchemyLabelKeys.stage]: labels[alchemyLabelKeys.stage] ?? "x",
168 [alchemyLabelKeys.id]: labels[alchemyLabelKeys.id] ?? "x",
169});
170
171export const hasOwnershipMarker = (
172 sources: Record<string, string | undefined> | null | undefined,
173) => Object.keys(sources ?? {}).some((key) => key.startsWith("alchemy-"));
174
175export const ownedByAlchemy = (
176 id: string,
177 sources: Record<string, string | undefined> | null | undefined,
178) => hasAlchemyLabels(id, stringMapOf(sources));
179
180export const scheduleKey = (
181 info: datapipelines.GoogleCloudDatapipelinesV1ScheduleSpec | undefined,
182) => ({
183 schedule: info?.schedule ?? "",
184 timeZone: info?.timeZone ?? "",
185});
186
187export const replaceOnIdentity = (input: {
188 previousId?: string;
189 nextId?: string;
190 previousLocation?: string;
191 nextLocation?: string;
192 previousType?: string;
193 nextType?: string;
194}) => {
195 if (
196 input.previousId !== undefined &&
197 input.nextId !== undefined &&
198 input.previousId !== input.nextId
199 ) {
200 return { action: "replace" as const, deleteFirst: false };
201 }
202 if (
203 input.previousLocation !== undefined &&
204 input.nextLocation !== undefined &&
205 input.previousLocation !== input.nextLocation
206 ) {
207 return { action: "replace" as const, deleteFirst: false };
208 }
209 if (
210 input.previousType !== undefined &&
211 input.nextType !== undefined &&
212 input.previousType !== input.nextType
213 ) {
214 return { action: "replace" as const, deleteFirst: false };
215 }
216 return undefined;
217};
218
219export const getPipeline = (name: string) =>
220 name.length === 0
221 ? Effect.succeed(undefined)
222 : datapipelines
223 .getProjectsLocationsPipelines({ name })
224 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
225
226export const ignoreMissing = <A, R>(
227 effect: Effect.Effect<
228 A,
229 datapipelines.DeleteProjectsLocationsPipelinesError,
230 R
231 >,
232) => effect.pipe(Effect.catchTag("NotFound", () => Effect.void));
233
234const emptyList = <A>() => Effect.succeed([] as A[]);
235
236export const listPipelinesAt = (parent: string) =>
237 datapipelines.listProjectsLocationsPipelines
238 .pages({
239 parent,
240 pageSize: 200,
241 })
242 .pipe(
243 Stream.flatMap((page) => Stream.fromIterable(page.pipelines ?? [])),
244 Stream.runCollect,
245 Effect.map((chunk) => Array.from(chunk)),
246 Effect.catchTag("NotFound", () =>
247 emptyList<datapipelines.GoogleCloudDatapipelinesV1Pipeline>(),
248 ),
249 );
250
251export const listOwnedPipelines = (project: string, region: string) =>
252 Effect.gen(function* () {
253 const pages = yield* Effect.forEach(
254 [...new Set([region, ...LIST_LOCATIONS])],
255 (location) => listPipelinesAt(locationParent(project, location)),
256 { concurrency: 2 },
257 );
258 const seen = new Set<string>();
259 const owned: datapipelines.GoogleCloudDatapipelinesV1Pipeline[] = [];
260 for (const row of pages.flat()) {
261 const name = row.name ?? "";
262 if (name.length > 0 && seen.has(name)) continue;
263 if (!hasOwnershipMarker(row.pipelineSources)) continue;
264 if (name.length > 0) seen.add(name);
265 owned.push(row);
266 }
267 return owned;
268 });
269
270export const findOwnedPipeline = (
271 id: string,
272 project: string,
273 region: string,
274 name?: string,
275) =>
276 Effect.gen(function* () {
277 const existing = yield* getPipeline(name ?? "");
278 if (existing !== undefined) return existing;
279 const rows = yield* listOwnedPipelines(project, region);
280 for (const row of rows) {
281 if (yield* ownedByAlchemy(id, row.pipelineSources)) {
282 return row;
283 }
284 }
285 return undefined;
286 });