cleanup.ts · Source

cleanup.tsapps/worker/src/processors/files/cleanup.tsraw ← files8,618 B← Back to the app
1/**
2 * apps/worker — S3 orphan-object sweeper (TASK-0104 §1/§3, spec file
3 * `apps/worker/src/processors/files/cleanup.ts`).
4 *
5 * A file object can outlive its database rows without a purge (a crashed
6 * finalize, a partial transaction, an externally-deleted row). The sweeper
7 * reconciles the `files/` prefix of the FILES bucket against the
8 * `File` + `FileVersion` ledger:
9 *
10 *   - every object key under `files/{ws}/{fileId}/{v}/{name}` must be a
11 *     live object key: a non-deleted File row's current `File.key`, OR a
12 *     FileVersion key whose File is non-deleted (retained old versions —
13 *     reclaimed by "delete old versions"/purge);
14 *   - any other `files/` object is an ORPHAN;
15 *   - an orphan is deleted only after a GRACE PERIOD (default 7 days — the
16 *     spec's "7-day grace" — the window for a crashed pipeline to finish and
17 *     register its FileVersion row);
18 *   - every deleted object is audited `file.orphan_purged {key, reason}`
19 *     (actor SYSTEM/worker).
20 *
21 * Two entry points share this core:
22 *   - the root-level `scripts/storage-orphans.ts` (`pnpm storage:orphans`)
23 *     — DRY-RUN by default (reports orphans without deleting; `--purge`
24 *     makes it act);
25 *   - the daily `maintenance:orphans` repeatable (grace-aware, deletes).
26 *
27 * Dependency note: this module imports ONLY `@prisma/client` (types) +
28 * `@factory/storage` (S3Client) so the ROOT `scripts/storage-orphans.ts`
29 * can import it (the root has no `pino` hoist — logging is injected).
30 *
31 * Scope note: only the `files/` prefix is swept. Thumbs are derived (a
32 * thumb without a version row is re-generated on next view; a thumb for a
33 * purged file is deleted by the purge itself), so sweeping them would only
34 * add noise.
35 */
36import type { PrismaClient } from "@prisma/client";
37import { S3Client } from "@factory/storage";
38 
39/** Minimal structural logger (pino's Logger satisfies it; the root CLI
40 *  passes a console-backed one). */
41export interface OrphanLogger {
42  info(obj: Record<string, unknown>, msg?: string): void;
43  warn(obj: Record<string, unknown>, msg?: string): void;
44}
45 
46const consoleLogger: OrphanLogger = {
47  info: (obj, msg) => console.log(msg ? `${msg} ${JSON.stringify(obj)}` : JSON.stringify(obj)),
48  warn: (obj, msg) => console.warn(msg ? `${msg} ${JSON.stringify(obj)}` : JSON.stringify(obj)),
49};
50 
51/** Spec's "7-day grace" (docs/tasks/phase-09-files.md TASK-0104 §1). */
52export const ORPHAN_GRACE_DAYS_DEFAULT = 7;
53 
54export interface OrphanSweeperOptions {
55  s3: S3Client;
56  prisma: PrismaClient;
57  /** Grace before an orphan object may be deleted. 0 = delete immediately
58   *  (used by tests). Default 7 days. */
59  graceDays?: number;
60  /** Report only (never delete). Default false (the maintenance run);
61   *  true for the `storage:orphans` CLI. */
62  dryRun?: boolean;
63  /** Bound the object count per run (the run is re-entrant; a bounded batch
64   *  keeps the daily cron short). Default 10_000. */
65  maxObjects?: number;
66  /** Restrict the sweep to one file's objects (key path segment 3 = this
67   *  fileId). Used by the `storage:orphans --file <id>` acceptance check so
68   *  a purge can be verified in isolation from unrelated pre-existing
69   *  detritus in the bucket. */
70  fileId?: string;
71  /** Logger. Defaults to a console logger (the worker passes pino). */
72  log?: OrphanLogger;
73}
74 
75export interface OrphanReport {
76  scannedKeys: number;
77  /** Set when the sweep was scoped to one file (`fileId`). */
78  fileId?: string;
79  orphansFound: number;
80  deleted: number;
81  /** Orphans kept because they are inside the grace period (or because the
82   *  run was a dry-run / the object vanished in the meantime). */
83  retained: number;
84  dryRun: boolean;
85  keys: string[];
86}
87 
88export interface ParsedOrphanKey {
89  workspaceId: string;
90  fileId: string;
91  version: number | null;
92  name: string;
93}
94 
95/**
96 * Parse a files-bucket object key into `{workspaceId, fileId, version, name}`.
97 * The pipeline key is `files/{ws}/{fileId}/{version}/{name}` (buildFileKey),
98 * so a key that does not match (including a missing version segment) is
99 * malformed → orphan.
100 */
101export function parseOrphanKey(key: string): ParsedOrphanKey | null {
102  const m = /^files\/([^/]+)\/([^/]+)\/(\d+)\/(.+)$/.exec(key);
103  if (!m) return null;
104  const version = Number(m[3]);
105  return {
106    workspaceId: m[1]!,
107    fileId: m[2]!,
108    version: Number.isSafeInteger(version) && version >= 1 ? version : null,
109    name: m[4]!,
110  };
111}
112 
113/**
114 * Live ledger: the exact set of object keys that are NOT orphans —
115 *   - each non-deleted File row's CURRENT key (`File.key`), plus
116 *   - each FileVersion key whose File is non-deleted (the retained old
117 *     versions that a replace created and that "delete old versions"/purge
118 *     will eventually reclaim).
119 * (A FileVersion's key always names the same (fileId, version) as the key
120 * parse yields, so exact-key membership is the precise check.)
121 */
122async function loadLiveLedger(prisma: PrismaClient): Promise<Set<string>> {
123  const [files, versions] = await Promise.all([
124    prisma.file.findMany({
125      where: { deletedAt: null },
126      select: { key: true },
127    }),
128    prisma.fileVersion.findMany({
129      where: { file: { deletedAt: null } },
130      select: { key: true },
131    }),
132  ]);
133  const live = new Set<string>();
134  for (const f of files) live.add(f.key);
135  for (const v of versions) live.add(v.key);
136  return live;
137}
138 
139export async function runOrphanSweep(opts: OrphanSweeperOptions): Promise<OrphanReport> {
140  const { s3, prisma } = opts;
141  const log = opts.log ?? consoleLogger;
142  const graceDays = opts.graceDays ?? ORPHAN_GRACE_DAYS_DEFAULT;
143  const dryRun = opts.dryRun ?? false;
144  const maxObjects = opts.maxObjects ?? 10_000;
145  const t0 = Date.now();
146 
147  // 1. List ALL objects under `files/` (paginated past the 1000-key cap).
148  const keys: string[] = [];
149  let token: string | undefined;
150  do {
151    const page = await s3.listObjects("files/", 1000, token);
152    keys.push(...page.keys);
153    token = page.truncated ? page.nextToken : undefined;
154    if (keys.length >= maxObjects) break;
155  } while (token);
156 
157  // 2. Load the live ledger.
158  const live = await loadLiveLedger(prisma);
159 
160  // 3. Classify each key; delete orphans past grace.
161  const now = Date.now();
162  const graceMs = graceDays * 24 * 3600 * 1000;
163  const report: OrphanReport = {
164    scannedKeys: keys.length,
165    orphansFound: 0,
166    deleted: 0,
167    retained: 0,
168    dryRun,
169    keys: [],
170  };
171  if (opts.fileId !== undefined) report.fileId = opts.fileId;
172 
173  for (const key of keys.slice(0, maxObjects)) {
174    const parsed = parseOrphanKey(key);
175    // Scoped sweep (--file): ignore every object that is not this file's.
176    if (opts.fileId && parsed?.fileId !== opts.fileId) continue;
177    let reason: string;
178    if (!parsed) {
179      reason = "malformed_key";
180    } else if (live.has(key)) {
181      continue; // the key maps to a live file/version → NOT an orphan
182    } else {
183      reason = "no_ledger_row";
184    }
185 
186    report.orphansFound += 1;
187    report.keys.push(key);
188 
189    if (dryRun) {
190      report.retained += 1;
191      continue;
192    }
193 
194    // Grace anchor = the object's Last-Modified (a fresh orphan — a crashed
195    // finalize about to land its FileVersion row — is NOT deleted).
196    let lastModified = 0;
197    try {
198      const head = await s3.headObject(key);
199      lastModified = head?.lastModified ? new Date(head.lastModified).getTime() : 0;
200    } catch {
201      // Vanished in the meantime — nothing to do.
202      report.retained += 1;
203      continue;
204    }
205    if (graceMs > 0 && now - lastModified < graceMs) {
206      report.retained += 1;
207      continue;
208    }
209 
210    await s3.deleteObject(key);
211    report.deleted += 1;
212    const ws = parsed?.workspaceId;
213    const fileId = parsed?.fileId;
214    log.warn(
215      { key, reason, ws, fileId, ageDays: lastModified ? Math.floor((now - lastModified) / 86_400_000) : 0 },
216      "orphan object purged"
217    );
218    // Audit (best-effort — the object delete is already done; a failed
219    // audit must not re-raise and lose the run's other deletes).
220    try {
221      await prisma.auditLog.create({
222        data: {
223          actorType: "SYSTEM",
224          actorId: "worker",
225          workspaceId: ws ?? null,
226          resourceType: "file",
227          resourceId: fileId ?? "unknown",
228          action: "file.orphan_purged",
229          metadata: { key, reason },
230        },
231      });
232    } catch (err) {
233      log.warn({ err: (err as Error).message, key }, "orphan purge audit failed (non-fatal)");
234    }
235  }
236 
237  log.info({ ...report, ms: Date.now() - t0 }, "orphan sweep done");
238  return report;
239}
240