/**
 * AI Software Factory — Integration: Files pipeline (Phase 9, TASK-0102)
 *
 * The worker pipeline against the REAL stack: Testcontainers Postgres/Redis +
 * the host MinIO (buckets `factory-files` / `factory-thumbs`) + the host
 * ClamAV unix socket (`/run/clamav/clamd.ctl`).
 *
 * It drives `runFilesPipeline` DIRECTLY (the exported handler seam — no live
 * BullMQ Job, no NestJS app boot): the upload is created the same way the
 * API does it (a `File` row in SCANNING + the object PUT to the bucket at
 * the spec key layout `files/{ws}/{fileId}/{version}/{name}`), then the
 * pipeline runs in-process. This isolates the worker's contract — scan →
 * metadata → thumbnails → finalize — which is TASK-0102's scope.
 *
 * Acceptance (docs/tasks/phase-09-files.md TASK-0102):
 *   1. clean PNG  → READY + metadata.dimensions + sha256 + 3 thumbnails
 *                   (160/480/1280, `thumbnailKey` = 480) + FileVersion row +
 *                   `file.uploaded` (exactly ONCE from the worker) +
 *                   `file.finalized`.
 *   2. EICAR      → VIRUS_DETECTED + scanReport{engine:"clamav",infected,
 *                   signature} + NO thumbnails + `file.virus_detected` +
 *                   NO `file.finalized` + admin Notification row.
 *   3. corrupt    → FAILED + `metadata.pipeline.error` (corrupt_object).
 *   3b. video     → READY + metadata.media{duration ±0.5 s, codec} +
 *                   first-frame PNG (1280).
 *   3c. PDF       → READY + metadata.pdf.pages + first-page PNG (1280).
 *   4. dev-skip   → no scanner configured → READY + scanReport
 *                   {engine:"none",skipped:true}.
 *   5. idempotent → re-running a READY file is a no-op (no second
 *                   `file.uploaded`).
 *   6. retryable  → the S3 object is gone (404) → the pipeline THROWS a
 *                   RetryableError (BullMQ would back off); the row stays
 *                   SCANNING.
 *
 * Gated on the host MinIO (the same gate as files.integration.test.ts);
 * ClamAV is required for this spec (it is the pipeline's scanner) — if the
 * socket is absent the whole file is skipped with a clear message.
 */
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { createHash, randomBytes } from "node:crypto";
import { mkdtemp, readFile, writeFile } from "node:fs/promises";
import os from "node:os";
import path from "node:path";
import { fileURLToPath, pathToFileURL } from "node:url";
import { PrismaClient } from "@prisma/client";
import { Redis } from "ioredis";
import { buildFileKey, planThumbnails } from "@factory/storage";
import { getDatabaseUrl, getRedisUrl } from "./helpers/containers";
import { deploySchema } from "./helpers/api";

const HERE = path.dirname(fileURLToPath(import.meta.url));
const FIX = path.join(HERE, "fixtures", "files");

function fileUrl(p: string): string {
  return `file://${p.startsWith("/") ? p : `/${p}`}`;
}

// Import the worker's pipeline + clients from SOURCE (the worker ships no
// build for the monorepo test path — same source-import convention the
// integration helper uses for the API app module).
const { runFilesPipeline } = await import(
  /* webpackIgnore: true */
  fileUrl(path.resolve(HERE, "..", "..", "apps", "worker", "src", "processors", "files", "processor.ts"))
);
const { S3Client } = await import(
  /* webpackIgnore: true */
  fileUrl(path.resolve(HERE, "..", "..", "apps", "worker", "src", "lib", "s3.ts"))
);
const { RetryableError } = await import(
  /* webpackIgnore: true */
  fileUrl(path.resolve(HERE, "..", "..", "apps", "worker", "src", "processors", "files", "errors.ts"))
);

// ── host gates ──────────────────────────────────────────────────────────────

