1import * as Cloudflare from "@/Cloudflare";
2import { CloudflareEnvironment } from "@/Cloudflare/CloudflareEnvironment";
3import * as Provider from "@/Provider";
4import * as Test from "@/Test/Alchemy";
5import * as queues from "@distilled.cloud/cloudflare/queues";
6import { describe, expect } from "alchemy-test";
7import * as Effect from "effect/Effect";
8import { MinimumLogLevel } from "effect/References";
9import * as Schedule from "effect/Schedule";
10const { test } = Test.make({ providers: Cloudflare.providers() });
11
12const logLevel = Effect.provideService(
13 MinimumLogLevel,
14 process.env.DEBUG ? "Debug" : "Info",
15);
16
17// The scoped API token the test harness mints propagates eventually-
18// consistently across Cloudflare's edge — ride out 403 blips
19// (`Forbidden`, declared in the distilled error union) on the test's
20// own out-of-band verification calls.
21const getSubscription = (accountId: string, subscriptionId: string) =>
22 queues.getSubscription({ accountId, subscriptionId }).pipe(
23 Effect.retry({
24 while: (e) => e._tag === "Forbidden",
25 schedule: Schedule.fixed("500 millis"),
26 times: 20,
27 }),
28 );
29
30const expectGone = (accountId: string, subscriptionId: string) =>
31 getSubscription(accountId, subscriptionId).pipe(
32 Effect.flatMap(() =>
33 Effect.fail({ _tag: "SubscriptionNotDeleted" } as const),
34 ),
35 // A missing subscription surfaces as `SubscriptionNotFound`
36 // (HTTP 404 "No subscription with this ID") — that's the success
37 // condition here.
38 Effect.catchTag("SubscriptionNotFound", () => Effect.void),
39 Effect.retry({
40 while: (e) => e._tag === "SubscriptionNotDeleted",
41 schedule: Schedule.max([
42 Schedule.fixed("500 millis"),
43 Schedule.recurs(20),
44 ]),
45 }),
46 );
47
48// Sequential: Cloudflare allows only ONE subscription per source per
49// account, and every test here uses the same `{ type: "r2" }` (or kv)
50// source. Run concurrently they adopt/patch/delete each other's
51// subscription via the AlreadyExists fallback and fail with queueId
52// mismatches and SubscriptionNotFound.
53describe.sequential(
54 "Subscription",
55 { tags: ["provider:cloudflare", "provider:cloudflare:queue", "live"] },
56 () => {
57 test.provider(
58 "create r2 event subscription into a queue and destroy it",
59 (stack) =>
60 Effect.gen(function* () {
61 const { accountId } = yield* yield* CloudflareEnvironment;
62
63 yield* stack.destroy();
64
65 const deployed = yield* stack.deploy(
66 Effect.gen(function* () {
67 const queue = yield* Cloudflare.Queues.Queue("SubQueue", {
68 name: "alchemy-test-sub-queue",
69 });
70 // Compose with an R2 bucket — the subscription delivers
71 // account-level R2 events (e.g. this bucket's lifecycle)
72 // into the queue.
73 const bucket = yield* Cloudflare.R2.Bucket("SubBucket", {
74 forceDestroy: true,
75 name: "alchemy-test-sub-bucket",
76 });
77 const subscription = yield* Cloudflare.Queues.Subscription(
78 "R2Events",
79 {
80 source: { type: "r2" },
81 events: ["bucket.created", "bucket.deleted"],
82 queueId: queue.queueId,
83 },
84 );
85 return { queue, bucket, subscription };
86 }),
87 );
88
89 const sub = deployed.subscription;
90 expect(sub.subscriptionId).toBeDefined();
91 expect(sub.accountId).toEqual(accountId);
92 expect(sub.source).toEqual({ type: "r2" });
93 expect([...sub.events].sort()).toEqual([
94 "bucket.created",
95 "bucket.deleted",
96 ]);
97 expect(sub.enabled).toBe(true);
98 expect(sub.queueId).toEqual(deployed.queue.queueId);
99
100 // Out-of-band verification straight against the API.
101 const live = yield* getSubscription(accountId, sub.subscriptionId);
102 expect(live.id).toEqual(sub.subscriptionId);
103 expect(live.destination.queueId).toEqual(deployed.queue.queueId);
104 expect([...live.events].sort()).toEqual([
105 "bucket.created",
106 "bucket.deleted",
107 ]);
108 expect(live.enabled).toBe(true);
109
110 yield* stack.destroy();
111
112 yield* expectGone(accountId, sub.subscriptionId);
113 }).pipe(logLevel),
114 { tags: ["provider:cloudflare:r2"] },
115 );
116
117 test.provider(
118 "update mutable props in place (same subscriptionId)",
119 (stack) =>
120 Effect.gen(function* () {
121 const { accountId } = yield* yield* CloudflareEnvironment;
122
123 yield* stack.destroy();
124
125 const initial = yield* stack.deploy(
126 Effect.gen(function* () {
127 const queueA = yield* Cloudflare.Queues.Queue("SubQueueA", {
128 name: "alchemy-test-sub-queue-a",
129 });
130 const queueB = yield* Cloudflare.Queues.Queue("SubQueueB", {
131 name: "alchemy-test-sub-queue-b",
132 });
133 const subscription = yield* Cloudflare.Queues.Subscription(
134 "UpdateSub",
135 {
136 name: "alchemy-sub-update",
137 source: { type: "r2" },
138 events: ["bucket.created"],
139 queueId: queueA.queueId,
140 },
141 );
142 return { queueA, queueB, subscription };
143 }),
144 );
145
146 expect(initial.subscription.name).toEqual("alchemy-sub-update");
147 expect(initial.subscription.enabled).toBe(true);
148 expect(initial.subscription.queueId).toEqual(initial.queueA.queueId);
149
150 // Mutate everything mutable: name, events, enabled, and the
151 // destination queue.
152 const updated = yield* stack.deploy(
153 Effect.gen(function* () {
154 const queueA = yield* Cloudflare.Queues.Queue("SubQueueA", {
155 name: "alchemy-test-sub-queue-a",
156 });
157 const queueB = yield* Cloudflare.Queues.Queue("SubQueueB", {
158 name: "alchemy-test-sub-queue-b",
159 });
160 const subscription = yield* Cloudflare.Queues.Subscription(
161 "UpdateSub",
162 {
163 name: "alchemy-sub-update-v2",
164 source: { type: "r2" },
165 events: ["bucket.created", "bucket.deleted"],
166 queueId: queueB.queueId,
167 enabled: false,
168 },
169 );
170 return { queueA, queueB, subscription };
171 }),
172 );
173
174 // Same subscription mutated in place — not a replacement.
175 expect(updated.subscription.subscriptionId).toEqual(
176 initial.subscription.subscriptionId,
177 );
178 expect(updated.subscription.name).toEqual("alchemy-sub-update-v2");
179 expect(updated.subscription.enabled).toBe(false);
180 expect(updated.subscription.queueId).toEqual(updated.queueB.queueId);
181 expect([...updated.subscription.events].sort()).toEqual([
182 "bucket.created",
183 "bucket.deleted",
184 ]);
185
186 const live = yield* getSubscription(
187 accountId,
188 updated.subscription.subscriptionId,
189 );
190 expect(live.name).toEqual("alchemy-sub-update-v2");
191 expect(live.enabled).toBe(false);
192 expect(live.destination.queueId).toEqual(updated.queueB.queueId);
193 expect([...live.events].sort()).toEqual([
194 "bucket.created",
195 "bucket.deleted",
196 ]);
197
198 yield* stack.destroy();
199
200 yield* expectGone(accountId, updated.subscription.subscriptionId);
201 }).pipe(logLevel),
202 );
203
204 test.provider(
205 "replaces the subscription when the source changes",
206 (stack) =>
207 Effect.gen(function* () {
208 const { accountId } = yield* yield* CloudflareEnvironment;
209
210 yield* stack.destroy();
211
212 const initial = yield* stack.deploy(
213 Effect.gen(function* () {
214 const queue = yield* Cloudflare.Queues.Queue("SubQueueR", {
215 name: "alchemy-test-sub-queue-r",
216 });
217 const subscription = yield* Cloudflare.Queues.Subscription(
218 "ReplaceSub",
219 {
220 source: { type: "kv" },
221 events: ["namespace.created"],
222 queueId: queue.queueId,
223 },
224 );
225 return { queue, subscription };
226 }),
227 );
228
229 expect(initial.subscription.source).toEqual({ type: "kv" });
230
231 const replaced = yield* stack.deploy(
232 Effect.gen(function* () {
233 const queue = yield* Cloudflare.Queues.Queue("SubQueueR", {
234 name: "alchemy-test-sub-queue-r",
235 });
236 const subscription = yield* Cloudflare.Queues.Subscription(
237 "ReplaceSub",
238 {
239 source: { type: "r2" },
240 events: ["bucket.created"],
241 queueId: queue.queueId,
242 },
243 );
244 return { queue, subscription };
245 }),
246 );
247
248 // The source is fixed at creation — changing it produces a new
249 // subscription identity.
250 expect(replaced.subscription.subscriptionId).not.toEqual(
251 initial.subscription.subscriptionId,
252 );
253 expect(replaced.subscription.source).toEqual({ type: "r2" });
254
255 // The old subscription is gone after the replacement settles.
256 yield* expectGone(accountId, initial.subscription.subscriptionId);
257
258 const live = yield* getSubscription(
259 accountId,
260 replaced.subscription.subscriptionId,
261 );
262 expect(live.events).toEqual(["bucket.created"]);
263
264 yield* stack.destroy();
265
266 yield* expectGone(accountId, replaced.subscription.subscriptionId);
267 }).pipe(logLevel),
268 );
269
270 // Canonical `list()` test (account collection): deploy a subscription,
271 // then enumerate every subscription in the account via the typed provider
272 // and assert the deployed one is present in the exhaustively-paginated
273 // result.
274 test.provider("list enumerates the deployed subscription", (stack) =>
275 Effect.gen(function* () {
276 yield* stack.destroy();
277
278 const deployed = yield* stack.deploy(
279 Effect.gen(function* () {
280 const queue = yield* Cloudflare.Queues.Queue("ListSubQueue", {
281 name: "alchemy-test-list-sub-queue",
282 });
283 const subscription = yield* Cloudflare.Queues.Subscription(
284 "ListR2Events",
285 {
286 source: { type: "r2" },
287 events: ["bucket.created"],
288 queueId: queue.queueId,
289 },
290 );
291 return { queue, subscription };
292 }),
293 );
294
295 const provider = yield* Provider.findProvider(
296 Cloudflare.Queues.Subscription,
297 );
298 const all = yield* provider.list();
299
300 expect(
301 all.some(
302 (s) => s.subscriptionId === deployed.subscription.subscriptionId,
303 ),
304 ).toBe(true);
305
306 yield* stack.destroy();
307 }).pipe(logLevel),
308 );
309 },
310);