| 1 | import { WorkflowEntrypoint, type WorkflowEvent, type WorkflowStep } from "cloudflare:workers"; |
| 2 | import type { Env } from "../env.js"; |
| 3 | import { AUTHORED_PAGE, assembleExport, exportAccount, exportAuthored, exportRepository, failExport, startExport, type ExportJob } from "./export.js"; |
| 4 | import { deleteAccountObjects, destroyRepository, ghostPage, type DeletionCleanup } from "./lifecycle.js"; |
| 5 | |
| 6 | export type AccountJob = ExportJob | DeletionCleanup; |
| 7 | |
| 8 | const RETRY = { retries: { limit: 5, delay: "10 seconds", backoff: "exponential" }, timeout: "15 minutes" } as const; |
| 9 | /** Bundling a large repository, or putting a large archive together, takes longer. */ |
| 10 | const LONG = { ...RETRY, timeout: "60 minutes" } as const; |
| 11 | /** An overall cap on paged steps, so a broken page can't run forever (enough for 100,000 repositories). */ |
| 12 | const MAX_PAGES = 4000; |
| 13 | |
| 14 | /** |
| 15 | * The background part of the account lifecycle (accounts/): building an export (instance `export-<id>`), and |
| 16 | * finishing a deletion (instance `delete-<account id>`): wiping the deleted repositories, the account's own |
| 17 | * R2 objects, and attributing its work in every other repository to the ghost. Each step is retried on |
| 18 | * failure; an export that still fails is marked failed with the reason. |
| 19 | */ |
| 20 | export class AccountWorkflow extends WorkflowEntrypoint<Env, AccountJob> { |
| 21 | override async run(event: WorkflowEvent<AccountJob>, step: WorkflowStep): Promise<unknown> { |
| 22 | const job = event.payload; |
| 23 | if (job.kind === "export") return this.export(job.exportId, step); |
| 24 | return this.cleanup(job, step); |
| 25 | } |
| 26 | |
| 27 | private async export(id: string, step: WorkflowStep): Promise<unknown> { |
| 28 | try { |
| 29 | const plan = await step.do("start", RETRY, () => startExport(this.env, id)); |
| 30 | await step.do("account", RETRY, () => exportAccount(this.env, id)); |
| 31 | for (const repoId of plan.owned) await step.do(`repository ${repoId}`, LONG, () => exportRepository(this.env, id, repoId)); |
| 32 | for (let i = 0; i < plan.visible.length; i += AUTHORED_PAGE) { |
| 33 | const page = plan.visible.slice(i, i + AUTHORED_PAGE); |
| 34 | await step.do(`authored ${i}`, RETRY, () => exportAuthored(this.env, id, page)); |
| 35 | } |
| 36 | await step.do("assemble", LONG, () => assembleExport(this.env, id)); |
| 37 | return { ready: true }; |
| 38 | } catch (e) { |
| 39 | const message = e instanceof Error ? e.message : String(e); |
| 40 | await step.do("fail", RETRY, () => failExport(this.env, id, message)); |
| 41 | return { failed: message }; |
| 42 | } |
| 43 | } |
| 44 | |
| 45 | private async cleanup(job: DeletionCleanup, step: WorkflowStep): Promise<unknown> { |
| 46 | for (const repoId of job.repoIds) await step.do(`destroy ${repoId}`, LONG, () => destroyRepository(this.env, repoId)); |
| 47 | for (let i = 0; i < 10; i++) { |
| 48 | if (await step.do(`objects ${i}`, RETRY, () => deleteAccountObjects(this.env, job.accountId))) break; |
| 49 | } |
| 50 | let after: string | null = ""; |
| 51 | for (let i = 0; after !== null && i < MAX_PAGES; i++) { |
| 52 | const from: string = after; |
| 53 | after = await step.do(`ghost ${i}`, RETRY, () => ghostPage(this.env, job, from)); |
| 54 | } |
| 55 | return { done: true }; |
| 56 | } |
| 57 | } |