1import * as ds from "@distilled.cloud/gcp/datastream_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 { createPhysicalName } from "../../PhysicalName.ts";
7import { tagRecord } from "../../Tags.ts";
8import { stripInternalLabels } from "../Labels.ts";
9import { waitForOperation as waitForLongRunning } from "../Operation.ts";
10
11export const MAX_NAME_LENGTH = 63;
12
13export class ResourceNotResolved extends Data.TaggedError(
14 "GCP.Datastream.ResourceNotResolved",
15)<{
16 name: string;
17}> {}
18
19export class ResourceStillExists extends Data.TaggedError(
20 "GCP.Datastream.ResourceStillExists",
21)<{
22 name: string;
23}> {}
24
25export const lastSegment = (value: string) => {
26 const trimmed = value.replace(/\/+$/, "");
27 const parts = trimmed.split("/");
28 return parts[parts.length - 1] || trimmed;
29};
30
31export const rfc1035 = (name: string, fallback = "ds"): string => {
32 let next = name
33 .toLowerCase()
34 .replace(/[^a-z0-9-]/g, "-")
35 .replace(/-+/g, "-")
36 .replace(/^-+|-+$/g, "");
37 if (!/^[a-z]/.test(next)) next = `d${next}`;
38 next = next.slice(0, MAX_NAME_LENGTH).replace(/-+$/g, "");
39 if (next.length === 0) return fallback;
40 if (!/[a-z0-9]$/.test(next)) next = `${next.slice(0, MAX_NAME_LENGTH - 1)}0`;
41 return next.slice(0, MAX_NAME_LENGTH);
42};
43
44export const normalizeLocation = (
45 location: string | undefined,
46 defaultLocation: string,
47) => lastSegment(location ?? defaultLocation).toLowerCase();
48
49export const locationParent = (project: string, location: string) =>
50 `projects/${project}/locations/${lastSegment(location).toLowerCase()}`;
51
52export const toPhysicalId = (
53 id: string,
54 explicit: string | undefined,
55 existing: string | undefined,
56 fallback = "ds",
57) =>
58 Effect.gen(function* () {
59 if (explicit !== undefined) return rfc1035(explicit, fallback);
60 if (existing !== undefined) return existing;
61 return rfc1035(
62 yield* createPhysicalName({
63 id,
64 maxLength: MAX_NAME_LENGTH,
65 lowercase: true,
66 }),
67 fallback,
68 );
69 });
70
71export const parseName = (
72 name: string,
73 collection: string,
74 defaultLocation: string,
75) => {
76 const parts = name.split("/").filter((part) => part.length > 0);
77 const collectionAt = parts.lastIndexOf(collection);
78 const locationsAt = parts.lastIndexOf("locations");
79 const projectsAt = parts.lastIndexOf("projects");
80 return {
81 project:
82 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
83 location:
84 locationsAt >= 0 && parts[locationsAt + 1]
85 ? parts[locationsAt + 1]!
86 : defaultLocation,
87 id:
88 collectionAt >= 0 && parts[collectionAt + 1]
89 ? parts[collectionAt + 1]!
90 : lastSegment(name),
91 parent:
92 collectionAt > 0
93 ? parts.slice(0, collectionAt).join("/")
94 : parts.slice(0, Math.max(0, parts.length - 1)).join("/"),
95 };
96};
97
98export const expandParent = (
99 value: string,
100 project: string,
101 location: string,
102 collection: string,
103) => {
104 if (value.includes("/")) return value.replace(/\/+$/, "");
105 return `projects/${project}/locations/${location}/${collection}/${value}`;
106};
107
108export const connectionProfileOf = (
109 value: string,
110 project: string,
111 location: string,
112) => expandParent(value, project, location, "connectionProfiles");
113
114export const privateConnectionOf = (
115 value: string,
116 project: string,
117 location: string,
118) => expandParent(value, project, location, "privateConnections");
119
120export const networkOf = (value: string, project: string) => {
121 const trimmed = value.replace(/\/+$/, "");
122 const projectsAt = trimmed.indexOf("projects/");
123 if (projectsAt >= 0) return trimmed.slice(projectsAt);
124 return `projects/${project}/global/networks/${trimmed}`;
125};
126
127export const userLabels = (
128 labels: Record<string, string | undefined> | null | undefined,
129): Record<string, string> => stripInternalLabels(tagRecord(labels));
130
131export const hasAlchemyLabelMap = (
132 labels: Record<string, string | undefined> | null | undefined,
133) => Object.keys(labels ?? {}).some((key) => key.startsWith("alchemy-"));
134
135export const canonical = (value: unknown): unknown => {
136 if (value === undefined || value === null) return undefined;
137 if (typeof value === "boolean" || typeof value === "number") return value;
138 if (typeof value === "string") {
139 return value.length === 0 ? undefined : value;
140 }
141 if (Array.isArray(value)) {
142 const items = value.map(canonical).filter((item) => item !== undefined);
143 return items.length === 0 ? undefined : items;
144 }
145 if (typeof value === "object") {
146 const entries = Object.entries(value as Record<string, unknown>)
147 .map(([key, item]) => [key, canonical(item)] as const)
148 .filter(([, item]) => item !== undefined)
149 .sort(([left], [right]) => left.localeCompare(right));
150 if (entries.length === 0) return undefined;
151 return Object.fromEntries(entries);
152 }
153 return undefined;
154};
155
156export const fingerprint = (value: unknown): string =>
157 JSON.stringify(canonical(value) ?? null);
158
159export const fieldMask = (fields: Array<string | false | undefined>) =>
160 fields
161 .filter((field): field is string => typeof field === "string")
162 .join(",");
163
164export const emptyMessage = (value: unknown) =>
165 value === undefined ? undefined : {};
166
167export const replaceOnIdentity = (input: {
168 previousId: string | undefined;
169 nextId: string | undefined;
170 previousLocation: string;
171 nextLocation: string;
172 extra?: boolean;
173 previousParent?: string;
174 nextParent?: string;
175}) => {
176 const parentChanged =
177 (input.previousParent ?? "") !== "" &&
178 (input.nextParent ?? "") !== "" &&
179 (input.previousParent ?? "") !== (input.nextParent ?? "");
180 const replace =
181 (input.extra ?? false) ||
182 parentChanged ||
183 (input.previousId !== undefined &&
184 input.nextId !== undefined &&
185 input.nextId !== input.previousId) ||
186 input.previousLocation !== input.nextLocation;
187 if (!replace) return undefined;
188 const samePhysical =
189 input.previousLocation === input.nextLocation &&
190 !parentChanged &&
191 input.previousId !== undefined &&
192 input.nextId === input.previousId;
193 return {
194 action: "replace" as const,
195 deleteFirst: samePhysical,
196 };
197};
198
199/**
200 * Wait for a Datastream long-running operation. Private connections peer a
201 * VPC and can take 10–20 minutes.
202 * `ALREADY_EXISTS` (a concurrent create won) always succeeds;
203 * `notFoundOk` also accepts a `NOT_FOUND` result or an operation that has
204 * already been garbage-collected (deletes).
205 */
206export const waitForOperation = (
207 operation: ds.Operation,
208 options?: { notFoundOk?: boolean },
209) =>
210 waitForLongRunning(
211 operation,
212 (name) => ds.getProjectsLocationsOperations({ name }),
213 { budget: "30 minutes" },
214 ).pipe(
215 Effect.catchIf(
216 (error) =>
217 error._tag === "GCP.OperationFailed" &&
218 (error.code === 6 ||
219 (options?.notFoundOk === true && error.code === 5)),
220 () => Effect.void,
221 ),
222 Effect.catchIf(
223 (error) => options?.notFoundOk === true && error._tag === "NotFound",
224 () => Effect.void,
225 ),
226 );
227
228export const waitUntilExists = <A, E extends { readonly _tag: string }, R>(
229 get: Effect.Effect<A | undefined, E, R>,
230 name: string,
231) =>
232 get.pipe(
233 Effect.flatMap((value) =>
234 value
235 ? Effect.succeed(value)
236 : Effect.fail(new ResourceNotResolved({ name })),
237 ),
238 Effect.retry({
239 while: (error) => error._tag === "GCP.Datastream.ResourceNotResolved",
240 times: 8,
241 schedule: Schedule.spaced("1 second"),
242 }),
243 );
244
245export const waitUntilGone = <A, E extends { readonly _tag: string }, R>(
246 get: Effect.Effect<A | undefined, E, R>,
247 name: string,
248 options?: { times?: number; interval?: `${number} seconds` },
249) =>
250 get.pipe(
251 Effect.flatMap((value) =>
252 value === undefined
253 ? Effect.void
254 : Effect.fail(new ResourceStillExists({ name })),
255 ),
256 Effect.retry({
257 while: (error) => error._tag === "GCP.Datastream.ResourceStillExists",
258 times: options?.times ?? 10,
259 schedule: Schedule.spaced(options?.interval ?? "4 seconds"),
260 }),
261 );
262
263export const settleOperation = (
264 operation: ds.Operation | undefined,
265 options?: { notFoundOk?: boolean },
266) =>
267 operation === undefined ? Effect.void : waitForOperation(operation, options);
268
269export const collectPages = <Page, A, E, R>(
270 pages: Stream.Stream<Page, E, R>,
271 items: (page: Page) => readonly A[] | undefined,
272) =>
273 pages.pipe(
274 Stream.flatMap((page) => Stream.fromIterable(items(page) ?? [])),
275 Stream.runCollect,
276 Effect.map((chunk) => Array.from(chunk)),
277 );
278
279/**
280 * List across every location via the `locations/-` wildcard, falling back to
281 * the default region where the API rejects the wildcard. A missing parent
282 * (`NotFound`) lists as empty; any other error fails.
283 */
284export const listAtLocation = <A, E extends { readonly _tag: string }, R>(
285 project: string,
286 region: string,
287 list: (parent: string) => Effect.Effect<A[], E, R>,
288) =>
289 list(`projects/${project}/locations/-`).pipe(
290 Effect.catchIf(
291 (error) => error._tag === "NotFound" || error._tag === "BadRequest",
292 () =>
293 list(`projects/${project}/locations/${region}`).pipe(
294 Effect.catchIf(
295 (error) => error._tag === "NotFound",
296 () => Effect.succeed<A[]>([]),
297 ),
298 ),
299 ),
300 );