1import * as eventarc from "@distilled.cloud/gcp/eventarc_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 type { GcpOpContext } from "@distilled.cloud/gcp/Protocol";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import { tagRecord } from "../../Tags.ts";
9import { stripInternalLabels } from "../Labels.ts";
10import {
11 type LongRunningOperation,
12 waitForOperation as waitForLongRunningOperation,
13} from "../Operation.ts";
14
15export const MAX_NAME_LENGTH = 63;
16
17export class EventarcNotResolved extends Data.TaggedError(
18 "GCP.Eventarc.NotResolved",
19)<{
20 name: string;
21}> {}
22
23export class EventarcStillExists extends Data.TaggedError(
24 "GCP.Eventarc.StillExists",
25)<{
26 name: string;
27}> {}
28
29export type LoggingConfig = {
30 /**
31 * Minimum log severity sent to Cloud Logging (`NONE`, `DEBUG`,
32 * `INFO`, `NOTICE`, `WARNING`, `ERROR`, `CRITICAL`, `ALERT`,
33 * `EMERGENCY`).
34 */
35 logSeverity?: eventarc.LoggingConfigLogSeverityEnum | (string & {});
36};
37
38export const lastSegment = (value: string) => {
39 const trimmed = value.replace(/\/+$/, "");
40 const parts = trimmed.split("/");
41 return parts[parts.length - 1] || trimmed;
42};
43
44export const rfc1035 = (
45 name: string,
46 fallback = "resource",
47 maxLength = MAX_NAME_LENGTH,
48): string => {
49 let next = name
50 .toLowerCase()
51 .replace(/[^a-z0-9-]/g, "-")
52 .replace(/-+/g, "-")
53 .replace(/^-+/, "")
54 .replace(/-+$/, "");
55 if (!/^[a-z]/.test(next)) {
56 next = `e${next}`;
57 }
58 next = next.slice(0, maxLength).replace(/-+$/, "");
59 return next.length > 0 ? next : fallback;
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/${location}`;
69
70export const resourceName = (
71 project: string,
72 location: string,
73 collection: string,
74 id: string,
75) => `projects/${project}/locations/${location}/${collection}/${id}`;
76
77export const parseName = (
78 name: string,
79 collection: string,
80 defaultLocation: string,
81) => {
82 const parts = name.split("/").filter((part) => part.length > 0);
83 const collectionAt = parts.lastIndexOf(collection);
84 const locationsAt = parts.lastIndexOf("locations");
85 const projectsAt = parts.lastIndexOf("projects");
86 return {
87 project:
88 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
89 location:
90 locationsAt >= 0 && parts[locationsAt + 1]
91 ? parts[locationsAt + 1]!
92 : defaultLocation,
93 id:
94 collectionAt >= 0 && parts[collectionAt + 1]
95 ? parts[collectionAt + 1]!
96 : lastSegment(name),
97 };
98};
99
100export const toPhysicalId = (
101 id: string,
102 explicit: string | undefined,
103 existing: string | undefined,
104 fallback = "resource",
105 maxLength = MAX_NAME_LENGTH,
106) =>
107 Effect.gen(function* () {
108 if (explicit !== undefined) return rfc1035(explicit, fallback, maxLength);
109 if (existing !== undefined) return existing;
110 return rfc1035(
111 yield* createPhysicalName({
112 id,
113 maxLength,
114 lowercase: true,
115 }),
116 fallback,
117 maxLength,
118 );
119 });
120
121export const userLabels = (
122 labels: Record<string, string | undefined> | null | undefined,
123): Record<string, string> => stripInternalLabels(tagRecord(labels));
124
125export const userAnnotations = (
126 annotations: Record<string, string | undefined> | null | undefined,
127): Record<string, string> => tagRecord(annotations);
128
129export const hasAlchemyLabelKeys = (
130 labels: Record<string, string | undefined> | null | undefined,
131) => Object.keys(labels ?? {}).some((key) => key.startsWith("alchemy-"));
132
133export const expandResource = (
134 value: string,
135 project: string,
136 location: string,
137 collection: string,
138) => {
139 const trimmed = value.replace(/\/+$/, "");
140 if (trimmed.includes(`/${collection}/`)) return trimmed;
141 return resourceName(project, location, collection, lastSegment(trimmed));
142};
143
144export const expandTopic = (value: string, project: string) => {
145 const trimmed = value.replace(/\/+$/, "");
146 if (trimmed.includes("/topics/")) return trimmed;
147 return `projects/${project}/topics/${lastSegment(trimmed)}`;
148};
149
150export const toLoggingConfig = (
151 config: LoggingConfig | eventarc.LoggingConfig | undefined,
152): LoggingConfig | undefined => {
153 const logSeverity = config?.logSeverity;
154 if (logSeverity === undefined || logSeverity.length === 0) return undefined;
155 return { logSeverity };
156};
157
158export const loggingKey = (
159 config: LoggingConfig | eventarc.LoggingConfig | undefined,
160) => toLoggingConfig(config)?.logSeverity ?? "";
161
162export const cryptoKeyKey = (name: string | undefined) =>
163 name === undefined || name.length === 0 ? "" : name;
164
165export const textKey = (value: string | undefined) => value ?? "";
166
167export const sameJson = (left: unknown, right: unknown) =>
168 JSON.stringify(canonical(left) ?? null) ===
169 JSON.stringify(canonical(right) ?? null);
170
171export const canonical = (value: unknown): unknown => {
172 if (value === undefined || value === null) return undefined;
173 if (typeof value === "boolean" || typeof value === "number") return value;
174 if (typeof value === "string") {
175 return value.length === 0 ? undefined : value;
176 }
177 if (Array.isArray(value)) {
178 const items = value.map(canonical).filter((item) => item !== undefined);
179 return items.length === 0 ? undefined : items;
180 }
181 if (typeof value === "object") {
182 const entries = Object.entries(value as Record<string, unknown>)
183 .map(([key, item]) => [key, canonical(item)] as const)
184 .filter(([, item]) => item !== undefined)
185 .sort(([left], [right]) => left.localeCompare(right));
186 if (entries.length === 0) return undefined;
187 return Object.fromEntries(entries);
188 }
189 return undefined;
190};
191
192export const changedFields = (
193 pairs: ReadonlyArray<readonly [string, boolean]>,
194) => pairs.filter(([, changed]) => changed).map(([field]) => field);
195
196export const compact = <T extends Record<string, unknown>>(value: T): T => {
197 const next: Record<string, unknown> = {};
198 for (const [key, entry] of Object.entries(value)) {
199 if (entry !== undefined && entry !== "") next[key] = entry;
200 }
201 return next as T;
202};
203
204/**
205 * Wait on an Eventarc long-running operation. Pipelines and enrollments
206 * routinely take ~4 minutes. `ALREADY_EXISTS` (a create race) counts as
207 * success; `NOT_FOUND` does too when `notFoundOk` (deletes).
208 */
209export const waitForOperation = (
210 operation: eventarc.GoogleLongrunningOperation,
211 options?: { notFoundOk?: boolean },
212) =>
213 waitForLongRunningOperation(
214 operation,
215 (name) => {
216 const get = eventarc.getProjectsLocationsOperations({ name });
217 const observe: Effect.Effect<
218 LongRunningOperation,
219 eventarc.GetProjectsLocationsOperationsError,
220 GcpOpContext
221 > =
222 options?.notFoundOk === true
223 ? get.pipe(
224 Effect.catchTag("NotFound", () =>
225 Effect.succeed<LongRunningOperation>({ name, done: true }),
226 ),
227 )
228 : get.pipe(
229 // A just-returned operation can briefly 404 on read.
230 Effect.retry({
231 while: (error) => error._tag === "NotFound",
232 times: 5,
233 schedule: Schedule.exponential("250 millis"),
234 }),
235 );
236 return observe;
237 },
238 { budget: "15 minutes" },
239 ).pipe(
240 // google.rpc.Code ALREADY_EXISTS = 6, NOT_FOUND = 5.
241 Effect.catchTag("GCP.OperationFailed", (error) =>
242 error.code === 6 || (options?.notFoundOk === true && error.code === 5)
243 ? Effect.succeed<LongRunningOperation>(operation)
244 : Effect.fail(error),
245 ),
246 );
247
248export const retryOnTransient = <A, E extends { readonly _tag: string }, R>(
249 effect: Effect.Effect<A, E, R>,
250) =>
251 effect.pipe(
252 Effect.retry({
253 while: (error) =>
254 error._tag === "Conflict" || error._tag === "GCP.OperationFailed",
255 times: 8,
256 schedule: Schedule.spaced("3 seconds"),
257 }),
258 );
259
260export const waitUntilPresent = <A, E, R>(
261 get: Effect.Effect<A | undefined, E, R>,
262 name: string,
263) =>
264 get.pipe(
265 Effect.flatMap((value) =>
266 value
267 ? Effect.succeed(value)
268 : Effect.fail(new EventarcNotResolved({ name })),
269 ),
270 Effect.retry({
271 while: (error) => error instanceof EventarcNotResolved,
272 times: 8,
273 schedule: Schedule.spaced("1 second"),
274 }),
275 );
276
277export const waitUntilGone = <A, E, R>(
278 get: Effect.Effect<A | undefined, E, R>,
279 name: string,
280) =>
281 get.pipe(
282 Effect.flatMap((value) =>
283 value === undefined
284 ? Effect.void
285 : Effect.fail(new EventarcStillExists({ name })),
286 ),
287 Effect.retry({
288 while: (error) => error instanceof EventarcStillExists,
289 times: 10,
290 schedule: Schedule.spaced("2 seconds"),
291 }),
292 );
293
294export const collectPages = <
295 Page,
296 Item,
297 E extends { readonly _tag: string },
298 R,
299>(
300 stream: Stream.Stream<Page, E, R>,
301 pick: (page: Page) => readonly Item[] | undefined,
302) =>
303 stream.pipe(
304 Stream.flatMap((page) => Stream.fromIterable(pick(page) ?? [])),
305 Stream.runCollect,
306 Effect.map((chunk) => Array.from(chunk)),
307 // A missing parent location lists as empty.
308 Effect.catchIf(
309 (error): error is E & { readonly _tag: "NotFound" } =>
310 error._tag === "NotFound",
311 () => Effect.succeed<Item[]>([]),
312 ),
313 );