const S3_ENDPOINT = process.env.S3_ENDPOINT || "http://localhost:9000";
// NOTE: `||` (not `??`) — the repo `.env` sets `CLAMD_SOCKET=` (EMPTY), which
// vite loads into process.env; `??` would keep the empty string and every run
// would skip the scan. The host daemon's socket is the spec's scanner.
const CLAMD_SOCKET = process.env.CLAMD_SOCKET || "/run/clamav/clamd.ctl";

async function up(url: string, socket: string): Promise<boolean> {
  try {
    const res = await fetch(`${url}/minio/health/live`, { signal: AbortSignal.timeout(3_000) });
    const minio = res.ok;
    if (socket) {
      // The clamd unix socket must EXIST to be usable (the scan is the
      // contract). `access` — not `readFile` — because a socket cannot be
      // read as a file (readFile fails even when the socket is live).
      await import("node:fs/promises").then((m) =>
        m.access(socket).then(
          () => {},
          () => {
            throw new Error(`clamd socket missing: ${socket}`);
          }
        )
      );
    }
    return minio;
  } catch {
    return false;
  }
}
const infraReady = await up(S3_ENDPOINT, CLAMD_SOCKET);

// ── fixtures (loaded once) ──────────────────────────────────────────────────

async function readFixture(name: string): Promise<Buffer> {
  return readFile(path.join(FIX, name));
}
// EICAR (the standard 69-byte test signature — ClamAV flags it).
const EICAR = Buffer.from(
  "X5O!P%@AP[4\\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*",
  "latin1"
);
// A corrupted PNG: valid signature + IHDR (so the magic-byte sniff and
// dimension read succeed) but the compressed IDAT data is truncated → sharp
// refuses to render it (the two-stage failure the spec wants: FAILED at the
// thumbnail step, not at the sniff).
function corruptPng(full: Buffer): Buffer {
  return full.subarray(0, Math.floor(full.length * 0.55));
}

// ── state ───────────────────────────────────────────────────────────────────

let prisma: PrismaClient;
let redis: Redis;
let filesClient: InstanceType<typeof S3Client>;
let thumbsClient: InstanceType<typeof S3Client>;
let tempDir: string;
let env: import("../../apps/worker/src/config").WorkerEnv;
let envNoScanner: import("../../apps/worker/src/config").WorkerEnv;

interface Seed {
  id: string;
  key: string;
  workspaceId: string;
  ownerId: string;
  name: string;
}

/** Create a SCANNING File row + PUT the object to the bucket (API parity).
 *  A real owner user is created so the virus path's admin notification has a
 *  recipient (notifyWorkspaceAdmins targets the workspace owner + ADMIN/OWNER
 *  members). */
async function seed(bytes: Buffer, name: string, mime: string): Promise<Seed> {
  const ownerId = `user_it_${randomBytes(6).toString("hex")}`;
  const workspaceId = `wsp_it_${randomBytes(6).toString("hex")}`;
  const id = `file_it_${randomBytes(8).toString("hex")}`;
  const key = buildFileKey({ workspaceId, fileId: id, version: 1, name });
  await prisma.user.create({
    data: {
      id: ownerId,
      email: `${id}@factory.local`,
      name: "Pipeline Owner"
    }
  });
  await prisma.workspace.create({
    data: {
      id: workspaceId,
      name: `WS ${id}`,
      slug: `slug-${workspaceId}`,
      ownerId,
      billingPlan: "paid"
    }
  });
  await prisma.file.create({
    data: {
      id,
      workspaceId,
      name,
      key,
      size: bytes.length,
      mimeType: mime,
      status: "SCANNING",
      uploadedBy: ownerId
    }
  });
  await filesClient.putObject(key, bytes, mime);
  return { id, key, workspaceId, ownerId, name };
}

