1/**
2 * The delete-purge alarm job (DESIGN.md §2.3 Delete).
3 *
4 * `DELETE /repos/:o/:r` responds 204 immediately; this job then runs from
5 * the Repo DO's alarm until everything is gone:
6 *
7 * 1. **Registry row** — removed FIRST, freeing the `owner/name` for reuse
8 * right away (the fork pin below never reserves a name).
9 * 2. **R2 prefix** — a bounded list+delete loop over `{repoId}/`, but only
10 * while no live fork rows reference this repo (`fork_of` count): forks
11 * reference keys under the parent's prefix by full key (immutable,
12 * shared), so a forked repo's prefix is retained until its forks are
13 * gone.
14 * 3. **SQLite** — dropped via `deleteAll` once the pin clears and the
15 * prefix is drained.
16 *
17 * The job is idempotent and alarm-re-armable: every step tolerates having
18 * already run. A single alarm run deletes at most
19 * {@link MAX_PAGES_PER_RUN} × 1000 R2 objects and reports `"continue"` so
20 * the caller re-arms the alarm instead of blowing the 15-minute budget.
21 */
22import type { BlobStoreError, BlobStoreShape } from "../BlobStore.ts";
23import { RuntimeContext } from "../../RuntimeContext.ts";
24import * as Effect from "effect/Effect";
25import * as Stream from "effect/Stream";
26import { StoreError } from "../Protocol/Store.ts";
27import { repoPrefix } from "../Store/Keys.ts";
28
29/** R2 `list` pages drained per alarm run (1000 keys per page). */
30export const MAX_PAGES_PER_RUN = 10;
31
32/** Outcome of one purge run. */
33export type PurgeOutcome =
34 | {
35 /** Everything is gone (R2 prefix, SQLite, Registry row). */
36 readonly _tag: "done";
37 }
38 | {
39 /**
40 * More R2 keys remain (or the prefix is fork-pinned); the caller
41 * must re-arm the alarm and run again.
42 */
43 readonly _tag: "continue";
44 /** True when the prefix was retained because `fork_count > 0`. */
45 readonly forkPinned: boolean;
46 };
47
48/** Dependencies of {@link runPurgeJob}. */
49export interface PurgeJobOptions {
50 /** The repo's ULID (R2 key prefix). */
51 readonly repoId: string;
52 /** The swappable bulk-byte store. */
53 readonly blobs: BlobStoreShape;
54 /** Reads the current fork count from the Registry. */
55 readonly forkCount: Effect.Effect<number, StoreError>;
56 /** Drops the DO's entire SQLite/KV state (`storage.deleteAll`). */
57 readonly deleteAllStorage: Effect.Effect<void, StoreError>;
58 /** Removes the Registry row (FIRST step — frees the name; idempotent). */
59 readonly removeRegistryRow: Effect.Effect<void, StoreError>;
60}
61
62/** Folds R2 errors into the job's typed error. */
63const r2ToStore =
64 (what: string) =>
65 <A>(
66 effect: Effect.Effect<A, BlobStoreError, RuntimeContext>,
67 ): Effect.Effect<A, StoreError> =>
68 effect.pipe(
69 Effect.mapError(
70 (error) => new StoreError({ reason: `${what}: ${error.reason}` }),
71 ),
72 Effect.provide(RuntimeContext.phantom),
73 );
74
75/**
76 * Runs one bounded purge round. The registry row is removed FIRST — the
77 * `owner/name` frees immediately, even while forks pin the R2 prefix.
78 * Returns `"done"` when the repo is fully gone; `"continue"` when the
79 * caller must re-arm the alarm (more R2 keys, or the prefix is pinned by
80 * live forks — the DO's SQLite survives until the pin clears so the alarm
81 * keeps re-arming with the repoId at hand).
82 */
83export const runPurgeJob = (
84 options: PurgeJobOptions,
85): Effect.Effect<PurgeOutcome, StoreError> =>
86 Effect.gen(function* () {
87 const prefix = repoPrefix(options.repoId);
88
89 // 1. Free the name IMMEDIATELY — even while fork-pinned. The pin only
90 // protects the R2 prefix (forks read shared keys under it); the
91 // `owner/name` must never stay reserved for the lifetime of a fork.
92 // `forkCount` derives from live `fork_of` rows, so it keeps working
93 // after this row is gone. Idempotent across alarm retries.
94 yield* options.removeRegistryRow;
95
96 const forks = yield* options.forkCount;
97 if (forks > 0) {
98 // Fork-retention pin: leave the R2 prefix in place and keep the DO's
99 // SQLite (config carries the repoId; jobs row keeps re-arming) until
100 // the last fork is purged.
101 return { _tag: "continue", forkPinned: true } satisfies PurgeOutcome;
102 }
103
104 // 2. Bounded R2 prefix drain.
105 // Bounded drain: take at most MAX_PAGES_PER_RUN pages worth of keys
106 // from the listing stream, delete in 1000-key batches, and treat a
107 // short take as fully drained.
108 const cap = MAX_PAGES_PER_RUN * 1000;
109 const keys = yield* r2ToStore(`blob list ${prefix}`)(
110 Stream.runCollect(
111 options.blobs.list(prefix).pipe(
112 Stream.take(cap),
113 Stream.map((meta) => meta.key),
114 ),
115 ),
116 );
117 for (let at = 0; at < keys.length; at += 1000) {
118 yield* r2ToStore(`blob delete under ${prefix}`)(
119 options.blobs.delete(keys.slice(at, at + 1000)),
120 );
121 }
122 const drained = keys.length < cap;
123 if (!drained) {
124 return { _tag: "continue", forkPinned: false } satisfies PurgeOutcome;
125 }
126
127 // 3. Drop the DO's own state (the registry row is already gone — see
128 // step 1; a crash that loses this `deleteAll` is harmless because a
129 // re-created `owner/name` mints a fresh repoId and therefore a fresh
130 // DO, so the stale storage is never addressed again). NOTE: this
131 // wipes the `jobs` table — the caller must not touch SQLite after
132 // this point.
133 yield* options.deleteAllStorage;
134
135 return { _tag: "done" } satisfies PurgeOutcome;
136 });