files-pipeline.mddocs/development/files-pipeline.md← development10,720 B← Back to the app

Files Pipeline (worker) — Runbook

How the TASK-0102 worker pipeline works, how to run/verify it locally, and the operational knobs. Sibling docs: storage.md (state machine + event contract), testing.md, and the task spec docs/tasks/phase-09-files.md (TASK-0102).

The pipeline turns a File row left in SCANNING by the upload session (TASK-0101) into a usable, scanned, thumbnailed asset. It runs in the worker, driven by the BullMQ files queue (process-file jobs), not in the API.

1. Pipeline at a glance

process-file {fileId, version}
      │
      ▼
┌──────────┐   ┌────────┐   ┌────────────┐   ┌────────────┐   ┌─────────────┐
│ download │ →  │  scan  │ →  │  metadata  │ →  │ thumbnails │ →  │  finalize   │
└──────────┘   └────────┘   └────────────┘   └────────────┘   └─────────────┘
  S3 → temp     ClamAV        mime re-sniff     image/pdf/video   status +
  (streamed,    zCONTSCAN/    sha256, dims/     160/480/1280      events +
  sha256 in)    zINSTREAM     dur, pdf pages    PNGs → thumbs/    FileVersion

Seam. runFilesPipeline(deps, data) in apps/worker/src/processors/files/processor.ts is the exported handler — it does the whole run and returns {kind: "ready"|"virus"|"failed"} or {kind:"noop", reason:"missing_row"|"terminal"}. The BullMQ worker (createFilesWorker) is a thin wrapper: failed/noop → job.discard(); RetryableError → rethrown so BullMQ backs off; anything else is wrapped as RetryableError.
Per-step guard. Every step is wrapped by guard(step, fn): a FatalError marks the row FAILED (with metadata.pipeline.error) and returns {kind:"failed"} (no retry); a RetryableError is rethrown (BullMQ backoff). Each step emits a file:{fileId}:step:{name} log with its duration.
Idempotency. Re-running a READY or VIRUS_DETECTED row is a no-op ({kind:"noop", reason:"terminal"}). file.uploaded is emitted only on a fresh transition to READY (never on a no-op), which is what makes the at-least-once delivery safe.

Status machine

UPLOADING → SCANNING → READY ─────────┐
                  ├→ VIRUS_DETECTED   │  (terminal)
                  └→ FAILED           │
READY → FAILED  (only via a later re-scan; not a BullMQ retry)
VIRUS_DETECTED / FAILED = terminal

Enforced by assertTransition (steps/ + status.ts); an invalid pair throws and is logged as a pipeline invariant violation.

2. Step semantics

Step	What it does	On failure
download	`s3.downloadTo(key, tempPath)` — streams the object to the temp dir, computing `sha256` in the same pass (a 500 MB object is never held in RAM).	S3 404 → fatal `upload_not_found`. 5xx / `fetch failed` / `ECONN` / `UND_ERR` → retryable.
scan	See §4 ClamAV.	Transport error / timeout → retryable. Infected → terminal `VIRUS_DETECTED` (not an error).
metadata	Re-sniff mime from the leading bytes (final authority), then per-type: image → sharp dimensions (+EXIF orientation); video/audio → ffprobe duration + codec; PDF → pdfjs page count.	Undecodable/timeout → non-fatal: warning in `metadata.pipeline.warnings`, the file is still `READY` (a file is downloadable without rich metadata). A valid-magic-but-truncated object that sharp can't render is fatal at the thumbnails step.
thumbnails	`generateThumbs` writes the planned artifacts to the `thumbs/` bucket at `thumbs/{ws}/{fileId}/{size}/{stem}.{ext}`.	Render failure → fatal `corrupt_object` (row `FAILED`, error mentions `thumbnails`).
finalize	Transitions the row, writes `metadata` (sha256, dimensions/media/pdf, `pipeline{scannedAt,sha256,durationMs,steps}`), upserts the `FileVersion` row, emits events.	— (best-effort publishes; a Redis blip does not fail the run).

Thumbnails. Images → 160/480/1280 px wide, same format (svg/avif/heic/ tiff → png), never upscaling (a size is skipped when source width ≤ target). PDF → first-page PNG (1280w). Video → first-frame PNG (1280w). Audio/other → none. File.thumbnailKey = the 480 variant (primary UI thumb).

Events. On a fresh READY: file.uploaded (once) and file.finalized. On infected: file.virus_detected (never file.uploaded/file.finalized). Envelope on the factory:events stream: {event, payload: JSON.stringify(...), ts}.

3. Failure policy

Retryable (BullMQ exponential backoff, 2 s base, 3 attempts): S3 5xx, fetch transport errors, Redis blips, ClamAV transport/timeout. The row is not transitioned — it stays where it was so the retry re-runs the pipeline.
Fatal (row → FAILED, job done, no retry): upload_not_found (S3 404), corrupt_object (undecodable render), mime_mismatch, disk_full (temp-dir disk guard), and any unclassified error (unclassified). A corrupt object will never become valid, so retrying is pointless.

4. ClamAV

Two transports, selected by env:

