1import * as dataform from "@distilled.cloud/gcp/dataform_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 { waitForOperation as waitForGcpOperation } from "../Operation.ts";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import { tagRecord } from "../../Tags.ts";
9import {
10 alchemyLabelKeys,
11 createInternalLabels,
12 hasAlchemyLabels,
13 stripInternalLabels,
14} from "../Labels.ts";
15import { isTransientGcpError } from "../Errors.ts";
16
17export const MAX_ID_LENGTH = 63;
18export const MAX_DISPLAY_NAME_LENGTH = 63;
19
20export class DataformPending extends Data.TaggedError("GCP.Dataform.Pending")<{
21 name: string;
22}> {}
23
24export class DataformStillExists extends Data.TaggedError(
25 "GCP.Dataform.StillExists",
26)<{
27 name: string;
28}> {}
29
30export const lastSegment = (value: string) => {
31 const trimmed = value.replace(/\/+$/, "");
32 const parts = trimmed.split("/");
33 return parts[parts.length - 1] || trimmed;
34};
35
36export const locationParent = (project: string, location: string) =>
37 `projects/${project}/locations/${location}`;
38
39export const normalizeLocation = (
40 location: string | undefined,
41 defaultLocation: string,
42) => lastSegment(location ?? defaultLocation).toLowerCase();
43
44export const parseResourceName = (
45 name: string,
46 collection: string,
47 defaultLocation: string,
48) => {
49 const parts = name.split("/").filter((part) => part.length > 0);
50 const collectionAt = parts.lastIndexOf(collection);
51 const locationsAt = parts.lastIndexOf("locations");
52 const projectsAt = parts.lastIndexOf("projects");
53 return {
54 project:
55 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
56 location:
57 locationsAt >= 0 && parts[locationsAt + 1]
58 ? parts[locationsAt + 1]!
59 : defaultLocation,
60 id:
61 collectionAt >= 0 && parts[collectionAt + 1]
62 ? parts[collectionAt + 1]!
63 : lastSegment(name),
64 parent:
65 collectionAt > 0
66 ? parts.slice(0, collectionAt).join("/")
67 : parts.slice(0, Math.max(0, parts.length - 1)).join("/"),
68 };
69};
70
71export const expandParent = (
72 value: string,
73 project: string,
74 location: string,
75 collection: string,
76) => {
77 if (value.includes("/")) return value.replace(/\/+$/, "");
78 return `${locationParent(project, location)}/${collection}/${value}`;
79};
80
81export const expandRepository = (
82 value: string,
83 project: string,
84 location: string,
85) => expandParent(value, project, location, "repositories");
86
87export const rfc1035 = (name: string, maxLength = MAX_ID_LENGTH): string => {
88 let next = name
89 .toLowerCase()
90 .replace(/[^a-z0-9-]/g, "-")
91 .replace(/-+/g, "-")
92 .replace(/^-+|-+$/g, "");
93 if (!/^[a-z]/.test(next)) next = `d${next}`;
94 next = next.slice(0, maxLength).replace(/-+$/g, "");
95 if (next.length === 0) return "dataform";
96 if (!/[a-z0-9]$/.test(next)) next = `${next.slice(0, maxLength - 1)}0`;
97 return next.slice(0, maxLength);
98};
99
100export const toPhysicalId = (
101 id: string,
102 explicit: string | undefined,
103 existing: string | undefined,
104 maxLength = MAX_ID_LENGTH,
105) =>
106 Effect.gen(function* () {
107 if (explicit !== undefined) return rfc1035(explicit, maxLength);
108 if (existing !== undefined) return existing;
109 return rfc1035(
110 yield* createPhysicalName({
111 id,
112 maxLength,
113 lowercase: true,
114 }),
115 maxLength,
116 );
117 });
118
119export const userLabels = (
120 labels: Record<string, string | undefined> | null | undefined,
121): Record<string, string> => stripInternalLabels(tagRecord(labels));
122
123export const hasAlchemyLabelMap = (
124 labels: Record<string, string | undefined> | null | undefined,
125) => Object.keys(labels ?? {}).some((key) => key.startsWith("alchemy-"));
126
127export const sameText = (left: string | undefined, right: string | undefined) =>
128 (left ?? "") === (right ?? "");
129
130export const sameJson = (left: unknown, right: unknown) =>
131 JSON.stringify(left ?? null) === JSON.stringify(right ?? null);
132
133export const updateMaskOf = (...fields: Array<string | undefined>) =>
134 fields.filter((field): field is string => field !== undefined).join(",");
135
136export const replaceOnIdentity = (input: {
137 previousId?: string;
138 nextId?: string;
139 previousLocation?: string;
140 nextLocation?: string;
141 previousParent?: string;
142 nextParent?: string;
143 extra?: boolean;
144}) => {
145 const locationChanged =
146 (input.previousLocation ?? "") !== "" &&
147 (input.nextLocation ?? "") !== "" &&
148 lastSegment(input.previousLocation ?? "").toLowerCase() !==
149 lastSegment(input.nextLocation ?? "").toLowerCase();
150 const parentChanged =
151 (input.previousParent ?? "") !== "" &&
152 (input.nextParent ?? "") !== "" &&
153 (input.previousParent ?? "") !== (input.nextParent ?? "");
154 const idChanged =
155 input.previousId !== undefined &&
156 input.nextId !== undefined &&
157 input.previousId !== input.nextId;
158 if (input.extra === true || locationChanged || parentChanged || idChanged) {
159 const samePhysical =
160 !locationChanged &&
161 !parentChanged &&
162 input.previousId !== undefined &&
163 input.nextId === input.previousId;
164 return {
165 action: "replace" as const,
166 deleteFirst: samePhysical,
167 };
168 }
169 return undefined;
170};
171
172const markerOf = (stack: string, stage: string, id: string) =>
173 `[alchemy ${alchemyLabelKeys.stack}=${stack} ${alchemyLabelKeys.stage}=${stage} ${alchemyLabelKeys.id}=${id}]`;
174
175const compactMarkerOf = (stack: string, stage: string, id: string) =>
176 `[alc ${stack} ${stage} ${id}]`;
177
178const shrinkMarker = (
179 labels: Record<string, string>,
180 maxLength: number,
181 build: (stack: string, stage: string, id: string) => string,
182) => {
183 let stack = labels[alchemyLabelKeys.stack] ?? "x";
184 let stage = labels[alchemyLabelKeys.stage] ?? "x";
185 let id = labels[alchemyLabelKeys.id] ?? "x";
186 let marker = build(stack, stage, id);
187 while (
188 marker.length > maxLength &&
189 (stack.length > 1 || stage.length > 1 || id.length > 1)
190 ) {
191 if (stack.length >= stage.length && stack.length >= id.length) {
192 stack = stack.slice(0, -1);
193 } else if (stage.length >= id.length) {
194 stage = stage.slice(0, -1);
195 } else {
196 id = id.slice(0, -1);
197 }
198 marker = build(stack, stage, id);
199 }
200 return marker.slice(0, maxLength);
201};
202
203export const encodeOwnershipLine = (
204 labels: Record<string, string>,
205 text: string | undefined,
206 maxLength = MAX_DISPLAY_NAME_LENGTH,
207): string => {
208 const trimmed = text?.replace(/[\r\n]+/g, " ").trim();
209 const reserved =
210 trimmed && trimmed.length > 0
211 ? Math.min(trimmed.length + 1, Math.max(0, maxLength - 24))
212 : 0;
213 const budget = Math.max(16, maxLength - reserved);
214 const marker =
215 budget < 54
216 ? shrinkMarker(labels, budget, compactMarkerOf)
217 : shrinkMarker(labels, budget, markerOf);
218 if (!trimmed) return marker.slice(0, maxLength);
219 return `${marker} ${trimmed}`.slice(0, maxLength);
220};
221
222export const parseOwnership = (
223 text: string | undefined,
224): {
225 labels: Record<string, string>;
226 text: string | undefined;
227} => {
228 if (text?.startsWith("[alc ")) {
229 const end = text.indexOf("]");
230 if (end < 0) return { labels: {}, text };
231 const parts = text.slice("[alc ".length, end).trim().split(/\s+/);
232 const labels: Record<string, string> = {};
233 if (parts[0]) labels[alchemyLabelKeys.stack] = parts[0]!;
234 if (parts[1]) labels[alchemyLabelKeys.stage] = parts[1]!;
235 if (parts[2]) labels[alchemyLabelKeys.id] = parts[2]!;
236 const rest = text.slice(end + 1).replace(/^[\s\n]+/, "");
237 return { labels, text: rest.length > 0 ? rest : undefined };
238 }
239 if (!text?.startsWith("[alchemy ")) {
240 return { labels: {}, text };
241 }
242 const end = text.indexOf("]");
243 if (end < 0) return { labels: {}, text };
244 const labels: Record<string, string> = {};
245 for (const part of text.slice("[alchemy ".length, end).split(/\s+/)) {
246 const eq = part.indexOf("=");
247 if (eq > 0) {
248 labels[part.slice(0, eq)] = part.slice(eq + 1);
249 }
250 }
251 const rest = text.slice(end + 1).replace(/^[\s\n]+/, "");
252 return { labels, text: rest.length > 0 ? rest : undefined };
253};
254
255export const hasOwnershipMarker = (text: string | undefined) =>
256 Object.keys(parseOwnership(text).labels).some((key) =>
257 key.startsWith("alchemy-"),
258 );
259
260const prefixMatch = (expected: string, observed: string) =>
261 expected === observed ||
262 expected.startsWith(observed) ||
263 observed.startsWith(expected);
264
265export const ownedByAlchemy = (id: string, text: string | undefined) =>
266 Effect.gen(function* () {
267 const expected = yield* createInternalLabels(id);
268 const { labels } = parseOwnership(text);
269 if (!hasOwnershipMarker(text)) return false;
270 const exact = yield* hasAlchemyLabels(id, labels);
271 if (exact) return true;
272 return (
273 prefixMatch(
274 expected[alchemyLabelKeys.stack] ?? "",
275 labels[alchemyLabelKeys.stack] ?? "",
276 ) &&
277 prefixMatch(
278 expected[alchemyLabelKeys.stage] ?? "",
279 labels[alchemyLabelKeys.stage] ?? "",
280 ) &&
281 prefixMatch(
282 expected[alchemyLabelKeys.id] ?? "",
283 labels[alchemyLabelKeys.id] ?? "",
284 )
285 );
286 });
287
288export const mergeOwnedVars = (
289 vars: Record<string, string> | undefined,
290 ownership: Record<string, string>,
291): Record<string, string> => ({ ...vars, ...ownership });
292
293export const retryTransient = <A, E extends { readonly _tag: string }, R>(
294 effect: Effect.Effect<A, E, R>,
295) =>
296 effect.pipe(
297 Effect.retry({
298 while: isTransientGcpError,
299 times: 8,
300 schedule: Schedule.exponential("250 millis"),
301 }),
302 );
303
304export class DeleteNotConfirmed extends Data.TaggedError(
305 "GCP.Dataform.DeleteNotConfirmed",
306)<{}> {}
307
308/** Poll until the resource is gone; fails if it is still readable after ~60s. */
309export const waitUntilGone = <A, E, R>(
310 get: Effect.Effect<A | undefined, E, R>,
311) =>
312 get.pipe(
313 Effect.repeat({
314 schedule: Schedule.spaced("2 seconds"),
315 until: (value) => value === undefined,
316 times: 30,
317 }),
318 Effect.flatMap((value): Effect.Effect<void, DeleteNotConfirmed> =>
319 value === undefined ? Effect.void : Effect.fail(new DeleteNotConfirmed()),
320 ),
321 );
322
323export const waitUntilExists = <A, E extends { readonly _tag: string }, R>(
324 get: Effect.Effect<A | undefined, E, R>,
325 name: string,
326) =>
327 get.pipe(
328 Effect.flatMap((value) =>
329 value !== undefined
330 ? Effect.succeed(value)
331 : Effect.fail(new DataformPending({ name })),
332 ),
333 Effect.retry({
334 while: (error) => error._tag === "GCP.Dataform.Pending",
335 times: 8,
336 schedule: Schedule.exponential("250 millis"),
337 }),
338 );
339
340/**
341 * Wait for a Dataform operation (folder tree deletes, repository moves).
342 * An operation that finished with `ALREADY_EXISTS` (6) is a lost create
343 * race; `notFoundOk` also accepts `NOT_FOUND` (5).
344 */
345export const waitForOperation = (
346 operation: dataform.Operation,
347 options?: { notFoundOk?: boolean },
348) =>
349 waitForGcpOperation(
350 operation,
351 (name) =>
352 dataform
353 .getProjectsLocationsOperations({ name })
354 .pipe(
355 Effect.catchTag("NotFound", (error) =>
356 options?.notFoundOk === true
357 ? Effect.succeed<dataform.Operation>({ name, done: true })
358 : Effect.fail(error),
359 ),
360 ),
361 { budget: "10 minutes" },
362 ).pipe(
363 Effect.catchIf(
364 (error) =>
365 error._tag === "GCP.OperationFailed" &&
366 (error.code === 6 ||
367 (options?.notFoundOk === true && error.code === 5)),
368 () => Effect.void,
369 ),
370 Effect.asVoid,
371 );
372
373export const collectPages = <
374 Page,
375 Item,
376 E extends { readonly _tag: string },
377 R,
378>(
379 pages: Stream.Stream<Page, E, R>,
380 items: (page: Page) => readonly Item[] | null | undefined,
381) =>
382 pages.pipe(
383 Stream.flatMap((page) => Stream.fromIterable(items(page) ?? [])),
384 Stream.runCollect,
385 Effect.map((chunk) => Array.from(chunk)),
386 );
387
388const emptyList = <A>() => Effect.succeed<A[]>([]);
389
390const emptyOnMissing = <A, E extends { readonly _tag: string }, R>(
391 effect: Effect.Effect<A[], E, R>,
392) =>
393 effect.pipe(
394 Effect.catchIf(
395 (error) => error._tag === "NotFound",
396 () => emptyList<A>(),
397 ),
398 );
399
400export const listRepositories = (project: string, location: string) =>
401 emptyOnMissing(
402 collectPages(
403 dataform.listProjectsLocationsRepositories.pages({
404 parent: locationParent(project, location),
405 pageSize: 1000,
406 }),
407 (page) => page.repositories ?? [],
408 ).pipe(
409 Effect.catchIf(
410 () => true,
411 () =>
412 collectPages(
413 dataform.listProjectsLocationsRepositories.pages({
414 parent: `projects/${project}/locations/-`,
415 pageSize: 1000,
416 }),
417 (page) => page.repositories ?? [],
418 ),
419 ),
420 ),
421 );
422
423export const listOwnedRepositories = (project: string, location: string) =>
424 listRepositories(project, location).pipe(
425 Effect.map((repos) =>
426 repos.filter((repo) => hasAlchemyLabelMap(repo.labels)),
427 ),
428 );
429
430export const listTeamFolders = (project: string, location: string) =>
431 emptyOnMissing(
432 collectPages(
433 dataform.searchProjectsLocationsTeamFolders.pages({
434 location: locationParent(project, location),
435 pageSize: 1000,
436 }),
437 (page) =>
438 (page.results ?? []).flatMap((result) =>
439 result.teamFolder ? [result.teamFolder] : [],
440 ),
441 ),
442 );
443
444export const listFolders = (project: string, location: string) =>
445 Effect.gen(function* () {
446 const parent = locationParent(project, location);
447 const seen = new Set<string>();
448 const folders: dataform.Folder[] = [];
449 const queue: string[] = [];
450
451 const ingest = (folder: dataform.Folder | undefined) => {
452 const name = folder?.name;
453 if (folder === undefined || name === undefined || seen.has(name)) {
454 return;
455 }
456 seen.add(name);
457 folders.push(folder);
458 queue.push(name);
459 };
460
461 const root = yield* emptyOnMissing(
462 collectPages(
463 dataform.queryUserRootContentsProjectsLocations.pages({
464 location: parent,
465 pageSize: 1000,
466 }),
467 (page) => page.entries ?? [],
468 ),
469 );
470 for (const entry of root) ingest(entry.folder);
471
472 const teams = yield* listTeamFolders(project, location);
473 for (const team of teams) {
474 if (team.name) queue.push(team.name);
475 }
476
477 let steps = 0;
478 while (queue.length > 0 && steps < 50) {
479 steps += 1;
480 const name = queue.shift()!;
481 const entries = name.includes("/teamFolders/")
482 ? yield* emptyOnMissing(
483 collectPages(
484 dataform.queryContentsProjectsLocationsTeamFolders.pages({
485 teamFolder: name,
486 pageSize: 1000,
487 }),
488 (page) => page.entries ?? [],
489 ),
490 )
491 : yield* emptyOnMissing(
492 collectPages(
493 dataform.queryFolderContentsProjectsLocationsFolders.pages({
494 folder: name,
495 pageSize: 1000,
496 }),
497 (page) => page.entries ?? [],
498 ),
499 );
500 for (const entry of entries) ingest(entry.folder);
501 }
502
503 return folders;
504 });
505
506export const listReleaseConfigs = (parent: string) =>
507 parent.length === 0
508 ? emptyList<dataform.ReleaseConfig>()
509 : emptyOnMissing(
510 collectPages(
511 dataform.listProjectsLocationsRepositoriesReleaseConfigs.pages({
512 parent,
513 pageSize: 1000,
514 }),
515 (page) => page.releaseConfigs ?? [],
516 ),
517 );
518
519export const listWorkflowConfigs = (parent: string) =>
520 parent.length === 0
521 ? emptyList<dataform.WorkflowConfig>()
522 : emptyOnMissing(
523 collectPages(
524 dataform.listProjectsLocationsRepositoriesWorkflowConfigs.pages({
525 parent,
526 pageSize: 1000,
527 }),
528 (page) => page.workflowConfigs ?? [],
529 ),
530 );
531
532export const listWorkspaces = (parent: string) =>
533 parent.length === 0
534 ? emptyList<dataform.Workspace>()
535 : emptyOnMissing(
536 collectPages(
537 dataform.listProjectsLocationsRepositoriesWorkspaces.pages({
538 parent,
539 pageSize: 1000,
540 }),
541 (page) => page.workspaces ?? [],
542 ),
543 );
544
545export const listWorkflowInvocations = (parent: string) =>
546 parent.length === 0
547 ? emptyList<dataform.WorkflowInvocation>()
548 : emptyOnMissing(
549 collectPages(
550 dataform.listProjectsLocationsRepositoriesWorkflowInvocations.pages({
551 parent,
552 pageSize: 1000,
553 }),
554 (page) => page.workflowInvocations ?? [],
555 ),
556 );
557
558export const forEachOwnedRepository = <A, E, R>(
559 project: string,
560 location: string,
561 fn: (repo: dataform.Repository) => Effect.Effect<readonly A[], E, R>,
562) =>
563 listOwnedRepositories(project, location).pipe(
564 Effect.flatMap((repos) =>
565 Effect.forEach(repos, fn, { concurrency: 4 }).pipe(
566 Effect.map((chunks) => chunks.flat()),
567 ),
568 ),
569 );
570
571export const deleteFolderTree = (name: string) =>
572 dataform.deleteProjectsLocationsFolders({ name }).pipe(
573 Effect.catchTag("Conflict", () =>
574 dataform
575 .deleteTreeProjectsLocationsFolders({
576 name,
577 body: { force: true },
578 })
579 .pipe(
580 Effect.flatMap((operation) =>
581 waitForOperation(operation, { notFoundOk: true }),
582 ),
583 ),
584 ),
585 Effect.catchTag("NotFound", () => Effect.void),
586 Effect.asVoid,
587 );
588
589export const deleteTeamFolderTree = (name: string) =>
590 dataform.deleteProjectsLocationsTeamFolders({ name }).pipe(
591 Effect.catchTag("Conflict", () =>
592 dataform
593 .deleteTreeProjectsLocationsTeamFolders({
594 name,
595 body: { force: true },
596 })
597 .pipe(
598 Effect.flatMap((operation) =>
599 waitForOperation(operation, { notFoundOk: true }),
600 ),
601 ),
602 ),
603 Effect.catchTag("NotFound", () => Effect.void),
604 Effect.asVoid,
605 );