1import * as Cloudflare from "@/Cloudflare";
2import { CloudflareEnvironment } from "@/Cloudflare/CloudflareEnvironment";
3import * as Output from "@/Output";
4import * as Test from "@/Test/Alchemy";
5import * as accounts from "@distilled.cloud/cloudflare/accounts";
6import * as r2 from "@distilled.cloud/cloudflare/r2";
7import * as queues from "@distilled.cloud/cloudflare/queues";
8import * as Retry from "@distilled.cloud/cloudflare/Retry";
9import { expect } from "alchemy-test";
10import * as Effect from "effect/Effect";
11import * as Redacted from "effect/Redacted";
12import * as Schedule from "effect/Schedule";
13import { createHash } from "node:crypto";
14
15const { test } = Test.make({ providers: Cloudflare.providers() });
16const missingJobId =
17 "0000000000000000000000000000000000000000000000000000000000000000";
18
19const program = (options: {
20 job: boolean;
21 paused?: boolean;
22 overwrite?: boolean;
23 reference?: boolean;
24 sourcePrefix?: string;
25 targetName?: string;
26 otherJob?: boolean;
27 subscription?: "direct" | "ref";
28}) =>
29 Effect.gen(function* () {
30 const { accountId } = yield* yield* CloudflareEnvironment;
31 const source = yield* Cloudflare.R2.Bucket("Source", {
32 forceDestroy: true,
33 });
34 const target = yield* Cloudflare.R2.Bucket("Target", {
35 name: options.targetName,
36 forceDestroy: true,
37 });
38 const token = yield* Cloudflare.ApiToken.AccountApiToken("Credentials", {
39 policies: [
40 {
41 effect: "allow",
42 permissionGroups: [
43 "Workers R2 Storage Read",
44 "Workers R2 Storage Write",
45 ],
46 resources: { [`com.cloudflare.api.account.${accountId}`]: "*" },
47 },
48 ],
49 });
50 const secret = {
51 accessKeyId: token.tokenId.pipe(Output.map(Redacted.make)),
52 secretAccessKey: token.value.pipe(
53 Output.mapEffect((value) =>
54 Effect.sync(() =>
55 Redacted.make(
56 createHash("sha256").update(Redacted.value(value)).digest("hex"),
57 ),
58 ),
59 ),
60 ),
61 };
62 const job = options.job
63 ? yield* Cloudflare.R2.SuperSlurperJob("Migration", {
64 source: {
65 vendor: "r2",
66 bucket: source.bucketName,
67 secret,
68 pathPrefix: options.sourcePrefix,
69 },
70 target: { vendor: "r2", bucket: target.bucketName, secret },
71 paused: options.paused,
72 overwrite: options.overwrite,
73 })
74 : undefined;
75 const otherJob = options.otherJob
76 ? yield* Cloudflare.R2.SuperSlurperJob("OtherMigration", {
77 source: { vendor: "r2", bucket: source.bucketName, secret },
78 target: { vendor: "r2", bucket: target.bucketName, secret },
79 paused: true,
80 })
81 : undefined;
82 const reference = options.reference
83 ? yield* Cloudflare.R2.SuperSlurperJob.ref("Migration")
84 : undefined;
85 const queue = options.subscription
86 ? yield* Cloudflare.Queues.Queue("JobEventsQueue")
87 : undefined;
88 const subscription =
89 options.subscription && job && queue
90 ? yield* Cloudflare.Queues.Subscription("JobEvents", {
91 source:
92 options.subscription === "ref"
93 ? yield* Cloudflare.R2.SuperSlurperJob.ref("Migration")
94 : job,
95 events: ["job.resumed", "job.completed", "job.aborted"],
96 queueId: queue.queueId,
97 })
98 : undefined;
99 return {
100 source,
101 target,
102 token,
103 job,
104 otherJob,
105 reference,
106 queue,
107 subscription,
108 };
109 });
110
111const getJob = (identity: { accountId: string; jobId: string }) =>
112 r2
113 .getSuperSlurperJob({
114 accountId: identity.accountId,
115 jobId: identity.jobId,
116 })
117 .pipe(Retry.none);
118
119const terminal = (status: r2.GetSuperSlurperJobResponse["status"]) =>
120 status === "aborted" || status === "completed";
121
122test.provider(
123 "observes absence and types the API's absent-job mutation failures",
124 (stack) =>
125 Effect.gen(function* () {
126 yield* stack.destroy();
127 const { accountId } = yield* yield* CloudflareEnvironment;
128 const identity = { accountId, jobId: missingJobId };
129 const missing = yield* getJob(identity).pipe(Effect.flip);
130 expect(missing._tag).toBe("SuperSlurperJobOperationFailed");
131 expect(missing.message).toBe("Internal Server Error");
132 const invalidSecret = {
133 accessKeyId: "00000000000000000000000000000000",
134 secretAccessKey:
135 "0000000000000000000000000000000000000000000000000000000000000000",
136 };
137 const rejected = yield* r2
138 .createSuperSlurperJob({
139 accountId,
140 source: {
141 vendor: "r2",
142 bucket: "alchemy-slurper-missing-source",
143 secret: invalidSecret,
144 },
145 target: {
146 vendor: "r2",
147 bucket: "alchemy-slurper-missing-target",
148 secret: invalidSecret,
149 },
150 })
151 .pipe(Retry.none, Effect.flip);
152 expect(rejected._tag).toBe("SuperSlurperPreconnectivityFailed");
153 expect(rejected.message).toBe(
154 "Preconnectivity failed, please verify tokens and try again",
155 );
156 for (const operation of [
157 r2.abortSuperSlurperJob,
158 r2.pauseSuperSlurperJob,
159 r2.resumeSuperSlurperJob,
160 ]) {
161 const error = yield* operation(identity).pipe(Retry.none, Effect.flip);
162 expect(error._tag).toBe("SuperSlurperJobOperationFailed");
163 expect(error.message).toBe("Internal Server Error");
164 }
165 yield* stack.destroy();
166 }).pipe(Effect.ensuring(stack.destroy().pipe(Effect.orDie))),
167 {
168 tags: ["provider:cloudflare", "provider:cloudflare:r2", "live"],
169 timeout: 120_000,
170 exclusive: true,
171 },
172);
173
174test.provider(
175 "creates, pauses, resumes, preserves terminal identity, replaces and cancels without deleting objects",
176 (stack) =>
177 Effect.gen(function* () {
178 yield* stack.destroy();
179 const { accountId } = yield* yield* CloudflareEnvironment;
180 const buckets = yield* stack.deploy(program({ job: false }));
181 yield* r2.uploadBucketObject({
182 accountId,
183 bucketName: buckets.source.bucketName,
184 objectName: "migration.txt",
185 body: "super-slurper-source",
186 });
187 yield* r2.uploadBucketObject({
188 accountId,
189 bucketName: buckets.target.bucketName,
190 objectName: "sentinel.txt",
191 body: "keep-existing-target-data",
192 });
193 const created = yield* stack.deploy(program({ job: true, paused: true }));
194 const job = created.job!;
195 expect(job.accountId).toBe(accountId);
196 expect(job.jobId).toBeTruthy();
197 const paused = yield* getJob(job);
198 expect(paused.id).toBe(job.jobId);
199 expect(paused.source?.bucket).toBe(buckets.source.bucketName);
200 expect(paused.target?.bucket).toBe(buckets.target.bucketName);
201 expect(paused.status === "paused" || terminal(paused.status)).toBe(true);
202
203 const noop = yield* stack.deploy(
204 program({ job: true, paused: true, reference: true }),
205 );
206 expect(noop.job!.jobId).toBe(job.jobId);
207 expect(noop.reference!.jobId).toBe(job.jobId);
208 expect(noop.reference!.accountId).toBe(accountId);
209
210 const resumed = yield* stack.deploy(program({ job: true }));
211 expect(resumed.job!.jobId).toBe(job.jobId);
212 const running = yield* getJob(job);
213 expect(running.status === "running" || terminal(running.status)).toBe(
214 true,
215 );
216 if (!terminal(running.status)) {
217 yield* r2
218 .abortSuperSlurperJob({ accountId, jobId: job.jobId })
219 .pipe(Retry.none);
220 }
221 const finished = yield* getJob(job).pipe(
222 Effect.repeat({
223 schedule: Schedule.spaced("2 seconds"),
224 until: (value) => terminal(value.status),
225 times: 8,
226 }),
227 );
228 expect(terminal(finished.status)).toBe(true);
229
230 const terminalRedeploy = yield* stack.deploy(
231 program({ job: true, paused: true }),
232 );
233 expect(terminalRedeploy.job!.jobId).toBe(job.jobId);
234 expect((yield* getJob(job)).status).toBe(finished.status);
235
236 const replaced = yield* stack.deploy(
237 program({ job: true, paused: true, overwrite: true }),
238 );
239 const replacement = replaced.job!;
240 expect(replacement.jobId).not.toBe(job.jobId);
241 expect((yield* getJob(job)).status).toBe(finished.status);
242 const changedSource = yield* stack.deploy(
243 program({
244 job: true,
245 paused: true,
246 overwrite: true,
247 sourcePrefix: "migration",
248 otherJob: true,
249 }),
250 );
251 const filtered = changedSource.job!;
252 const otherJob = changedSource.otherJob!;
253 expect(filtered.jobId).not.toBe(replacement.jobId);
254 expect(otherJob.jobId).not.toBe(job.jobId);
255 expect(otherJob.jobId).not.toBe(filtered.jobId);
256 const firstPage = yield* r2.listSuperSlurperJobs({
257 accountId,
258 limit: 1,
259 offset: 0,
260 });
261 const secondPage = yield* r2.listSuperSlurperJobs({
262 accountId,
263 limit: 1,
264 offset: 1,
265 });
266 expect(firstPage.result).toHaveLength(1);
267 expect(secondPage.result).toHaveLength(1);
268 expect(firstPage.result[0]!.id).not.toBe(secondPage.result[0]!.id);
269 expect((yield* getJob(filtered)).source?.pathPrefix).toBe("migration");
270 const otherBeforeRemoval = yield* getJob(otherJob);
271 const removed = yield* stack.deploy(
272 program({ job: false, otherJob: true }),
273 );
274 expect(removed.otherJob!.jobId).toBe(otherJob.jobId);
275 expect((yield* getJob(otherJob)).status).toBe(otherBeforeRemoval.status);
276 expect(terminal((yield* getJob(filtered)).status)).toBe(true);
277 expect(removed.source.bucketName).toBe(buckets.source.bucketName);
278 expect(removed.target.bucketName).toBe(buckets.target.bucketName);
279 const cancelled = yield* getJob(replacement).pipe(
280 Effect.repeat({
281 schedule: Schedule.spaced("2 seconds"),
282 until: (value) => terminal(value.status),
283 times: 8,
284 }),
285 );
286 expect(terminal(cancelled.status)).toBe(true);
287 const objects = yield* r2.listBucketObjects({
288 accountId,
289 bucketName: buckets.target.bucketName,
290 });
291 expect(
292 objects.result.some((object) => object.key === "sentinel.txt"),
293 ).toBe(true);
294 const sourceObjects = yield* r2.listBucketObjects({
295 accountId,
296 bucketName: buckets.source.bucketName,
297 });
298 expect(
299 sourceObjects.result.some((object) => object.key === "migration.txt"),
300 ).toBe(true);
301
302 yield* stack.destroy();
303 for (const bucketName of [
304 buckets.source.bucketName,
305 buckets.target.bucketName,
306 ]) {
307 const error = yield* r2
308 .getBucket({ accountId, bucketName })
309 .pipe(Effect.flip);
310 expect(error._tag).toBe("NoSuchBucket");
311 }
312 expect((yield* getJob(job)).status).toBe(finished.status);
313 expect(terminal((yield* getJob(replacement)).status)).toBe(true);
314 expect(terminal((yield* getJob(otherJob)).status)).toBe(true);
315 const tokenGone = yield* accounts
316 .getToken({
317 accountId,
318 tokenId: buckets.token.tokenId,
319 })
320 .pipe(Effect.flip);
321 expect(["TokenNotFound", "InvalidRoute"]).toContain(tokenGone._tag);
322 yield* Effect.log({
323 jobs: [job.jobId, replacement.jobId, filtered.jobId, otherJob.jobId],
324 status: "all terminal; buckets and token destroyed",
325 });
326 yield* stack.destroy();
327 }).pipe(Effect.ensuring(stack.destroy().pipe(Effect.orDie))),
328 {
329 tags: [
330 "provider:cloudflare",
331 "provider:cloudflare:account",
332 "provider:cloudflare:apitoken",
333 "provider:cloudflare:queue",
334 "provider:cloudflare:r2",
335 "live",
336 ],
337 timeout: 120_000,
338 exclusive: true,
339 },
340);
341
342test.provider(
343 "an unresolved target bucket identity replaces the migration",
344 (stack) =>
345 Effect.gen(function* () {
346 yield* stack.destroy();
347 const initial = yield* stack.deploy(
348 program({
349 job: true,
350 paused: true,
351 targetName: "alchemy-slurper-target-replacement-a",
352 }),
353 );
354 const replaced = yield* stack.deploy(
355 program({
356 job: true,
357 paused: true,
358 targetName: "alchemy-slurper-target-replacement-b",
359 }),
360 );
361 expect(replaced.job!.jobId).not.toBe(initial.job!.jobId);
362 expect((yield* getJob(replaced.job!)).target?.bucket).toBe(
363 replaced.target.bucketName,
364 );
365 expect(terminal((yield* getJob(initial.job!)).status)).toBe(true);
366 yield* stack.destroy();
367 expect(terminal((yield* getJob(replaced.job!)).status)).toBe(true);
368 }).pipe(Effect.ensuring(stack.destroy().pipe(Effect.orDie))),
369 {
370 tags: [
371 "provider:cloudflare",
372 "provider:cloudflare:apitoken",
373 "provider:cloudflare:queue",
374 "provider:cloudflare:r2",
375 "live",
376 ],
377 timeout: 120_000,
378 exclusive: true,
379 },
380);
381
382test.provider(
383 "job resources and refs deliver account-wide subscription events without taking ownership",
384 (stack) =>
385 Effect.gen(function* () {
386 yield* stack.destroy();
387 const { accountId } = yield* yield* CloudflareEnvironment;
388 const buckets = yield* stack.deploy(program({ job: false }));
389 yield* r2.uploadBucketObject({
390 accountId,
391 bucketName: buckets.source.bucketName,
392 objectName: "event.txt",
393 body: "migration event",
394 });
395 const created = yield* stack.deploy(
396 program({ job: true, paused: true, subscription: "direct" }),
397 );
398 const job = created.job!;
399 const subscription = created.subscription!;
400 expect(subscription.source).toEqual({ type: "superSlurper" });
401 const live = yield* queues.getSubscription({
402 accountId,
403 subscriptionId: subscription.subscriptionId,
404 });
405 expect(live.source).toEqual(
406 expect.objectContaining({ type: "superSlurper" }),
407 );
408 yield* queues.createConsumer({
409 accountId,
410 queueId: created.queue!.queueId,
411 type: "http_pull",
412 });
413 const referenced = yield* stack.deploy(
414 program({ job: true, paused: true, subscription: "ref" }),
415 );
416 expect(referenced.subscription!.subscriptionId).toBe(
417 subscription.subscriptionId,
418 );
419 const resumed = yield* stack.deploy(
420 program({ job: true, subscription: "ref" }),
421 );
422 expect(resumed.job!.jobId).toBe(job.jobId);
423 const bodies: string[] = [];
424 const delivered = () =>
425 bodies.some(
426 (body) =>
427 body.includes("cf.superSlurper.job.") &&
428 body.includes(job.jobId) &&
429 body.includes(subscription.subscriptionId),
430 );
431 yield* Effect.gen(function* () {
432 const batch = yield* queues
433 .pullMessage({
434 accountId,
435 queueId: created.queue!.queueId,
436 batchSize: 100,
437 visibilityTimeoutMs: 1000,
438 })
439 .pipe(
440 Effect.retry({
441 while: (error) => error._tag === "QueueHttpPullNotEnabled",
442 schedule: Schedule.spaced("2 seconds"),
443 times: 8,
444 }),
445 );
446 for (const message of batch.messages ?? [])
447 if (message.body) bodies.push(message.body);
448 const acks = (batch.messages ?? []).flatMap((message) =>
449 message.leaseId ? [{ leaseId: message.leaseId }] : [],
450 );
451 if (acks.length)
452 yield* queues.ackMessage({
453 accountId,
454 queueId: created.queue!.queueId,
455 acks,
456 });
457 // A `job.resumed` emitted before the new subscription propagates is
458 // lost, so emit a fresh one until an event is delivered.
459 if (!delivered() && (yield* getJob(job)).status === "running") {
460 yield* r2.pauseSuperSlurperJob(job).pipe(Retry.none, Effect.ignore);
461 yield* r2.resumeSuperSlurperJob(job).pipe(Retry.none, Effect.ignore);
462 }
463 }).pipe(
464 Effect.repeat({
465 schedule: Schedule.spaced("5 seconds"),
466 times: 10,
467 until: delivered,
468 }),
469 Effect.timeout("60 seconds"),
470 );
471 expect(delivered()).toBe(true);
472 yield* stack.deploy(program({ job: true, paused: true }));
473 expect((yield* getJob(job)).id).toBe(job.jobId);
474 const gone = yield* queues
475 .getSubscription({
476 accountId,
477 subscriptionId: subscription.subscriptionId,
478 })
479 .pipe(Effect.flip);
480 expect(gone._tag).toBe("SubscriptionNotFound");
481 yield* stack.destroy();
482 expect(terminal((yield* getJob(job)).status)).toBe(true);
483 }).pipe(Effect.ensuring(stack.destroy().pipe(Effect.orDie))),
484 {
485 tags: [
486 "provider:cloudflare",
487 "provider:cloudflare:apitoken",
488 "provider:cloudflare:queue",
489 "provider:cloudflare:r2",
490 "live",
491 ],
492 timeout: 120_000,
493 exclusive: true,
494 },
495);
496
497// A one-object migration remained running after the bounded 24-second probe.
498test.provider.skipIf(
499 process.env.CLOUDFLARE_TEST_SUPER_SLURPER_COMPLETION !== "1",
500)(
501 "completes a real object transfer and never restarts the completed job",
502 (stack) =>
503 Effect.gen(function* () {
504 yield* stack.destroy();
505 const { accountId } = yield* yield* CloudflareEnvironment;
506 const buckets = yield* stack.deploy(program({ job: false }));
507 yield* r2.uploadBucketObject({
508 accountId,
509 bucketName: buckets.source.bucketName,
510 objectName: "migration.txt",
511 body: "super-slurper-source",
512 });
513 const created = yield* stack.deploy(program({ job: true }));
514 const job = created.job!;
515 const completed = yield* getJob(job).pipe(
516 Effect.repeat({
517 schedule: Schedule.spaced("5 seconds"),
518 until: (value) => terminal(value.status),
519 times: 8,
520 }),
521 );
522 expect(completed.status).toBe("completed");
523 const migrated = yield* r2.listBucketObjects({
524 accountId,
525 bucketName: buckets.target.bucketName,
526 });
527 expect(
528 migrated.result.some((object) => object.key === "migration.txt"),
529 ).toBe(true);
530 const redeployed = yield* stack.deploy(
531 program({ job: true, paused: true }),
532 );
533 expect(redeployed.job!.jobId).toBe(job.jobId);
534 expect((yield* getJob(job)).status).toBe("completed");
535 yield* stack.destroy();
536 expect((yield* getJob(job)).status).toBe("completed");
537 }),
538 {
539 tags: [
540 "provider:cloudflare",
541 "provider:cloudflare:apitoken",
542 "provider:cloudflare:queue",
543 "provider:cloudflare:r2",
544 "live",
545 ],
546 timeout: 120_000,
547 exclusive: true,
548 },
549);