Config	Transport	Protocol
`CLAMD_SOCKET` set	unix socket	LOCAL `zCONTSCAN <path>\0` → one null-terminated verdict (`<path>: OK` / `<path>: <sig> FOUND` / `ERROR: msg`). No INIC/INOK. This is what `clamdscan` does.
`CLAMD_HOST`/`CLAMAV_PORT` set, no socket	TCP	REMOTE `zINSTREAM\n` → `INIC` → `INOK\n` → base64 stream → `INOC\n` → verdict.
neither	dev mode	`WARN file scan skipped`, `scanReport {engine:"none", skipped:true, reason}`, run continues.

Production is fail-closed: assertScannerConfigured runs at worker boot and refuses to start when NODE_ENV=production and no scanner is configured.

Setting up a local socket (bare metal / this host)

sudo service clamav-daemon status
# the socket path is /run/clamav/clamd.ctl (NOT the path in clamd.conf)
sudo -u clamav clamdscan --version   # sanity
CLAMD_SOCKET=/run/clamav/clamd.ctl   # in .env

The daemon runs as user clamav; the worker runs as a different user. The temp dir must be traversable + readable by clamav, otherwise zCONTSCAN accepts the connection and then times out (no error). The worker chmods its temp dir to 0755 in createTempFile for exactly this reason (see files/temp.ts). If you see clamav scan timeout on every file, check the temp-dir permissions first.
The zVERSION command hangs on both the unix socket and TCP (verified on 1.5.4), so getClamdVersion is best-effort with a 3 s timeout and returns "unknown". This is expected, not a bug.
Docker clamav/clamav images: the TCP INIC commands hang there (PING works) while the unix socket works — so for the CI Docker profile prefer the unix-socket mount or the CLAMAV_HOST TCP path only against a known-good clamd. Record which you use in the task notes.
EICAR is the standard 69-byte test signature (not malware); it is the virus path's fixture.

5. Temp dir, disk guard, cleanup

FILES_TEMP_DIR (default os.tmpdir()/factory-files) — where objects are downloaded for scan + metadata. It must be noexec (the bytes are never executed — only read by parsers + clamd).
Each file gets a private random-UUID path. The processor deletes it in a finally; a 5-minute cleanup cron (cleanupTempFiles) reaps anything older than FILES_TEMP_MAX_AGE_MS as a crash backstop (a SIGKILL mid-pipeline still gets reaped).
Disk guard: assertDiskSpace throws fatal disk_full when free space on the temp filesystem < FILES_TEMP_FREE_MIN_BYTES (512 MB default), so a full disk can't wedge the worker.

6. Environment (worker)

Var	Default	Meaning
`CLAMD_SOCKET`	—	unix-socket path (primary scanner). Empty = dev skip.
`CLAMAV_HOST` / `CLAMAV_PORT`	— / 3310	TCP scanner (used only when no socket).
`CLAMAV_TIMEOUT_MS`	30000	Per-scan wall timeout.
`FILES_WORKER_CONCURRENCY`	4	BullMQ `concurrency` for the `files` queue.
`FILES_TEMP_DIR`	`os.tmpdir()/factory-files`	Temp dir (noexec).
`FILES_TEMP_MAX_AGE_MS`	600000	Cleanup-cron age threshold.
`FILES_TEMP_FREE_MIN_BYTES`	536870912 (512 MB)	Disk-guard threshold.
`S3_BUCKET_FILES` / `S3_BUCKET_THUMBS`	`factory-files`/`factory-thumbs`	Source + thumbnail buckets.

7. Running the tests

Unit (no infra): apps/worker src/processors/files/__tests__/ — status-transition table, retryable/fatal matrix, temp cleanup, scan-gate, and a fake-clamd (unix + TCP) test for lib/clamav.ts. pnpm exec vitest run.
Integration (tests/integration/files-pipeline.integration.test.ts): real stack — Testcontainers PG/Redis + host MinIO + host ClamAV socket. Drives runFilesPipeline directly (no live Job / app boot). Scenarios: clean PNG, EICAR, corrupt PNG, video, PDF, dev-skip, 404-retryable, dead-endpoint, missing-row. Gated on MinIO + the clamd socket (skips with a clear message otherwise). Run via sg docker -c '…vitest run files-pipeline.integration.test.ts'.

8. Adding a new thumbnail kind

Add the source type to planThumbnails in packages/storage/src/thumbnails.ts (return {size, key, targetWidth, mime, source}).
Add a renderer in apps/worker/src/processors/files/generateThumbs.ts matching that source tag.
Wire the metadata (if it has one) in steps/metadata.ts. The plan is the pure contract; the worker only executes it.

9. Dependency justifications (AGENTS.md rule 17)

Dep	Why	License
`sharp`	Image decode/resize (dimensions + thumbnails). CJS `export =`; via a structural `SharpInstance` interface (`lib/sharp.ts`) because the ESM type surface lags the runtime.	Apache-2.0
`pdfjs-dist`	PDF page count + page-1 rasterize (with `canvas`). `mupdf` was rejected — it is AGPL, which conflicts with the product's licensing.	Apache-2.0
`canvas` (node-canvas)	The 2D context pdfjs renders page 1 into.	MIT
`ffmpeg`/`ffprobe` (system)	Video first-frame + duration/codec. No npm wrapper — calls the binaries directly (`fluent-ffmpeg` was not needed).	LGPL (system package)