1import * as pubsub from "@distilled.cloud/gcp/pubsub_v1";
2import * as we from "@distilled.cloud/gcp/workspaceevents_v1";
3import * as Data from "effect/Data";
4import * as Effect from "effect/Effect";
5import * as Schedule from "effect/Schedule";
6import * as Stream from "effect/Stream";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import { tagRecord } from "../../Tags.ts";
9import {
10 alchemyLabelKeys,
11 createInternalLabels,
12 hasAlchemyLabels,
13 sanitizeLabelValue,
14} from "../Labels.ts";
15import { waitForOperation as waitForGcpOperation } from "../Operation.ts";
16
17export const SUBSCRIPTION_PREFIX = "subscriptions/";
18export const TASK_PREFIX = "tasks/";
19export const CONFIG_COLLECTION = "pushNotificationConfigs";
20export const MAX_CONFIG_ID_LENGTH = 63;
21export const MAX_TOPIC_ID_LENGTH = 255;
22export const MAX_TOKEN_LENGTH = 8000;
23
24export const WESUB_LABEL_KEYS = {
25 stack: "alchemy-wesub-stack",
26 stage: "alchemy-wesub-stage",
27 id: "alchemy-wesub-id",
28} as const;
29
30export const LIST_EVENT_TYPES = [
31 "google.workspace.chat.message.v1.created",
32 "google.workspace.chat.message.v1.updated",
33 "google.workspace.chat.message.v1.deleted",
34 "google.workspace.chat.membership.v1.created",
35 "google.workspace.chat.membership.v1.updated",
36 "google.workspace.chat.membership.v1.deleted",
37 "google.workspace.chat.reaction.v1.created",
38 "google.workspace.chat.reaction.v1.deleted",
39 "google.workspace.chat.space.v1.updated",
40 "google.workspace.chat.space.v1.deleted",
41 "google.workspace.meet.conference.v2.started",
42 "google.workspace.meet.conference.v2.ended",
43 "google.workspace.meet.participant.v2.joined",
44 "google.workspace.meet.participant.v2.left",
45 "google.workspace.meet.recording.v2.fileGenerated",
46 "google.workspace.meet.transcript.v2.fileGenerated",
47 "google.workspace.drive.file.v3.created",
48 "google.workspace.drive.file.v3.updated",
49 "google.workspace.drive.file.v3.deleted",
50] as const;
51
52export class ResourceNotResolved extends Data.TaggedError(
53 "GCP.WorkspaceEvents.ResourceNotResolved",
54)<{
55 name: string;
56}> {}
57
58export class ResourceStillExists extends Data.TaggedError(
59 "GCP.WorkspaceEvents.ResourceStillExists",
60)<{
61 name: string;
62}> {}
63
64export const lastSegment = (value: string) => {
65 const trimmed = value.replace(/\/+$/, "");
66 const parts = trimmed.split("/");
67 return parts[parts.length - 1] || trimmed;
68};
69
70export const sameText = (left: string | undefined, right: string | undefined) =>
71 (left ?? "") === (right ?? "");
72
73export const sameBoolean = (
74 left: boolean | undefined,
75 right: boolean | undefined,
76) => (left ?? false) === (right ?? false);
77
78export const jsonEqual = (left: unknown, right: unknown) =>
79 JSON.stringify(left ?? null) === JSON.stringify(right ?? null);
80
81export const sortedStrings = (values: readonly string[] | undefined) =>
82 [...(values ?? [])].slice().sort();
83
84export const updateMaskOf = (...fields: Array<string | undefined>) =>
85 fields.filter((field): field is string => field !== undefined).join(",");
86
87export const rfc1035 = (
88 name: string,
89 fallback = "config",
90 maxLength = MAX_CONFIG_ID_LENGTH,
91): string => {
92 let next = name
93 .toLowerCase()
94 .replace(/[^a-z0-9-]/g, "-")
95 .replace(/-+/g, "-")
96 .replace(/^-+/, "")
97 .replace(/-+$/, "");
98 if (!/^[a-z]/.test(next)) next = `w${next}`;
99 next = next.slice(0, maxLength).replace(/-+$/, "");
100 if (next.length === 0) return fallback;
101 if (!/[a-z0-9]$/.test(next)) {
102 next = `${next.slice(0, maxLength - 1)}0`;
103 }
104 return next.slice(0, maxLength);
105};
106
107export const toPhysicalId = (
108 id: string,
109 explicit: string | undefined,
110 existing: string | undefined,
111 fallback: string,
112 maxLength: number,
113) =>
114 Effect.gen(function* () {
115 if (explicit !== undefined && explicit.length > 0) {
116 return rfc1035(explicit, fallback, maxLength);
117 }
118 if (existing !== undefined && existing.length > 0) {
119 return existing;
120 }
121 return rfc1035(
122 yield* createPhysicalName({
123 id,
124 maxLength,
125 lowercase: true,
126 }),
127 fallback,
128 maxLength,
129 );
130 });
131
132export const wesubLabels = (id: string) =>
133 Effect.gen(function* () {
134 const internal = yield* createInternalLabels(id);
135 const labels: Record<string, string> = {
136 [WESUB_LABEL_KEYS.stack]:
137 internal[alchemyLabelKeys.stack] ?? sanitizeLabelValue("x"),
138 [WESUB_LABEL_KEYS.stage]:
139 internal[alchemyLabelKeys.stage] ?? sanitizeLabelValue("x"),
140 [WESUB_LABEL_KEYS.id]:
141 internal[alchemyLabelKeys.id] ?? sanitizeLabelValue("x"),
142 };
143 return labels;
144 });
145
146export const hasWesubLabels = (
147 labels: Record<string, string | undefined> | null | undefined,
148) =>
149 Object.keys(labels ?? {}).some(
150 (key) => key.startsWith("alchemy-wesub-") || key.startsWith("alchemy-"),
151 );
152
153const prefixMatch = (expected: string, observed: string) =>
154 expected === observed ||
155 expected.startsWith(observed) ||
156 observed.startsWith(expected);
157
158export const ownedByWesubLabels = (
159 id: string,
160 labels: Record<string, string | undefined> | null | undefined,
161) =>
162 Effect.gen(function* () {
163 const expected = yield* wesubLabels(id);
164 const observed = tagRecord(labels);
165 if (!hasWesubLabels(observed)) return false;
166 const exact =
167 observed[WESUB_LABEL_KEYS.stack] === expected[WESUB_LABEL_KEYS.stack] &&
168 observed[WESUB_LABEL_KEYS.stage] === expected[WESUB_LABEL_KEYS.stage] &&
169 observed[WESUB_LABEL_KEYS.id] === expected[WESUB_LABEL_KEYS.id];
170 if (exact) return true;
171 return (
172 prefixMatch(
173 expected[WESUB_LABEL_KEYS.stack] ?? "",
174 observed[WESUB_LABEL_KEYS.stack] ?? "",
175 ) &&
176 prefixMatch(
177 expected[WESUB_LABEL_KEYS.stage] ?? "",
178 observed[WESUB_LABEL_KEYS.stage] ?? "",
179 ) &&
180 prefixMatch(
181 expected[WESUB_LABEL_KEYS.id] ?? "",
182 observed[WESUB_LABEL_KEYS.id] ?? "",
183 )
184 );
185 });
186
187export const toSubscriptionName = (value: string) => {
188 if (value.length === 0) return value;
189 return value.startsWith(SUBSCRIPTION_PREFIX)
190 ? value
191 : `${SUBSCRIPTION_PREFIX}${value}`;
192};
193
194export const subscriptionIdOf = (name: string) =>
195 name.startsWith(SUBSCRIPTION_PREFIX)
196 ? name.slice(SUBSCRIPTION_PREFIX.length)
197 : lastSegment(name);
198
199export const toTaskName = (value: string) => {
200 if (value.length === 0) return value;
201 return value.startsWith(TASK_PREFIX) ? value : `${TASK_PREFIX}${value}`;
202};
203
204export const taskIdOf = (name: string) =>
205 name.startsWith(TASK_PREFIX)
206 ? name.slice(TASK_PREFIX.length)
207 : lastSegment(name);
208
209export const toConfigName = (task: string, configId: string) =>
210 `${toTaskName(task)}/${CONFIG_COLLECTION}/${configId}`;
211
212export const configIdOf = (name: string) => {
213 const parts = name.split("/").filter((part) => part.length > 0);
214 const at = parts.lastIndexOf(CONFIG_COLLECTION);
215 return at >= 0 && parts[at + 1] ? parts[at + 1]! : lastSegment(name);
216};
217
218export const parentOfConfig = (name: string) => {
219 const parts = name.split("/").filter((part) => part.length > 0);
220 const at = parts.lastIndexOf(CONFIG_COLLECTION);
221 return at > 0 ? parts.slice(0, at).join("/") : toTaskName(lastSegment(name));
222};
223
224export const expandTopic = (value: string, project: string) => {
225 const trimmed = value.replace(/\/+$/, "");
226 if (trimmed.includes("/topics/")) return trimmed;
227 return `projects/${project}/topics/${lastSegment(trimmed)}`;
228};
229
230export const listFilter = (
231 eventTypes: readonly string[],
232 targetResource?: string,
233) => {
234 const events = eventTypes.map((type) => `event_types:"${type}"`).join(" OR ");
235 if (targetResource !== undefined && targetResource.length > 0) {
236 return `(${events}) AND target_resource="${targetResource}"`;
237 }
238 return events;
239};
240
241export const catchMissing = <A, E extends { readonly _tag: string }, R>(
242 effect: Effect.Effect<A, E, R>,
243) =>
244 effect.pipe(
245 Effect.catchIf(
246 (error): error is E & { readonly _tag: "NotFound" } =>
247 error._tag === "NotFound",
248 () => Effect.succeed(undefined),
249 ),
250 );
251
252export const ignoreMissing = <E extends { readonly _tag: string }, R>(
253 effect: Effect.Effect<unknown, E, R>,
254) =>
255 effect.pipe(
256 Effect.catchIf(
257 (error): error is E & { readonly _tag: "NotFound" } =>
258 error._tag === "NotFound",
259 () => Effect.void,
260 ),
261 );
262
263export const getSubscription = (name: string) =>
264 name.length === 0
265 ? Effect.succeed(undefined)
266 : catchMissing(we.getSubscriptions({ name: toSubscriptionName(name) }));
267
268export const getConfig = (name: string, tenant?: string) =>
269 name.length === 0
270 ? Effect.succeed(undefined)
271 : catchMissing(
272 we.getTasksPushNotificationConfigs({
273 name,
274 tenant,
275 }),
276 );
277
278export const getTopic = (name: string) =>
279 name.length === 0
280 ? Effect.succeed(undefined)
281 : pubsub
282 .getProjectsTopics({ topic: name })
283 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
284
285export const ensureTopic = (name: string, labels: Record<string, string>) =>
286 Effect.gen(function* () {
287 let current = yield* getTopic(name);
288 if (current === undefined) {
289 const created = yield* pubsub
290 .createProjectsTopics({
291 name,
292 body: { labels },
293 })
294 .pipe(Effect.catchTag("Conflict", () => getTopic(name)));
295 current = created ?? undefined;
296 }
297 if (current === undefined) return undefined;
298 const observed = tagRecord(current.labels);
299 const desired = { ...observed, ...labels };
300 const changed = Object.entries(desired).some(
301 ([key, value]) => observed[key] !== value,
302 );
303 if (!changed) return current;
304 return yield* pubsub
305 .patchProjectsTopics({
306 name,
307 body: {
308 topic: { name, labels: desired },
309 updateMask: "labels",
310 },
311 })
312 .pipe(
313 Effect.catchTag(["NotFound", "Conflict"], () =>
314 Effect.succeed(current),
315 ),
316 );
317 });
318
319export const stringField = (
320 value: unknown,
321 key: string,
322): string | undefined => {
323 if (value === null || value === undefined || typeof value !== "object") {
324 return undefined;
325 }
326 const field = (value as Record<string, unknown>)[key];
327 return typeof field === "string" && field.length > 0 ? field : undefined;
328};
329
330export const operationResourceName = (operation: we.Operation) =>
331 stringField(operation.response, "name") ??
332 stringField(operation.metadata, "name");
333
334/**
335 * Wait for a Workspace Events long-running operation (subscription create/update/delete).
336 */
337export const waitForOperation = (operation: we.Operation) =>
338 Effect.gen(function* () {
339 yield* waitForGcpOperation(
340 operation,
341 (name) => we.getOperations({ name }),
342 { budget: "10 minutes" },
343 );
344 // Return the finished operation so callers can read its `response`.
345 if (operation.done === true || !operation.name) return operation;
346 return yield* we.getOperations({ name: operation.name });
347 });
348
349export const waitUntilExists = <A, E, R>(
350 get: Effect.Effect<A | undefined, E, R>,
351 name: string,
352): Effect.Effect<NonNullable<A>, E | ResourceNotResolved, R> =>
353 get.pipe(
354 Effect.filterOrFail(
355 (value): value is NonNullable<A> => value != null,
356 () => new ResourceNotResolved({ name }),
357 ),
358 Effect.retry({
359 while: (error) => error instanceof ResourceNotResolved,
360 times: 8,
361 schedule: Schedule.spaced("1 second"),
362 }),
363 );
364
365export const waitUntilGone = <A, E, R>(
366 get: Effect.Effect<A | undefined, E, R>,
367 name: string,
368): Effect.Effect<void, E | ResourceStillExists, R> =>
369 get.pipe(
370 Effect.filterOrFail(
371 (value) => value === undefined,
372 () => new ResourceStillExists({ name }),
373 ),
374 Effect.retry({
375 while: (error) => error instanceof ResourceStillExists,
376 times: 8,
377 schedule: Schedule.spaced("2 seconds"),
378 }),
379 Effect.asVoid,
380 );
381
382export const collectPages = <Page, A, E, R>(
383 pages: Stream.Stream<Page, E, R>,
384 items: (page: Page) => readonly A[] | undefined,
385) =>
386 pages.pipe(
387 Stream.flatMap((page) => Stream.fromIterable(items(page) ?? [])),
388 Stream.runCollect,
389 Effect.map((chunk) => Array.from(chunk)),
390 );
391
392export const listSubscriptions = (filter: string) =>
393 collectPages(
394 we.listSubscriptions.pages({
395 filter,
396 pageSize: 100,
397 }),
398 (page) => page.subscriptions,
399 ).pipe(
400 Effect.catchTag("NotFound", () => Effect.succeed([] as we.Subscription[])),
401 );
402
403export const listOwnedSubscriptions = () =>
404 Effect.gen(function* () {
405 const found = new Map<string, we.Subscription>();
406 const chunkSize = 6;
407 for (let index = 0; index < LIST_EVENT_TYPES.length; index += chunkSize) {
408 const chunk = LIST_EVENT_TYPES.slice(index, index + chunkSize);
409 const page = yield* listSubscriptions(listFilter(chunk));
410 for (const subscription of page) {
411 const name = subscription.name;
412 if (name !== undefined && name.length > 0) {
413 found.set(name, subscription);
414 }
415 }
416 }
417 const owned: we.Subscription[] = [];
418 for (const subscription of found.values()) {
419 const topic = subscription.notificationEndpoint?.pubsubTopic;
420 if (topic === undefined || topic.length === 0) continue;
421 const observed = yield* getTopic(topic);
422 if (hasWesubLabels(observed?.labels)) {
423 owned.push(subscription);
424 }
425 }
426 return owned;
427 });
428
429export const findSubscription = (
430 name: string,
431 eventTypes: readonly string[],
432 targetResource: string | undefined,
433) =>
434 Effect.gen(function* () {
435 const existing = yield* getSubscription(name);
436 if (existing !== undefined) return existing;
437 if (eventTypes.length === 0) return undefined;
438 const listed = yield* listSubscriptions(
439 listFilter(eventTypes, targetResource),
440 );
441 return listed[0];
442 });
443
444export const listConfigs = (parent: string, tenant?: string) =>
445 collectPages(
446 we.listTasksPushNotificationConfigs.pages({
447 parent,
448 tenant,
449 pageSize: 100,
450 }),
451 (page) => page.configs,
452 ).pipe(
453 Effect.catchTag("NotFound", () =>
454 Effect.succeed([] as we.TaskPushNotificationConfig[]),
455 ),
456 );