/** Every `factory:events` entry for one file id, decoded. */
async function eventsFor(id: string): Promise<Array<{ event: string; payload: any }>> {
  const raw = await redis.xrange("factory:events", "-", "+");
  const out: Array<{ event: string; payload: any }> = [];
  for (const [, fields] of raw) {
    const kv: Record<string, string> = {};
    for (let i = 0; i < fields.length; i += 2) kv[fields[i]] = fields[i + 1];
    if (kv.event && kv.payload) {
      try {
        const p = JSON.parse(kv.payload);
        if (p.fileId === id) out.push({ event: kv.event, payload: p });
      } catch {
        /* not JSON — ignore */
      }
    }
  }
  return out;
}

const countEvents = (evts: Array<{ event: string }>, name: string): number =>
  evts.filter((e) => e.event === name).length;

function baseEnv(over: Record<string, unknown> = {}): import("../../apps/worker/src/config").WorkerEnv {
  return {
    NODE_ENV: "test",
    REDIS_URL: "ignored",
    DATABASE_URL: "ignored",
    S3_ENDPOINT,
    S3_REGION: process.env.S3_REGION ?? "us-east-1",
    S3_BUCKET_FILES: process.env.S3_BUCKET_FILES ?? "factory-files",
    S3_BUCKET_THUMBS: process.env.S3_BUCKET_THUMBS ?? "factory-thumbs",
    S3_BUCKET: process.env.S3_BUCKET_FILES ?? "factory-files",
    S3_ACCESS_KEY: process.env.S3_ACCESS_KEY ?? "minioadmin",
    S3_SECRET_KEY: process.env.S3_SECRET_KEY ?? "minioadmin",
    S3_PUBLIC_BASE: "http://localhost:9001",
    SEARCH_URL: "http://localhost:9200",
    SEARCH_USERNAME: "admin",
    SEARCH_PASSWORD: "changeme",
    SEARCH_VECTOR_DIMENSION: 1536,
    SEARCH_KNN_M: 16,
    SEARCH_KNN_EF_CONSTRUCTION: 100,
    MASTER_ENCRYPTION_KEY: "",
    AI_API_KEY: "",
    AI_PROVIDER: "openai",
    AI_BASE_URL: "",
    AI_MODEL: "gpt-4o-mini",
    AI_EMBEDDING_PROVIDER: "",
    AI_EMBEDDING_MODEL: "text-embedding-3-small",
    SMTP_HOST: "", // no mail in this spec (the Notification row is the assertion)
    SMTP_PORT: 587,
    SMTP_USER: "",
    SMTP_PASS: "",
    SMTP_FROM: "noreply@factory.local",
    SMTP_STARTTLS: true,
    CLAMAV_HOST: "",
    CLAMAV_PORT: 3310,
    CLAMD_SOCKET,
    FILES_WORKER_CONCURRENCY: 4,
    CLAMAV_TIMEOUT_MS: 30_000,
    FILES_TEMP_DIR: tempDir,
    FILES_TEMP_MAX_AGE_MS: 600_000,
    FILES_TEMP_FREE_MIN_BYTES: 1, // this spec manages its own tiny fixtures
    PUBLIC_URL: "http://localhost:3000",
    WORKER_CONCURRENCY: 10,
    WORKER_ID: "test-worker",
    WORKER_RETENTION_EXECUTE: false,
    HEALTH_PORT: 4003,
    ...over
  } as import("../../apps/worker/src/config").WorkerEnv;
}

