retention.ts · Source

retention.tsapps/worker/src/processors/retention.tsraw ← processors22,457 B← Back to the app
1/**
2 * apps/worker — the ORG-SCOPED RETENTION core (TASK-0133).
3 *
4 * Single implementation of the purge engine, shared by:
5 *   - the DAILY job (`retention.run`, 04:00 UTC, all orgs, mode from the
6 *     `WORKER_RETENTION_EXECUTE` gate — dry by default), and
7 *   - the ADMIN "Run now / Run dry" trigger (the API enqueues the SAME job
8 *     with `{orgId, scope, mode, runId, actor}`; the result lands in the
9 *     Redis report `worker:retention:report:{runId}` the console polls).
10 *
11 * Semantics (spec `docs/tasks/phase-14-enterprise.md` TASK-0133):
12 *   - one `RetentionPolicy` per (org, scope); defaults: trash PURGE 30 d,
13 *     comments PURGE 90 d, audit_logs HOLD (never purge while HOLD), files
14 *     PURGE 0 d (0 = no purge).
15 *   - PURGE → batched transactions (100 ids/batch, each batch its own
16 *     `retention.purged {scope, counts}` audit row).
17 *   - HOLD → skipped + `retention.hold_skipped` audit (audit_logs HOLD is the
18 *     "legal hold" requirement — audit rows are NEVER purged while it is on).
19 *   - per-scope isolation: a throw in one scope does not stop the others
20 *     (per-scope try/catch + `retention.failed` audit for the broken one).
21 *   - org isolation: every query is scoped to the org's workspaces (+ the
22 *     shared org-scope OR-branch for audit rows — `@factory/shared/events/
23 *     audit-scope`, the SAME builder the API viewer uses).
24 *
25 * The old global `runRetention` sweep (maintenance.processor) is REPLACED
26 * (recorded conflict: it had no org isolation — a global 30 d trash purge —
27 * and a global 365 d audit purge; the spec's per-org policies supersede it.
28 * The function remains exported for the deprecation window).
29 *
30 * Recorded decisions:
31 *  - `Page.trashedAt` is the trash clock (the spec text says `Page.deletedAt`;
32 *    the live trash path + existing sweep key off `trashedAt`).
33 *  - `retention.purged` counts shape is the spec's `{pages, blocks, comments,
34 *    files, bytesFreed}` — the audit_logs scope has no dedicated slot and
35 *    reuses `comments` (the scope field disambiguates; documented).
36 *  - files: `bytesFreed` sums the deleted rows' `File.size` (current version).
37 */
38import type { PrismaClient, Prisma } from "@prisma/client";
39import type { Redis } from "ioredis";
40import { asJson } from "../lib/prisma.js";
41import { getLogger } from "../logger.js";
42import {
43  auditScopedWhere,
44  type AuditFilters
45} from "@factory/shared/events/audit-scope";
46import { planPageVersionPrune, versionRetentionFromSettings } from "@factory/shared";
47 
48const log = getLogger("retention");
49 
50export type RetentionScope = "trash" | "comments" | "audit_logs" | "files";
51export const RETENTION_SCOPES: RetentionScope[] = ["trash", "comments", "audit_logs", "files"];
52 
53/** The spec defaults (applied when the org has no policy row for a scope). */
54export const RETENTION_DEFAULTS: Record<RetentionScope, { type: "PURGE" | "HOLD"; retainDays: number }> = {
55  trash: { type: "PURGE", retainDays: 30 },
56  comments: { type: "PURGE", retainDays: 90 },
57  audit_logs: { type: "HOLD", retainDays: 365 },
58  files: { type: "PURGE", retainDays: 0 }
59};
60 
61export const RETENTION_BATCH_SIZE = 100;
62 
63export interface ScopeCounts {
64  pages: number;
65  blocks: number;
66  comments: number;
67  files: number;
68  bytesFreed: number;
69}
70 
71export interface RetentionActor {
72  type: "USER" | "SYSTEM";
73  id: string;
74}
75 
76export interface RetentionRunOptions {
77  prisma: PrismaClient;
78  redis?: Redis;
79  orgId: string;
80  scope: RetentionScope;
81  mode: "dry" | "execute";
82  /** The triggering admin (admin runs) or the worker identity (cron). */
83  actor: RetentionActor;
84  /** When set, the run report is written here (the console polls it). */
85  runId?: string;
86}
87 
88interface ScopePolicy {
89  type: "PURGE" | "HOLD";
90  retainDays: number;
91  config: Record<string, unknown>;
92}
93 
94async function orgWorkspaceIds(prisma: PrismaClient, orgId: string): Promise<string[]> {
95  const rows = await prisma.workspace.findMany({ where: { organizationId: orgId }, select: { id: true } });
96  return rows.map((r) => r.id);
97}
98 
99/** The effective policy for (org, scope) — the row, or the spec default. */
100export async function policyFor(
101  prisma: PrismaClient,
102  orgId: string,
103  scope: RetentionScope
104): Promise<ScopePolicy> {
105  const row = await prisma.retentionPolicy.findUnique({
106    where: { organizationId_scope: { organizationId: orgId, scope } }
107  });
108  const def = RETENTION_DEFAULTS[scope];
109  if (!row) return { ...def, config: {} };
110  return {
111    type: row.type,
112    retainDays: row.retainDays ?? def.retainDays,
113    config: (row.config as Record<string, unknown> | null) ?? {}
114  };
115}
116 
117/** Persist `config.lastRun`/`lastDryRun` on the policy row (best-effort). */
118async function touchLastRun(
119  prisma: PrismaClient,
120  orgId: string,
121  scope: RetentionScope,
122  mode: "dry" | "execute",
123  summary: Record<string, unknown>
124): Promise<void> {
125  const key = mode === "execute" ? "lastRun" : "lastDryRun";
126  const value = { at: new Date().toISOString(), ...summary };
127  const existing = await prisma.retentionPolicy.findUnique({
128    where: { organizationId_scope: { organizationId: orgId, scope } }
129  });
130  try {
131    if (existing) {
132      const cfg = (existing.config as Record<string, unknown> | null) ?? {};
133      await prisma.retentionPolicy.update({
134        where: { id: existing.id },
135        data: { config: asJson({ ...cfg, [key]: value }) }
136      });
137    } else {
138      await prisma.retentionPolicy.create({
139        data: {
140          organizationId: orgId,
141          scope,
142          type: RETENTION_DEFAULTS[scope].type,
143          retainDays: RETENTION_DEFAULTS[scope].retainDays,
144          config: asJson({ [key]: value })
145        }
146      });
147    }
148  } catch (err) {
149    log.warn({ err: (err as Error).message, orgId, scope }, "retention lastRun update failed");
150  }
151}
152 
153async function writeAudit(
154  prisma: PrismaClient,
155  actor: RetentionActor,
156  orgId: string,
157  action: string,
158  metadata: Record<string, unknown>
159): Promise<void> {
160  await prisma.auditLog.create({
161    data: {
162      actorType: actor.type,
163      actorId: actor.id,
164      workspaceId: null,
165      resourceType: "organization",
166      resourceId: orgId,
167      action,
168      metadata: asJson({ orgId, ...metadata })
169    }
170  });
171}
172 
173/** The admin-console report (the poll endpoint reads it; TTL 7 d). */
174async function writeRunReport(
175  redis: Redis | undefined,
176  runId: string | undefined,
177  orgId: string,
178  report: Record<string, unknown>
179): Promise<void> {
180  if (!redis || !runId) return;
181  await redis
182    .set(`worker:retention:report:${runId}`, JSON.stringify({ orgId, ...report }), "EX", 7 * 86_400)
183    .catch((err) => log.warn({ err: (err as Error).message, runId }, "retention report write failed"));
184}
185 
186// ─────────────────────────────── purge cores ────────────────────────────────
187 
188/**
189 * trash — pages with `trashedAt < cutoff` in the org's workspaces.
190 * Cascade map (schema): Block/PageFavorite/PageVersion/Comment/Share CASCADE
191 * on pageId; DatabaseRecord.pageId is SET-NULL (explicit deleteMany, mirroring
192 * the page service's `deletePermanent`); File.pageId is a plain ref (the files
193 * scope owns file rows). Blocks are deleted explicitly so the audit counts
194 * them (the cascade would also work, but counting the cascade is not
195 * deterministic).
196 */
197async function purgeTrash(
198  prisma: PrismaClient,
199  orgId: string,
200  wsIds: string[],
201  cutoff: Date,
202  execute: boolean,
203  actor: RetentionActor
204): Promise<ScopeCounts> {
205  const where: Prisma.PageWhereInput = {
206    workspaceId: { in: wsIds },
207    trashedAt: { lt: cutoff }
208  };
209  const ids = await prisma.page.findMany({ where, select: { id: true } });
210  const counts: ScopeCounts = { pages: 0, blocks: 0, comments: 0, files: 0, bytesFreed: 0 };
211  if (execute) {
212    for (let i = 0; i < ids.length; i += RETENTION_BATCH_SIZE) {
213      const batch = ids.slice(i, i + RETENTION_BATCH_SIZE).map((p) => p.id);
214      const blockCount = await prisma.$transaction(async (tx) => {
215        // Explicit deletes (children → parent). BlockVersion CASCADEs on
216        // blockId; the rest CASCADE on pageId (Block/Comment/Share/Favorite/
217        // PageVersion). DatabaseRecord.pageId is SET-NULL → explicit delete.
218        await tx.databaseRecord.deleteMany({ where: { pageId: { in: batch } } });
219        const blocks = await tx.block.deleteMany({ where: { pageId: { in: batch } } });
220        await tx.page.deleteMany({ where: { id: { in: batch } } });
221        return blocks.count;
222      });
223      counts.pages += batch.length;
224      counts.blocks += blockCount;
225      await writeAudit(prisma, actor, orgId, "retention.purged", {
226        scope: "trash",
227        counts: { pages: batch.length, blocks: blockCount }
228      });
229    }
230  }
231  // Dry-run: report the would-be page count (blocks are not enumerable
232  // without deleting — the UI shows the page count).
233  return { ...counts, pages: execute ? counts.pages : ids.length };
234}
235 
236/**
237 * comments — root comments older than the cutoff with NO reply newer than
238 * the cutoff (a recent reply keeps the whole thread). Replies + reactions of
239 * purged threads go with them.
240 */
241async function purgeComments(
242  prisma: PrismaClient,
243  orgId: string,
244  wsIds: string[],
245  cutoff: Date,
246  execute: boolean,
247  actor: RetentionActor
248): Promise<ScopeCounts> {
249  const candidates = await prisma.comment.findMany({
250    where: { workspaceId: { in: wsIds }, createdAt: { lt: cutoff } },
251    select: { id: true }
252  });
253  const ids = candidates.map((c) => c.id);
254  // A thread survives when ANY reply is newer than the cutoff.
255  let survivors: string[] = [];
256  if (ids.length) {
257    const recent = await prisma.commentReply.findMany({
258      where: { commentId: { in: ids }, createdAt: { gt: cutoff } },
259      select: { commentId: true },
260      distinct: ["commentId"]
261    });
262    survivors = recent.map((r) => r.commentId);
263  }
264  const toPurge = ids.filter((id) => !survivors.includes(id));
265  const counts: ScopeCounts = { pages: 0, blocks: 0, comments: 0, files: 0, bytesFreed: 0 };
266  if (execute) {
267    for (let i = 0; i < toPurge.length; i += RETENTION_BATCH_SIZE) {
268      const batch = toPurge.slice(i, i + RETENTION_BATCH_SIZE);
269      await prisma.$transaction(async (tx) => {
270        // Reactions cascade on reply; delete reactions (comment + replies) →
271        // replies → comments explicitly (countable, deterministic).
272        await tx.commentReaction.deleteMany({ where: { commentId: { in: batch } } });
273        const replyRes = await tx.commentReply.deleteMany({ where: { commentId: { in: batch } } });
274        await tx.comment.deleteMany({ where: { id: { in: batch } } });
275        void replyRes;
276      });
277      counts.comments += batch.length;
278      await writeAudit(prisma, actor, orgId, "retention.purged", {
279        scope: "comments",
280        counts: { comments: batch.length }
281      });
282    }
283  }
284  // Dry-run: report the would-be count (root comments minus kept threads).
285  return { ...counts, comments: execute ? counts.comments : toPurge.length };
286}
287 
288/**
289 * audit_logs — the shared org-scope where (the SAME builder as the viewer)
290 * ∧ `createdAt < cutoff`. HOLD → the caller skips (hold_skipped). The counts
291 * slot reuses `comments` (the spec shape has no audit slot — recorded).
292 */
293async function purgeAuditLogs(
294  prisma: PrismaClient,
295  orgId: string,
296  wsIds: string[],
297  cutoff: Date,
298  execute: boolean,
299  actor: RetentionActor
300): Promise<ScopeCounts> {
301  const filters: AuditFilters = { to: cutoff };
302  const where = auditScopedWhere(orgId, wsIds, filters) as Prisma.AuditLogWhereInput;
303  const total = await prisma.auditLog.count({ where });
304  const counts: ScopeCounts = { pages: 0, blocks: 0, comments: 0, files: 0, bytesFreed: 0 };
305  if (execute) {
306    let remaining = total;
307    while (remaining > 0) {
308      const ids = await prisma.auditLog.findMany({ where, select: { id: true }, take: RETENTION_BATCH_SIZE });
309      if (!ids.length) break;
310      const res = await prisma.auditLog.deleteMany({ where: { id: { in: ids.map((r) => r.id) } } });
311      counts.comments += res.count; // recorded slot reuse
312      remaining -= res.count;
313      await writeAudit(prisma, actor, orgId, "retention.purged", {
314        scope: "audit_logs",
315        counts: { auditLogs: res.count }
316      });
317    }
318  }
319  // Dry-run: report the would-be row count.
320  return { ...counts, comments: execute ? counts.comments : total };
321}
322 
323/**
324 * files — `File.deletedAt < cutoff` in the org's workspaces. Versions go
325 * with the file; `bytesFreed` sums the deleted rows' current `size`.
326 * (S3 object deletion is the files lifecycle policy's job, as in the old
327 * sweep — the DB rows + accounting are what retention owns here.)
328 */
329async function purgeFiles(
330  prisma: PrismaClient,
331  orgId: string,
332  wsIds: string[],
333  cutoff: Date,
334  execute: boolean,
335  actor: RetentionActor
336): Promise<ScopeCounts> {
337  const where: Prisma.FileWhereInput = {
338    workspaceId: { in: wsIds },
339    deletedAt: { lt: cutoff }
340  };
341  const rows = await prisma.file.findMany({ where, select: { id: true, size: true } });
342  const counts: ScopeCounts = { pages: 0, blocks: 0, comments: 0, files: 0, bytesFreed: 0 };
343  if (execute) {
344    for (let i = 0; i < rows.length; i += RETENTION_BATCH_SIZE) {
345      const batch = rows.slice(i, i + RETENTION_BATCH_SIZE);
346      const bytes = batch.reduce((s, f) => s + (f.size ?? 0), 0);
347      await prisma.$transaction(async (tx) => {
348        await tx.fileVersion.deleteMany({ where: { fileId: { in: batch.map((f) => f.id) } } });
349        await tx.file.deleteMany({ where: { id: { in: batch.map((f) => f.id) } } });
350      });
351      counts.files += batch.length;
352      counts.bytesFreed += bytes;
353      await writeAudit(prisma, actor, orgId, "retention.purged", {
354        scope: "files",
355        counts: { files: batch.length, bytesFreed: bytes }
356      });
357    }
358  }
359  // Dry-run: report the would-be count + the bytes that would be freed.
360  return {
361    ...counts,
362    files: execute ? counts.files : rows.length,
363    bytesFreed: execute ? counts.bytesFreed : rows.reduce((s, f) => s + (f.size ?? 0), 0)
364  };
365}
366 
367// ─────────────────────────────── run entry ──────────────────────────────────
368 
369/**
370 * Run ONE (org, scope) in the given mode. Returns the report object (the
371 * admin poll endpoint + the cron log both consume it).
372 */
373export async function runRetentionScope(args: RetentionRunOptions): Promise<Record<string, unknown>> {
374  const { prisma, redis, orgId, scope, mode, actor, runId } = args;
375  const execute = mode === "execute";
376  const t0 = Date.now();
377  const policy = await policyFor(prisma, orgId, scope);
378  const base = {
379    runId: runId ?? null,
380    orgId,
381    scope,
382    mode,
383    type: policy.type,
384    retainDays: policy.retainDays,
385    startedAt: new Date().toISOString()
386  };
387 
388  // HOLD (or files retainDays 0 = no-purge) → skip.
389  if (policy.type === "HOLD" || (scope === "files" && policy.retainDays <= 0)) {
390    await writeAudit(prisma, actor, orgId, "retention.hold_skipped", {
391      scope,
392      reason: policy.type === "HOLD" ? "hold" : "no_purge"
393    });
394    const report = { ...base, status: "skipped", counts: null, reason: policy.type === "HOLD" ? "hold" : "no_purge" };
395    await writeRunReport(redis, runId, orgId, report);
396    log.info({ ...report }, "retention scope skipped (hold)");
397    return report;
398  }
399 
400  const cutoff = new Date(Date.now() - Math.max(policy.retainDays, 0) * 86_400_000);
401  const wsIds = await orgWorkspaceIds(prisma, orgId);
402 
403  let counts: ScopeCounts;
404  if (scope === "trash") counts = await purgeTrash(prisma, orgId, wsIds, cutoff, execute, actor);
405  else if (scope === "comments") counts = await purgeComments(prisma, orgId, wsIds, cutoff, execute, actor);
406  else if (scope === "audit_logs") counts = await purgeAuditLogs(prisma, orgId, wsIds, cutoff, execute, actor);
407  else counts = await purgeFiles(prisma, orgId, wsIds, cutoff, execute, actor);
408 
409  const report = {
410    ...base,
411    status: execute ? "purged" : "dry_run",
412    cutoff: cutoff.toISOString(),
413    counts,
414    durationMs: Date.now() - t0
415  };
416  if (!execute) {
417    // The spec's dry-run audit: `retention.dry_run {scope, count, oldest,
418    // newest}` — `count` is the primary items for the scope (pages for
419    // trash, comments for comments, audit rows for audit_logs, files for
420    // files). oldest/newest are the boundary timestamps of the purge set.
421    const count =
422      scope === "trash"
423        ? counts.pages
424        : scope === "comments"
425          ? counts.comments
426          : scope === "files"
427            ? counts.files
428            : counts.comments; // the audit_logs slot reuse (recorded)
429    await writeAudit(prisma, actor, orgId, "retention.dry_run", { scope, count, counts });
430  }
431  await touchLastRun(prisma, orgId, scope, mode, { counts });
432  await writeRunReport(redis, runId, orgId, report);
433  log.info({ ...report }, "retention scope done");
434  return report;
435}
436 
437/**
438 * Run ALL scopes for an org (the daily 04:00 job, per org). Per-scope
439 * isolation: a throw in one scope → `retention.failed` audit + the others
440 * still run. Returns the per-scope report map.
441 */
442export async function runOrgRetention(
443  args: Omit<RetentionRunOptions, "scope"> & { scopes?: RetentionScope[] }
444): Promise<Record<string, Record<string, unknown>>> {
445  const { prisma, redis, orgId, mode, actor, runId, scopes } = args;
446  const out: Record<string, Record<string, unknown>> = {};
447  for (const scope of scopes ?? RETENTION_SCOPES) {
448    try {
449      out[scope] = await runRetentionScope({ ...args, scope });
450    } catch (err) {
451      const msg = (err as Error).message;
452      log.error({ err: msg, orgId, scope }, "retention scope failed");
453      await prisma.auditLog
454        .create({
455          data: {
456            actorType: actor.type,
457            actorId: actor.id,
458            workspaceId: null,
459            resourceType: "organization",
460            resourceId: orgId,
461            action: "retention.failed",
462            metadata: asJson({ orgId, scope, error: msg })
463          }
464        })
465        .catch(() => undefined);
466      out[scope] = { orgId, scope, mode, status: "failed", error: msg };
467      // The admin poll must not see a forever-"processing" report.
468      await writeRunReport(redis, runId, orgId, { orgId, scope, mode, status: "failed", error: msg });
469    }
470  }
471  return out;
472}
473 
474/** The daily job body: every org, per-org policies, mode from the gate. */
475export async function runRetentionDaily(
476  prisma: PrismaClient,
477  redis: Redis,
478  env: { WORKER_RETENTION_EXECUTE: boolean }
479): Promise<Record<string, unknown>> {
480  const mode = env.WORKER_RETENTION_EXECUTE ? "execute" : "dry";
481  const orgs = await prisma.organization.findMany({ select: { id: true } });
482  const perOrg: Record<string, Record<string, unknown>> = {};
483  for (const org of orgs) {
484    perOrg[org.id] = await runOrgRetention({
485      prisma,
486      redis,
487      orgId: org.id,
488      mode,
489      actor: { type: "SYSTEM", id: "worker:maintenance" }
490    });
491  }
492  log.info({ orgs: orgs.length, mode }, "retention daily sweep done");
493  return { orgs: orgs.length, mode, perOrg };
494}
495 
496 
497// ─────────────────────── page-version prune (T-152) ────────────────────────
498//
499// Per-WORKSPACE revision retention (distinct from the org RetentionPolicy
500// engine above, which owns trash/comments/audit/files): prune old PageVersion
501// snapshots per page. Policy lives in `workspace.settings.versions`
502// ({ maxVersions, keepRecentDays }) — parsed by the shared
503// `versionRetentionFromSettings` (defaults: keep 100 newest + 7 d grace).
504// Idempotent: a second run finds nothing left to prune.
505 
506export interface PageVersionPruneResult {
507  workspaceId: string;
508  pages: number;
509  versions: number;
510  maxVersions: number;
511  keepRecentDays: number;
512}
513 
514/**
515 * Prune page versions for ONE workspace per its settings policy.
516 * Writes one audit row (`page.versions_pruned`) when anything was removed.
517 */
518export async function pruneWorkspacePageVersions(
519  prisma: PrismaClient,
520  workspaceId: string,
521  actor: RetentionActor
522): Promise<PageVersionPruneResult> {
523  const ws = await prisma.workspace.findUnique({ where: { id: workspaceId }, select: { settings: true } });
524  if (!ws) return { workspaceId, pages: 0, versions: 0, maxVersions: 0, keepRecentDays: 0 };
525  const policy = versionRetentionFromSettings(ws.settings);
526  const pages = await prisma.page.findMany({ where: { workspaceId }, select: { id: true } });
527  let prunedPages = 0;
528  let prunedVersions = 0;
529  for (const page of pages) {
530    const rows = await prisma.pageVersion.findMany({
531      where: { pageId: page.id },
532      select: { id: true, version: true, createdAt: true }
533    });
534    const toPrune = planPageVersionPrune(rows, policy);
535    if (toPrune.length === 0) continue;
536    const res = await prisma.pageVersion.deleteMany({ where: { pageId: page.id, id: { in: toPrune } } });
537    prunedPages++;
538    prunedVersions += res.count;
539  }
540  if (prunedVersions > 0) {
541    await prisma.auditLog.create({
542      data: {
543        actorType: actor.type,
544        actorId: actor.id,
545        workspaceId,
546        resourceType: "page",
547        resourceId: workspaceId,
548        action: "page.versions_pruned",
549        metadata: asJson({ pages: prunedPages, versions: prunedVersions, ...policy })
550      }
551    });
552  }
553  log.info({ workspaceId, pages: prunedPages, versions: prunedVersions }, "page-version prune done");
554  return { workspaceId, pages: prunedPages, versions: prunedVersions, maxVersions: policy.maxVersions, keepRecentDays: policy.keepRecentDays };
555}
556 
557/** Daily sweep helper: every workspace (the `retention` repeatable calls it
558 *  after the org sweep — same 04:00 run, one extra pass). */
559export async function pruneAllPageVersions(
560  prisma: PrismaClient,
561  actor: RetentionActor
562): Promise<{ workspaces: number; pages: number; versions: number }> {
563  const wss = await prisma.workspace.findMany({ select: { id: true } });
564  let pages = 0;
565  let versions = 0;
566  for (const ws of wss) {
567    const r = await pruneWorkspacePageVersions(prisma, ws.id, actor);
568    pages += r.pages;
569    versions += r.versions;
570  }
571  return { workspaces: wss.length, pages, versions };
572}
573