1import * as r2 from "@distilled.cloud/cloudflare/r2";
2import * as Retry from "@distilled.cloud/cloudflare/Retry";
3import * as Effect from "effect/Effect";
4import * as Redacted from "effect/Redacted";
5import * as Schedule from "effect/Schedule";
6import { deepEqual, isResolved } from "../../Diff.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { CloudflareEnvironment } from "../CloudflareEnvironment.ts";
10import type { Providers } from "../Providers.ts";
11
12/** S3-compatible credentials with access to a migration bucket. */
13export interface SuperSlurperCredentials {
14 /** Access key ID. Changing credentials replaces the migration. */
15 accessKeyId: Redacted.Redacted<string>;
16 /** Secret access key. Changing credentials replaces the migration. */
17 secretAccessKey: Redacted.Redacted<string>;
18}
19
20/** Object selection shared by all source vendors. */
21export interface SuperSlurperSourceSelection {
22 /** Name of the source bucket. Immutable. */
23 bucket: string;
24 /** Object keys to migrate. Omit to migrate all matching objects. Immutable. */
25 keys?: string[];
26 /** Restrict migration to this object key prefix. Immutable. */
27 pathPrefix?: string;
28}
29
30/** AWS S3 or another HTTPS S3-compatible source. */
31export interface SuperSlurperS3Source extends SuperSlurperSourceSelection {
32 /** Select the S3 source API. */
33 vendor: "s3";
34 /** Source credentials with read access. Immutable. */
35 secret: SuperSlurperCredentials;
36 /** Custom S3-compatible HTTPS endpoint. Immutable. */
37 endpoint?: string;
38 /** Source bucket region. Immutable. */
39 region?: string;
40}
41
42/** Google Cloud Storage source. */
43export interface SuperSlurperGcsSource extends SuperSlurperSourceSelection {
44 /** Select Google Cloud Storage. */
45 vendor: "gcs";
46 /** Service account credentials with source read access. Immutable. */
47 secret: {
48 /** Service account email. */
49 clientEmail: Redacted.Redacted<string>;
50 /** Service account private key in PEM format. */
51 privateKey: Redacted.Redacted<string>;
52 };
53}
54
55/** Cloudflare R2 source. */
56export interface SuperSlurperR2Source extends SuperSlurperSourceSelection {
57 /** Select Cloudflare R2. */
58 vendor: "r2";
59 /** S3-compatible R2 credentials with source read access. Immutable. */
60 secret: SuperSlurperCredentials;
61 /** Source jurisdiction. Immutable. @default "default" */
62 jurisdiction?: "default" | "eu" | "us" | "fedramp";
63}
64
65/** Supported source bucket configurations. */
66export type SuperSlurperSource =
67 | SuperSlurperS3Source
68 | SuperSlurperGcsSource
69 | SuperSlurperR2Source;
70
71/** R2 destination for migrated objects. */
72export interface SuperSlurperTarget {
73 /** Name of the destination R2 bucket. Immutable. */
74 bucket: string;
75 /** The destination vendor must be R2. */
76 vendor: "r2";
77 /** S3-compatible R2 credentials with destination write access. Immutable. */
78 secret: SuperSlurperCredentials;
79 /** Destination jurisdiction. Immutable. @default "default" */
80 jurisdiction?: "default" | "eu" | "us" | "fedramp";
81}
82
83export interface SuperSlurperJobProps {
84 /** Source bucket and credentials. Any change replaces the job. */
85 source: SuperSlurperSource;
86 /** Destination R2 bucket and credentials. Any change replaces the job. */
87 target: SuperSlurperTarget;
88 /** Overwrite existing destination objects. Immutable. @default false */
89 overwrite?: boolean;
90 /**
91 * Pause an active job, or resume it when false. Terminal jobs are never
92 * restarted. A new job starts running before the pause request is applied,
93 * so this is not a guarantee that no objects are transferred.
94 * @default false
95 */
96 paused?: boolean;
97}
98
99export interface SuperSlurperJobAttributes {
100 /** Cloudflare account that owns the migration. */
101 accountId: string;
102 /** Server-assigned migration ID. Stable until replacement. */
103 jobId: string;
104 /** Last observed job status; not a completion guarantee. */
105 status: r2.GetSuperSlurperJobResponse["status"];
106 /** Creation timestamp reported by Cloudflare. */
107 createdAt: string | undefined;
108 /** Completion timestamp, if the job has finished. */
109 finishedAt: string | undefined;
110}
111
112export type SuperSlurperJob = Resource<
113 "Cloudflare.R2.SuperSlurperJob",
114 SuperSlurperJobProps,
115 SuperSlurperJobAttributes,
116 never,
117 Providers
118>;
119
120/**
121 * Starts a bulk migration from S3, GCS, or R2 into an R2 bucket.
122 *
123 * Deployment returns the job identity without waiting for the migration.
124 * Completed and aborted jobs remain resources: redeploying does not run them
125 * again. Changing source, target, credentials, or overwrite replaces the job.
126 * Destroy only cancels this resource's active job; it never deletes copied
127 * objects or Cloudflare's job history. Transfers already in flight may finish.
128 *
129 * Cloudflare exposes no job name, ownership tags, or idempotency key. The
130 * persisted job ID is the only ownership record; similar jobs are never
131 * adopted. If a create response or the subsequent state write is lost, the
132 * job cannot be recovered automatically and retrying may create a second job.
133 * Inspect the account's migration history and cancel that orphan explicitly.
134 * Account-wide adoption and cancellation are intentionally not supported.
135 * If reading a cached ID fails with Cloudflare's ambiguous server error,
136 * paginated job history is checked for that exact ID before declaring it absent.
137 *
138 * ### Migrating from S3
139 * **Example:** Copy an S3 prefix into R2
140 * ```typescript
141 * const target = yield* Cloudflare.R2.Bucket("Archive");
142 * const job = yield* Cloudflare.R2.SuperSlurperJob("Migration", {
143 * source: {
144 * vendor: "s3",
145 * bucket: "legacy-archive",
146 * region: "us-east-1",
147 * pathPrefix: "photos/",
148 * secret: {
149 * accessKeyId: yield* Config.Redacted("AWS_ACCESS_KEY_ID"),
150 * secretAccessKey: yield* Config.Redacted("AWS_SECRET_ACCESS_KEY"),
151 * },
152 * },
153 * target: {
154 * vendor: "r2",
155 * bucket: target.bucketName,
156 * secret: {
157 * accessKeyId: yield* Config.Redacted("R2_ACCESS_KEY_ID"),
158 * secretAccessKey: yield* Config.Redacted("R2_SECRET_ACCESS_KEY"),
159 * },
160 * },
161 * });
162 * ```
163 *
164 * ### Migrating from GCS or R2
165 * **Example:** Google Cloud Storage source
166 * ```typescript
167 * const source: Cloudflare.R2.SuperSlurperSource = {
168 * vendor: "gcs",
169 * bucket: "legacy-media",
170 * secret: {
171 * clientEmail: yield* Config.Redacted("GCS_CLIENT_EMAIL"),
172 * privateKey: yield* Config.Redacted("GCS_PRIVATE_KEY"),
173 * },
174 * };
175 * ```
176 *
177 * **Example:** R2 source and paused desired state
178 * ```typescript
179 * yield* Cloudflare.R2.SuperSlurperJob("Migration", {
180 * source: {
181 * vendor: "r2",
182 * bucket: "old-media",
183 * secret: {
184 * accessKeyId: yield* Config.Redacted("SOURCE_R2_ACCESS_KEY_ID"),
185 * secretAccessKey: yield* Config.Redacted("SOURCE_R2_SECRET_ACCESS_KEY"),
186 * },
187 * },
188 * target: {
189 * vendor: "r2",
190 * bucket: "new-media",
191 * secret: {
192 * accessKeyId: yield* Config.Redacted("TARGET_R2_ACCESS_KEY_ID"),
193 * secretAccessKey: yield* Config.Redacted("TARGET_R2_SECRET_ACCESS_KEY"),
194 * },
195 * },
196 * paused: true,
197 * });
198 * ```
199 *
200 * ### Referencing a Migration
201 * **Example:** Read a persisted job identity without creating a new job
202 * ```typescript
203 * const job = yield* Cloudflare.R2.SuperSlurperJob.ref("Migration");
204 * return { accountId: job.accountId, jobId: job.jobId };
205 * ```
206 *
207 * @resource
208 * @product R2
209 * @category Storage & Databases
210 */
211export const SuperSlurperJob = Resource<SuperSlurperJob>(
212 "Cloudflare.R2.SuperSlurperJob",
213);
214
215export const SuperSlurperJobProvider = () =>
216 Provider.succeed(SuperSlurperJob, {
217 stables: ["accountId", "jobId"],
218 nuke: { skip: true },
219 diff: Effect.fn(function* ({ olds, news, output }) {
220 const { accountId } = yield* yield* CloudflareEnvironment;
221 if (output && output.accountId !== accountId) {
222 return { action: "replace" } as const;
223 }
224 if (
225 !("source" in news) ||
226 !isResolved(news.source) ||
227 !isResolved(news.target) ||
228 !isResolved(news.overwrite)
229 ) {
230 return { action: "replace" } as const;
231 }
232 if (
233 !deepEqual(olds.source, news.source) ||
234 !deepEqual(olds.target, news.target) ||
235 (olds.overwrite ?? false) !== (news.overwrite ?? false)
236 ) {
237 return { action: "replace" } as const;
238 }
239 }),
240 read: Effect.fn(function* ({ output }) {
241 if (!output) return undefined;
242 const observed = yield* observe(output);
243 return observed ? attributes(output, observed) : undefined;
244 }),
245 reconcile: Effect.fn(function* ({ news, output }) {
246 const { accountId } = yield* yield* CloudflareEnvironment;
247 let identity = output;
248 let observed = identity ? yield* observe(identity) : undefined;
249 if (!observed) {
250 const created = yield* r2
251 .createSuperSlurperJob({
252 accountId,
253 source: sourceRequest(news.source),
254 target: {
255 ...news.target,
256 secret: credentialsRequest(news.target.secret),
257 },
258 overwrite: news.overwrite ?? false,
259 })
260 .pipe(
261 Retry.none,
262 // Preconnectivity rejects before creating a job; fresh tokens may not have propagated.
263 Effect.retry({
264 while: (error) =>
265 error._tag === "SuperSlurperPreconnectivityFailed",
266 schedule: Schedule.spaced("3 seconds"),
267 times: 8,
268 }),
269 );
270 if (!created.id) {
271 return yield* Effect.fail(
272 new Error("Super Slurper did not return a job ID"),
273 );
274 }
275 identity = attributes(
276 { accountId, jobId: created.id },
277 { status: "running" },
278 );
279 observed = { id: created.id, status: "running" };
280 }
281 const current = identity!;
282 if (news.paused && observed.status === "running") {
283 yield* mutate(current, "pause");
284 observed = (yield* observe(current)) ?? observed;
285 } else if (!news.paused && observed.status === "paused") {
286 yield* mutate(current, "resume");
287 observed = (yield* observe(current)) ?? observed;
288 }
289 return attributes(current, observed);
290 }),
291 delete: Effect.fn(function* ({ output }) {
292 const observed = yield* observe(output);
293 if (observed?.status === "running" || observed?.status === "paused") {
294 yield* mutate(output, "abort");
295 }
296 }),
297 });
298
299const observe = (identity: { accountId: string; jobId: string }) =>
300 r2
301 .getSuperSlurperJob({
302 accountId: identity.accountId,
303 jobId: identity.jobId,
304 })
305 .pipe(
306 Retry.none,
307 Effect.map((job) => (job.id ? job : undefined)),
308 Effect.catchTag("SuperSlurperJobOperationFailed", () =>
309 findInHistory(identity),
310 ),
311 );
312
313const findInHistory = Effect.fn(function* (identity: {
314 accountId: string;
315 jobId: string;
316}) {
317 const limit = 50;
318 const seen = new Set<string>();
319 for (let offset = 0; ;) {
320 // The SDK's single-page paginator does not follow Slurper's offset parameter.
321 const page = yield* r2
322 .listSuperSlurperJobs({
323 accountId: identity.accountId,
324 limit,
325 offset,
326 })
327 .pipe(Retry.none);
328 const job = page.result.find((item) => item.id === identity.jobId);
329 if (job) return job;
330 if (page.result.length === 0) return undefined;
331 const ids = page.result.flatMap((item) => (item.id ? [item.id] : []));
332 if (ids.length === 0 || ids.every((id) => seen.has(id))) {
333 return yield* Effect.fail(
334 new Error("Super Slurper job history pagination did not advance"),
335 );
336 }
337 for (const id of ids) seen.add(id);
338 offset += page.result.length;
339 }
340});
341
342const mutate = (
343 identity: { accountId: string; jobId: string },
344 action: "pause" | "resume" | "abort",
345) => {
346 const operation =
347 action === "pause"
348 ? r2.pauseSuperSlurperJob
349 : action === "resume"
350 ? r2.resumeSuperSlurperJob
351 : r2.abortSuperSlurperJob;
352 return operation({
353 accountId: identity.accountId,
354 jobId: identity.jobId,
355 }).pipe(
356 Retry.none,
357 Effect.asVoid,
358 Effect.catchTag("SuperSlurperJobOperationFailed", (error) =>
359 Effect.gen(function* () {
360 const current = yield* observe(identity);
361 if (
362 !current ||
363 current.status === "completed" ||
364 current.status === "aborted" ||
365 (action === "pause" && current.status === "paused") ||
366 (action === "resume" && current.status === "running")
367 ) {
368 return;
369 }
370 return yield* Effect.fail(error);
371 }),
372 ),
373 );
374};
375
376const attributes = (
377 identity: { accountId: string; jobId: string },
378 job: r2.GetSuperSlurperJobResponse,
379): SuperSlurperJobAttributes => ({
380 accountId: identity.accountId,
381 jobId: identity.jobId,
382 status: job.status,
383 createdAt: job.createdAt ?? undefined,
384 finishedAt: job.finishedAt ?? undefined,
385});
386
387const credentialsRequest = (secret: SuperSlurperCredentials) => ({
388 accessKeyId: Redacted.value(secret.accessKeyId),
389 secretAccessKey: Redacted.value(secret.secretAccessKey),
390});
391
392const sourceRequest = (
393 source: SuperSlurperSource,
394): r2.SuperSlurperJobsCreateRequestSource =>
395 source.vendor === "gcs"
396 ? {
397 ...source,
398 secret: {
399 clientEmail: Redacted.value(source.secret.clientEmail),
400 privateKey: Redacted.value(source.secret.privateKey),
401 },
402 }
403 : { ...source, secret: credentialsRequest(source.secret) };