files-pipeline.integration.test.ts

files-pipeline.integration.test.tstests/integration/files-pipeline.integration.test.tsraw ← integration25,471 B← Back to the app
1/**
2 * AI Software Factory — Integration: Files pipeline (Phase 9, TASK-0102)
3 *
4 * The worker pipeline against the REAL stack: Testcontainers Postgres/Redis +
5 * the host MinIO (buckets `factory-files` / `factory-thumbs`) + the host
6 * ClamAV unix socket (`/run/clamav/clamd.ctl`).
7 *
8 * It drives `runFilesPipeline` DIRECTLY (the exported handler seam — no live
9 * BullMQ Job, no NestJS app boot): the upload is created the same way the
10 * API does it (a `File` row in SCANNING + the object PUT to the bucket at
11 * the spec key layout `files/{ws}/{fileId}/{version}/{name}`), then the
12 * pipeline runs in-process. This isolates the worker's contract — scan →
13 * metadata → thumbnails → finalize — which is TASK-0102's scope.
14 *
15 * Acceptance (docs/tasks/phase-09-files.md TASK-0102):
16 *   1. clean PNG  → READY + metadata.dimensions + sha256 + 3 thumbnails
17 *                   (160/480/1280, `thumbnailKey` = 480) + FileVersion row +
18 *                   `file.uploaded` (exactly ONCE from the worker) +
19 *                   `file.finalized`.
20 *   2. EICAR      → VIRUS_DETECTED + scanReport{engine:"clamav",infected,
21 *                   signature} + NO thumbnails + `file.virus_detected` +
22 *                   NO `file.finalized` + admin Notification row.
23 *   3. corrupt    → FAILED + `metadata.pipeline.error` (corrupt_object).
24 *   3b. video     → READY + metadata.media{duration ±0.5 s, codec} +
25 *                   first-frame PNG (1280).
26 *   3c. PDF       → READY + metadata.pdf.pages + first-page PNG (1280).
27 *   4. dev-skip   → no scanner configured → READY + scanReport
28 *                   {engine:"none",skipped:true}.
29 *   5. idempotent → re-running a READY file is a no-op (no second
30 *                   `file.uploaded`).
31 *   6. retryable  → the S3 object is gone (404) → the pipeline THROWS a
32 *                   RetryableError (BullMQ would back off); the row stays
33 *                   SCANNING.
34 *
35 * Gated on the host MinIO (the same gate as files.integration.test.ts);
36 * ClamAV is required for this spec (it is the pipeline's scanner) — if the
37 * socket is absent the whole file is skipped with a clear message.
38 */
39import { afterAll, beforeAll, describe, expect, it } from "vitest";
40import { createHash, randomBytes } from "node:crypto";
41import { mkdtemp, readFile, writeFile } from "node:fs/promises";
42import os from "node:os";
43import path from "node:path";
44import { fileURLToPath, pathToFileURL } from "node:url";
45import { PrismaClient } from "@prisma/client";
46import { Redis } from "ioredis";
47import { buildFileKey, planThumbnails } from "@factory/storage";
48import { getDatabaseUrl, getRedisUrl } from "./helpers/containers";
49import { deploySchema } from "./helpers/api";
50 
51const HERE = path.dirname(fileURLToPath(import.meta.url));
52const FIX = path.join(HERE, "fixtures", "files");
53 
54function fileUrl(p: string): string {
55  return `file://${p.startsWith("/") ? p : `/${p}`}`;
56}
57 
58// Import the worker's pipeline + clients from SOURCE (the worker ships no
59// build for the monorepo test path — same source-import convention the
60// integration helper uses for the API app module).
61const { runFilesPipeline } = await import(
62  /* webpackIgnore: true */
63  fileUrl(path.resolve(HERE, "..", "..", "apps", "worker", "src", "processors", "files", "processor.ts"))
64);
65const { S3Client } = await import(
66  /* webpackIgnore: true */
67  fileUrl(path.resolve(HERE, "..", "..", "apps", "worker", "src", "lib", "s3.ts"))
68);
69const { RetryableError } = await import(
70  /* webpackIgnore: true */
71  fileUrl(path.resolve(HERE, "..", "..", "apps", "worker", "src", "processors", "files", "errors.ts"))
72);
73 
74// ── host gates ──────────────────────────────────────────────────────────────
75 
76const S3_ENDPOINT = process.env.S3_ENDPOINT || "http://localhost:9000";
77// NOTE: `||` (not `??`) — the repo `.env` sets `CLAMD_SOCKET=` (EMPTY), which
78// vite loads into process.env; `??` would keep the empty string and every run
79// would skip the scan. The host daemon's socket is the spec's scanner.
80const CLAMD_SOCKET = process.env.CLAMD_SOCKET || "/run/clamav/clamd.ctl";
81 
82async function up(url: string, socket: string): Promise<boolean> {
83  try {
84    const res = await fetch(`${url}/minio/health/live`, { signal: AbortSignal.timeout(3_000) });
85    const minio = res.ok;
86    if (socket) {
87      // The clamd unix socket must EXIST to be usable (the scan is the
88      // contract). `access` — not `readFile` — because a socket cannot be
89      // read as a file (readFile fails even when the socket is live).
90      await import("node:fs/promises").then((m) =>
91        m.access(socket).then(
92          () => {},
93          () => {
94            throw new Error(`clamd socket missing: ${socket}`);
95          }
96        )
97      );
98    }
99    return minio;
100  } catch {
101    return false;
102  }
103}
104const infraReady = await up(S3_ENDPOINT, CLAMD_SOCKET);
105 
106// ── fixtures (loaded once) ──────────────────────────────────────────────────
107 
108async function readFixture(name: string): Promise<Buffer> {
109  return readFile(path.join(FIX, name));
110}
111// EICAR (the standard 69-byte test signature — ClamAV flags it).
112const EICAR = Buffer.from(
113  "X5O!P%@AP[4\\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*",
114  "latin1"
115);
116// A corrupted PNG: valid signature + IHDR (so the magic-byte sniff and
117// dimension read succeed) but the compressed IDAT data is truncated → sharp
118// refuses to render it (the two-stage failure the spec wants: FAILED at the
119// thumbnail step, not at the sniff).
120function corruptPng(full: Buffer): Buffer {
121  return full.subarray(0, Math.floor(full.length * 0.55));
122}
123 
124// ── state ───────────────────────────────────────────────────────────────────
125 
126let prisma: PrismaClient;
127let redis: Redis;
128let filesClient: InstanceType<typeof S3Client>;
129let thumbsClient: InstanceType<typeof S3Client>;
130let tempDir: string;
131let env: import("../../apps/worker/src/config").WorkerEnv;
132let envNoScanner: import("../../apps/worker/src/config").WorkerEnv;
133 
134interface Seed {
135  id: string;
136  key: string;
137  workspaceId: string;
138  ownerId: string;
139  name: string;
140}
141 
142/** Create a SCANNING File row + PUT the object to the bucket (API parity).
143 *  A real owner user is created so the virus path's admin notification has a
144 *  recipient (notifyWorkspaceAdmins targets the workspace owner + ADMIN/OWNER
145 *  members). */
146async function seed(bytes: Buffer, name: string, mime: string): Promise<Seed> {
147  const ownerId = `user_it_${randomBytes(6).toString("hex")}`;
148  const workspaceId = `wsp_it_${randomBytes(6).toString("hex")}`;
149  const id = `file_it_${randomBytes(8).toString("hex")}`;
150  const key = buildFileKey({ workspaceId, fileId: id, version: 1, name });
151  await prisma.user.create({
152    data: {
153      id: ownerId,
154      email: `${id}@factory.local`,
155      name: "Pipeline Owner"
156    }
157  });
158  await prisma.workspace.create({
159    data: {
160      id: workspaceId,
161      name: `WS ${id}`,
162      slug: `slug-${workspaceId}`,
163      ownerId,
164      billingPlan: "paid"
165    }
166  });
167  await prisma.file.create({
168    data: {
169      id,
170      workspaceId,
171      name,
172      key,
173      size: bytes.length,
174      mimeType: mime,
175      status: "SCANNING",
176      uploadedBy: ownerId
177    }
178  });
179  await filesClient.putObject(key, bytes, mime);
180  return { id, key, workspaceId, ownerId, name };
181}
182 
183/** Every `factory:events` entry for one file id, decoded. */
184async function eventsFor(id: string): Promise<Array<{ event: string; payload: any }>> {
185  const raw = await redis.xrange("factory:events", "-", "+");
186  const out: Array<{ event: string; payload: any }> = [];
187  for (const [, fields] of raw) {
188    const kv: Record<string, string> = {};
189    for (let i = 0; i < fields.length; i += 2) kv[fields[i]] = fields[i + 1];
190    if (kv.event && kv.payload) {
191      try {
192        const p = JSON.parse(kv.payload);
193        if (p.fileId === id) out.push({ event: kv.event, payload: p });
194      } catch {
195        /* not JSON — ignore */
196      }
197    }
198  }
199  return out;
200}
201 
202const countEvents = (evts: Array<{ event: string }>, name: string): number =>
203  evts.filter((e) => e.event === name).length;
204 
205function baseEnv(over: Record<string, unknown> = {}): import("../../apps/worker/src/config").WorkerEnv {
206  return {
207    NODE_ENV: "test",
208    REDIS_URL: "ignored",
209    DATABASE_URL: "ignored",
210    S3_ENDPOINT,
211    S3_REGION: process.env.S3_REGION ?? "us-east-1",
212    S3_BUCKET_FILES: process.env.S3_BUCKET_FILES ?? "factory-files",
213    S3_BUCKET_THUMBS: process.env.S3_BUCKET_THUMBS ?? "factory-thumbs",
214    S3_BUCKET: process.env.S3_BUCKET_FILES ?? "factory-files",
215    S3_ACCESS_KEY: process.env.S3_ACCESS_KEY ?? "minioadmin",
216    S3_SECRET_KEY: process.env.S3_SECRET_KEY ?? "minioadmin",
217    S3_PUBLIC_BASE: "http://localhost:9001",
218    SEARCH_URL: "http://localhost:9200",
219    SEARCH_USERNAME: "admin",
220    SEARCH_PASSWORD: "changeme",
221    SEARCH_VECTOR_DIMENSION: 1536,
222    SEARCH_KNN_M: 16,
223    SEARCH_KNN_EF_CONSTRUCTION: 100,
224    MASTER_ENCRYPTION_KEY: "",
225    AI_API_KEY: "",
226    AI_PROVIDER: "openai",
227    AI_BASE_URL: "",
228    AI_MODEL: "gpt-4o-mini",
229    AI_EMBEDDING_PROVIDER: "",
230    AI_EMBEDDING_MODEL: "text-embedding-3-small",
231    SMTP_HOST: "", // no mail in this spec (the Notification row is the assertion)
232    SMTP_PORT: 587,
233    SMTP_USER: "",
234    SMTP_PASS: "",
235    SMTP_FROM: "noreply@factory.local",
236    SMTP_STARTTLS: true,
237    CLAMAV_HOST: "",
238    CLAMAV_PORT: 3310,
239    CLAMD_SOCKET,
240    FILES_WORKER_CONCURRENCY: 4,
241    CLAMAV_TIMEOUT_MS: 30_000,
242    FILES_TEMP_DIR: tempDir,
243    FILES_TEMP_MAX_AGE_MS: 600_000,
244    FILES_TEMP_FREE_MIN_BYTES: 1, // this spec manages its own tiny fixtures
245    PUBLIC_URL: "http://localhost:3000",
246    WORKER_CONCURRENCY: 10,
247    WORKER_ID: "test-worker",
248    WORKER_RETENTION_EXECUTE: false,
249    HEALTH_PORT: 4003,
250    ...over
251  } as import("../../apps/worker/src/config").WorkerEnv;
252}
253 
254describe.skipIf(!infraReady)(
255  "Files pipeline integration (TASK-0102 — worker, real stack)",
256  () => {
257    beforeAll(async () => {
258      const [dbUrl, redisUrl] = await Promise.all([getDatabaseUrl(), getRedisUrl()]);
259      await deploySchema(dbUrl);
260      prisma = new PrismaClient({ datasources: { db: { url: dbUrl } } });
261      await prisma.$connect();
262      redis = new Redis(redisUrl, { maxRetriesPerRequest: null });
263      const cfg = {
264        endpoint: S3_ENDPOINT,
265        region: process.env.S3_REGION ?? "us-east-1",
266        accessKey: process.env.S3_ACCESS_KEY ?? "minioadmin",
267        secretKey: process.env.S3_SECRET_KEY ?? "minioadmin"
268      };
269      filesClient = new S3Client({ ...cfg, bucket: process.env.S3_BUCKET_FILES ?? "factory-files" });
270      thumbsClient = new S3Client({
271        ...cfg,
272        bucket: process.env.S3_BUCKET_THUMBS ?? "factory-thumbs"
273      });
274      tempDir = await mkdtemp(path.join(os.tmpdir(), "factory-files-it-"));
275      env = baseEnv();
276      envNoScanner = baseEnv({ CLAMD_SOCKET: "", CLAMAV_HOST: "" });
277    }, 180_000);
278 
279    afterAll(async () => {
280      await redis?.quit().catch(() => {});
281      await prisma?.$disconnect().catch(() => {});
282    });
283 
284    // ── 1. happy path: clean PNG → READY + metadata + 3 thumbnails ───────
285 
286    it("clean 1400×900 PNG → READY, dimensions + sha256 + 3 thumbnails, FileVersion, file.uploaded×1 + file.finalized×1", async () => {
287      const hero = await readFixture("hero-1400.png");
288      const s = await seed(hero, "hero.png", "image/png");
289      const sha = createHash("sha256").update(hero).digest("hex");
290 
291      const res = await runFilesPipeline(
292        { prisma, s3: filesClient, s3Thumbs: thumbsClient, redis, env },
293        { fileId: s.id, version: 1, workspaceId: s.workspaceId }
294      );
295      expect(res).toEqual({ kind: "ready" });
296 
297      const row = await prisma.file.findUniqueOrThrow({ where: { id: s.id } });
298      expect(row.status).toBe("READY");
299      expect(row.mimeType).toBe("image/png");
300      expect(row.size).toBe(hero.length);
301      const meta = row.metadata as any;
302      expect(meta.sha256).toBe(sha);
303      expect(meta.dimensions).toMatchObject({ width: 1400, height: 900 });
304      expect(meta.pipeline?.steps).toBeInstanceOf(Array);
305      expect(meta.pipeline?.steps?.length).toBeGreaterThanOrEqual(4); // download,scan,metadata,thumbnails
306      expect(row.scanReport).toMatchObject({ engine: "clamav", scanned: true, infected: false });
307 
308      // thumbnails: all three sizes exist in the thumbs bucket; thumbnailKey
309      // is the 480 variant (the spec's primary key).
310      const plan = planThumbnails({
311        mime: "image/png",
312        workspaceId: s.workspaceId,
313        fileId: s.id,
314        name: "hero.png",
315        sourceWidth: 1400
316      });
317      expect(plan.map((p) => p.size).sort()).toEqual(["1280", "160", "480"]);
318      for (const item of plan) {
319        const head = await thumbsClient.headObject(item.key);
320        expect(head, `missing thumbnail ${item.key}`).toBeTruthy();
321      }
322      expect(row.thumbnailKey).toBe(plan.find((p) => p.size === "480")!.key);
323 
324      // thumbnailUrl: a 15-min signed GET that returns 200 with the 480 PNG
325      // (acceptance: "thumbnailUrl returns 200 with the right image
326      // signature" — PNG magic bytes).
327      expect(row.thumbnailUrl).toBeTruthy();
328      const turl = await fetch(row.thumbnailUrl!);
329      expect(turl.status).toBe(200);
330      const tbytes = Buffer.from(await turl.arrayBuffer());
331      expect(tbytes.subarray(0, 8).toString("hex")).toBe(
332        "89504e470d0a1a0a" // PNG signature
333      );
334 
335      // FileVersion row (the worker owns the upsert — TASK-0101 never creates
336      // it; this is the recorded reconciliation).
337      const fv = await prisma.fileVersion.findUnique({
338        where: { fileId_version: { fileId: s.id, version: 1 } }
339      });
340      expect(fv).toMatchObject({ key: s.key, size: hero.length });
341 
342      // event contract: `file.uploaded` exactly once (the worker's; the API
343      // also emits one at complete() — see the task file), `file.finalized`
344      // once.
345      const evts = await eventsFor(s.id);
346      expect(countEvents(evts, "file.uploaded")).toBe(1);
347      expect(countEvents(evts, "file.finalized")).toBe(1);
348      expect(evts.find((e) => e.event === "file.uploaded")!.payload).toMatchObject({
349        fileId: s.id,
350        mimeType: "image/png",
351        size: hero.length
352      });
353 
354      // idempotency: a re-run is a no-op (no second file.uploaded).
355      const res2 = await runFilesPipeline(
356        { prisma, s3: filesClient, s3Thumbs: thumbsClient, redis, env },
357        { fileId: s.id, version: 1, workspaceId: s.workspaceId }
358      );
359      expect(res2).toEqual({ kind: "noop", reason: "terminal" });
360      const evts2 = await eventsFor(s.id);
361      expect(countEvents(evts2, "file.uploaded")).toBe(1); // unchanged
362    }, 60_000);
363 
364    // ── 2. EICAR → VIRUS_DETECTED (quarantine) ────────────────────────────
365 
366    it("EICAR → VIRUS_DETECTED + scanReport + file.virus_detected + admin Notification, NO thumbnails, NO file.finalized", async () => {
367      const s = await seed(EICAR, "report.txt", "text/plain");
368      const res = await runFilesPipeline(
369        { prisma, s3: filesClient, s3Thumbs: thumbsClient, redis, env },
370        { fileId: s.id, version: 1, workspaceId: s.workspaceId }
371      );
372      expect(res).toEqual({ kind: "virus" });
373 
374      const row = await prisma.file.findUniqueOrThrow({ where: { id: s.id } });
375      expect(row.status).toBe("VIRUS_DETECTED");
376      expect(row.scanReport).toMatchObject({
377        engine: "clamav",
378        infected: true
379      });
380      expect(String(row.scanReport?.signature)).toContain("Eicar-Test-Signature");
381      expect(row.thumbnailKey).toBeNull();
382      // the object stays (forensics); no thumbnails were generated.
383      expect(await filesClient.headObject(s.key)).toBeTruthy();
384      const thumbPlan = planThumbnails({
385        mime: "text/plain",
386        workspaceId: s.workspaceId,
387        fileId: s.id,
388        name: "report.txt"
389      });
390      expect(thumbPlan).toEqual([]);
391 
392      // event contract: file.virus_detected fired; file.finalized did NOT.
393      const evts = await eventsFor(s.id);
394      expect(countEvents(evts, "file.virus_detected")).toBe(1);
395      expect(countEvents(evts, "file.finalized")).toBe(0);
396      // file.uploaded: the API would emit one at complete(); the worker does
397      // NOT emit one for a quarantined file.
398      expect(countEvents(evts, "file.uploaded")).toBe(0);
399 
400      // a quarantined file must not be versioned.
401      const fv = await prisma.fileVersion.findUnique({
402        where: { fileId_version: { fileId: s.id, version: 1 } }
403      });
404      expect(fv).toBeNull();
405 
406      // the workspace owner gets an in-app admin notification (no mail —
407      // SMTP is empty in this spec; the row is the assertion). `link` is a
408      // Json column ({type:"file", id:fileId}); nested keys are addressed
409      // with `path` (contains/string_contains don't search object values).
410      const notes = await prisma.notification.findMany({
411        where: { workspaceId: s.workspaceId, link: { path: ["id"], equals: s.id } }
412      });
413      expect(notes.length).toBeGreaterThanOrEqual(1);
414      expect(notes[0].title).toMatch(/Malicious file blocked/i);
415    }, 60_000);
416 
417    // ── 3. corrupt object → FAILED (fatal, no retry) ──────────────────────
418 
419    it("corrupt PNG (IHDR ok, IDAT truncated) → FAILED + metadata.pipeline.error, no thumbnails", async () => {
420      const hero = await readFixture("hero-1400.png");
421      const s = await seed(corruptPng(hero), "broken.png", "image/png");
422      const res = await runFilesPipeline(
423        { prisma, s3: filesClient, s3Thumbs: thumbsClient, redis, env },
424        { fileId: s.id, version: 1, workspaceId: s.workspaceId }
425      );
426      expect(res).toEqual({ kind: "failed" });
427 
428      const row = await prisma.file.findUniqueOrThrow({ where: { id: s.id } });
429      expect(row.status).toBe("FAILED");
430      expect(String((row.metadata as any)?.pipeline?.error)).toContain("thumbnails");
431      expect(row.thumbnailKey).toBeNull();
432      const fv = await prisma.fileVersion.findUnique({
433        where: { fileId_version: { fileId: s.id, version: 1 } }
434      });
435      expect(fv).toBeNull();
436    }, 60_000);
437 
438    // ── 3b. video → media metadata + first-frame thumbnail ────────────────
439    //     Acceptance: duration within ±0.5 s of ffprobe truth + first-frame
440    //     PNG at the planned 1280 key.
441 
442    it("3s h264 MP4 → READY + metadata.media{duration≈3, codec h264} + first-frame PNG", async () => {
443      const clip = await readFixture("clip.mp4");
444      const s = await seed(clip, "clip.mp4", "video/mp4");
445      const sha = createHash("sha256").update(clip).digest("hex");
446 
447      const res = await runFilesPipeline(
448        { prisma, s3: filesClient, s3Thumbs: thumbsClient, redis, env },
449        { fileId: s.id, version: 1, workspaceId: s.workspaceId }
450      );
451      expect(res).toEqual({ kind: "ready" });
452 
453      const row = await prisma.file.findUniqueOrThrow({ where: { id: s.id } });
454      expect(row.status).toBe("READY");
455      expect(row.mimeType).toBe("video/mp4"); // re-sniff confirmed the magic bytes
456      const meta = row.metadata as {
457        sha256?: string;
458        media?: { duration: number; codec: string };
459      };
460      expect(meta.sha256).toBe(sha);
461      expect(meta.media).toBeDefined();
462      expect(meta.media!.duration).toBeGreaterThanOrEqual(2.5); // ±0.5 s
463      expect(meta.media!.duration).toBeLessThanOrEqual(3.5);
464      expect(meta.media!.codec).toBe("h264");
465 
466      const plan = planThumbnails({
467        mime: "video/mp4",
468        workspaceId: s.workspaceId,
469        fileId: s.id,
470        name: "clip.mp4"
471      });
472      expect(plan.length).toBe(1);
473      expect(row.thumbnailKey).toBe(plan[0]!.key);
474      const head = await thumbsClient.headObject(plan[0]!.key);
475      expect(head.size).toBeGreaterThan(0);
476    }, 60_000);
477 
478    // ── 3c. PDF → page count + first-page thumbnail ────────────────────────
479    //     Acceptance: `metadata.pdf.pages` correct + first-page PNG.
480 
481    it("1-page PDF → READY + metadata.pdf.pages=1 + first-page PNG (pdfjs)", async () => {
482      const pdf = await readFixture("doc.pdf");
483      const s = await seed(pdf, "doc.pdf", "application/pdf");
484      const sha = createHash("sha256").update(pdf).digest("hex");
485 
486      const res = await runFilesPipeline(
487        { prisma, s3: filesClient, s3Thumbs: thumbsClient, redis, env },
488        { fileId: s.id, version: 1, workspaceId: s.workspaceId }
489      );
490      expect(res).toEqual({ kind: "ready" });
491 
492      const row = await prisma.file.findUniqueOrThrow({ where: { id: s.id } });
493      expect(row.status).toBe("READY");
494      const meta = row.metadata as { sha256?: string; pdf?: { pages: number } };
495      expect(meta.sha256).toBe(sha);
496      expect(meta.pdf).toEqual({ pages: 1 });
497 
498      const plan = planThumbnails({
499        mime: "application/pdf",
500        workspaceId: s.workspaceId,
501        fileId: s.id,
502        name: "doc.pdf"
503      });
504      expect(plan.length).toBe(1);
505      expect(plan[0]!.source).toBe("pdf-page-1");
506      expect(row.thumbnailKey).toBe(plan[0]!.key);
507      const head = await thumbsClient.headObject(plan[0]!.key);
508      expect(head.size).toBeGreaterThan(0);
509    }, 60_000);
510 
511    // ── 4. no scanner configured → dev-skip, still READY ──────────────────
512 
513    it("no scanner (dev mode) → READY + scanReport{engine:'none',skipped:true}, thumbnails still generated", async () => {
514      const hero = await readFixture("hero-1400.png");
515      const s = await seed(hero, "dev-skip.png", "image/png");
516      const res = await runFilesPipeline(
517        { prisma, s3: filesClient, s3Thumbs: thumbsClient, redis, env: envNoScanner },
518        { fileId: s.id, version: 1, workspaceId: s.workspaceId }
519      );
520      expect(res).toEqual({ kind: "ready" });
521 
522      const row = await prisma.file.findUniqueOrThrow({ where: { id: s.id } });
523      expect(row.status).toBe("READY");
524      expect(row.scanReport).toMatchObject({ engine: "none", skipped: true });
525      expect(row.thumbnailKey).toBeTruthy(); // thumbs are independent of the scan
526    }, 60_000);
527 
528    // ── 6a. retryable failure: storage unreachable → RetryableError ──────
529    //     (a transport blip — BullMQ backs off and re-drives the job; the row
530    //     is NOT transitioned.)
531 
532    it("storage unreachable (fetch fails) → throws RetryableError, row stays SCANNING", async () => {
533      const hero = await readFixture("hero-1400.png");
534      const s = await seed(hero, "retryable.png", "image/png");
535      // A client pointed at a dead port: the signed fetch rejects with a
536      // TypeError (ECONNREFUSED) → classified retryable.
537      const dead = new S3Client({
538        endpoint: "http://127.0.0.1:1",
539        region: process.env.S3_REGION ?? "us-east-1",
540        bucket: "factory-files",
541        accessKey: "minioadmin",
542        secretKey: "minioadmin"
543      });
544      await expect(
545        runFilesPipeline(
546          { prisma, s3: dead, s3Thumbs: thumbsClient, redis, env },
547          { fileId: s.id, version: 1, workspaceId: s.workspaceId }
548        )
549      ).rejects.toBeInstanceOf(RetryableError);
550      // a retryable failure must NOT transition the row (BullMQ re-drives it).
551      const row = await prisma.file.findUniqueOrThrow({ where: { id: s.id } });
552      expect(row.status).toBe("SCANNING");
553    }, 60_000);
554 
555    // ── 6b. fatal: the object is gone (404) → FAILED, no retry ────────────
556    //     (a retry cannot resurrect a deleted object — classifyError maps
557    //     S3 404 to fatal 'upload_not_found'.)
558 
559    it("S3 object deleted (404) → {kind:'failed'}, row FAILED (upload_not_found)", async () => {
560      const hero = await readFixture("hero-1400.png");
561      const s = await seed(hero, "fatal404.png", "image/png");
562      await filesClient.deleteObject(s.key); // the object was reaped
563      const res = await runFilesPipeline(
564        { prisma, s3: filesClient, s3Thumbs: thumbsClient, redis, env },
565        { fileId: s.id, version: 1, workspaceId: s.workspaceId }
566      );
567      expect(res).toEqual({ kind: "failed" });
568      const row = await prisma.file.findUniqueOrThrow({ where: { id: s.id } });
569      expect(row.status).toBe("FAILED");
570      expect(String((row.metadata as any)?.pipeline?.error)).toContain("download");
571    }, 60_000);
572 
573    // ── 5. missing row → no-op (row deleted between enqueue and run) ─────
574 
575    it("file row missing → {kind:'noop', reason:'missing_row'} (no throw)", async () => {
576      const ghost = `file_ghost_${randomBytes(6).toString("hex")}`;
577      const res = await runFilesPipeline(
578        { prisma, s3: filesClient, s3Thumbs: thumbsClient, redis, env },
579        { fileId: ghost, version: 1 }
580      );
581      expect(res).toEqual({ kind: "noop", reason: "missing_row" });
582    }, 30_000);
583  }
584);
585