1import * as pubsublite from "@distilled.cloud/gcp/pubsublite_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 { GcpEnvironment } from "../Environment.ts";
8import {
9 alchemyLabelKeys,
10 createInternalLabels,
11 hasAlchemyLabels,
12} from "../Labels.ts";
13
14// Zonal default is left fixed: zones are not derivable from `GcpEnvironment.region` without assuming a suffix.
15export const DEFAULT_ZONE = "us-central1-a";
16export const MAX_ID_LENGTH = 255;
17export const DEFAULT_THROUGHPUT_CAPACITY = "4";
18export const DEFAULT_PARTITION_COUNT = "1";
19export const DEFAULT_PUBLISH_MIB = 4;
20export const DEFAULT_SUBSCRIBE_MIB = 4;
21/** 30 GiB — Pub/Sub Lite minimum storage per partition. */
22export const DEFAULT_PER_PARTITION_BYTES = "32212254720";
23
24export const LIST_LOCATIONS = [
25 "us-central1",
26 "us-central1-a",
27 "us-central1-b",
28 "us-central1-c",
29 "us-central1-f",
30] as const;
31
32const OWNERSHIP_MARKER = "+alc.";
33
34export class ResourceNotResolved extends Data.TaggedError(
35 "GCP.PubSubLite.NotResolved",
36)<{
37 name: string;
38}> {}
39
40export const lastSegment = (value: string) => {
41 const trimmed = value.replace(/\/+$/, "");
42 const parts = trimmed.split("/");
43 return parts[parts.length - 1] || trimmed;
44};
45
46export const normalizeLocation = (
47 location: string | undefined,
48 fallback: string,
49) => lastSegment(location ?? fallback).toLowerCase();
50
51export const regionOf = (location: string) => {
52 const parts = location.split("-").filter((part) => part.length > 0);
53 const tail = parts[parts.length - 1] ?? "";
54 if (/^[a-z]$/.test(tail) && parts.length > 1) {
55 return parts.slice(0, -1).join("-");
56 }
57 return location;
58};
59
60export const parentOf = (project: string, location: string) =>
61 `projects/${project}/locations/${location}`;
62
63export const resourceName = (
64 project: string,
65 location: string,
66 collection: string,
67 id: string,
68) => `${parentOf(project, location)}/${collection}/${id}`;
69
70export const parseName = (name: string, collection: string) => {
71 const parts = name.split("/").filter((part) => part.length > 0);
72 const collectionAt = parts.lastIndexOf(collection);
73 const locationsAt = parts.lastIndexOf("locations");
74 const projectsAt = parts.lastIndexOf("projects");
75 return {
76 project:
77 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
78 location:
79 locationsAt >= 0 && parts[locationsAt + 1] ? parts[locationsAt + 1]! : "",
80 id:
81 collectionAt >= 0 && parts[collectionAt + 1]
82 ? parts[collectionAt + 1]!
83 : lastSegment(name),
84 parent:
85 collectionAt > 0
86 ? parts.slice(0, collectionAt).join("/")
87 : parts.slice(0, Math.max(0, parts.length - 1)).join("/"),
88 };
89};
90
91export const expandName = (
92 value: string,
93 project: string,
94 location: string,
95 collection: string,
96) => {
97 const trimmed = value.replace(/\/+$/, "");
98 if (trimmed.includes(`/${collection}/`)) return trimmed;
99 if (trimmed.startsWith("projects/")) return trimmed;
100 return resourceName(project, location, collection, lastSegment(trimmed));
101};
102
103export const sameText = (left: string | undefined, right: string | undefined) =>
104 (left ?? "") === (right ?? "");
105
106export const sameRef = (left: string | undefined, right: string | undefined) =>
107 lastSegment(left ?? "") === lastSegment(right ?? "");
108
109export const jsonEqual = (left: unknown, right: unknown) =>
110 JSON.stringify(left ?? null) === JSON.stringify(right ?? null);
111
112export const fieldMask = (fields: Array<string | false | undefined>) =>
113 fields
114 .filter((field): field is string => typeof field === "string")
115 .join(",");
116
117const sanitizeId = (value: string, maxLength: number) => {
118 let next = value.replace(/[^a-zA-Z0-9._~+%:-]/g, "-").replace(/-+/g, "-");
119 next = next.replace(/^[^a-zA-Z]+/, "");
120 if (next.length === 0) next = "topic";
121 if (!/^[a-zA-Z]/.test(next)) next = `p${next}`;
122 return next.slice(0, maxLength).replace(/[^a-zA-Z0-9._~+%:-]+$/g, "");
123};
124
125export const stripOwnershipId = (id: string) => {
126 const index = id.lastIndexOf(OWNERSHIP_MARKER);
127 return index >= 0 ? id.slice(0, index) : id;
128};
129
130export const parseOwnershipId = (
131 id: string | undefined,
132): Record<string, string> => {
133 if (!id) return {};
134 const index = id.lastIndexOf(OWNERSHIP_MARKER);
135 if (index < 0) return {};
136 const [stack, stage, ...idParts] = id
137 .slice(index + OWNERSHIP_MARKER.length)
138 .split(".");
139 const labels: Record<string, string> = {};
140 if (stack) labels[alchemyLabelKeys.stack] = stack;
141 if (stage) labels[alchemyLabelKeys.stage] = stage;
142 if (idParts.length > 0) {
143 labels[alchemyLabelKeys.id] = idParts.join(".");
144 }
145 return labels;
146};
147
148export const hasOwnershipMarker = (id: string | undefined) =>
149 Object.keys(parseOwnershipId(id)).some((key) => key.startsWith("alchemy-"));
150
151const fitMarker = (labels: Record<string, string>, maxLength: number) => {
152 let stack = labels[alchemyLabelKeys.stack] ?? "x";
153 let stage = labels[alchemyLabelKeys.stage] ?? "x";
154 let id = labels[alchemyLabelKeys.id] ?? "x";
155 let marker = `${OWNERSHIP_MARKER}${stack}.${stage}.${id}`;
156 while (
157 marker.length > maxLength &&
158 (stack.length > 1 || stage.length > 1 || id.length > 1)
159 ) {
160 if (id.length >= stack.length && id.length >= stage.length) {
161 id = id.slice(0, -1);
162 } else if (stack.length >= stage.length) {
163 stack = stack.slice(0, -1);
164 } else {
165 stage = stage.slice(0, -1);
166 }
167 marker = `${OWNERSHIP_MARKER}${stack}.${stage}.${id}`;
168 }
169 return marker.slice(0, maxLength);
170};
171
172export const encodeResourceId = (
173 labels: Record<string, string>,
174 base: string,
175 maxLength = MAX_ID_LENGTH,
176) => {
177 const stripped = sanitizeId(stripOwnershipId(base), maxLength);
178 const minMarker = `${OWNERSHIP_MARKER}x.x.x`.length;
179 const marker = fitMarker(
180 labels,
181 Math.max(minMarker, Math.min(120, maxLength - 8)),
182 );
183 const room = Math.max(1, maxLength - marker.length);
184 return sanitizeId(`${stripped.slice(0, room)}${marker}`, maxLength);
185};
186
187const prefixMatch = (expected: string, observed: string) =>
188 expected === observed ||
189 expected.startsWith(observed) ||
190 observed.startsWith(expected);
191
192export const ownedByAlchemy = (logicalId: string, resourceId: string) =>
193 Effect.gen(function* () {
194 if (!hasOwnershipMarker(resourceId)) return false;
195 const expected = yield* createInternalLabels(logicalId);
196 const labels = parseOwnershipId(resourceId);
197 const exact = yield* hasAlchemyLabels(logicalId, labels);
198 if (exact) return true;
199 return (
200 prefixMatch(
201 expected[alchemyLabelKeys.stack] ?? "",
202 labels[alchemyLabelKeys.stack] ?? "",
203 ) &&
204 prefixMatch(
205 expected[alchemyLabelKeys.stage] ?? "",
206 labels[alchemyLabelKeys.stage] ?? "",
207 ) &&
208 prefixMatch(
209 expected[alchemyLabelKeys.id] ?? "",
210 labels[alchemyLabelKeys.id] ?? "",
211 )
212 );
213 });
214
215export const toResourceId = (
216 logicalId: string,
217 requested: string | undefined,
218 existing: string | undefined,
219) =>
220 Effect.gen(function* () {
221 const labels = yield* createInternalLabels(logicalId);
222 if (requested !== undefined && requested.length > 0) {
223 return encodeResourceId(labels, lastSegment(requested));
224 }
225 if (existing !== undefined && existing.length > 0) {
226 return existing;
227 }
228 const generated = yield* createPhysicalName({
229 id: logicalId,
230 maxLength: 120,
231 lowercase: true,
232 forbiddenPrefixes: ["goog"],
233 });
234 return encodeResourceId(labels, generated);
235 });
236
237export const replaceOnIdentity = (input: {
238 previousId?: string;
239 nextId?: string;
240 previousLocation?: string;
241 nextLocation?: string;
242 extra?: boolean;
243}) => {
244 const previousId =
245 input.previousId !== undefined
246 ? stripOwnershipId(lastSegment(input.previousId))
247 : undefined;
248 const nextId =
249 input.nextId !== undefined
250 ? stripOwnershipId(lastSegment(input.nextId))
251 : undefined;
252 const idChanged =
253 previousId !== undefined && nextId !== undefined && previousId !== nextId;
254 const locationChanged =
255 input.previousLocation !== undefined &&
256 input.nextLocation !== undefined &&
257 input.previousLocation !== input.nextLocation;
258 if (!idChanged && !locationChanged && input.extra !== true) {
259 return undefined;
260 }
261 const samePhysical = !idChanged && !locationChanged;
262 return {
263 action: "replace" as const,
264 deleteFirst: samePhysical,
265 };
266};
267
268export const countOf = (value: number | string | undefined) => {
269 if (value === undefined || value === "")
270 return Number(DEFAULT_PARTITION_COUNT);
271 const parsed = Number(value);
272 return Number.isFinite(parsed) ? parsed : Number(DEFAULT_PARTITION_COUNT);
273};
274
275export const partitionBody = (input: {
276 count?: number | string;
277 capacity?: {
278 publishMibPerSec?: number;
279 subscribeMibPerSec?: number;
280 };
281}): pubsublite.PartitionConfig => ({
282 count: String(countOf(input.count)),
283 capacity: {
284 publishMibPerSec: input.capacity?.publishMibPerSec ?? DEFAULT_PUBLISH_MIB,
285 subscribeMibPerSec:
286 input.capacity?.subscribeMibPerSec ?? DEFAULT_SUBSCRIBE_MIB,
287 },
288});
289
290export const retentionBody = (input: {
291 perPartitionBytes?: string;
292 period?: string;
293}): pubsublite.RetentionConfig => {
294 const body: pubsublite.RetentionConfig = {
295 perPartitionBytes: input.perPartitionBytes ?? DEFAULT_PER_PARTITION_BYTES,
296 };
297 if (input.period !== undefined) body.period = input.period;
298 return body;
299};
300
301export const partitionKey = (config: pubsublite.PartitionConfig | undefined) =>
302 JSON.stringify({
303 count: String(countOf(config?.count)),
304 publish: config?.capacity?.publishMibPerSec ?? DEFAULT_PUBLISH_MIB,
305 subscribe: config?.capacity?.subscribeMibPerSec ?? DEFAULT_SUBSCRIBE_MIB,
306 });
307
308export const retentionKey = (config: pubsublite.RetentionConfig | undefined) =>
309 JSON.stringify({
310 perPartitionBytes: config?.perPartitionBytes ?? DEFAULT_PER_PARTITION_BYTES,
311 period: config?.period ?? "",
312 });
313
314export const catchMissing = <A, E extends { readonly _tag: string }, R>(
315 effect: Effect.Effect<A, E, R>,
316) =>
317 effect.pipe(
318 Effect.catchIf(
319 (error): error is E & { readonly _tag: "NotFound" } =>
320 error._tag === "NotFound",
321 () => Effect.succeed(undefined),
322 ),
323 );
324
325export const ignoreMissing = <E extends { readonly _tag: string }, R>(
326 effect: Effect.Effect<unknown, E, R>,
327) =>
328 effect.pipe(
329 Effect.catchIf(
330 (error): error is E & { readonly _tag: "NotFound" } =>
331 error._tag === "NotFound",
332 () => Effect.void,
333 ),
334 );
335
336/**
337 * Retry while a delete is blocked by a dependent (a reservation still
338 * referenced by a topic, a topic with attached subscriptions): `409`.
339 */
340export const retryInUse = <A, E extends { readonly _tag: string }, R>(
341 effect: Effect.Effect<A, E, R>,
342) =>
343 effect.pipe(
344 Effect.retry({
345 while: (error) => error._tag === "Conflict",
346 times: 8,
347 schedule: Schedule.exponential("300 millis"),
348 }),
349 );
350
351export class DeleteNotConfirmed extends Data.TaggedError(
352 "GCP.PubSubLite.DeleteNotConfirmed",
353)<{}> {}
354
355/** Poll until the resource is gone; fails if it is still readable after ~60s. */
356export const waitUntilGone = <A, E, R>(
357 get: Effect.Effect<A | undefined, E, R>,
358) =>
359 get.pipe(
360 Effect.repeat({
361 schedule: Schedule.spaced("2 seconds"),
362 until: (value) => value === undefined,
363 times: 30,
364 }),
365 Effect.flatMap((value): Effect.Effect<void, DeleteNotConfirmed> =>
366 value === undefined ? Effect.void : Effect.fail(new DeleteNotConfirmed()),
367 ),
368 );
369
370const emptyList = <A>() => Effect.succeed([] as A[]);
371
372const collectPages = <Page, Item, E extends { readonly _tag: string }, R>(
373 pages: Stream.Stream<Page, E, R>,
374 items: (page: Page) => readonly Item[] | null | undefined,
375) =>
376 pages.pipe(
377 Stream.flatMap((page) => Stream.fromIterable(items(page) ?? [])),
378 Stream.runCollect,
379 Effect.map((chunk) => Array.from(chunk) as Item[]),
380 Effect.catchIf(
381 (error): error is E & { readonly _tag: "NotFound" } =>
382 error._tag === "NotFound",
383 () => emptyList<Item>(),
384 ),
385 );
386
387const listParents = (project: string, region: string) =>
388 [...new Set<string>([region, ...LIST_LOCATIONS])].map((location) =>
389 parentOf(project, location),
390 );
391
392const listAcross = <Item, E, R>(
393 listAt: (parent: string) => Effect.Effect<readonly Item[], E, R>,
394) =>
395 Effect.gen(function* () {
396 const env = yield* GcpEnvironment.current;
397 const pages = yield* Effect.forEach(
398 listParents(env.project, env.region),
399 (parent) => listAt(parent),
400 { concurrency: 4 },
401 );
402 const seen = new Set<string>();
403 const found: Item[] = [];
404 for (const page of pages) {
405 for (const item of page) {
406 const key =
407 typeof item === "object" &&
408 item !== null &&
409 "name" in item &&
410 typeof (item as { name?: unknown }).name === "string"
411 ? ((item as { name: string }).name ?? "")
412 : JSON.stringify(item);
413 if (seen.has(key)) continue;
414 seen.add(key);
415 found.push(item as Item);
416 }
417 }
418 return found;
419 });
420
421export const getReservation = (name: string) =>
422 name.length === 0
423 ? Effect.succeed(undefined)
424 : catchMissing(pubsublite.getAdminProjectsLocationsReservations({ name }));
425
426export const getTopic = (name: string) =>
427 name.length === 0
428 ? Effect.succeed(undefined)
429 : catchMissing(pubsublite.getAdminProjectsLocationsTopics({ name }));
430
431export const getSubscription = (name: string) =>
432 name.length === 0
433 ? Effect.succeed(undefined)
434 : catchMissing(pubsublite.getAdminProjectsLocationsSubscriptions({ name }));
435
436export const listReservationsAt = (parent: string) =>
437 collectPages(
438 pubsublite.listAdminProjectsLocationsReservations.pages({
439 parent,
440 pageSize: 1000,
441 }),
442 (page) => page.reservations,
443 );
444
445export const listTopicsAt = (parent: string) =>
446 collectPages(
447 pubsublite.listAdminProjectsLocationsTopics.pages({
448 parent,
449 pageSize: 1000,
450 }),
451 (page) => page.topics,
452 );
453
454export const listSubscriptionsAt = (parent: string) =>
455 collectPages(
456 pubsublite.listAdminProjectsLocationsSubscriptions.pages({
457 parent,
458 pageSize: 1000,
459 }),
460 (page) => page.subscriptions,
461 );
462
463export const listOwnedReservations = () =>
464 listAcross(listReservationsAt).pipe(
465 Effect.map((items) =>
466 items.filter((item) => hasOwnershipMarker(lastSegment(item.name ?? ""))),
467 ),
468 );
469
470export const listOwnedTopics = () =>
471 listAcross(listTopicsAt).pipe(
472 Effect.map((items) =>
473 items.filter((item) => hasOwnershipMarker(lastSegment(item.name ?? ""))),
474 ),
475 );
476
477export const listOwnedSubscriptions = () =>
478 listAcross(listSubscriptionsAt).pipe(
479 Effect.map((items) =>
480 items.filter((item) => hasOwnershipMarker(lastSegment(item.name ?? ""))),
481 ),
482 );