processor.ts · Source

processor.tsapps/worker/src/processors/files/processor.tsraw ← files15,235 B← Back to the app
1/**
2 * apps/worker — files queue processor (Phase 9, TASK-0102).
3 *
4 * Job: `process-file { fileId, version, workspaceId }` on the `files`
5 * BullMQ queue (enqueued by the API on upload completion, TASK-0101).
6 *
7 * Pipeline (sequential, each step timed + logged as
8 * `file:{fileId}:step:{name}`):
9 *
10 *   download   → S3 GET streamed to the temp dir (sha256 in the same pass)
11 *   scan       → ClamAV over the local copy (unix socket `zCONTSCAN` /
12 *                TCP `zINSTREAM` / dev-skip when unconfigured)
13 *   metadata   → magic-byte re-sniff (final authority) + sharp dims /
14 *                ffprobe media / pdfjs pages (best-effort, capped)
15 *   thumbnails → images 3 sizes / pdf page 1 / video first frame → thumbs
16 *                bucket (via @factory/storage planThumbnails)
17 *   finalize   → status READY|VIRUS_DETECTED|FAILED + metadata.pipeline +
18 *                FileVersion upsert + `file.uploaded` (exactly once) /
19 *                `file.finalized` / `file.virus_detected` + audit
20 *
21 * Status machine (enforced — see status.ts):
22 *   UPLOADING → SCANNING → READY | VIRUS_DETECTED | FAILED
23 *   READY / VIRUS_DETECTED are terminal (a re-run is a NO-OP); any other
24 *   transition throws InvalidTransitionError (an invariant violation).
25 *
26 * Error policy (see errors.ts):
27 *   retryable (network, S3 5xx, clamd timeout) → THROWN → BullMQ backoff
28 *   fatal (corrupt object, missing row, disk guard) → row FAILED +
29 *   `metadata.pipeline.error`, job DISCARDED (no retry — it can't be fixed)
30 *
31 * The Phase-0 scaffold (../files.processor.ts) is superseded by this module.
32 */
33import { Worker, type Job } from "bullmq";
34import type { Redis } from "ioredis";
35import { z } from "zod";
36import type { PrismaClient, File } from "@prisma/client";
37import { QUEUES } from "../../queues.js";
38import { getLogger } from "../../logger.js";
39import type { WorkerEnv } from "../../config.js";
40import type { S3Client } from "../../lib/s3.js";
41import { detectMimeType } from "@factory/storage";
42import { asJson } from "../../lib/prisma.js";
43import { withJobOtel } from "../../lib/job-otel.js";
44import { audit } from "../../lib/notify.js";
45import { assertTransition } from "./status.js";
46import { classifyError, FatalError, RetryableError } from "./errors.js";
47import { createTempFile, removeTempFile } from "./temp.js";
48import { scanStep, type ScanOutcome } from "./steps/scan.js";
49import { metadataStep } from "./steps/metadata.js";
50import { thumbnailsStep } from "./steps/thumbnails.js";
51import { finalizeStep, type FileRowLike } from "./steps/finalize.js";
52import type { FilePipelineContext, StepName, ThumbsResult } from "./types.js";
53 
54const log = getLogger("files");
55 
56/** The exact job data TASK-0101 enqueues. */
57export const processFileJobSchema = z.object({
58  fileId: z.string().min(1),
59  version: z.number().int().positive().default(1),
60  workspaceId: z.string().min(1).optional()
61});
62 
63export type ProcessFileJob = z.infer<typeof processFileJobSchema>;
64 
65/** The dependencies the pipeline needs (dependency-injected for tests). */
66export interface FilesPipelineDeps {
67  prisma: PrismaClient;
68  s3: S3Client;
69  /** The thumbs-bucket client (same creds, `factory-thumbs`). */
70  s3Thumbs: S3Client;
71  redis: Redis;
72  env: WorkerEnv;
73}
74 
75/** The terminal outcome of a pipeline run. */
76export type PipelineRunResult =
77  | { kind: "ready" }
78  | { kind: "virus" }
79  | { kind: "failed" }
80  | { kind: "noop"; reason: "missing_row" | "terminal" };
81 
82/**
83 * Run the full files pipeline for one job. Exported (not just the Worker
84 * callback) so it is unit/integration-testable WITHOUT a live BullMQ Job —
85 * the other processors use the same "exported handler" seam (e.g.
86 * `handleSearchEvent`).
87 *
88 * Contract:
89 *   - returns `{kind:"ready"|"virus"|"failed"|"noop"}` on a NORMAL terminal
90 *     state (the caller completes the job — for "failed" the row is already
91 *     FAILED, for "noop" nothing changed);
92 *   - THROWS a RetryableError on a transient failure (network / S3 5xx /
93 *     clamd timeout) so BullMQ backs off;
94 *   - the temp file is ALWAYS cleaned up (finally).
95 */
96export async function runFilesPipeline(
97  deps: FilesPipelineDeps,
98  data: ProcessFileJob
99): Promise<PipelineRunResult> {
100  const { prisma, s3, s3Thumbs, redis, env } = deps;
101  const started = Date.now();
102  const { fileId, version } = data;
103  log.info({ fileId, version }, "processing file");
104 
105  const row = (await prisma.file.findUnique({ where: { id: fileId } })) as
106    | File
107    | null;
108  if (!row) {
109    // Row gone (deleted/cancelled) — nothing to do, no retry.
110    log.warn({ fileId }, "file row missing — discarding");
111    return { kind: "noop", reason: "missing_row" };
112  }
113  const prevStatus = row.status;
114 
115  // Re-run idempotency: READY / VIRUS_DETECTED are terminal. A later re-scan
116  // of a VIRUS_DETECTED file is an explicit TASK-0104 operator action, never
117  // a BullMQ retry.
118  if (prevStatus === "READY" || prevStatus === "VIRUS_DETECTED") {
119    log.info({ fileId, prevStatus }, "file terminal — no-op re-run");
120    return { kind: "noop", reason: "terminal" };
121  }
122 
123  const ctx: FilePipelineContext = {
124    job: { fileId, version, workspaceId: data.workspaceId ?? row.workspaceId },
125    prisma,
126    s3,
127    s3Thumbs,
128    redis,
129    env,
130    temp: { path: "", size: 0 },
131    stepTimings: new Map<string, number>(),
132    warnings: [],
133    extracted: { sniffedMime: null },
134    sha256: ""
135  };
136  const mark = (name: StepName, ms: number): void => {
137    ctx.stepTimings.set(name, ms);
138    log.info({ fileId, step: name, ms }, `file:${fileId}:step:${name}`);
139  };
140 
141  // A step failure that is FATAL marks the row FAILED and returns {failed}
142  // (done, no retry); a RETRYABLE failure is rethrown so BullMQ backs off.
143  const guard = async (step: string, fn: () => Promise<void>): Promise<void> => {
144    const t0 = Date.now();
145    try {
146      await fn();
147      mark(step as StepName, Date.now() - t0);
148    } catch (err) {
149      const classified = classifyError(err);
150      if (classified instanceof FatalError) {
151        log.error(
152          { fileId, step, code: classified.code, err: classified.message },
153          "pipeline step failed FATAL — marking FAILED (no retry)"
154        );
155        await markFailed(ctx, row, `${step}: ${classified.message}`, classified.code);
156        throw new FatalHandledMarker(classified.code, step);
157      }
158      log.warn(
159        { fileId, step, err: classified.message },
160        "pipeline step failed RETRYABLE — backing off"
161      );
162      throw classified; // BullMQ retries per job options
163    }
164  };
165 
166  try {
167    // Ensure SCANNING (a stranded UPLOADING row re-driven by maintenance).
168    if (prevStatus === "UPLOADING") {
169      assertTransition("UPLOADING", "SCANNING");
170      await prisma.file.update({ where: { id: fileId }, data: { status: "SCANNING" } });
171    }
172 
173    // 1) download (streamed to temp, sha256 in-pass) ───────────────────────
174    await guard("download", async () => {
175      const path = await createTempFile(env, env.FILES_TEMP_FREE_MIN_BYTES);
176      const { size, sha256 } = await s3.downloadTo(row.key, path);
177      ctx.temp = { path, size };
178      ctx.sha256 = sha256;
179    });
180 
181    // 2) scan ───────────────────────────────────────────────────────────────
182    let scan: ScanOutcome | undefined;
183    await guard("scan", async () => {
184      scan = await scanStep(ctx);
185    });
186    const scanOutcome = scan!;
187    if (scanOutcome.infected) {
188      // Terminal: VIRUS_DETECTED (no thumbs, no index, no file.uploaded).
189      assertTransition(prevStatus, "VIRUS_DETECTED");
190      await finalizeStep(ctx, {
191        outcome: { kind: "virus", scan: scanOutcome.report },
192        prevStatus,
193        file: row,
194        finalMime: row.mimeType,
195        size: ctx.temp.size
196      });
197      return { kind: "virus" };
198    }
199 
200    // 3) metadata (re-sniff = final authority) ──────────────────────────────
201    let sniffedMime: string | null = null;
202    await guard("metadata", async () => {
203      sniffedMime = await sniff(ctx);
204      const finalMime = sniffedMime ?? row.mimeType;
205      if (!sniffedMime) {
206        ctx.warnings.push("mime: no magic-byte signal — keeping declared type");
207      }
208      await metadataStep(ctx, finalMime);
209    });
210    const finalMime = sniffedMime ?? row.mimeType;
211 
212    // 4) thumbnails ─────────────────────────────────────────────────────────
213    let thumbs: ThumbsResult | undefined;
214    await guard("thumbnails", async () => {
215      thumbs = await thumbnailsStep(ctx, finalMime);
216    });
217 
218    // 5) finalize READY ─────────────────────────────────────────────────────
219    assertTransition(prevStatus, "READY");
220    await finalizeStep(ctx, {
221      outcome: {
222        kind: "ready",
223        extracted: ctx.extracted,
224        thumbs: thumbs!,
225        scan: scanOutcome.report
226      },
227      prevStatus,
228      file: row,
229      finalMime,
230      size: ctx.temp.size
231    });
232    await stampDuration(ctx, row, Date.now() - started);
233    log.info(
234      { fileId, size: ctx.temp.size, ms: Date.now() - started },
235      "file processed READY"
236    );
237    return { kind: "ready" };
238  } catch (err) {
239    // A FATAL was already handled (row FAILED) — surface a {failed} result so
240    // the caller completes the job WITHOUT a BullMQ retry.
241    if (err instanceof FatalHandledMarker) {
242      return { kind: "failed" };
243    }
244    // Retryable (or unexpected): rethrow → BullMQ backoff.
245    throw err;
246  } finally {
247    // The object never stays on disk past the run (the cron backstops a
248    // SIGKILL mid-run).
249    await removeTempFile(ctx.temp.path);
250  }
251}
252 
253export function createFilesWorker(opts: {
254  connection: Redis;
255  prisma: PrismaClient;
256  env: WorkerEnv;
257  s3: S3Client;
258  /** The thumbs-bucket client (same creds, `factory-thumbs`). */
259  s3Thumbs: S3Client;
260  concurrency?: number;
261}): Worker {
262  const { connection, prisma, env, s3, s3Thumbs } = opts;
263 
264  return new Worker(
265    QUEUES.files,
266    withJobOtel(QUEUES.files, async (job: Job) => {
267      const data = processFileJobSchema.parse(job.data);
268      try {
269        const result = await runFilesPipeline(
270          { prisma, s3, s3Thumbs, redis: connection, env },
271          data
272        );
273        // "failed" (handled fatal — row already FAILED) and "noop" (missing
274        // row / already terminal) → the job is done; discard keeps it out of
275        // the retry/DLQ path.
276        if (result.kind === "failed" || result.kind === "noop") {
277          job.discard();
278        }
279      } catch (err) {
280        // RetryableError → let BullMQ retry (job options at enqueue).
281        if (err instanceof RetryableError) throw err;
282        // Anything else is a genuine unexpected error → let BullMQ retry too
283        // (the pipeline's own fatal/retryable split already handles the
284        // classified cases; reaching here means an unclassified throw, which
285        // we treat as retryable-but-loud rather than silently completing).
286        throw new RetryableError(`unhandled pipeline error: ${(err as Error).message}`, err);
287      }
288    }),
289    {
290      connection,
291      // TASK-0102 §7: files pipeline is IO+CPU bound (scan/sharp/ffmpeg) →
292      // its own lower concurrency, not the general WORKER_CONCURRENCY.
293      concurrency: opts.concurrency ?? env.FILES_WORKER_CONCURRENCY
294      // NOTE: the retry policy (3 attempts, exponential 2 s backoff,
295      // removeOnComplete 500 / removeOnFail 200 — TASK-0102 §7) is applied at
296      // ENQUEUE time by the API (uploads.service.enqueueProcessFile), not here.
297      // BullMQ WorkerOptions has no per-worker defaultJobOptions; job options
298      // ride on the enqueued job.
299    }
300  );
301}
302 
303/**
304 * Internal marker: a FATAL failure was already handled (row FAILED + audit)
305 * and the job was discarded. Swallowed by BullMQ's failure path — it must
306 * NOT be treated as a new retryable error. (BullMQ sees a throw; the row and
307 * the discard already made the outcome final.)
308 */
309class FatalHandledMarker extends Error {
310  constructor(code: string, step: string) {
311    super(`fatal (handled) ${step} ${code}`);
312    this.name = "FatalHandledMarker";
313  }
314}
315 
316/** Magic-byte re-sniff of the leading bytes (final authority; null = no signal). */
317async function sniff(ctx: FilePipelineContext): Promise<string | null> {
318  const { open } = await import("node:fs/promises");
319  const fd = await open(ctx.temp.path, "r");
320  try {
321    const buf = Buffer.alloc(16 * 1024);
322    const { bytesRead } = await fd.read(buf, 0, buf.length, 0);
323    if (bytesRead === 0) return null;
324    const det = detectMimeType(buf.subarray(0, bytesRead));
325    return det.confidence === "unknown" ? null : det.mime;
326  } finally {
327    await fd.close();
328  }
329}
330 
331/** Mark the row FAILED with `metadata.pipeline.error` + audit (never throws). */
332async function markFailed(
333  ctx: FilePipelineContext,
334  row: File,
335  error: string,
336  code: string
337): Promise<void> {
338  try {
339    const cur = (await ctx.prisma.file.findUnique({ where: { id: row.id } })) as File | null;
340    if (!cur) return;
341    // A FAILED transition is legal from UPLOADING/SCANNING (and READY on
342    // re-scan); assert so an unexpected state is visible, not silent.
343    assertTransition(cur.status, "FAILED");
344    const meta = (cur.metadata as Record<string, unknown> | null) ?? {};
345    const pipeline = {
346      scannedAt: new Date().toISOString(),
347      sha256: ctx.sha256 || undefined,
348      durationMs: 0,
349      steps: [...ctx.stepTimings.entries()].map(([name, ms]) => ({ name, ms })),
350      error,
351      ...(ctx.warnings.length ? { warnings: [...ctx.warnings] } : {})
352    };
353    await ctx.prisma.file.update({
354      where: { id: row.id },
355      data: { status: "FAILED", metadata: asJson({ ...meta, pipeline }) }
356    });
357    await audit(ctx.prisma, {
358      workspaceId: row.workspaceId,
359      resourceType: "file",
360      resourceId: row.id,
361      action: "file.process_failed",
362      metadata: { code, error, key: row.key }
363    });
364  } catch (err) {
365    log.error({ fileId: row.id, err: (err as Error).message }, "markFailed failed (row may be inconsistent)");
366  }
367}
368 
369/** Stamp the total pipeline wall time into `metadata.pipeline.durationMs`. */
370async function stampDuration(
371  ctx: FilePipelineContext,
372  row: File,
373  totalMs: number
374): Promise<void> {
375  const cur = (await ctx.prisma.file.findUnique({ where: { id: row.id } })) as File | null;
376  if (!cur) return;
377  const meta = (cur.metadata as Record<string, unknown> | null) ?? {};
378  const pipeline = (meta.pipeline as Record<string, unknown> | undefined) ?? {};
379  await ctx.prisma.file.update({
380    where: { id: row.id },
381    data: { metadata: asJson({ ...meta, pipeline: { ...pipeline, durationMs: totalMs } }) }
382  });
383}
384 
385/** Re-export for tests (the status-machine error). */
386export type { FileRowLike };
387export { RetryableError, FatalError };
388