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 { Unowned } from "../../AdoptPolicy.ts";
6import { isResolved } from "../../Diff.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { GcpEnvironment } from "../Environment.ts";
10import type { Providers } from "../Providers.ts";
11import {
12 ensureTopic,
13 expandTopic,
14 findSubscription,
15 getSubscription,
16 getTopic,
17 lastSegment,
18 listOwnedSubscriptions,
19 operationResourceName,
20 ownedByWesubLabels,
21 sameText,
22 sortedStrings,
23 subscriptionIdOf,
24 toPhysicalId,
25 toSubscriptionName,
26 updateMaskOf,
27 waitForOperation,
28 waitUntilExists,
29 waitUntilGone,
30 wesubLabels,
31} from "./internal.ts";
32
33export type PayloadOptions = {
34 /**
35 * Whether the event payload includes data about the changed resource.
36 * Chat and Drive events only.
37 */
38 includeResource?: boolean;
39 /**
40 * Comma-separated fields to include when `includeResource` is true.
41 */
42 fieldMask?: string;
43};
44
45export type DriveOptions = {
46 /**
47 * Receive events about Drive files that are children of the target
48 * folder or shared drive. Immutable — changing it replaces the
49 * subscription.
50 */
51 includeDescendants?: boolean;
52};
53
54export type SubscriptionProps = {
55 /**
56 * Resource name `subscriptions/{subscription}` or the `{subscription}`
57 * id. Server-assigned on create. Immutable — changing it replaces the
58 * subscription.
59 */
60 subscriptionId?: string;
61 /**
62 * Google Workspace resource to monitor, as a full resource name
63 * (`//chat.googleapis.com/spaces/{space}`,
64 * `//drive.googleapis.com/files/{file}`,
65 * `//meet.googleapis.com/spaces/{space}`). Immutable — changing it
66 * replaces the subscription.
67 */
68 targetResource: string;
69 /**
70 * CloudEvents types to receive about the target resource. Lifecycle
71 * events are delivered automatically.
72 */
73 eventTypes: string[];
74 /**
75 * Pub/Sub topic that receives events
76 * (`projects/{project}/topics/{topic}` or a topic id). Created with
77 * Alchemy ownership labels when omitted. Immutable — changing it
78 * replaces the subscription. The topic must live in the same project.
79 */
80 pubsubTopic?: string;
81 /**
82 * Event payload options. Chat and Drive only. Immutable — changing
83 * them replaces the subscription.
84 */
85 payloadOptions?: PayloadOptions;
86 /**
87 * Drive-only options. Immutable — changing them replaces the
88 * subscription.
89 */
90 driveOptions?: DriveOptions;
91 /**
92 * RFC3339 expiration timestamp. Mutually exclusive with `ttl`.
93 */
94 expireTime?: string;
95 /**
96 * Time-to-live duration (for example `"86400s"`). Mutually exclusive
97 * with `expireTime`. Omitted uses the API maximum.
98 */
99 ttl?: string;
100};
101
102export type Subscription = Resource<
103 "GCP.WorkspaceEvents.Subscription",
104 SubscriptionProps,
105 {
106 /** Full resource name `subscriptions/{subscription}`. */
107 name: string;
108 /** Subscription id (last path segment). */
109 subscriptionId: string;
110 /** Project id used when the subscription was reconciled. */
111 project: string;
112 /** Monitored Google Workspace resource. */
113 targetResource: string | undefined;
114 /** CloudEvents types. */
115 eventTypes: string[];
116 /** Pub/Sub topic that receives events. */
117 pubsubTopic: string | undefined;
118 /** Event payload options. */
119 payloadOptions: PayloadOptions | undefined;
120 /** Drive-only options. */
121 driveOptions: DriveOptions | undefined;
122 /** RFC3339 expiration timestamp. */
123 expireTime: string | undefined;
124 /** Subscription state (`ACTIVE`, `SUSPENDED`, `DELETED`). */
125 state: string | undefined;
126 /** Error that suspended the subscription, if any. */
127 suspensionReason: string | undefined;
128 /** User who authorized the subscription. */
129 authority: string | undefined;
130 /** Service account that authorized the subscription. */
131 serviceAccountAuthority: string | undefined;
132 /** System-assigned unique identifier. */
133 uid: string | undefined;
134 /** RFC3339 creation timestamp. */
135 createTime: string | undefined;
136 /** RFC3339 last-update timestamp. */
137 updateTime: string | undefined;
138 /** Whether Alchemy created the Pub/Sub topic. */
139 managedTopic: boolean;
140 },
141 never,
142 Providers
143>;
144
145/**
146 * A Google Workspace Events subscription.
147 *
148 * Subscriptions have no labels field, so Alchemy stamps ownership onto
149 * the destination Pub/Sub topic (`alchemy-wesub-*`) for `list` / nuke.
150 * `targetResource`, the Pub/Sub topic, payload options, and Drive options
151 * are identity — changing them replaces the subscription. Event types and
152 * expiration update in place. Creating a subscription requires Workspace
153 * Events OAuth scopes for the target resource (Chat, Drive, or Meet).
154 *
155 * ### Creating a Subscription
156 * **Example:** Chat space messages to a Pub/Sub topic
157 * ```typescript
158 * const topic = yield* GCP.PubSub.Topic("ChatEvents", {});
159 * const sub = yield* GCP.WorkspaceEvents.Subscription("SpaceEvents", {
160 * targetResource: "//chat.googleapis.com/spaces/SPACE_ID",
161 * eventTypes: ["google.workspace.chat.message.v1.created"],
162 * pubsubTopic: topic.name,
163 * });
164 * ```
165 *
166 * **Example:** Generated topic
167 * ```typescript
168 * const sub = yield* GCP.WorkspaceEvents.Subscription("MeetEvents", {
169 * targetResource: "//meet.googleapis.com/spaces/SPACE_ID",
170 * eventTypes: ["google.workspace.meet.conference.v2.started"],
171 * });
172 * ```
173 *
174 * ### Updating a Subscription
175 * **Example:** Renew TTL and add an event type
176 * ```typescript
177 * // Same logical id, changed props: the engine updates it in place.
178 * const sub = yield* GCP.WorkspaceEvents.Subscription("SpaceEvents", {
179 * targetResource: "//chat.googleapis.com/spaces/SPACE_ID",
180 * eventTypes: [
181 * "google.workspace.chat.message.v1.created",
182 * "google.workspace.chat.message.v1.updated",
183 * ],
184 * pubsubTopic: topic.name,
185 * ttl: "86400s",
186 * });
187 * ```
188 *
189 * @resource
190 * @category WorkspaceEvents
191 */
192export const Subscription = Resource<Subscription>(
193 "GCP.WorkspaceEvents.Subscription",
194);
195
196export class SubscriptionNotResolved extends Data.TaggedError(
197 "GCP.WorkspaceEvents.SubscriptionNotResolved",
198)<{
199 name: string;
200}> {}
201
202const payloadOf = (
203 options: we.PayloadOptions | undefined,
204): PayloadOptions | undefined => {
205 if (options === undefined) return undefined;
206 return {
207 includeResource: options.includeResource,
208 fieldMask: options.fieldMask,
209 };
210};
211
212const driveOf = (
213 options: we.DriveOptions | undefined,
214): DriveOptions | undefined => {
215 if (options === undefined) return undefined;
216 return { includeDescendants: options.includeDescendants };
217};
218
219const toAttrs = (
220 subscription: we.Subscription,
221 project: string,
222 managedTopic: boolean,
223) => {
224 const name = subscription.name ?? "";
225 return {
226 name,
227 subscriptionId: subscriptionIdOf(name),
228 project,
229 targetResource: subscription.targetResource,
230 eventTypes: subscription.eventTypes ?? [],
231 pubsubTopic: subscription.notificationEndpoint?.pubsubTopic,
232 payloadOptions: payloadOf(subscription.payloadOptions),
233 driveOptions: driveOf(subscription.driveOptions),
234 expireTime: subscription.expireTime,
235 state: subscription.state,
236 suspensionReason: subscription.suspensionReason,
237 authority: subscription.authority,
238 serviceAccountAuthority: subscription.serviceAccountAuthority,
239 uid: subscription.uid,
240 createTime: subscription.createTime,
241 updateTime: subscription.updateTime,
242 managedTopic,
243 };
244};
245
246const lookupName = (
247 subscriptionId: string | undefined,
248 existingName: string | undefined,
249) => {
250 if (subscriptionId !== undefined && subscriptionId.length > 0) {
251 return toSubscriptionName(subscriptionId);
252 }
253 if (existingName !== undefined && existingName.length > 0) {
254 return toSubscriptionName(existingName);
255 }
256 return "";
257};
258
259const sameTopic = (left: string | undefined, right: string | undefined) => {
260 if (left === undefined || right === undefined) return left === right;
261 return left === right || lastSegment(left) === lastSegment(right);
262};
263
264export const SubscriptionProvider = () =>
265 Provider.succeed(Subscription, {
266 stables: [
267 "name",
268 "subscriptionId",
269 "project",
270 "uid",
271 "createTime",
272 "targetResource",
273 ],
274
275 diff: Effect.fn(function* ({ news, olds, output }) {
276 if (!isResolved(news)) return undefined;
277 const previousId = olds?.subscriptionId ?? output?.subscriptionId;
278 if (
279 previousId !== undefined &&
280 news.subscriptionId !== undefined &&
281 toSubscriptionName(news.subscriptionId) !==
282 toSubscriptionName(previousId) &&
283 news.subscriptionId !== output?.subscriptionId &&
284 toSubscriptionName(news.subscriptionId) !== output?.name
285 ) {
286 return { action: "replace" as const, deleteFirst: false };
287 }
288 const previousTarget = olds?.targetResource ?? output?.targetResource;
289 if (
290 previousTarget !== undefined &&
291 news.targetResource !== previousTarget
292 ) {
293 return { action: "replace" as const, deleteFirst: false };
294 }
295 const previousTopic = olds?.pubsubTopic ?? output?.pubsubTopic;
296 if (
297 news.pubsubTopic !== undefined &&
298 previousTopic !== undefined &&
299 !sameTopic(news.pubsubTopic, previousTopic)
300 ) {
301 return { action: "replace" as const, deleteFirst: false };
302 }
303 const previousPayload = olds?.payloadOptions ?? output?.payloadOptions;
304 if (
305 news.payloadOptions !== undefined &&
306 previousPayload !== undefined &&
307 (news.payloadOptions.includeResource !==
308 previousPayload.includeResource ||
309 (news.payloadOptions.fieldMask ?? "") !==
310 (previousPayload.fieldMask ?? ""))
311 ) {
312 return { action: "replace" as const, deleteFirst: false };
313 }
314 const previousDrive = olds?.driveOptions ?? output?.driveOptions;
315 if (
316 news.driveOptions?.includeDescendants !== undefined &&
317 previousDrive?.includeDescendants !== undefined &&
318 news.driveOptions.includeDescendants !==
319 previousDrive.includeDescendants
320 ) {
321 return { action: "replace" as const, deleteFirst: false };
322 }
323 return undefined;
324 }),
325
326 read: Effect.fn(function* ({ id, olds, output }) {
327 const env = yield* GcpEnvironment.current;
328 const name = lookupName(
329 olds?.subscriptionId ?? output?.subscriptionId,
330 output?.name,
331 );
332 let existing = yield* getSubscription(name);
333 if (existing === undefined) {
334 existing = yield* findSubscription(
335 name,
336 olds?.eventTypes ?? output?.eventTypes ?? [],
337 olds?.targetResource ?? output?.targetResource,
338 );
339 }
340 if (existing === undefined) return undefined;
341 const topic = yield* getTopic(
342 existing.notificationEndpoint?.pubsubTopic ?? "",
343 );
344 const attrs = toAttrs(
345 existing,
346 env.project,
347 output?.managedTopic === true,
348 );
349 return (yield* ownedByWesubLabels(id, topic?.labels))
350 ? attrs
351 : Unowned(attrs);
352 }),
353
354 list: () =>
355 Effect.gen(function* () {
356 const env = yield* GcpEnvironment.current;
357 const subscriptions = yield* listOwnedSubscriptions();
358 return subscriptions.map((subscription) =>
359 toAttrs(subscription, env.project, false),
360 );
361 }),
362
363 reconcile: Effect.fn(function* ({ id, news, output }) {
364 const env = yield* GcpEnvironment.current;
365 const labels = yield* wesubLabels(id);
366 const managedTopic = news.pubsubTopic === undefined;
367 const topicName = news.pubsubTopic
368 ? expandTopic(news.pubsubTopic, env.project)
369 : expandTopic(
370 output?.pubsubTopic ??
371 (yield* toPhysicalId(
372 `${id}-topic`,
373 undefined,
374 undefined,
375 "topic",
376 255,
377 )),
378 env.project,
379 );
380
381 yield* ensureTopic(topicName, labels);
382
383 const eventTypes = news.eventTypes;
384 const lookup = lookupName(
385 news.subscriptionId ?? output?.subscriptionId,
386 output?.name,
387 );
388 let current = yield* findSubscription(
389 lookup,
390 eventTypes,
391 news.targetResource,
392 );
393
394 if (current === undefined) {
395 const created = yield* we
396 .createSubscriptions({
397 body: {
398 targetResource: news.targetResource,
399 eventTypes,
400 notificationEndpoint: { pubsubTopic: topicName },
401 payloadOptions: news.payloadOptions,
402 driveOptions: news.driveOptions,
403 expireTime: news.expireTime,
404 ttl: news.ttl,
405 },
406 })
407 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
408 if (created !== undefined) {
409 // ALREADY_EXISTS (6): a concurrent create won the race.
410 const settled = yield* waitForOperation(created).pipe(
411 Effect.catchTag("GCP.OperationFailed", (error) =>
412 error.code === 6 ? Effect.succeed(created) : Effect.fail(error),
413 ),
414 );
415 const createdName = operationResourceName(settled) ?? lookup;
416 if (createdName.length > 0) {
417 current = yield* waitUntilExists(
418 getSubscription(createdName),
419 createdName,
420 );
421 }
422 }
423 if (current === undefined) {
424 current = yield* findSubscription(
425 lookup,
426 eventTypes,
427 news.targetResource,
428 );
429 }
430 }
431
432 if (current === undefined) {
433 return yield* new SubscriptionNotResolved({
434 name: lookup || news.targetResource,
435 });
436 }
437
438 const name = current.name ?? lookup;
439 const eventTypesChanged =
440 sortedStrings(current.eventTypes).join(",") !==
441 sortedStrings(eventTypes).join(",");
442 const expireChanged =
443 news.expireTime !== undefined &&
444 !sameText(current.expireTime, news.expireTime);
445 const ttlChanged =
446 news.ttl !== undefined && news.expireTime === undefined;
447 const mask = updateMaskOf(
448 eventTypesChanged ? "event_types" : undefined,
449 expireChanged ? "expire_time" : undefined,
450 ttlChanged ? "ttl" : undefined,
451 );
452 if (mask.length > 0) {
453 const patched = yield* we.patchSubscriptions({
454 name,
455 updateMask: mask,
456 body: {
457 eventTypes,
458 expireTime: news.expireTime,
459 ttl: news.ttl,
460 },
461 });
462 yield* waitForOperation(patched);
463 current = (yield* getSubscription(name)) ?? current;
464 }
465
466 if (current.state === "SUSPENDED") {
467 const reactivated = yield* we
468 .reactivateSubscriptions({ name, body: {} })
469 .pipe(
470 Effect.catchTag(["NotFound", "BadRequest", "Conflict"], () =>
471 Effect.succeed(undefined),
472 ),
473 );
474 if (reactivated !== undefined) {
475 yield* waitForOperation(reactivated);
476 current = (yield* getSubscription(name)) ?? current;
477 }
478 }
479
480 const fresh = (yield* getSubscription(name)) ?? current;
481 return toAttrs(fresh, env.project, managedTopic);
482 }),
483
484 delete: Effect.fn(function* ({ output }) {
485 const name = output.name || toSubscriptionName(output.subscriptionId);
486 if (name.length > 0) {
487 const operation = yield* we
488 .deleteSubscriptions({ name, allowMissing: true })
489 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
490 if (operation !== undefined) {
491 // NOT_FOUND (5): already gone.
492 yield* waitForOperation(operation).pipe(
493 Effect.catchTag("GCP.OperationFailed", (error) =>
494 error.code === 5 ? Effect.void : Effect.fail(error),
495 ),
496 );
497 }
498 yield* waitUntilGone(getSubscription(name), name);
499 }
500 if (output.managedTopic === true && output.pubsubTopic) {
501 yield* pubsub
502 .deleteProjectsTopics({ topic: output.pubsubTopic })
503 .pipe(Effect.catchTag("NotFound", () => Effect.void));
504 }
505 }),
506 });