1import * as analyticshub from "@distilled.cloud/gcp/analyticshub_v1";
2import { Retry as GcpRetry } from "@distilled.cloud/gcp/Retry";
3import * as Data from "effect/Data";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Schedule from "effect/Schedule";
7import * as Stream from "effect/Stream";
8import { createPhysicalName } from "../../PhysicalName.ts";
9import {
10 alchemyLabelKeys,
11 createInternalLabels,
12 hasAlchemyLabels,
13} from "../Labels.ts";
14import { isTransientGcpError } from "../Errors.ts";
15
16export const MAX_ID_LENGTH = 40;
17export const MAX_DISPLAY_NAME_LENGTH = 63;
18export const LIST_LOCATIONS = ["us-central1", "US", "EU"] as const;
19
20export class AnalyticshubNotResolved extends Data.TaggedError(
21 "GCP.AnalyticsHub.ResourceNotResolved",
22)<{
23 name: string;
24}> {}
25
26export const lastSegment = (value: string) => {
27 const trimmed = value.replace(/\/+$/, "");
28 const parts = trimmed.split("/");
29 return parts[parts.length - 1] || trimmed;
30};
31
32export const locationParent = (project: string, location: string) =>
33 `projects/${project}/locations/${location}`;
34
35export const normalizeLocation = (
36 location: string | undefined,
37 defaultLocation: string,
38) => lastSegment(location ?? defaultLocation);
39
40export const sameText = (left: string | undefined, right: string | undefined) =>
41 (left ?? "") === (right ?? "");
42
43export const sameBool = (
44 left: boolean | undefined,
45 right: boolean | undefined,
46) => (left === true) === (right === true);
47
48const canonical = (value: unknown): unknown => {
49 if (value === undefined || value === null) return undefined;
50 if (typeof value === "boolean" || typeof value === "number") return value;
51 if (typeof value === "string") return value.length === 0 ? undefined : value;
52 if (Array.isArray(value)) {
53 const items = value.map(canonical).filter((item) => item !== undefined);
54 return items.length === 0 ? undefined : items;
55 }
56 if (typeof value === "object") {
57 const entries = Object.entries(value as Record<string, unknown>)
58 .map(([key, item]) => [key, canonical(item)] as const)
59 .filter(([, item]) => item !== undefined)
60 .sort(([left], [right]) => left.localeCompare(right));
61 if (entries.length === 0) return undefined;
62 return Object.fromEntries(entries);
63 }
64 return undefined;
65};
66
67export const sameJson = (left: unknown, right: unknown) =>
68 JSON.stringify(canonical(left) ?? null) ===
69 JSON.stringify(canonical(right) ?? null);
70
71export const updateMaskOf = (...fields: Array<string | undefined>) =>
72 fields.filter((field): field is string => field !== undefined).join(",");
73
74export const parseResourceName = (name: string, collection: string) => {
75 const parts = name.split("/").filter((part) => part.length > 0);
76 const collectionAt = parts.lastIndexOf(collection);
77 const locationsAt = parts.lastIndexOf("locations");
78 const projectsAt = parts.lastIndexOf("projects");
79 return {
80 project:
81 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
82 location:
83 locationsAt >= 0 && parts[locationsAt + 1] ? parts[locationsAt + 1]! : "",
84 id:
85 collectionAt >= 0 && parts[collectionAt + 1]
86 ? parts[collectionAt + 1]!
87 : lastSegment(name),
88 parent:
89 collectionAt > 0
90 ? parts.slice(0, collectionAt).join("/")
91 : parts.slice(0, Math.max(0, parts.length - 1)).join("/"),
92 };
93};
94
95export const expandParent = (
96 value: string,
97 project: string,
98 location: string,
99 collection: string,
100) => {
101 if (value.includes("/")) return value.replace(/\/+$/, "");
102 return `projects/${project}/locations/${location}/${collection}/${value}`;
103};
104
105export const toPhysicalId = (
106 id: string,
107 explicit: string | undefined,
108 existing: string | undefined,
109 maxLength = MAX_ID_LENGTH,
110) =>
111 Effect.gen(function* () {
112 if (explicit !== undefined) return explicit;
113 if (existing !== undefined) return existing;
114 const generated = yield* createPhysicalName({
115 id,
116 maxLength,
117 lowercase: true,
118 delimiter: "_",
119 });
120 let next = generated.replace(/[^a-z0-9_]/g, "_").replace(/_+/g, "_");
121 if (!/^[a-z]/.test(next)) next = `a${next}`.slice(0, maxLength);
122 next = next.slice(0, maxLength).replace(/_+$/g, "");
123 return next.length > 0 ? next : "exchange";
124 });
125
126export const displayNameOf = (value: string | undefined, fallback: string) => {
127 const next = (value ?? fallback).trim();
128 return next.slice(0, MAX_DISPLAY_NAME_LENGTH);
129};
130
131export const encodeDescription = (
132 labels: Record<string, string>,
133 description: string | undefined,
134): string => {
135 const marker = `[alchemy ${alchemyLabelKeys.stack}=${labels[alchemyLabelKeys.stack]} ${alchemyLabelKeys.stage}=${labels[alchemyLabelKeys.stage]} ${alchemyLabelKeys.id}=${labels[alchemyLabelKeys.id]}]`;
136 return description && description.length > 0
137 ? `${marker}\n${description}`
138 : marker;
139};
140
141export const parseDescription = (
142 description: string | undefined,
143): {
144 labels: Record<string, string>;
145 description: string | undefined;
146} => {
147 if (!description?.startsWith("[alchemy ")) {
148 return { labels: {}, description };
149 }
150 const end = description.indexOf("]");
151 if (end < 0) return { labels: {}, description };
152 const labels: Record<string, string> = {};
153 for (const part of description.slice("[alchemy ".length, end).split(/\s+/)) {
154 const eq = part.indexOf("=");
155 if (eq > 0) {
156 labels[part.slice(0, eq)] = part.slice(eq + 1);
157 }
158 }
159 const rest = description.slice(end + 1).replace(/^\n/, "");
160 return { labels, description: rest.length > 0 ? rest : undefined };
161};
162
163export const hasOwnershipMarker = (description: string | undefined) =>
164 Object.keys(parseDescription(description).labels).some((key) =>
165 key.startsWith("alchemy-"),
166 );
167
168export const ownedById = (id: string, description: string | undefined) =>
169 Effect.gen(function* () {
170 const { labels } = parseDescription(description);
171 return yield* hasAlchemyLabels(id, labels);
172 });
173
174export const sharingKind = (
175 config: analyticshub.SharingEnvironmentConfig | undefined,
176) => (config?.dcrExchangeConfig !== undefined ? "dcr" : "default");
177
178export const replaceOnIdentity = (input: {
179 previousId?: string;
180 nextId?: string;
181 previousParent?: string;
182 nextParent?: string;
183 previousLocation?: string;
184 nextLocation?: string;
185 extra?: boolean;
186}) => {
187 const idChanged =
188 input.previousId !== undefined &&
189 input.nextId !== undefined &&
190 input.previousId !== input.nextId;
191 const parentChanged =
192 (input.previousParent ?? "") !== "" &&
193 (input.nextParent ?? "") !== "" &&
194 (input.previousParent ?? "") !== (input.nextParent ?? "");
195 const locationChanged =
196 (input.previousLocation ?? "") !== "" &&
197 (input.nextLocation ?? "") !== "" &&
198 (input.previousLocation ?? "").toLowerCase() !==
199 (input.nextLocation ?? "").toLowerCase();
200 if (!idChanged && !parentChanged && !locationChanged && !input.extra) {
201 return undefined;
202 }
203 return { action: "replace" as const, deleteFirst: false };
204};
205
206export const retryTransient = <A, E extends { readonly _tag: string }, R>(
207 effect: Effect.Effect<A, E, R>,
208) =>
209 effect.pipe(
210 Effect.retry({
211 while: (error) => isTransientGcpError(error) || error._tag === "NotFound",
212 times: 8,
213 schedule: Schedule.exponential("250 millis"),
214 }),
215 );
216
217export const ignoreGone = <A, E extends { readonly _tag: string }, R>(
218 effect: Effect.Effect<A, E, R>,
219) =>
220 effect.pipe(
221 Effect.catchIf(
222 (error): error is E & { readonly _tag: "NotFound" } =>
223 error._tag === "NotFound",
224 () => Effect.void,
225 ),
226 );
227
228const noRetryLayer = Layer.succeed(GcpRetry, { while: () => false });
229
230const isUnavailable = (error: { readonly _tag: string }) =>
231 error._tag === "Conflict" ||
232 error._tag === "InternalServerError" ||
233 error._tag === "ServiceUnavailable" ||
234 error._tag === "BadGateway" ||
235 error._tag === "GatewayTimeout";
236
237export const deleteRetry = <A, E extends { readonly _tag: string }, R>(
238 effect: Effect.Effect<A, E, R>,
239) =>
240 effect.pipe(
241 Effect.provide(noRetryLayer),
242 Effect.retry({
243 while: isUnavailable,
244 times: 12,
245 schedule: Schedule.spaced("3 seconds"),
246 }),
247 Effect.catchIf(
248 (error): error is E & { readonly _tag: "NotFound" } =>
249 error._tag === "NotFound",
250 () => Effect.void,
251 ),
252 );
253
254export const missingGet = (name: string) =>
255 name.length === 0
256 ? Effect.succeed(undefined)
257 : analyticshub
258 .getProjectsLocationsDataExchanges({ name })
259 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
260
261const emptyList = <A>() => Effect.succeed<A[]>([]);
262
263export const collectPages = <Page, Item, E, R>(
264 pages: Stream.Stream<Page, E, R>,
265 items: (page: Page) => readonly Item[] | null | undefined,
266) =>
267 pages.pipe(
268 Stream.flatMap((page) => Stream.fromIterable(items(page) ?? [])),
269 Stream.runCollect,
270 Effect.map((chunk): Item[] => Array.from(chunk)),
271 );
272
273export const listDataExchanges = (parent: string) =>
274 parent.length === 0
275 ? emptyList<analyticshub.DataExchange>()
276 : collectPages(
277 analyticshub.listProjectsLocationsDataExchanges.pages({
278 parent,
279 pageSize: 1000,
280 }),
281 (page) => page.dataExchanges,
282 ).pipe(
283 Effect.retry({
284 while: isUnavailable,
285 times: 8,
286 schedule: Schedule.spaced("2 seconds"),
287 }),
288 Effect.catchTag("NotFound", () =>
289 emptyList<analyticshub.DataExchange>(),
290 ),
291 );
292
293export const listListings = (parent: string) =>
294 parent.length === 0
295 ? emptyList<analyticshub.Listing>()
296 : collectPages(
297 analyticshub.listProjectsLocationsDataExchangesListings.pages({
298 parent,
299 pageSize: 1000,
300 }),
301 (page) => page.listings,
302 ).pipe(
303 Effect.retry({
304 while: isUnavailable,
305 times: 8,
306 schedule: Schedule.spaced("2 seconds"),
307 }),
308 Effect.catchTag("NotFound", () => emptyList<analyticshub.Listing>()),
309 );
310
311export const listQueryTemplates = (parent: string) =>
312 parent.length === 0
313 ? emptyList<analyticshub.QueryTemplate>()
314 : collectPages(
315 analyticshub.listProjectsLocationsDataExchangesQueryTemplates.pages({
316 parent,
317 pageSize: 1000,
318 }),
319 (page) => page.queryTemplates,
320 ).pipe(
321 Effect.retry({
322 while: isUnavailable,
323 times: 8,
324 schedule: Schedule.spaced("2 seconds"),
325 }),
326 Effect.catchTag("NotFound", () =>
327 emptyList<analyticshub.QueryTemplate>(),
328 ),
329 );
330
331export const listExchangesInProject = (project: string, region: string) =>
332 Effect.forEach(
333 [...new Set<string>([region, ...LIST_LOCATIONS])],
334 (location) => listDataExchanges(locationParent(project, location)),
335 { concurrency: 2 },
336 ).pipe(Effect.map((groups) => groups.flat()));
337
338export const listChildResources = <A, E, R>(
339 parents: readonly { name?: string }[],
340 list: (name: string) => Effect.Effect<A[], E, R>,
341) =>
342 Effect.forEach(
343 parents.filter((parent) => (parent.name ?? "").length > 0),
344 (parent) => list(parent.name!),
345 { concurrency: 4 },
346 ).pipe(Effect.map((groups) => groups.flat()));
347
348export const namedOf = <T extends { name?: string }>(items: readonly T[]) =>
349 items.filter((item) => (item.name ?? "").length > 0);
350
351export const ownershipLabels = (id: string) => createInternalLabels(id);