maintenance.processor.ts

maintenance.processor.tsapps/worker/src/processors/maintenance.processor.tsraw ← processors15,423 B← Back to the app
1/**
2 * apps/worker — maintenance queue processor (scheduled housekeeping).
3 *
4 * Repeatable jobs (registered in index.ts via BullMQ repeat):
5 *   retention  — daily 04:00  (id: maintenance:retention) → per-org policies
6 *   reconcile  — weekly 02:30 (id: maintenance:reconcile)  → search drift check
7 *   dlq-report — hourly      (id: maintenance:webhook-dlq) → stale failed deliveries
8 *   cleanup    — hourly      (id: maintenance:cleanup)     → expired sessions/tokens/invites/keys
9 *   resync     — ad-hoc      ({type:"workspace-resync", workspaceId}) from permission events
10 * Ad-hoc jobs:
11 *   retention.run — {orgId, scope, mode, runId, actor} (the org-admin console
12 *                   "Run now / Run dry" — the SAME core as the daily job)
13 *   audit.export  — {orgId, exportId} (the >10k-row async CSV export)
14 *
15 * All jobs are idempotent (each op is a count/compare or a scoped delete)
16 * and log their timing + counts.
17 */
18import { Worker, type Job } from "bullmq";
19import { withJobOtel } from "../lib/job-otel.js";
20 
21import type { Redis } from "ioredis";
22import { z } from "zod";
23import type { PrismaClient } from "@prisma/client";
24import { QUEUES, enqueue } from "../queues.js";
25import { getLogger } from "../logger.js";
26import type { WorkerEnv } from "../config.js";
27import { notifyWorkspaceAdmins } from "../lib/notify.js";
28import { runDigestRun } from "./email-decision.processor.js";
29import { S3Client } from "@factory/storage";
30import { runOrphanSweep } from "./files/cleanup.js";
31import { runAuditExport } from "./audit-export.js";
32import { runRetentionScope, runRetentionDaily, pruneAllPageVersions, pruneWorkspacePageVersions } from "./retention.js";
33import { runTaskDueSweep, runReviewDueSweep } from "./tasks.js";
34 
35const log = getLogger("maintenance");
36 
37const INVITE_EXPIRY_DAYS = 7;
38 
39export const maintenanceJobSchema = z.union([
40  z.object({ type: z.literal("retention") }),
41  // TASK-0133: the org-admin "Run now / Run dry" (one scope, the triggering
42  // admin carried in the job — audited as the admin, not the worker).
43  z.object({
44    type: z.literal("retention.run"),
45    orgId: z.string().min(1),
46    scope: z.enum(["trash", "comments", "audit_logs", "files"]),
47    mode: z.enum(["dry", "execute"]),
48    runId: z.string().min(1),
49    actor: z.object({ type: z.enum(["USER", "SYSTEM"]), id: z.string().min(1) })
50  }),
51  // TASK-0133: the >10k-row async CSV export (enqueued by the API).
52  z.object({
53    type: z.literal("audit.export"),
54    orgId: z.string().min(1),
55    exportId: z.string().min(1)
56  }),
57  z.object({ type: z.literal("reconcile") }),
58  // TASK-0096 §7: 15-minute contentHash reconcile sweep (enqueues the same
59  // search-index job `reconcile` — one sweep implementation, single writer).
60  z.object({ type: z.literal("search-reconcile") }),
61  z.object({ type: z.literal("dlq-report") }),
62  z.object({ type: z.literal("cleanup") }),
63  z.object({ type: z.literal("workspace-resync"), workspaceId: z.string().min(1) }),
64  // TASK-0094 R6: the hourly notification digest.
65  z.object({ type: z.literal("notification-digest") }),
66  // TASK-0104 §1: the daily S3 orphan-object sweep (grace-aware delete).
67  z.object({ type: z.literal("orphans") }),
68  // T-152: per-workspace page-version retention (ad-hoc, admin-triggered).
69  z.object({
70    type: z.literal("versions.prune"),
71    workspaceId: z.string().min(1),
72    actor: z.object({ type: z.enum(["USER", "SYSTEM"]), id: z.string().min(1) })
73  }),
74  // T-160: the hourly task due-reminder sweep (repeatable; ad-hoc for tests).
75  z.object({ type: z.literal("task-due-sweep") }),
76  // T-162/T-163: the hourly review-deadline reminder sweep.
77  z.object({ type: z.literal("review-due-sweep") })
78]);
79export type MaintenanceJobData = z.infer<typeof maintenanceJobSchema>;
80 
81export const MAINTENANCE_REPEATABLES = [
82  // TASK-0133: the spec's 04:00 UTC (was 03:00 — the per-org policy engine
83  // replaces the old global sweep; the orphans sweep stays offset at 03:30).
84  { id: "maintenance:retention", pattern: "0 4 * * *", data: { type: "retention" } },
85  { id: "maintenance:reconcile", pattern: "30 2 * * 0", data: { type: "reconcile" } },
86  // TASK-0096 §7: search drift repair every 15 minutes (the backstop for
87  // missed/late events; idempotent contentHash sweep).
88  { id: "maintenance:search-reconcile", pattern: "*/15 * * * *", data: { type: "search-reconcile" } },
89  { id: "maintenance:webhook-dlq", pattern: "0 * * * *", data: { type: "dlq-report" } },
90  { id: "maintenance:cleanup", pattern: "15 * * * *", data: { type: "cleanup" } },
91  // The notification digest (TASK-0094 R6): hourly, offset from the other
92  // hourly jobs to spread the load.
93  { id: "maintenance:notification-digest", pattern: "30 * * * *", data: { type: "notification-digest" } },
94  // TASK-0104 §1: the daily S3 orphan sweep (offset from the 03:00 retention
95  // run so the two heavy maintenance jobs don't overlap).
96  { id: "maintenance:orphans", pattern: "30 3 * * *", data: { type: "orphans" } },
97  // T-160: task due reminders — hourly (offset from the :15 cleanup and the
98  // :30 digest). The 1h look-back in the core covers the sweep timing.
99  { id: "maintenance:task-due", pattern: "45 * * * *", data: { type: "task-due-sweep" } },
100  { id: "maintenance:review-due", pattern: "46 * * * *", data: { type: "review-due-sweep" } }
101] as const;
102 
103/**
104 * Retention (TASK-0133) — the DAILY per-org policy engine.
105 *
106 * REPLACES the old global sweep (recorded conflict: it purged a global 30 d
107 * trash + 365 d audit with NO org isolation; the spec's per-org policies
108 * supersede it). Delegates to `runRetentionDaily` (retention.ts) — one
109 * implementation for the cron AND the admin "Run now / Run dry" trigger.
110 * Dry-run by default; WORKER_RETENTION_EXECUTE=true → execute.
111 *
112 * The legacy single-key report (`worker:retention:report`) is still written
113 * for backward compatibility (per-run reports live at
114 * `worker:retention:report:{runId}` — the org console polls those).
115 */
116export async function runRetention(
117  prisma: PrismaClient,
118  redis: Redis,
119  env: WorkerEnv
120): Promise<Record<string, unknown>> {
121  const report = await runRetentionDaily(prisma, redis, env);
122  // T-152: per-workspace page-version retention rides the same 04:00 run.
123  report.versions = await pruneAllPageVersions(prisma, { type: "SYSTEM", id: "worker:maintenance" });
124  await redis
125    .set("worker:retention:report", JSON.stringify({ ...report, at: new Date().toISOString() }), "EX", 7 * 86_400)
126    .catch(() => undefined);
127  return report;
128}
129 
130/**
131 * Search reconciliation (TASK-0096): enqueue a contentHash sweep on the
132 * `search-index` queue — the single search writer owns the sweep
133 * (runReconcile in search-index/processor.ts). The weekly `reconcile` job and
134 * the 15-min `search-reconcile` repeatable both funnel through this.
135 */
136export async function runReconcile(connection: Redis): Promise<{ enqueued: "search-index:reconcile" }> {
137  await enqueue(connection, QUEUES.searchIndex, { type: "reconcile" });
138  log.info({}, "search reconcile enqueued (search-index queue)");
139  return { enqueued: "search-index:reconcile" };
140}
141 
142/**
143 * Hourly webhook DLQ report: deliveries that are FAILED and older than 24h
144 * and not yet reported → admin notification (deduped by a redis flag per
145 * delivery so each is reported once).
146 */
147export async function runWebhookDlqReport(
148  prisma: PrismaClient,
149  connection: Redis,
150  env: WorkerEnv
151): Promise<{ stale: number; notified: number }> {
152  const t0 = Date.now();
153  const cutoff = new Date(Date.now() - 24 * 3_600_000);
154  const stale = await prisma.webhookDelivery.findMany({
155    where: { status: "FAILED", createdAt: { lt: cutoff } },
156    select: { id: true, webhookId: true, event: true, error: true },
157    take: 100
158  });
159  let notified = 0;
160  const byWorkspace = new Map<string, typeof stale>();
161  for (const d of stale) {
162    const webhook = await prisma.webhook.findUnique({
163      where: { id: d.webhookId },
164      select: { workspaceId: true }
165    });
166    if (!webhook) continue;
167    const flag = `worker:dlq-reported:${d.id}`;
168    const already = await connection.exists(flag).catch(() => 0);
169    if (already) continue;
170    await connection.set(flag, "1", "EX", 7 * 86_400).catch(() => undefined);
171    const list = byWorkspace.get(webhook.workspaceId) ?? [];
172    list.push(d);
173    byWorkspace.set(webhook.workspaceId, list);
174    notified++;
175  }
176  for (const [workspaceId, items] of byWorkspace) {
177    await notifyWorkspaceAdmins(
178      prisma,
179      connection,
180      env.SMTP_HOST,
181      workspaceId,
182      `${items.length} webhook delivery(s) dead-lettered`,
183      items
184        .slice(0, 5)
185        .map((d) => `• ${d.event}: ${d.error ?? "unknown error"}`)
186        .join("\n"),
187      undefined,
188      "INTEGRATION"
189    );
190  }
191  log.info({ stale: stale.length, notified, durationMs: Date.now() - t0 }, "webhook DLQ report done");
192  return { stale: stale.length, notified };
193}
194 
195/**
196 * Hourly cleanup: expired refresh tokens, expired invitations (7d), expired
197 * API keys (mark revoked). Sessions with expired tokens are revoked too.
198 */
199export async function runCleanup(prisma: PrismaClient): Promise<Record<string, number>> {
200  const t0 = Date.now();
201  const now = new Date();
202  const inviteCutoff = new Date(Date.now() - INVITE_EXPIRY_DAYS * 86_400_000);
203 
204  const refreshTokenRes = await prisma.refreshToken.deleteMany({
205    where: { expiresAt: { lt: now }, revokedAt: null }
206  });
207  const sessionRes = await prisma.session.updateMany({
208    where: { expiresAt: { lt: now }, revokedAt: null },
209    data: { revokedAt: now }
210  });
211  const invitationRes = await prisma.invitation.updateMany({
212    where: {
213      status: "PENDING",
214      expiresAt: { lt: now },
215      revokedAt: null,
216      createdAt: { lt: inviteCutoff }
217    },
218    data: { status: "EXPIRED" }
219  });
220  const apiKeyRes = await prisma.apiKey.updateMany({
221    where: { expiresAt: { lt: now }, revokedAt: null },
222    data: { revokedAt: now }
223  });
224 
225  const counts = {
226    expiredRefreshTokens: refreshTokenRes.count,
227    revokedSessions: sessionRes.count,
228    expiredInvitations: invitationRes.count,
229    revokedApiKeys: apiKeyRes.count
230  };
231  log.info({ ...counts, durationMs: Date.now() - t0 }, "cleanup done");
232  return counts;
233}
234 
235/**
236 * TASK-0104 §1: the daily S3 orphan sweep. Runs the grace-aware sweep over the
237 * `files/` prefix (default 7-day grace — a crashed finalize is given a week
238 * to land its FileVersion row before its object is reclaimed). Re-entrant +
239 * bounded (10k objects/run) so a huge tenant can't starve the cron.
240 *
241 * Builds the canonical `@factory/storage` S3Client (it has `listObjects`; the
242 * worker's `lib/s3.ts` client — the pipeline's — does not, and I won't add a
243 * third listing path to it) from the env, scoped to the files bucket.
244 */
245export async function runOrphans(
246  prisma: PrismaClient,
247  env: WorkerEnv
248): Promise<Record<string, unknown>> {
249  const s3 = new S3Client({
250    config: {
251      // Mirrors apps/api/src/modules/files/s3.ts makeFilesClient (the
252      // canonical files-bucket client) — the sweep must address the SAME
253      // bucket the pipeline writes to.
254      endpoint: env.S3_ENDPOINT.replace(/\/+$/, ""),
255      region: env.S3_REGION,
256      bucket: env.S3_BUCKET_FILES || env.S3_BUCKET,
257      accessKeyId: env.S3_ACCESS_KEY,
258      secretAccessKey: env.S3_SECRET_KEY,
259      forcePathStyle: true
260    }
261  });
262  const report = await runOrphanSweep({ s3, prisma, log });
263  return { ...report };
264}
265 
266// ---------------------------------------------------------------------------
267// Worker
268// ---------------------------------------------------------------------------
269 
270export function createMaintenanceWorker(opts: {
271  connection: Redis;
272  prisma: PrismaClient;
273  env: WorkerEnv;
274  concurrency?: number;
275}): Worker {
276  const { connection, prisma, env } = opts;
277  return new Worker(
278    QUEUES.maintenance,
279    withJobOtel(QUEUES.maintenance, async (job: Job) => {
280      const started = Date.now();
281      const data = maintenanceJobSchema.parse(job.data);
282      log.info({ jobId: job.id, type: data.type }, "maintenance job");
283      switch (data.type) {
284        case "retention":
285          return runRetention(prisma, connection, env);
286        case "retention.run": {
287          // TASK-0133: the org-admin "Run now / Run dry" (ONE scope). The
288          // report lands at `worker:retention:report:{runId}` (the console
289          // polls it); the purge is audited as the triggering admin.
290          const report = await runRetentionScope({
291            prisma,
292            redis: connection,
293            orgId: data.orgId,
294            scope: data.scope,
295            mode: data.mode,
296            actor: data.actor,
297            runId: data.runId
298          });
299          log.info({ jobId: job.id, orgId: data.orgId, scope: data.scope, ...report }, "retention.run done");
300          return report;
301        }
302        case "audit.export": {
303          // TASK-0133: the >10k-row async CSV export (cursor → CSV → S3).
304          const res = await runAuditExport(prisma, connection, data.orgId, data.exportId);
305          log.info({ jobId: job.id, ...res }, "audit export done");
306          return res;
307        }
308        case "reconcile":
309        case "search-reconcile":
310          return runReconcile(connection);
311        case "dlq-report":
312          return runWebhookDlqReport(prisma, connection, env);
313        case "cleanup":
314          return runCleanup(prisma);
315        case "notification-digest": {
316          // TASK-0094 R6: the hourly digest (one email per (user, workspace)).
317          const res = await runDigestRun(prisma, connection, env);
318          log.info({ jobId: job.id, enqueued: res.enqueued }, "notification digest done");
319          return res;
320        }
321        case "orphans": {
322          // TASK-0104 §1: the daily S3 orphan sweep (builds its own client).
323          const res = await runOrphans(prisma, env);
324          log.info({ jobId: job.id, ...res }, "orphan sweep done");
325          return res;
326        }
327        case "versions.prune": {
328          const res = await pruneWorkspacePageVersions(prisma, data.workspaceId, data.actor);
329          log.info({ jobId: job.id, ...res }, "versions.prune done");
330          return res;
331        }
332        case "task-due-sweep": {
333          // T-160: notify assignees of tasks due within the 24h window.
334          const res = await runTaskDueSweep(prisma, connection);
335          log.info({ jobId: job.id, ...res }, "task-due-sweep done");
336          return res;
337        }
338        case "review-due-sweep": {
339          // T-162/T-163: remind reviewers of record of review deadlines.
340          const res = await runReviewDueSweep(prisma, connection);
341          log.info({ jobId: job.id, ...res }, "review-due-sweep done");
342          return res;
343        }
344        case "workspace-resync": {
345          // Enqueued by the search-index consumer on workspace-scoped
346          // permission changes.
347          await enqueue(connection, QUEUES.searchIndex, {
348            type: "reindex",
349            workspaceId: data.workspaceId
350          });
351          log.info({ jobId: job.id, workspaceId: data.workspaceId }, "workspace resync enqueued");
352          return { resynced: data.workspaceId };
353        }
354      }
355      log.info({ jobId: job.id, ms: Date.now() - started }, "done");
356    }),
357    {
358      connection,
359      concurrency: opts.concurrency ?? env.WORKER_CONCURRENCY
360    }
361  );
362}
363