describe.skipIf(!infraReady)(
  "Files pipeline integration (TASK-0102 — worker, real stack)",
  () => {
    beforeAll(async () => {
      const [dbUrl, redisUrl] = await Promise.all([getDatabaseUrl(), getRedisUrl()]);
      await deploySchema(dbUrl);
      prisma = new PrismaClient({ datasources: { db: { url: dbUrl } } });
      await prisma.$connect();
      redis = new Redis(redisUrl, { maxRetriesPerRequest: null });
      const cfg = {
        endpoint: S3_ENDPOINT,
        region: process.env.S3_REGION ?? "us-east-1",
        accessKey: process.env.S3_ACCESS_KEY ?? "minioadmin",
        secretKey: process.env.S3_SECRET_KEY ?? "minioadmin"
      };
      filesClient = new S3Client({ ...cfg, bucket: process.env.S3_BUCKET_FILES ?? "factory-files" });
      thumbsClient = new S3Client({
        ...cfg,
        bucket: process.env.S3_BUCKET_THUMBS ?? "factory-thumbs"
      });
      tempDir = await mkdtemp(path.join(os.tmpdir(), "factory-files-it-"));
      env = baseEnv();
      envNoScanner = baseEnv({ CLAMD_SOCKET: "", CLAMAV_HOST: "" });
    }, 180_000);

    afterAll(async () => {
      await redis?.quit().catch(() => {});
      await prisma?.$disconnect().catch(() => {});
    });

    // ── 1. happy path: clean PNG → READY + metadata + 3 thumbnails ───────

    it("clean 1400×900 PNG → READY, dimensions + sha256 + 3 thumbnails, FileVersion, file.uploaded×1 + file.finalized×1", async () => {
      const hero = await readFixture("hero-1400.png");
      const s = await seed(hero, "hero.png", "image/png");
      const sha = createHash("sha256").update(hero).digest("hex");

      const res = await runFilesPipeline(
        { prisma, s3: filesClient, s3Thumbs: thumbsClient, redis, env },
        { fileId: s.id, version: 1, workspaceId: s.workspaceId }
      );
      expect(res).toEqual({ kind: "ready" });

      const row = await prisma.file.findUniqueOrThrow({ where: { id: s.id } });
      expect(row.status).toBe("READY");
      expect(row.mimeType).toBe("image/png");
      expect(row.size).toBe(hero.length);
      const meta = row.metadata as any;
      expect(meta.sha256).toBe(sha);
      expect(meta.dimensions).toMatchObject({ width: 1400, height: 900 });
      expect(meta.pipeline?.steps).toBeInstanceOf(Array);
      expect(meta.pipeline?.steps?.length).toBeGreaterThanOrEqual(4); // download,scan,metadata,thumbnails
      expect(row.scanReport).toMatchObject({ engine: "clamav", scanned: true, infected: false });

      // thumbnails: all three sizes exist in the thumbs bucket; thumbnailKey
      // is the 480 variant (the spec's primary key).
      const plan = planThumbnails({
        mime: "image/png",
        workspaceId: s.workspaceId,
        fileId: s.id,
        name: "hero.png",
        sourceWidth: 1400
      });
      expect(plan.map((p) => p.size).sort()).toEqual(["1280", "160", "480"]);
      for (const item of plan) {
        const head = await thumbsClient.headObject(item.key);
        expect(head, `missing thumbnail ${item.key}`).toBeTruthy();
      }
      expect(row.thumbnailKey).toBe(plan.find((p) => p.size === "480")!.key);

      // thumbnailUrl: a 15-min signed GET that returns 200 with the 480 PNG
      // (acceptance: "thumbnailUrl returns 200 with the right image
      // signature" — PNG magic bytes).
      expect(row.thumbnailUrl).toBeTruthy();
      const turl = await fetch(row.thumbnailUrl!);
      expect(turl.status).toBe(200);
      const tbytes = Buffer.from(await turl.arrayBuffer());
      expect(tbytes.subarray(0, 8).toString("hex")).toBe(
        "89504e470d0a1a0a" // PNG signature
      );

      // FileVersion row (the worker owns the upsert — TASK-0101 never creates
      // it; this is the recorded reconciliation).
      const fv = await prisma.fileVersion.findUnique({
        where: { fileId_version: { fileId: s.id, version: 1 } }
      });
      expect(fv).toMatchObject({ key: s.key, size: hero.length });

      // event contract: `file.uploaded` exactly once (the worker's; the API
      // also emits one at complete() — see the task file), `file.finalized`
      // once.
      const evts = await eventsFor(s.id);
      expect(countEvents(evts, "file.uploaded")).toBe(1);
      expect(countEvents(evts, "file.finalized")).toBe(1);
      expect(evts.find((e) => e.event === "file.uploaded")!.payload).toMatchObject({
        fileId: s.id,
        mimeType: "image/png",
        size: hero.length
      });

      // idempotency: a re-run is a no-op (no second file.uploaded).
      const res2 = await runFilesPipeline(
        { prisma, s3: filesClient, s3Thumbs: thumbsClient, redis, env },
        { fileId: s.id, version: 1, workspaceId: s.workspaceId }
      );
      expect(res2).toEqual({ kind: "noop", reason: "terminal" });
      const evts2 = await eventsFor(s.id);
      expect(countEvents(evts2, "file.uploaded")).toBe(1); // unchanged
    }, 60_000);

    // ── 2. EICAR → VIRUS_DETECTED (quarantine) ────────────────────────────

    it("EICAR → VIRUS_DETECTED + scanReport + file.virus_detected + admin Notification, NO thumbnails, NO file.finalized", async () => {
      const s = await seed(EICAR, "report.txt", "text/plain");
      const res = await runFilesPipeline(
        { prisma, s3: filesClient, s3Thumbs: thumbsClient, redis, env },
        { fileId: s.id, version: 1, workspaceId: s.workspaceId }
      );
      expect(res).toEqual({ kind: "virus" });

      const row = await prisma.file.findUniqueOrThrow({ where: { id: s.id } });
      expect(row.status).toBe("VIRUS_DETECTED");
      expect(row.scanReport).toMatchObject({
        engine: "clamav",
        infected: true
      });
      expect(String(row.scanReport?.signature)).toContain("Eicar-Test-Signature");
      expect(row.thumbnailKey).toBeNull();
      // the object stays (forensics); no thumbnails were generated.
      expect(await filesClient.headObject(s.key)).toBeTruthy();
      const thumbPlan = planThumbnails({
        mime: "text/plain",
        workspaceId: s.workspaceId,
        fileId: s.id,
        name: "report.txt"
      });
      expect(thumbPlan).toEqual([]);

      // event contract: file.virus_detected fired; file.finalized did NOT.
      const evts = await eventsFor(s.id);
      expect(countEvents(evts, "file.virus_detected")).toBe(1);
      expect(countEvents(evts, "file.finalized")).toBe(0);
      // file.uploaded: the API would emit one at complete(); the worker does
      // NOT emit one for a quarantined file.
      expect(countEvents(evts, "file.uploaded")).toBe(0);

      // a quarantined file must not be versioned.
      const fv = await prisma.fileVersion.findUnique({
        where: { fileId_version: { fileId: s.id, version: 1 } }
      });
      expect(fv).toBeNull();

      // the workspace owner gets an in-app admin notification (no mail —
      // SMTP is empty in this spec; the row is the assertion). `link` is a
      // Json column ({type:"file", id:fileId}); nested keys are addressed
      // with `path` (contains/string_contains don't search object values).
      const notes = await prisma.notification.findMany({
        where: { workspaceId: s.workspaceId, link: { path: ["id"], equals: s.id } }
      });
      expect(notes.length).toBeGreaterThanOrEqual(1);
      expect(notes[0].title).toMatch(/Malicious file blocked/i);
    }, 60_000);

    // ── 3. corrupt object → FAILED (fatal, no retry) ──────────────────────

    it("corrupt PNG (IHDR ok, IDAT truncated) → FAILED + metadata.pipeline.error, no thumbnails", async () => {
      const hero = await readFixture("hero-1400.png");
      const s = await seed(corruptPng(hero), "broken.png", "image/png");
      const res = await runFilesPipeline(
        { prisma, s3: filesClient, s3Thumbs: thumbsClient, redis, env },
        { fileId: s.id, version: 1, workspaceId: s.workspaceId }
      );
      expect(res).toEqual({ kind: "failed" });

      const row = await prisma.file.findUniqueOrThrow({ where: { id: s.id } });
      expect(row.status).toBe("FAILED");
      expect(String((row.metadata as any)?.pipeline?.error)).toContain("thumbnails");
      expect(row.thumbnailKey).toBeNull();
      const fv = await prisma.fileVersion.findUnique({
        where: { fileId_version: { fileId: s.id, version: 1 } }
      });
      expect(fv).toBeNull();
    }, 60_000);

    // ── 3b. video → media metadata + first-frame thumbnail ────────────────
    //     Acceptance: duration within ±0.5 s of ffprobe truth + first-frame
    //     PNG at the planned 1280 key.

    it("3s h264 MP4 → READY + metadata.media{duration≈3, codec h264} + first-frame PNG", async () => {
      const clip = await readFixture("clip.mp4");
      const s = await seed(clip, "clip.mp4", "video/mp4");
      const sha = createHash("sha256").update(clip).digest("hex");

      const res = await runFilesPipeline(
        { prisma, s3: filesClient, s3Thumbs: thumbsClient, redis, env },
        { fileId: s.id, version: 1, workspaceId: s.workspaceId }
      );
      expect(res).toEqual({ kind: "ready" });

      const row = await prisma.file.findUniqueOrThrow({ where: { id: s.id } });
      expect(row.status).toBe("READY");
      expect(row.mimeType).toBe("video/mp4"); // re-sniff confirmed the magic bytes
      const meta = row.metadata as {
        sha256?: string;
        media?: { duration: number; codec: string };
      };
      expect(meta.sha256).toBe(sha);
      expect(meta.media).toBeDefined();
      expect(meta.media!.duration).toBeGreaterThanOrEqual(2.5); // ±0.5 s
      expect(meta.media!.duration).toBeLessThanOrEqual(3.5);
      expect(meta.media!.codec).toBe("h264");

      const plan = planThumbnails({
        mime: "video/mp4",
        workspaceId: s.workspaceId,
        fileId: s.id,
        name: "clip.mp4"
      });
      expect(plan.length).toBe(1);
      expect(row.thumbnailKey).toBe(plan[0]!.key);
      const head = await thumbsClient.headObject(plan[0]!.key);
      expect(head.size).toBeGreaterThan(0);
    }, 60_000);

    // ── 3c. PDF → page count + first-page thumbnail ────────────────────────
    //     Acceptance: `metadata.pdf.pages` correct + first-page PNG.

    it("1-page PDF → READY + metadata.pdf.pages=1 + first-page PNG (pdfjs)", async () => {
      const pdf = await readFixture("doc.pdf");
      const s = await seed(pdf, "doc.pdf", "application/pdf");
      const sha = createHash("sha256").update(pdf).digest("hex");

      const res = await runFilesPipeline(
        { prisma, s3: filesClient, s3Thumbs: thumbsClient, redis, env },
        { fileId: s.id, version: 1, workspaceId: s.workspaceId }
      );
      expect(res).toEqual({ kind: "ready" });

      const row = await prisma.file.findUniqueOrThrow({ where: { id: s.id } });
      expect(row.status).toBe("READY");
      const meta = row.metadata as { sha256?: string; pdf?: { pages: number } };
      expect(meta.sha256).toBe(sha);
      expect(meta.pdf).toEqual({ pages: 1 });

      const plan = planThumbnails({
        mime: "application/pdf",
        workspaceId: s.workspaceId,
        fileId: s.id,
        name: "doc.pdf"
      });
      expect(plan.length).toBe(1);
      expect(plan[0]!.source).toBe("pdf-page-1");
      expect(row.thumbnailKey).toBe(plan[0]!.key);
      const head = await thumbsClient.headObject(plan[0]!.key);
      expect(head.size).toBeGreaterThan(0);
    }, 60_000);

    // ── 4. no scanner configured → dev-skip, still READY ──────────────────

    it("no scanner (dev mode) → READY + scanReport{engine:'none',skipped:true}, thumbnails still generated", async () => {
      const hero = await readFixture("hero-1400.png");
      const s = await seed(hero, "dev-skip.png", "image/png");
      const res = await runFilesPipeline(
        { prisma, s3: filesClient, s3Thumbs: thumbsClient, redis, env: envNoScanner },
        { fileId: s.id, version: 1, workspaceId: s.workspaceId }
      );
      expect(res).toEqual({ kind: "ready" });

      const row = await prisma.file.findUniqueOrThrow({ where: { id: s.id } });
      expect(row.status).toBe("READY");
      expect(row.scanReport).toMatchObject({ engine: "none", skipped: true });
      expect(row.thumbnailKey).toBeTruthy(); // thumbs are independent of the scan
    }, 60_000);

    // ── 6a. retryable failure: storage unreachable → RetryableError ──────
    //     (a transport blip — BullMQ backs off and re-drives the job; the row
    //     is NOT transitioned.)

    it("storage unreachable (fetch fails) → throws RetryableError, row stays SCANNING", async () => {
      const hero = await readFixture("hero-1400.png");
      const s = await seed(hero, "retryable.png", "image/png");
      // A client pointed at a dead port: the signed fetch rejects with a
      // TypeError (ECONNREFUSED) → classified retryable.
      const dead = new S3Client({
        endpoint: "http://127.0.0.1:1",
        region: process.env.S3_REGION ?? "us-east-1",
        bucket: "factory-files",
        accessKey: "minioadmin",
        secretKey: "minioadmin"
      });
      await expect(
        runFilesPipeline(
          { prisma, s3: dead, s3Thumbs: thumbsClient, redis, env },
          { fileId: s.id, version: 1, workspaceId: s.workspaceId }
        )
      ).rejects.toBeInstanceOf(RetryableError);
      // a retryable failure must NOT transition the row (BullMQ re-drives it).
      const row = await prisma.file.findUniqueOrThrow({ where: { id: s.id } });
      expect(row.status).toBe("SCANNING");
    }, 60_000);

    // ── 6b. fatal: the object is gone (404) → FAILED, no retry ────────────
    //     (a retry cannot resurrect a deleted object — classifyError maps
    //     S3 404 to fatal 'upload_not_found'.)

    it("S3 object deleted (404) → {kind:'failed'}, row FAILED (upload_not_found)", async () => {
      const hero = await readFixture("hero-1400.png");
      const s = await seed(hero, "fatal404.png", "image/png");
      await filesClient.deleteObject(s.key); // the object was reaped
      const res = await runFilesPipeline(
        { prisma, s3: filesClient, s3Thumbs: thumbsClient, redis, env },
        { fileId: s.id, version: 1, workspaceId: s.workspaceId }
      );
      expect(res).toEqual({ kind: "failed" });
      const row = await prisma.file.findUniqueOrThrow({ where: { id: s.id } });
      expect(row.status).toBe("FAILED");
      expect(String((row.metadata as any)?.pipeline?.error)).toContain("download");
    }, 60_000);

    // ── 5. missing row → no-op (row deleted between enqueue and run) ─────

    it("file row missing → {kind:'noop', reason:'missing_row'} (no throw)", async () => {
      const ghost = `file_ghost_${randomBytes(6).toString("hex")}`;
      const res = await runFilesPipeline(
        { prisma, s3: filesClient, s3Thumbs: thumbsClient, redis, env },
        { fileId: ghost, version: 1 }
      );
      expect(res).toEqual({ kind: "noop", reason: "missing_row" });
    }, 30_000);
  }
);