1import * as pubsub from "@distilled.cloud/gcp/pubsub_v1";
2import * as storage from "@distilled.cloud/gcp/storage_v1";
3import * as serviceusage from "@distilled.cloud/gcp/unstable/serviceusage_v1beta1";
4import * as Data from "effect/Data";
5import * as Effect from "effect/Effect";
6import * as Schedule from "effect/Schedule";
7import * as Stream from "effect/Stream";
8import { Unowned } from "../../AdoptPolicy.ts";
9import { isResolved } from "../../Diff.ts";
10import * as Provider from "../../Provider.ts";
11import { Resource } from "../../Resource.ts";
12import { tagRecord } from "../../Tags.ts";
13import { DeleteNotConfirmed } from "../Errors.ts";
14import { GcpEnvironment } from "../Environment.ts";
15import {
16 ALCHEMY_LABEL_PREFIX,
17 hasAlchemyLabels,
18 stripInternalLabels,
19 toLabels,
20} from "../Labels.ts";
21import { waitForOperation } from "../Operation.ts";
22import type { Providers } from "../Providers.ts";
23
24const DEFAULT_PAYLOAD_FORMAT = "JSON_API_V1";
25const CANONICAL_TOPIC_PREFIX = "//pubsub.googleapis.com/";
26const PUBLISHER_ROLE = "roles/pubsub.publisher";
27
28export type NotificationPayloadFormat = "JSON_API_V1" | "NONE";
29
30export type NotificationProps = {
31 /**
32 * Name of the Cloud Storage bucket this notification is attached to.
33 * Immutable — changing it replaces the notification.
34 */
35 bucketName: string;
36 /**
37 * Cloud Pub/Sub topic to publish events to. Accepts a topic id, a
38 * resource name (`projects/{project}/topics/{topic}`), or the canonical
39 * URI (`//pubsub.googleapis.com/projects/{project}/topics/{topic}`).
40 * Immutable — changing it replaces the notification.
41 */
42 topic: string;
43 /**
44 * Payload included in each Pub/Sub message.
45 * @default "JSON_API_V1"
46 */
47 payloadFormat?: NotificationPayloadFormat;
48 /**
49 * Event types to publish. Omit (or pass an empty list) to receive every
50 * event (`OBJECT_FINALIZE`, `OBJECT_METADATA_UPDATE`, `OBJECT_DELETE`,
51 * `OBJECT_ARCHIVE`). Immutable — changing it replaces the notification.
52 */
53 eventTypes?: string[];
54 /**
55 * If set, only objects whose names start with this prefix generate
56 * notifications. Immutable — changing it replaces the notification.
57 */
58 objectNamePrefix?: string;
59 /**
60 * Extra attributes attached to every published Pub/Sub message. Alchemy
61 * writes no ownership markers here (they would reach every subscriber);
62 * the config is tracked by its server-assigned id. Immutable — changing
63 * attributes replaces the notification.
64 */
65 customAttributes?: Record<string, string>;
66};
67
68export type Notification = Resource<
69 "GCP.Storage.Notification",
70 NotificationProps,
71 {
72 /** Parent bucket name. */
73 bucketName: string;
74 /** Server-assigned notification id. */
75 notificationId: string;
76 /** Canonical Pub/Sub topic URI. */
77 topic: string;
78 /** Topic id (last path segment). */
79 topicId: string;
80 /** Project id used when normalizing the topic. */
81 project: string;
82 /** Payload format (`JSON_API_V1` or `NONE`). */
83 payloadFormat: string;
84 /** Event types this notification is subscribed to (empty = all). */
85 eventTypes: string[];
86 /** Object-name prefix filter, if any. */
87 objectNamePrefix: string | undefined;
88 /** User custom attributes (Alchemy ownership keys stripped). */
89 customAttributes: Record<string, string>;
90 /** GCS self-link. */
91 selfLink: string | undefined;
92 /** HTTP etag. */
93 etag: string | undefined;
94 },
95 never,
96 Providers
97>;
98
99/**
100 * A Cloud Storage Pub/Sub notification configuration on a bucket.
101 *
102 * GCS has no update API for notifications — every user-facing field is
103 * immutable and changing it replaces the config. The Cloud Storage
104 * service account is granted `roles/pubsub.publisher` on the topic
105 * during reconcile so events can actually be published.
106 *
107 * Notifications carry no labels, and `customAttributes` are delivered on
108 * every message, so Alchemy does not stamp ownership markers. `read` finds
109 * the config by its recorded id; without one, an identical config on the
110 * bucket is reported as unowned.
111 *
112 * ### Creating a Notification
113 * **Example:** Notify on every object event
114 * ```typescript
115 * const bucket = yield* GCP.Storage.Bucket("assets", {
116 * forceDestroy: true,
117 * });
118 * const topic = yield* GCP.PubSub.Topic("events", {});
119 * const notification = yield* GCP.Storage.Notification("object-events", {
120 * bucketName: bucket.bucketName,
121 * topic: topic.name,
122 * });
123 * ```
124 *
125 * **Example:** Filtered events, prefix, and custom attributes
126 * ```typescript
127 * const notification = yield* GCP.Storage.Notification("uploads", {
128 * bucketName: bucket.bucketName,
129 * topic: topic.name,
130 * payloadFormat: "JSON_API_V1",
131 * eventTypes: ["OBJECT_FINALIZE"],
132 * objectNamePrefix: "uploads/",
133 * customAttributes: { env: "prod" },
134 * });
135 * ```
136 *
137 * @resource
138 * @category Storage
139 */
140export const Notification = Resource<Notification>("GCP.Storage.Notification");
141
142export class NotificationNotResolved extends Data.TaggedError(
143 "GCP.Storage.NotificationNotResolved",
144)<{
145 bucketName: string;
146 notificationId?: string;
147}> {}
148
149export class NotificationServiceAccountMissing extends Data.TaggedError(
150 "GCP.Storage.NotificationServiceAccountMissing",
151)<{
152 project: string;
153}> {}
154
155const topicIdOf = (topic: string) => topic.split("/").pop() ?? topic;
156
157const toCanonicalTopic = (topic: string, project: string) => {
158 if (topic.startsWith(CANONICAL_TOPIC_PREFIX)) return topic;
159 if (topic.startsWith("projects/")) {
160 return `${CANONICAL_TOPIC_PREFIX}${topic}`;
161 }
162 return `${CANONICAL_TOPIC_PREFIX}projects/${project}/topics/${topic}`;
163};
164
165const toPubsubName = (topic: string, project: string) => {
166 const canonical = toCanonicalTopic(topic, project);
167 return canonical.startsWith(CANONICAL_TOPIC_PREFIX)
168 ? canonical.slice(CANONICAL_TOPIC_PREFIX.length)
169 : canonical;
170};
171
172const topicKey = (topic: string, project: string) =>
173 toPubsubName(topic, project);
174
175const userAttributes = (
176 attributes: Record<string, string | undefined> | null | undefined,
177): Record<string, string> => stripInternalLabels(tagRecord(attributes));
178
179const sameAttributes = (
180 left: Record<string, string> | undefined,
181 right: Record<string, string> | undefined,
182) => {
183 const a = left ?? {};
184 const b = right ?? {};
185 const keys = new Set([...Object.keys(a), ...Object.keys(b)]);
186 for (const key of keys) {
187 if (a[key] !== b[key]) return false;
188 }
189 return true;
190};
191
192const normalizeEventTypes = (eventTypes: string[] | undefined) =>
193 [...(eventTypes ?? [])].map((event) => event.toUpperCase()).sort();
194
195const sameEventTypes = (
196 left: string[] | undefined,
197 right: string[] | undefined,
198) =>
199 normalizeEventTypes(left).join("\0") ===
200 normalizeEventTypes(right).join("\0");
201
202const toAttrs = (
203 notification: storage.Notification,
204 bucketName: string,
205 project: string,
206) => {
207 const topic = notification.topic ?? "";
208 return {
209 bucketName,
210 notificationId: notification.id ?? "",
211 topic,
212 topicId: topicIdOf(topic),
213 project,
214 payloadFormat: notification.payload_format ?? DEFAULT_PAYLOAD_FORMAT,
215 eventTypes: notification.event_types ?? [],
216 objectNamePrefix: notification.object_name_prefix,
217 customAttributes: userAttributes(notification.custom_attributes),
218 selfLink: notification.selfLink,
219 etag: notification.etag,
220 };
221};
222
223const matchesDesired = (
224 notification: storage.Notification,
225 news: NotificationProps,
226 project: string,
227) => {
228 const desiredTopic = toCanonicalTopic(news.topic, project);
229 const observedTopic = notification.topic ?? "";
230 const payloadFormat = news.payloadFormat ?? DEFAULT_PAYLOAD_FORMAT;
231 return (
232 topicKey(observedTopic, project) === topicKey(desiredTopic, project) &&
233 (notification.payload_format ?? DEFAULT_PAYLOAD_FORMAT) === payloadFormat &&
234 sameEventTypes(notification.event_types, news.eventTypes) &&
235 (notification.object_name_prefix ?? "") === (news.objectNamePrefix ?? "") &&
236 sameAttributes(
237 userAttributes(notification.custom_attributes),
238 toLabels(news.customAttributes),
239 )
240 );
241};
242
243const immutableChanged = (
244 news: NotificationProps,
245 olds: Partial<NotificationProps> | undefined,
246 output: Notification["Attributes"] | undefined,
247 project: string,
248) => {
249 const previousBucket = olds?.bucketName ?? output?.bucketName;
250 if (previousBucket !== undefined && news.bucketName !== previousBucket) {
251 return true;
252 }
253 const previousTopic = olds?.topic ?? output?.topic;
254 if (
255 previousTopic !== undefined &&
256 topicKey(news.topic, project) !== topicKey(previousTopic, project)
257 ) {
258 return true;
259 }
260 const previousPayload =
261 olds?.payloadFormat ?? output?.payloadFormat ?? DEFAULT_PAYLOAD_FORMAT;
262 if ((news.payloadFormat ?? DEFAULT_PAYLOAD_FORMAT) !== previousPayload) {
263 return true;
264 }
265 if (
266 !sameEventTypes(news.eventTypes, olds?.eventTypes ?? output?.eventTypes)
267 ) {
268 return true;
269 }
270 const previousPrefix =
271 olds?.objectNamePrefix ?? output?.objectNamePrefix ?? "";
272 if ((news.objectNamePrefix ?? "") !== previousPrefix) {
273 return true;
274 }
275 const previousAttrs = olds?.customAttributes ?? output?.customAttributes;
276 if (!sameAttributes(toLabels(news.customAttributes), previousAttrs)) {
277 return true;
278 }
279 return false;
280};
281
282const getById = (bucketName: string, notificationId: string) =>
283 storage
284 .getNotifications({
285 bucket: bucketName,
286 notification: notificationId,
287 })
288 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
289
290const listOnBucket = (bucketName: string) =>
291 storage.listNotifications({ bucket: bucketName }).pipe(
292 Effect.map((page) => page.items ?? []),
293 Effect.catchTag("NotFound", () =>
294 Effect.succeed([] as Array<storage.Notification>),
295 ),
296 );
297
298/** A config on the bucket identical to the desired one (props match). */
299const findMatching = (
300 bucketName: string,
301 props: NotificationProps,
302 project: string,
303) =>
304 listOnBucket(bucketName).pipe(
305 Effect.map((items) =>
306 items.find((item) => matchesDesired(item, props, project)),
307 ),
308 );
309
310const waitUntilGone = (bucketName: string, notificationId: string) =>
311 getById(bucketName, notificationId).pipe(
312 Effect.map((existing) =>
313 existing === undefined ? ("gone" as const) : ("found" as const),
314 ),
315 Effect.repeat({
316 schedule: Schedule.spaced("1 second"),
317 until: (status) => status === "gone",
318 times: 30,
319 }),
320 Effect.flatMap((status): Effect.Effect<void, DeleteNotConfirmed> =>
321 status === "gone"
322 ? Effect.void
323 : Effect.fail(
324 new DeleteNotConfirmed({
325 resource: `${bucketName}/notificationConfigs/${notificationId}`,
326 }),
327 ),
328 ),
329 );
330
331const deleteById = (bucketName: string, notificationId: string) =>
332 storage
333 .deleteNotifications({
334 bucket: bucketName,
335 notification: notificationId,
336 })
337 .pipe(Effect.catchTag("NotFound", () => Effect.void));
338
339const ensureGcsServiceIdentity = (project: string, email: string) =>
340 Effect.gen(function* () {
341 const projectNumber = email.match(/^service-(\d+)@/)?.[1] ?? project;
342 const operation = yield* serviceusage
343 .generateServiceIdentityServices({
344 parent: `projects/${projectNumber}/services/storage.googleapis.com`,
345 })
346 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
347 if (operation === undefined) return;
348 yield* waitForOperation(
349 operation,
350 (name) => serviceusage.getOperations({ name }),
351 { budget: "2 minutes", interval: "2 seconds" },
352 );
353 });
354
355const ensureTopicPublisher = (project: string, topic: string) =>
356 Effect.gen(function* () {
357 const serviceAccount = yield* storage.getProjectsServiceAccount({
358 projectId: project,
359 });
360 const email = serviceAccount.email_address;
361 if (!email) {
362 return yield* new NotificationServiceAccountMissing({ project });
363 }
364 yield* ensureGcsServiceIdentity(project, email);
365 const resource = toPubsubName(topic, project);
366 const member = `serviceAccount:${email}`;
367 const grant = Effect.gen(function* () {
368 const policy = yield* pubsub.getIamPolicyProjectsTopics({
369 resource,
370 });
371 const bindings = (policy.bindings ?? []).map((binding) => ({
372 ...binding,
373 members: [...(binding.members ?? [])],
374 }));
375 const publisher = bindings.find(
376 (binding) => binding.role === PUBLISHER_ROLE,
377 );
378 if (publisher?.members?.includes(member)) {
379 return;
380 }
381 if (publisher) {
382 publisher.members = [...(publisher.members ?? []), member];
383 } else {
384 bindings.push({ role: PUBLISHER_ROLE, members: [member] });
385 }
386 yield* pubsub.setIamPolicyProjectsTopics({
387 resource,
388 body: {
389 policy: {
390 ...policy,
391 bindings,
392 },
393 },
394 });
395 });
396 yield* grant.pipe(
397 Effect.retry({
398 while: (error) =>
399 error._tag === "Conflict" || error._tag === "IamMemberNotFound",
400 times: 8,
401 schedule: Schedule.spaced("2 seconds"),
402 }),
403 );
404 });
405
406export const NotificationProvider = () =>
407 Provider.succeed(Notification, {
408 stables: ["notificationId", "bucketName", "project", "selfLink"],
409
410 diff: Effect.fn(function* ({ news, olds, output }) {
411 if (!isResolved(news)) return undefined;
412 if (output === undefined && olds === undefined) return undefined;
413 const env = yield* GcpEnvironment.current;
414 if (immutableChanged(news, olds, output, env.project)) {
415 return { action: "replace" as const, deleteFirst: true };
416 }
417 return undefined;
418 }),
419
420 read: Effect.fn(function* ({ id, olds, output }) {
421 const env = yield* GcpEnvironment.current;
422 const bucketName = olds?.bucketName ?? output?.bucketName;
423 if (!bucketName) return undefined;
424 let existing =
425 output?.notificationId !== undefined
426 ? yield* getById(bucketName, output.notificationId)
427 : undefined;
428 if (existing === undefined && olds !== undefined) {
429 existing = yield* findMatching(bucketName, olds, env.project);
430 }
431 if (existing === undefined) return undefined;
432 const attrs = toAttrs(existing, bucketName, env.project);
433 // Configs created before markers were dropped still carry them.
434 const owned =
435 (existing.id !== undefined && existing.id === output?.notificationId) ||
436 (yield* hasAlchemyLabels(id, tagRecord(existing.custom_attributes)));
437 return owned ? attrs : Unowned(attrs);
438 }),
439
440 list: () =>
441 Effect.gen(function* () {
442 const env = yield* GcpEnvironment.current;
443 const buckets = yield* storage.listBuckets
444 .items({ project: env.project, maxResults: 1000 })
445 .pipe(
446 Stream.runCollect,
447 Effect.map((chunk) => Array.from(chunk)),
448 );
449 const pages = yield* Effect.forEach(
450 buckets,
451 (bucket) => {
452 const bucketName = bucket.name;
453 if (!bucketName) {
454 return Effect.succeed([] as Array<Notification["Attributes"]>);
455 }
456 // Only configs from before ownership markers were dropped are
457 // identifiable; the rest go away with their bucket.
458 return listOnBucket(bucketName).pipe(
459 Effect.map((items) =>
460 items
461 .filter((item) =>
462 Object.keys(item.custom_attributes ?? {}).some((key) =>
463 key.startsWith(ALCHEMY_LABEL_PREFIX),
464 ),
465 )
466 .map((item) => toAttrs(item, bucketName, env.project)),
467 ),
468 );
469 },
470 { concurrency: 8 },
471 );
472 return pages.flat();
473 }),
474
475 reconcile: Effect.fn(function* ({ id, news, output }) {
476 const env = yield* GcpEnvironment.current;
477 const bucketName = news.bucketName;
478 const payloadFormat = news.payloadFormat ?? DEFAULT_PAYLOAD_FORMAT;
479 const canonicalTopic = toCanonicalTopic(news.topic, env.project);
480 const desiredAttributes = toLabels(news.customAttributes);
481
482 let current: storage.Notification | undefined;
483 if (output?.notificationId) {
484 current = yield* getById(
485 output.bucketName ?? bucketName,
486 output.notificationId,
487 );
488 }
489 if (current === undefined) {
490 current = yield* findMatching(bucketName, news, env.project);
491 }
492
493 if (current === undefined) {
494 yield* ensureTopicPublisher(env.project, canonicalTopic);
495 const created = yield* storage
496 .insertNotifications({
497 bucket: bucketName,
498 body: {
499 topic: canonicalTopic,
500 payload_format: payloadFormat,
501 event_types:
502 news.eventTypes && news.eventTypes.length > 0
503 ? news.eventTypes
504 : undefined,
505 object_name_prefix: news.objectNamePrefix,
506 custom_attributes: desiredAttributes,
507 },
508 })
509 .pipe(
510 Effect.retry({
511 while: (error) => error._tag === "NotificationTopicUnavailable",
512 times: 8,
513 schedule: Schedule.spaced("2 seconds"),
514 }),
515 );
516 current = created;
517 }
518
519 if (current === undefined || !current.id) {
520 return yield* new NotificationNotResolved({
521 bucketName,
522 notificationId: output?.notificationId,
523 });
524 }
525
526 return toAttrs(current, bucketName, env.project);
527 }),
528
529 delete: Effect.fn(function* ({ output }) {
530 const bucketName = output.bucketName;
531 const notificationId = output.notificationId;
532 if (!bucketName || !notificationId) return;
533 yield* deleteById(bucketName, notificationId);
534 yield* waitUntilGone(bucketName, notificationId);
535 }),
536 });