1import * as Cloudflare from "@/Cloudflare";
2import { CloudflareEnvironment } from "@/Cloudflare/CloudflareEnvironment";
3import * as Provider from "@/Provider";
4import * as Test from "@/Test/Alchemy";
5import * as r2 from "@distilled.cloud/cloudflare/r2";
6import { expect } from "alchemy-test";
7import * as Effect from "effect/Effect";
8import { MinimumLogLevel } from "effect/References";
9import * as Schedule from "effect/Schedule";
10
11const { test } = Test.make({ providers: Cloudflare.providers() });
12
13const logLevel = Effect.provideService(
14 MinimumLogLevel,
15 process.env.DEBUG ? "Debug" : "Info",
16);
17
18// The scoped API token the test harness mints propagates eventually-
19// consistently across Cloudflare's edge — ride out 403 blips (`Forbidden`,
20// declared in the distilled error union) on out-of-band verification calls.
21const getNotification = (
22 accountId: string,
23 bucketName: string,
24 queueId: string,
25) =>
26 r2.getBucketEventNotification({ accountId, bucketName, queueId }).pipe(
27 Effect.retry({
28 while: (e) => e._tag === "Forbidden",
29 schedule: Schedule.exponential("500 millis"),
30 times: 8,
31 }),
32 );
33
34// A deleted configuration surfaces as `EventNotificationConfigNotFound`
35// (code 11011) or `NoEventNotificationConfig` (code 11015) — and once the
36// stack's bucket/queue are destroyed, as `BucketNotFound`/`QueueNotFound`.
37// All of those are the success condition here.
38const expectGone = (accountId: string, bucketName: string, queueId: string) =>
39 getNotification(accountId, bucketName, queueId).pipe(
40 Effect.flatMap(() =>
41 Effect.fail({ _tag: "NotificationNotDeleted" } as const),
42 ),
43 Effect.catchTag(
44 [
45 "NoEventNotificationConfig",
46 "EventNotificationConfigNotFound",
47 "BucketNotFound",
48 "QueueNotFound",
49 ],
50 () => Effect.void,
51 ),
52 Effect.retry({
53 while: (e) => e._tag === "NotificationNotDeleted",
54 schedule: Schedule.max([
55 Schedule.exponential("500 millis"),
56 Schedule.recurs(10),
57 ]),
58 }),
59 );
60
61// One program deploying the bucket, the queue, and the notification joining
62// them. The notification's props reference both resources' outputs, so the
63// engine orders notification-last on deploy (and first on destroy).
64const program = (opts: {
65 rules: Cloudflare.R2.BucketEventNotificationRule[];
66}) =>
67 Effect.gen(function* () {
68 const bucket = yield* Cloudflare.R2.Bucket("EventBucket", {
69 forceDestroy: true,
70 });
71 const queue = yield* Cloudflare.Queues.Queue("EventQueueA");
72 const notification = yield* Cloudflare.R2.BucketEventNotification(
73 "Notification",
74 {
75 bucketName: bucket.bucketName,
76 queueId: queue.queueId,
77 rules: opts.rules,
78 },
79 );
80 return { bucket, queue, notification };
81 });
82
83// Replacement variant — both queues stay deployed across both steps; only
84// the notification's target queue flips, so the replacement is isolated to
85// the notification itself.
86const replacementProgram = (opts: {
87 rules: Cloudflare.R2.BucketEventNotificationRule[];
88 target: "A" | "B";
89}) =>
90 Effect.gen(function* () {
91 const bucket = yield* Cloudflare.R2.Bucket("EventBucket", {
92 forceDestroy: true,
93 });
94 const queueA = yield* Cloudflare.Queues.Queue("EventQueueA");
95 const queueB = yield* Cloudflare.Queues.Queue("EventQueueB");
96 const target = opts.target === "B" ? queueB : queueA;
97 const notification = yield* Cloudflare.R2.BucketEventNotification(
98 "Notification",
99 {
100 bucketName: bucket.bucketName,
101 queueId: target.queueId,
102 rules: opts.rules,
103 },
104 );
105 return { bucket, queueA, queueB, notification };
106 });
107
108test.provider(
109 "create, update rules in place, and delete",
110 (stack) =>
111 Effect.gen(function* () {
112 const { accountId } = yield* yield* CloudflareEnvironment;
113
114 yield* stack.destroy();
115
116 // Create — a single prefix-scoped rule.
117 const initial = yield* stack.deploy(
118 program({
119 rules: [
120 {
121 actions: ["PutObject", "DeleteObject"],
122 prefix: "incoming/",
123 },
124 ],
125 }),
126 );
127
128 expect(initial.notification.accountId).toEqual(accountId);
129 expect(initial.notification.bucketName).toEqual(
130 initial.bucket.bucketName,
131 );
132 expect(initial.notification.queueId).toEqual(initial.queue.queueId);
133 expect(initial.notification.jurisdiction).toEqual("default");
134 expect(initial.notification.rules).toHaveLength(1);
135 expect([...initial.notification.rules[0].actions].sort()).toEqual([
136 "DeleteObject",
137 "PutObject",
138 ]);
139 expect(initial.notification.rules[0].prefix).toEqual("incoming/");
140 expect(initial.notification.rules[0].ruleId).toBeDefined();
141
142 const live = yield* getNotification(
143 accountId,
144 initial.bucket.bucketName,
145 initial.queue.queueId,
146 );
147 // The endpoint echoes the queue ID in dashed-UUID form.
148 expect(live.queueId?.replace(/-/g, "")).toEqual(initial.queue.queueId);
149 expect(live.rules ?? []).toHaveLength(1);
150 expect(live.rules?.[0].prefix).toEqual("incoming/");
151
152 // Update — replace the rule set in place (same bucket + queue pair).
153 const updated = yield* stack.deploy(
154 program({
155 rules: [
156 {
157 actions: ["PutObject", "CompleteMultipartUpload"],
158 prefix: "images/",
159 suffix: ".png",
160 description: "new PNG images",
161 },
162 {
163 actions: ["DeleteObject", "LifecycleDeletion"],
164 // R2 rejects rules whose key ranges overlap ("invalid
165 // overlap"), even with disjoint actions — scope this rule
166 // to a separate prefix.
167 prefix: "logs/",
168 },
169 ],
170 }),
171 );
172
173 // Same configuration mutated in place — not a replacement.
174 expect(updated.notification.bucketName).toEqual(
175 initial.notification.bucketName,
176 );
177 expect(updated.notification.queueId).toEqual(
178 initial.notification.queueId,
179 );
180 expect(updated.notification.rules).toHaveLength(2);
181
182 const liveUpdated = yield* getNotification(
183 accountId,
184 updated.bucket.bucketName,
185 updated.queue.queueId,
186 );
187 const liveRules = [...(liveUpdated.rules ?? [])].sort((a, b) =>
188 (a.prefix ?? "").localeCompare(b.prefix ?? ""),
189 );
190 expect(liveRules).toHaveLength(2);
191 expect(liveRules[0].prefix).toEqual("images/");
192 expect(liveRules[0].suffix).toEqual(".png");
193 expect(liveRules[0].description).toEqual("new PNG images");
194 expect(liveRules[1].prefix).toEqual("logs/");
195 expect([...liveRules[1].actions].sort()).toEqual([
196 "DeleteObject",
197 "LifecycleDeletion",
198 ]);
199
200 // Redeploying identical props is a no-op (same rule IDs survive).
201 const noop = yield* stack.deploy(
202 program({
203 rules: [
204 {
205 actions: ["PutObject", "CompleteMultipartUpload"],
206 prefix: "images/",
207 suffix: ".png",
208 description: "new PNG images",
209 },
210 {
211 actions: ["DeleteObject", "LifecycleDeletion"],
212 // R2 rejects rules whose key ranges overlap ("invalid
213 // overlap"), even with disjoint actions — scope this rule
214 // to a separate prefix.
215 prefix: "logs/",
216 },
217 ],
218 }),
219 );
220 expect(noop.notification.rules.map((rule) => rule.ruleId).sort()).toEqual(
221 updated.notification.rules.map((rule) => rule.ruleId).sort(),
222 );
223
224 yield* stack.destroy();
225
226 yield* expectGone(
227 accountId,
228 initial.bucket.bucketName,
229 initial.queue.queueId,
230 );
231
232 // Destroy again — delete must be idempotent (already gone).
233 yield* stack.destroy();
234 }).pipe(logLevel),
235 {
236 tags: [
237 "provider:cloudflare",
238 "provider:cloudflare:queue",
239 "provider:cloudflare:r2",
240 "live",
241 ],
242 timeout: 300_000,
243 },
244);
245
246test.provider(
247 "changing the queue triggers a replacement",
248 (stack) =>
249 Effect.gen(function* () {
250 const { accountId } = yield* yield* CloudflareEnvironment;
251
252 yield* stack.destroy();
253
254 const initial = yield* stack.deploy(
255 replacementProgram({
256 target: "A",
257 rules: [{ actions: ["PutObject"] }],
258 }),
259 );
260 expect(initial.notification.queueId).toEqual(initial.queueA.queueId);
261
262 // Point the notification at a different queue — the (bucket, queue)
263 // pair is the configuration's identity, so this replaces it.
264 const replaced = yield* stack.deploy(
265 replacementProgram({
266 target: "B",
267 rules: [{ actions: ["PutObject"] }],
268 }),
269 );
270
271 expect(replaced.queueB.queueId).not.toEqual(initial.queueA.queueId);
272 expect(replaced.notification.queueId).toEqual(replaced.queueB.queueId);
273
274 // The old pair's configuration is gone; the new pair's is live.
275 yield* expectGone(
276 accountId,
277 initial.bucket.bucketName,
278 initial.queueA.queueId,
279 );
280 const live = yield* getNotification(
281 accountId,
282 replaced.bucket.bucketName,
283 replaced.queueB.queueId,
284 );
285 // The endpoint echoes the queue ID in dashed-UUID form.
286 expect(live.queueId?.replace(/-/g, "")).toEqual(replaced.queueB.queueId);
287
288 yield* stack.destroy();
289
290 yield* expectGone(
291 accountId,
292 replaced.bucket.bucketName,
293 replaced.queueB.queueId,
294 );
295 }).pipe(logLevel),
296 {
297 tags: [
298 "provider:cloudflare",
299 "provider:cloudflare:queue",
300 "provider:cloudflare:r2",
301 "live",
302 ],
303 timeout: 300_000,
304 },
305);
306
307test.provider(
308 "list enumerates deployed bucket event notifications",
309 (stack) =>
310 Effect.gen(function* () {
311 yield* stack.destroy();
312
313 const deployed = yield* stack.deploy(
314 program({
315 rules: [
316 { actions: ["PutObject", "DeleteObject"], prefix: "incoming/" },
317 ],
318 }),
319 );
320
321 // Parent fan-out over every R2 bucket, then each bucket's
322 // event-notification queues — the deployed (bucket, queue) pair must
323 // appear, hydrated into the exact `read` Attributes shape.
324 const provider = yield* Provider.findProvider(
325 Cloudflare.R2.BucketEventNotification,
326 );
327 const all = yield* provider.list();
328
329 const found = all.find(
330 (n) =>
331 n.bucketName === deployed.bucket.bucketName &&
332 n.queueId === deployed.queue.queueId,
333 );
334 expect(found).toBeDefined();
335 expect(found?.accountId).toBeDefined();
336 expect(found?.jurisdiction).toEqual("default");
337 expect(found?.rules.length).toBeGreaterThanOrEqual(1);
338 expect(found?.rules.some((rule) => rule.prefix === "incoming/")).toBe(
339 true,
340 );
341
342 yield* stack.destroy();
343 }).pipe(logLevel),
344 {
345 tags: [
346 "provider:cloudflare",
347 "provider:cloudflare:queue",
348 "provider:cloudflare:r2",
349 "live",
350 ],
351 timeout: 300_000,
352 },
353);