1import * as r2 from "@distilled.cloud/cloudflare/r2";
2import * as Effect from "effect/Effect";
3import * as Predicate from "effect/Predicate";
4import * as Schedule from "effect/Schedule";
5
6import { Unowned } from "../../AdoptPolicy.ts";
7import { deepEqual } from "../../Diff.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import { CloudflareEnvironment } from "../CloudflareEnvironment.ts";
11import type { Providers } from "../Providers.ts";
12import type { Bucket } from "./Bucket.ts";
13
14const TypeId = "Cloudflare.R2.BucketEventNotification" as const;
15type TypeId = typeof TypeId;
16
17/**
18 * An R2 object action that can trigger an event notification.
19 *
20 * - `PutObject` — an object is uploaded via a single PUT
21 * - `CopyObject` — an object is copied into the bucket
22 * - `DeleteObject` — an object is deleted
23 * - `CompleteMultipartUpload` — a multipart upload completes
24 * - `LifecycleDeletion` — an object is deleted by a lifecycle rule
25 */
26export type BucketEventNotificationAction =
27 | "PutObject"
28 | "CopyObject"
29 | "DeleteObject"
30 | "CompleteMultipartUpload"
31 | "LifecycleDeletion";
32
33/**
34 * A single notification rule. An event message is delivered to the queue
35 * when an object matching the rule's `prefix`/`suffix` undergoes one of
36 * the rule's `actions`.
37 */
38export interface BucketEventNotificationRule {
39 /**
40 * Object actions that trigger a notification for this rule.
41 */
42 actions: BucketEventNotificationAction[];
43 /**
44 * Only notify for objects whose key starts with this prefix.
45 * @default "" (match all objects)
46 */
47 prefix?: string;
48 /**
49 * Only notify for objects whose key ends with this suffix.
50 * @default "" (match all objects)
51 */
52 suffix?: string;
53 /**
54 * Human-readable description of the rule. Cloudflare auto-generates one
55 * when omitted.
56 */
57 description?: string;
58}
59
60export interface BucketEventNotificationProps {
61 /**
62 * Name of the R2 bucket that produces the events. Pass
63 * `bucket.bucketName` from a `Cloudflare.R2.Bucket`.
64 *
65 * Immutable — changing the bucket triggers a replacement.
66 */
67 bucketName: string;
68 /**
69 * ID of the Queue that receives the event messages. Pass
70 * `queue.queueId` from a `Cloudflare.Queues.Queue`.
71 *
72 * Immutable — changing the queue triggers a replacement.
73 */
74 queueId: string;
75 /**
76 * Jurisdiction of the bucket (must match the bucket's own jurisdiction).
77 *
78 * Immutable — changing the jurisdiction triggers a replacement.
79 * @default "default"
80 */
81 jurisdiction?: Bucket.Jurisdiction;
82 /**
83 * Rules that decide which object events are delivered to the queue.
84 * The full set is replaced on every update (the provider clears the
85 * pair's configuration and re-creates the desired rules when they
86 * drift).
87 *
88 * Note: Cloudflare rejects rule sets whose `prefix`/`suffix` key ranges
89 * overlap ("invalid overlap"), even when their actions are disjoint —
90 * scope each rule to a distinct key range.
91 */
92 rules: BucketEventNotificationRule[];
93}
94
95export interface BucketEventNotificationAttributes {
96 /** Name of the bucket that produces the events. */
97 bucketName: string;
98 /** ID of the queue that receives the event messages. */
99 queueId: string;
100 /** Name of the queue that receives the event messages, if reported. */
101 queueName: string | undefined;
102 /** Account the configuration lives in. */
103 accountId: string;
104 /** Jurisdiction of the bucket. */
105 jurisdiction: Bucket.Jurisdiction;
106 /** The active notification rules, including server-assigned rule IDs. */
107 rules: BucketEventNotification.Rule[];
108}
109
110export type BucketEventNotification = Resource<
111 TypeId,
112 BucketEventNotificationProps,
113 BucketEventNotificationAttributes,
114 never,
115 Providers
116>;
117
118/**
119 * Event notifications for a Cloudflare R2 bucket, delivered to a Queue.
120 *
121 * When objects in the bucket are created, deleted, or copied, R2 publishes
122 * an event message to the configured Queue. One configuration exists per
123 * (bucket, queue) pair and holds a list of rules; consume the messages with
124 * a Queue consumer Worker.
125 *
126 * The configuration's identity is the (bucket, queue) pair — changing
127 * either triggers a replacement, while rule changes are applied in place
128 * (the provider converges the pair's configuration to exactly the
129 * declared rule set).
130 * ### Notifying a Queue
131 * **Example:** Notify on every upload and delete
132 * ```typescript
133 * const bucket = yield* Cloudflare.R2.Bucket("Uploads");
134 * const queue = yield* Cloudflare.Queues.Queue("UploadEvents");
135 *
136 * yield* Cloudflare.R2.BucketEventNotification("UploadNotifications", {
137 * bucketName: bucket.bucketName,
138 * queueId: queue.queueId,
139 * rules: [
140 * {
141 * actions: ["PutObject", "CompleteMultipartUpload", "DeleteObject"],
142 * },
143 * ],
144 * });
145 * ```
146 *
147 * **Example:** Scope notifications to a key prefix and suffix
148 * ```typescript
149 * yield* Cloudflare.R2.BucketEventNotification("ImageNotifications", {
150 * bucketName: bucket.bucketName,
151 * queueId: queue.queueId,
152 * rules: [
153 * {
154 * actions: ["PutObject"],
155 * prefix: "images/",
156 * suffix: ".png",
157 * description: "new PNG images",
158 * },
159 * ],
160 * });
161 * ```
162 *
163 * ### Multiple rules
164 * **Example:** Separate rules per key range
165 * ```typescript
166 * // Rules must cover non-overlapping key ranges — Cloudflare rejects
167 * // overlapping prefixes/suffixes even when the actions are disjoint.
168 * yield* Cloudflare.R2.BucketEventNotification("Notifications", {
169 * bucketName: bucket.bucketName,
170 * queueId: queue.queueId,
171 * rules: [
172 * { actions: ["PutObject"], prefix: "incoming/" },
173 * { actions: ["DeleteObject", "LifecycleDeletion"], prefix: "logs/" },
174 * ],
175 * });
176 * ```
177 *
178 * @see https://developers.cloudflare.com/r2/buckets/event-notifications/
179 *
180 * @resource
181 * @product R2
182 * @category Storage & Databases
183 */
184export const BucketEventNotification =
185 Resource<BucketEventNotification>(TypeId);
186
187export declare namespace BucketEventNotification {
188 export type Rule = {
189 actions: BucketEventNotificationAction[];
190 prefix: string;
191 suffix: string;
192 description: string | undefined;
193 ruleId: string | undefined;
194 createdAt: string | undefined;
195 };
196}
197
198/**
199 * Returns true if the given value is an BucketEventNotification resource.
200 */
201export const isBucketEventNotification = (
202 value: unknown,
203): value is BucketEventNotification =>
204 Predicate.hasProperty(value, "Type") && value.Type === TypeId;
205
206export const BucketEventNotificationProvider = () =>
207 Provider.succeed(BucketEventNotification, {
208 stables: ["bucketName", "queueId", "accountId", "jurisdiction"],
209
210 diff: Effect.fn(function* ({ olds = {}, news }) {
211 const o = olds as Partial<BucketEventNotificationProps>;
212 const n = news as BucketEventNotificationProps;
213 // No prior props to compare against — let the engine decide.
214 if (o.bucketName === undefined) return undefined;
215 // bucketName/queueId are Input<string>; compare only once both
216 // sides are concrete strings.
217 if (
218 typeof o.bucketName === "string" &&
219 typeof n.bucketName === "string" &&
220 o.bucketName !== n.bucketName
221 ) {
222 return { action: "replace" } as const;
223 }
224 if (
225 typeof o.queueId === "string" &&
226 typeof n.queueId === "string" &&
227 o.queueId !== n.queueId
228 ) {
229 return { action: "replace" } as const;
230 }
231 if ((o.jurisdiction ?? "default") !== (n.jurisdiction ?? "default")) {
232 return { action: "replace" } as const;
233 }
234 return undefined;
235 }),
236
237 read: Effect.fn(function* ({ output, olds }) {
238 const { accountId } = yield* yield* CloudflareEnvironment;
239 const acct = output?.accountId ?? accountId;
240 // The (bucket, queue) pair is the configuration's identity; cold
241 // reads derive it from the last-persisted props.
242 const bucketName =
243 output?.bucketName ??
244 (typeof olds?.bucketName === "string" ? olds.bucketName : undefined);
245 const queueId =
246 output?.queueId ??
247 (typeof olds?.queueId === "string" ? olds.queueId : undefined);
248 if (bucketName === undefined || queueId === undefined) return undefined;
249 const jurisdiction =
250 output?.jurisdiction ?? olds?.jurisdiction ?? "default";
251
252 const observed = yield* getConfiguration(
253 acct,
254 bucketName,
255 queueId,
256 jurisdiction,
257 );
258 if (!observed) return undefined;
259
260 const attrs = toAttributes(
261 observed,
262 acct,
263 bucketName,
264 queueId,
265 jurisdiction,
266 );
267 // Event notification configs carry no ownership markers. With no
268 // prior output we cannot prove we created this configuration —
269 // brand it `Unowned` so takeover is gated behind the adopt policy.
270 return output ? attrs : Unowned(attrs);
271 }),
272
273 reconcile: Effect.fn(function* ({ news, output }) {
274 const { accountId } = yield* yield* CloudflareEnvironment;
275 const acct = output?.accountId ?? accountId;
276 // Inputs have been resolved to concrete strings by Plan.
277 const bucketName = news.bucketName as string;
278 const queueId = news.queueId as string;
279 const jurisdiction = news.jurisdiction ?? "default";
280
281 // 1. Observe — the configuration may or may not exist; `output` is
282 // only a cache of the identity, never proof of existence.
283 const observed = yield* getConfiguration(
284 acct,
285 bucketName,
286 queueId,
287 jurisdiction,
288 );
289
290 // 2/3. Ensure + sync — skip the API entirely when the observed rules
291 // already match the desired set.
292 if (!observed || !sameRules(observed.rules ?? [], news.rules)) {
293 // The PUT *appends* rules to the queue's configuration (it is not
294 // a full replace — verified against the live API), so on drift
295 // the existing configuration must be cleared first. The brief gap
296 // is unavoidable; each step is independently idempotent.
297 if (observed) {
298 yield* r2
299 .deleteBucketEventNotification({
300 accountId: acct,
301 bucketName,
302 queueId,
303 jurisdiction,
304 })
305 .pipe(
306 Effect.catchTag(
307 [
308 "EventNotificationConfigNotFound",
309 "BucketNotFound",
310 "QueueNotFound",
311 ],
312 () => Effect.void,
313 ),
314 );
315 }
316 yield* r2
317 .putBucketEventNotification({
318 accountId: acct,
319 bucketName,
320 queueId,
321 jurisdiction,
322 rules: news.rules.map((rule) => ({
323 actions: [...rule.actions],
324 prefix: rule.prefix,
325 suffix: rule.suffix,
326 description: rule.description,
327 })),
328 })
329 .pipe(
330 // A freshly-created bucket or queue can briefly 404 on the
331 // event-notification endpoint — ride out the consistency lag.
332 Effect.retry({
333 while: (e) =>
334 e._tag === "BucketNotFound" || e._tag === "QueueNotFound",
335 schedule: r2EventNotificationConsistencySchedule,
336 }),
337 );
338 }
339
340 // 4. Return — re-read so the attributes carry the server-assigned
341 // per-rule IDs. The read can lag the PUT briefly.
342 if (news.rules.length === 0) {
343 // An empty rule set reads back as "no configuration".
344 return toAttributes(
345 { queueId, queueName: undefined, rules: [] },
346 acct,
347 bucketName,
348 queueId,
349 jurisdiction,
350 );
351 }
352 const final = yield* r2
353 .getBucketEventNotification({
354 accountId: acct,
355 bucketName,
356 queueId,
357 jurisdiction,
358 })
359 .pipe(
360 Effect.retry({
361 while: (e) =>
362 e._tag === "NoEventNotificationConfig" ||
363 e._tag === "EventNotificationConfigNotFound" ||
364 e._tag === "BucketNotFound" ||
365 e._tag === "QueueNotFound",
366 schedule: r2EventNotificationConsistencySchedule,
367 }),
368 );
369 return toAttributes(final, acct, bucketName, queueId, jurisdiction);
370 }),
371
372 delete: Effect.fn(function* ({ output }) {
373 // Idempotent — the configuration, bucket, or queue may already be
374 // gone (all typed not-found variants in the operation's union).
375 yield* r2
376 .deleteBucketEventNotification({
377 accountId: output.accountId,
378 bucketName: output.bucketName,
379 queueId: output.queueId,
380 jurisdiction: output.jurisdiction,
381 })
382 .pipe(
383 Effect.catchTag(
384 [
385 "EventNotificationConfigNotFound",
386 "BucketNotFound",
387 "QueueNotFound",
388 ],
389 () => Effect.void,
390 ),
391 );
392 }),
393
394 // Parent fan-out: a notification configuration is keyed by (bucket,
395 // queue) and there is no account-wide enumeration. Enumerate every R2
396 // bucket (account-scoped), then list each bucket's event-notification
397 // queues — one (bucket, queue) pair is one BucketEventNotification.
398 list: Effect.fn(function* () {
399 const { accountId } = yield* yield* CloudflareEnvironment;
400 const buckets = yield* listAllBuckets(accountId);
401 const perBucket = yield* Effect.forEach(
402 buckets,
403 (bucket) => {
404 const bucketName = bucket.name;
405 if (bucketName == null) {
406 return Effect.succeed([] as BucketEventNotificationAttributes[]);
407 }
408 const jurisdiction = (bucket.jurisdiction ??
409 "default") as Bucket.Jurisdiction;
410 return r2
411 .listBucketEventNotifications({
412 accountId,
413 bucketName,
414 jurisdiction,
415 })
416 .pipe(
417 Effect.map((res) =>
418 (res.queues ?? [])
419 .filter(
420 (q): q is typeof q & { queueId: string } =>
421 q.queueId != null,
422 )
423 .map((q): BucketEventNotificationAttributes => ({
424 bucketName,
425 // The endpoint echoes the queue ID in dashed-UUID
426 // form; normalise to the undashed form
427 // `Queue.queueId` uses so list items match `read`.
428 queueId: q.queueId.replace(/-/g, ""),
429 queueName: q.queueName ?? undefined,
430 accountId,
431 jurisdiction,
432 rules: (q.rules ?? []).map(toRuleAttributes),
433 })),
434 ),
435 // A bucket with no event-notification config (or a bucket
436 // that vanished mid-enumeration) is not an error — skip it.
437 Effect.catchTag(
438 [
439 "NoEventNotificationConfig",
440 "BucketNotFound",
441 "NoSuchBucket",
442 // Plan-gated / partial buckets reject the route.
443 "InvalidRoute",
444 ],
445 () => Effect.succeed([] as BucketEventNotificationAttributes[]),
446 ),
447 );
448 },
449 { concurrency: 10 },
450 );
451 return perBucket.flat();
452 }),
453 });
454
455// R2 can make a freshly-created bucket/queue visible to its own endpoints
456// before the event-notification endpoints accept it. Retry only that narrow
457// not-found lag with a bounded budget.
458const r2EventNotificationConsistencySchedule = Schedule.max([
459 Schedule.exponential(250),
460 Schedule.recurs(6),
461]);
462
463type ObservedBucket = NonNullable<r2.ListBucketsResponse["buckets"]>[number];
464
465// `listBuckets` is a non-paginated distilled op that returns at most one
466// page (default ordering by name). Page through it exhaustively with the
467// `startAfter` cursor so `list()` enumerates *every* bucket in the account.
468const listAllBuckets = (accountId: string) =>
469 Effect.gen(function* () {
470 const pageSize = 1000;
471 const all: ObservedBucket[] = [];
472 let startAfter: string | undefined;
473 while (true) {
474 const { buckets } = yield* r2.listBuckets({
475 accountId,
476 perPage: pageSize,
477 startAfter,
478 });
479 const page = buckets ?? [];
480 all.push(...page);
481 const last = page.at(-1)?.name;
482 if (page.length < pageSize || last == null) break;
483 startAfter = last;
484 }
485 return all;
486 });
487
488const getConfiguration = (
489 accountId: string,
490 bucketName: string,
491 queueId: string,
492 jurisdiction: Bucket.Jurisdiction,
493) =>
494 r2
495 .getBucketEventNotification({
496 accountId,
497 bucketName,
498 queueId,
499 jurisdiction,
500 })
501 .pipe(
502 Effect.map(
503 (config): r2.GetBucketEventNotificationResponse | undefined => config,
504 ),
505 // "Gone" comes in several typed flavors: the bucket has no event
506 // notification config at all (`NoEventNotificationConfig`, code
507 // 11015), no config for this queue (`EventNotificationConfigNotFound`,
508 // code 11011), or the bucket/queue themselves are gone.
509 Effect.catchTag(
510 [
511 "NoEventNotificationConfig",
512 "EventNotificationConfigNotFound",
513 "BucketNotFound",
514 "QueueNotFound",
515 ],
516 () => Effect.succeed(undefined),
517 ),
518 );
519
520type ObservedRule = NonNullable<
521 r2.GetBucketEventNotificationResponse["rules"]
522>[number];
523
524const toRuleAttributes = (
525 rule: ObservedRule,
526): BucketEventNotification.Rule => ({
527 // Distilled widens generated string enums to open unions; the API only
528 // returns the known action variants.
529 actions: [...rule.actions] as BucketEventNotificationAction[],
530 prefix: rule.prefix ?? "",
531 suffix: rule.suffix ?? "",
532 description: rule.description ?? undefined,
533 ruleId: rule.ruleId ?? undefined,
534 createdAt: rule.createdAt ?? undefined,
535});
536
537const toAttributes = (
538 config: r2.GetBucketEventNotificationResponse,
539 accountId: string,
540 bucketName: string,
541 queueId: string,
542 jurisdiction: Bucket.Jurisdiction,
543): BucketEventNotificationAttributes => ({
544 bucketName,
545 // The endpoint echoes the queue ID in dashed-UUID form while the Queues
546 // API uses the undashed form — keep the canonical input ID so attributes
547 // compare cleanly against `Queue.queueId`.
548 queueId,
549 queueName: config.queueName ?? undefined,
550 accountId,
551 jurisdiction,
552 rules: (config.rules ?? []).map(toRuleAttributes),
553});
554
555/**
556 * A rule's identity within the configuration: its matched actions (order-
557 * insensitive) plus prefix/suffix. The server-assigned `ruleId` and
558 * auto-generated description are deliberately excluded.
559 */
560const ruleIdentity = (rule: {
561 actions: readonly string[];
562 prefix?: string | null;
563 suffix?: string | null;
564}) =>
565 JSON.stringify({
566 actions: [...rule.actions].sort(),
567 prefix: rule.prefix ?? "",
568 suffix: rule.suffix ?? "",
569 });
570
571/**
572 * Compares the observed rules against the desired rules as an unordered
573 * set. Descriptions are compared only when the desired rule specifies one —
574 * Cloudflare auto-generates a description otherwise.
575 */
576const sameRules = (
577 observed: readonly ObservedRule[],
578 desired: readonly BucketEventNotificationRule[],
579): boolean => {
580 if (observed.length !== desired.length) return false;
581 const observedKeys = observed.map(ruleIdentity).sort();
582 const desiredKeys = desired.map(ruleIdentity).sort();
583 if (!deepEqual(observedKeys, desiredKeys)) return false;
584 return desired.every(
585 (rule) =>
586 rule.description === undefined ||
587 observed.some(
588 (candidate) =>
589 ruleIdentity(candidate) === ruleIdentity(rule) &&
590 (candidate.description ?? "") === rule.description,
591 ),
592 );
593};