1import * as firebasedataconnect from "@distilled.cloud/gcp/firebasedataconnect_v1";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Stream from "effect/Stream";
5import * as Schedule from "effect/Schedule";
6import { createPhysicalName } from "../../PhysicalName.ts";
7import { tagRecord } from "../../Tags.ts";
8import { stripInternalLabels } from "../Labels.ts";
9import {
10 OperationFailed,
11 waitForOperation as waitForLongRunning,
12 type LongRunningOperation,
13} from "../Operation.ts";
14
15export const DEFAULT_SCHEMA_ID = "main";
16export const MAX_NAME_LENGTH = 63;
17// Full connector names are capped at 200 chars; a 63-char service id leaves
18// room for a 48-char connector id.
19export const MAX_CONNECTOR_ID_LENGTH = 48;
20
21export class ResourceNotResolved extends Data.TaggedError(
22 "GCP.FirebaseDataConnect.ResourceNotResolved",
23)<{
24 name: string;
25}> {}
26
27export class ResourceStillExists extends Data.TaggedError(
28 "GCP.FirebaseDataConnect.ResourceStillExists",
29)<{
30 name: string;
31}> {}
32
33export class ResourceReconciling extends Data.TaggedError(
34 "GCP.FirebaseDataConnect.ResourceReconciling",
35)<{
36 name: string;
37}> {}
38
39export const lastSegment = (value: string) => {
40 const trimmed = value.replace(/\/+$/, "");
41 const parts = trimmed.split("/");
42 return parts[parts.length - 1] || trimmed;
43};
44
45export const rfc1035 = (
46 name: string,
47 fallback = "dataconnect",
48 maxLength = MAX_NAME_LENGTH,
49): string => {
50 let next = name
51 .toLowerCase()
52 .replace(/[^a-z0-9-]/g, "-")
53 .replace(/-+/g, "-")
54 .replace(/^-+|-+$/g, "");
55 if (!/^[a-z]/.test(next)) next = `a${next}`;
56 next = next.slice(0, maxLength).replace(/-+$/g, "");
57 if (next.length === 0) return fallback.slice(0, maxLength);
58 if (!/[a-z0-9]$/.test(next)) next = `${next.slice(0, maxLength - 1)}0`;
59 return next.slice(0, maxLength);
60};
61
62export const normalizeLocation = (
63 location: string | undefined,
64 defaultLocation: string,
65) => lastSegment(location ?? defaultLocation).toLowerCase();
66
67export const parentOf = (project: string, location: string) =>
68 `projects/${project}/locations/${lastSegment(location).toLowerCase()}`;
69
70export const toPhysicalId = (
71 id: string,
72 explicit: string | undefined,
73 existing: string | undefined,
74 fallback = "dataconnect",
75 maxLength = MAX_NAME_LENGTH,
76) =>
77 Effect.gen(function* () {
78 if (explicit !== undefined) return rfc1035(explicit, fallback, maxLength);
79 if (existing !== undefined) return existing;
80 return rfc1035(
81 yield* createPhysicalName({
82 id,
83 maxLength,
84 lowercase: true,
85 }),
86 fallback,
87 maxLength,
88 );
89 });
90
91export const parseName = (
92 name: string,
93 collection: string,
94 defaultLocation: string,
95) => {
96 const parts = name.split("/").filter((part) => part.length > 0);
97 const collectionAt = parts.lastIndexOf(collection);
98 const locationsAt = parts.lastIndexOf("locations");
99 const projectsAt = parts.lastIndexOf("projects");
100 const servicesAt = parts.lastIndexOf("services");
101 return {
102 project:
103 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
104 location:
105 locationsAt >= 0 && parts[locationsAt + 1]
106 ? parts[locationsAt + 1]!
107 : defaultLocation,
108 serviceId:
109 servicesAt >= 0 && parts[servicesAt + 1] ? parts[servicesAt + 1]! : "",
110 id:
111 collectionAt >= 0 && parts[collectionAt + 1]
112 ? parts[collectionAt + 1]!
113 : lastSegment(name),
114 parent:
115 collectionAt > 0
116 ? parts.slice(0, collectionAt).join("/")
117 : parts.slice(0, Math.max(0, parts.length - 1)).join("/"),
118 };
119};
120
121export const expandParent = (
122 value: string,
123 project: string,
124 location: string,
125 collection: string,
126) => {
127 if (value.includes("/")) return value.replace(/\/+$/, "");
128 return `projects/${project}/locations/${location}/${collection}/${value}`;
129};
130
131export const userLabels = (
132 labels: Record<string, string | undefined> | null | undefined,
133): Record<string, string> => stripInternalLabels(tagRecord(labels));
134
135export const stringMap = (
136 value: Record<string, string | undefined> | null | undefined,
137): Record<string, string> =>
138 Object.fromEntries(
139 Object.entries(value ?? {}).filter(
140 (entry): entry is [string, string] => typeof entry[1] === "string",
141 ),
142 );
143
144export const hasAlchemyLabelMap = (
145 labels: Record<string, string | undefined> | null | undefined,
146) => Object.keys(labels ?? {}).some((key) => key.startsWith("alchemy-"));
147
148export const canonical = (value: unknown): unknown => {
149 if (value === undefined || value === null) return undefined;
150 if (typeof value === "boolean" || typeof value === "number") return value;
151 if (typeof value === "string") {
152 return value.length === 0 ? undefined : value;
153 }
154 if (Array.isArray(value)) {
155 const items = value.map(canonical).filter((item) => item !== undefined);
156 return items.length === 0 ? undefined : items;
157 }
158 if (typeof value === "object") {
159 const entries = Object.entries(value as Record<string, unknown>)
160 .map(([key, item]) => [key, canonical(item)] as const)
161 .filter(([, item]) => item !== undefined)
162 .sort(([left], [right]) => left.localeCompare(right));
163 if (entries.length === 0) return undefined;
164 return Object.fromEntries(entries);
165 }
166 return undefined;
167};
168
169export const fingerprint = (value: unknown): string =>
170 JSON.stringify(canonical(value) ?? null);
171
172export const fieldMask = (fields: Array<string | false | undefined>) =>
173 fields
174 .filter((field): field is string => typeof field === "string")
175 .join(",");
176
177export const sameText = (left: string | undefined, right: string | undefined) =>
178 (left ?? "") === (right ?? "");
179
180export const replaceOnIdentity = (input: {
181 previousId: string | undefined;
182 nextId: string | undefined;
183 previousLocation: string;
184 nextLocation: string;
185 extra?: boolean;
186 previousParent?: string;
187 nextParent?: string;
188}) => {
189 const parentChanged =
190 (input.previousParent ?? "") !== "" &&
191 (input.nextParent ?? "") !== "" &&
192 (input.previousParent ?? "") !== (input.nextParent ?? "");
193 const replace =
194 (input.extra ?? false) ||
195 parentChanged ||
196 (input.previousId !== undefined &&
197 input.nextId !== undefined &&
198 input.nextId !== input.previousId) ||
199 input.previousLocation !== input.nextLocation;
200 if (!replace) return undefined;
201 const samePhysical =
202 input.previousLocation === input.nextLocation &&
203 !parentChanged &&
204 input.previousId !== undefined &&
205 input.nextId === input.previousId;
206 return {
207 action: "replace" as const,
208 deleteFirst: samePhysical,
209 };
210};
211
212export const waitUntilExists = <A, E, R>(
213 get: Effect.Effect<A | undefined, E, R>,
214 name: string,
215): Effect.Effect<NonNullable<A>, E | ResourceNotResolved, R> =>
216 get.pipe(
217 Effect.filterOrFail(
218 (value): value is NonNullable<A> => value !== undefined,
219 () => new ResourceNotResolved({ name }),
220 ),
221 Effect.retry({
222 while: (error) => error instanceof ResourceNotResolved,
223 times: 8,
224 schedule: Schedule.spaced("1 second"),
225 }),
226 );
227
228export const waitUntilReady = <A extends { reconciling?: boolean }, E, R>(
229 get: Effect.Effect<A | undefined, E, R>,
230 name: string,
231): Effect.Effect<
232 NonNullable<A>,
233 E | ResourceNotResolved | ResourceReconciling,
234 R
235> =>
236 get.pipe(
237 Effect.filterOrFail(
238 (value): value is NonNullable<A> => value !== undefined,
239 () => new ResourceNotResolved({ name }),
240 ),
241 Effect.filterOrFail(
242 (value) => value.reconciling !== true,
243 () => new ResourceReconciling({ name }),
244 ),
245 Effect.retry({
246 while: (error) =>
247 error instanceof ResourceNotResolved ||
248 error instanceof ResourceReconciling,
249 times: 10,
250 schedule: Schedule.spaced("3 seconds"),
251 }),
252 );
253
254export const waitUntilGone = <A, E, R>(
255 get: Effect.Effect<A | undefined, E, R>,
256 name: string,
257): Effect.Effect<void, E | ResourceStillExists, R> =>
258 get.pipe(
259 Effect.filterOrFail(
260 (value) => value === undefined,
261 () => new ResourceStillExists({ name }),
262 ),
263 Effect.retry({
264 while: (error) => error instanceof ResourceStillExists,
265 times: 10,
266 schedule: Schedule.spaced("3 seconds"),
267 }),
268 Effect.asVoid,
269 );
270
271/** Every location at once: the API accepts the `locations/-` wildcard. */
272export const listAtLocation = <A, E, R>(
273 project: string,
274 list: (parent: string) => Effect.Effect<A[], E, R>,
275) => list(`projects/${project}/locations/-`);
276
277/** Children of every parent in every location (`locations/-/{nested}`). */
278export const listAtNested = <A, E, R>(
279 project: string,
280 nested: string,
281 list: (parent: string) => Effect.Effect<A[], E, R>,
282) => list(`projects/${project}/locations/-/${nested}`);
283
284export const listLabeledPages = <Page, A, E, R>(
285 pages: Stream.Stream<Page, E, R>,
286 items: (page: Page) => readonly A[] | undefined,
287 labelsOf: (item: A) => Record<string, string | undefined> | null | undefined,
288) =>
289 pages.pipe(
290 Stream.flatMap((page) => Stream.fromIterable(items(page) ?? [])),
291 Stream.filter((item) => hasAlchemyLabelMap(labelsOf(item))),
292 Stream.runCollect,
293 Effect.map((chunk) => Array.from(chunk)),
294 );
295
296/** Services that provision Cloud SQL take up to ~20 minutes. */
297const OPERATION_BUDGET = "30 minutes";
298
299/**
300 * Wait for a long-running operation. ALREADY_EXISTS (code 6) means a
301 * concurrent create won the race; reconcile observes the resource next.
302 */
303export const waitForOperation = (operation: LongRunningOperation) =>
304 waitForLongRunning(
305 operation,
306 (name) => firebasedataconnect.getProjectsLocationsOperations({ name }),
307 {
308 budget: OPERATION_BUDGET,
309 },
310 ).pipe(
311 Effect.catchIf(
312 (error) => error._tag === "GCP.OperationFailed" && error.code === 6,
313 () => Effect.succeed(operation),
314 ),
315 );
316
317/**
318 * Wait for a delete operation. A vanished operation or NOT_FOUND (code 5)
319 * means the resource is already gone.
320 */
321export const waitForDeleteOperation = (operation: LongRunningOperation) =>
322 waitForOperation(operation).pipe(
323 Effect.catchIf(
324 (error) =>
325 error._tag === "NotFound" ||
326 (error._tag === "GCP.OperationFailed" && error.code === 5),
327 () => Effect.succeed(operation),
328 ),
329 );
330
331/**
332 * Re-run a patch-and-wait whose operation failed with a stale etag
333 * (ABORTED, code 10). Sibling updates (the parent service, another schema
334 * or connector) bump the server etag while the patch is in flight.
335 */
336export const retryStaleEtag = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
337 effect.pipe(
338 Effect.retry({
339 while: (error) => error instanceof OperationFailed && error.code === 10,
340 times: 5,
341 schedule: Schedule.spaced("5 seconds"),
342 }),
343 );