/**
 * apps/worker — files queue processor (Phase 9, TASK-0102).
 *
 * Job: `process-file { fileId, version, workspaceId }` on the `files`
 * BullMQ queue (enqueued by the API on upload completion, TASK-0101).
 *
 * Pipeline (sequential, each step timed + logged as
 * `file:{fileId}:step:{name}`):
 *
 *   download   → S3 GET streamed to the temp dir (sha256 in the same pass)
 *   scan       → ClamAV over the local copy (unix socket `zCONTSCAN` /
 *                TCP `zINSTREAM` / dev-skip when unconfigured)
 *   metadata   → magic-byte re-sniff (final authority) + sharp dims /
 *                ffprobe media / pdfjs pages (best-effort, capped)
 *   thumbnails → images 3 sizes / pdf page 1 / video first frame → thumbs
 *                bucket (via @factory/storage planThumbnails)
 *   finalize   → status READY|VIRUS_DETECTED|FAILED + metadata.pipeline +
 *                FileVersion upsert + `file.uploaded` (exactly once) /
 *                `file.finalized` / `file.virus_detected` + audit
 *
 * Status machine (enforced — see status.ts):
 *   UPLOADING → SCANNING → READY | VIRUS_DETECTED | FAILED
 *   READY / VIRUS_DETECTED are terminal (a re-run is a NO-OP); any other
 *   transition throws InvalidTransitionError (an invariant violation).
 *
 * Error policy (see errors.ts):
 *   retryable (network, S3 5xx, clamd timeout) → THROWN → BullMQ backoff
 *   fatal (corrupt object, missing row, disk guard) → row FAILED +
 *   `metadata.pipeline.error`, job DISCARDED (no retry — it can't be fixed)
 *
 * The Phase-0 scaffold (../files.processor.ts) is superseded by this module.
 */
import { Worker, type Job } from "bullmq";
import type { Redis } from "ioredis";
import { z } from "zod";
import type { PrismaClient, File } from "@prisma/client";
import { QUEUES } from "../../queues.js";
import { getLogger } from "../../logger.js";
import type { WorkerEnv } from "../../config.js";
import type { S3Client } from "../../lib/s3.js";
import { detectMimeType } from "@factory/storage";
import { asJson } from "../../lib/prisma.js";
import { withJobOtel } from "../../lib/job-otel.js";
import { audit } from "../../lib/notify.js";
import { assertTransition } from "./status.js";
import { classifyError, FatalError, RetryableError } from "./errors.js";
import { createTempFile, removeTempFile } from "./temp.js";
import { scanStep, type ScanOutcome } from "./steps/scan.js";
import { metadataStep } from "./steps/metadata.js";
import { thumbnailsStep } from "./steps/thumbnails.js";
import { finalizeStep, type FileRowLike } from "./steps/finalize.js";
import type { FilePipelineContext, StepName, ThumbsResult } from "./types.js";

const log = getLogger("files");

/** The exact job data TASK-0101 enqueues. */
export const processFileJobSchema = z.object({
  fileId: z.string().min(1),
  version: z.number().int().positive().default(1),
  workspaceId: z.string().min(1).optional()
});

export type ProcessFileJob = z.infer<typeof processFileJobSchema>;

/** The dependencies the pipeline needs (dependency-injected for tests). */
export interface FilesPipelineDeps {
  prisma: PrismaClient;
  s3: S3Client;
  /** The thumbs-bucket client (same creds, `factory-thumbs`). */
  s3Thumbs: S3Client;
  redis: Redis;
  env: WorkerEnv;
}

/** The terminal outcome of a pipeline run. */
export type PipelineRunResult =
  | { kind: "ready" }
  | { kind: "virus" }
  | { kind: "failed" }
  | { kind: "noop"; reason: "missing_row" | "terminal" };

/**
 * Run the full files pipeline for one job. Exported (not just the Worker
 * callback) so it is unit/integration-testable WITHOUT a live BullMQ Job —
 * the other processors use the same "exported handler" seam (e.g.
 * `handleSearchEvent`).
 *
 * Contract:
 *   - returns `{kind:"ready"|"virus"|"failed"|"noop"}` on a NORMAL terminal
 *     state (the caller completes the job — for "failed" the row is already
 *     FAILED, for "noop" nothing changed);
 *   - THROWS a RetryableError on a transient failure (network / S3 5xx /
 *     clamd timeout) so BullMQ backs off;
 *   - the temp file is ALWAYS cleaned up (finally).
 */
export async function runFilesPipeline(
  deps: FilesPipelineDeps,
  data: ProcessFileJob
): Promise<PipelineRunResult> {
  const { prisma, s3, s3Thumbs, redis, env } = deps;
  const started = Date.now();
  const { fileId, version } = data;
  log.info({ fileId, version }, "processing file");

  const row = (await prisma.file.findUnique({ where: { id: fileId } })) as
    | File
    | null;
  if (!row) {
    // Row gone (deleted/cancelled) — nothing to do, no retry.
    log.warn({ fileId }, "file row missing — discarding");
    return { kind: "noop", reason: "missing_row" };
  }
  const prevStatus = row.status;

  // Re-run idempotency: READY / VIRUS_DETECTED are terminal. A later re-scan
  // of a VIRUS_DETECTED file is an explicit TASK-0104 operator action, never
  // a BullMQ retry.
  if (prevStatus === "READY" || prevStatus === "VIRUS_DETECTED") {
    log.info({ fileId, prevStatus }, "file terminal — no-op re-run");
    return { kind: "noop", reason: "terminal" };
  }

  const ctx: FilePipelineContext = {
    job: { fileId, version, workspaceId: data.workspaceId ?? row.workspaceId },
    prisma,
    s3,
    s3Thumbs,
    redis,
    env,
    temp: { path: "", size: 0 },
    stepTimings: new Map<string, number>(),
    warnings: [],
    extracted: { sniffedMime: null },
    sha256: ""
  };
  const mark = (name: StepName, ms: number): void => {
    ctx.stepTimings.set(name, ms);
    log.info({ fileId, step: name, ms }, `file:${fileId}:step:${name}`);
  };

  // A step failure that is FATAL marks the row FAILED and returns {failed}
  // (done, no retry); a RETRYABLE failure is rethrown so BullMQ backs off.
  const guard = async (step: string, fn: () => Promise<void>): Promise<void> => {
    const t0 = Date.now();
    try {
      await fn();
      mark(step as StepName, Date.now() - t0);
    } catch (err) {
      const classified = classifyError(err);
      if (classified instanceof FatalError) {
        log.error(
          { fileId, step, code: classified.code, err: classified.message },
          "pipeline step failed FATAL — marking FAILED (no retry)"
        );
        await markFailed(ctx, row, `${step}: ${classified.message}`, classified.code);
        throw new FatalHandledMarker(classified.code, step);
      }
      log.warn(
        { fileId, step, err: classified.message },
        "pipeline step failed RETRYABLE — backing off"
      );
      throw classified; // BullMQ retries per job options
    }
  };

  try {
    // Ensure SCANNING (a stranded UPLOADING row re-driven by maintenance).
    if (prevStatus === "UPLOADING") {
      assertTransition("UPLOADING", "SCANNING");
      await prisma.file.update({ where: { id: fileId }, data: { status: "SCANNING" } });
    }

    // 1) download (streamed to temp, sha256 in-pass) ───────────────────────
    await guard("download", async () => {
      const path = await createTempFile(env, env.FILES_TEMP_FREE_MIN_BYTES);
      const { size, sha256 } = await s3.downloadTo(row.key, path);
      ctx.temp = { path, size };
      ctx.sha256 = sha256;
    });

    // 2) scan ───────────────────────────────────────────────────────────────
    let scan: ScanOutcome | undefined;
    await guard("scan", async () => {
      scan = await scanStep(ctx);
    });
    const scanOutcome = scan!;
    if (scanOutcome.infected) {
      // Terminal: VIRUS_DETECTED (no thumbs, no index, no file.uploaded).
      assertTransition(prevStatus, "VIRUS_DETECTED");
      await finalizeStep(ctx, {
        outcome: { kind: "virus", scan: scanOutcome.report },
        prevStatus,
        file: row,
        finalMime: row.mimeType,
        size: ctx.temp.size
      });
      return { kind: "virus" };
    }

    // 3) metadata (re-sniff = final authority) ──────────────────────────────
    let sniffedMime: string | null = null;
    await guard("metadata", async () => {
      sniffedMime = await sniff(ctx);
      const finalMime = sniffedMime ?? row.mimeType;
      if (!sniffedMime) {
        ctx.warnings.push("mime: no magic-byte signal — keeping declared type");
      }
      await metadataStep(ctx, finalMime);
    });
    const finalMime = sniffedMime ?? row.mimeType;

    // 4) thumbnails ─────────────────────────────────────────────────────────
    let thumbs: ThumbsResult | undefined;
    await guard("thumbnails", async () => {
      thumbs = await thumbnailsStep(ctx, finalMime);
    });

    // 5) finalize READY ─────────────────────────────────────────────────────
    assertTransition(prevStatus, "READY");
    await finalizeStep(ctx, {
      outcome: {
        kind: "ready",
        extracted: ctx.extracted,
        thumbs: thumbs!,
        scan: scanOutcome.report
      },
      prevStatus,
      file: row,
      finalMime,
      size: ctx.temp.size
    });
    await stampDuration(ctx, row, Date.now() - started);
    log.info(
      { fileId, size: ctx.temp.size, ms: Date.now() - started },
      "file processed READY"
    );
    return { kind: "ready" };
  } catch (err) {
    // A FATAL was already handled (row FAILED) — surface a {failed} result so
    // the caller completes the job WITHOUT a BullMQ retry.
    if (err instanceof FatalHandledMarker) {
      return { kind: "failed" };
    }
    // Retryable (or unexpected): rethrow → BullMQ backoff.
    throw err;
  } finally {
    // The object never stays on disk past the run (the cron backstops a
    // SIGKILL mid-run).
    await removeTempFile(ctx.temp.path);
  }
}

export function createFilesWorker(opts: {
  connection: Redis;
  prisma: PrismaClient;
  env: WorkerEnv;
  s3: S3Client;
  /** The thumbs-bucket client (same creds, `factory-thumbs`). */
  s3Thumbs: S3Client;
  concurrency?: number;
}): Worker {
  const { connection, prisma, env, s3, s3Thumbs } = opts;

  return new Worker(
    QUEUES.files,
    withJobOtel(QUEUES.files, async (job: Job) => {
      const data = processFileJobSchema.parse(job.data);
      try {
        const result = await runFilesPipeline(
          { prisma, s3, s3Thumbs, redis: connection, env },
          data
        );
        // "failed" (handled fatal — row already FAILED) and "noop" (missing
        // row / already terminal) → the job is done; discard keeps it out of
        // the retry/DLQ path.
        if (result.kind === "failed" || result.kind === "noop") {
          job.discard();
        }
      } catch (err) {
        // RetryableError → let BullMQ retry (job options at enqueue).
        if (err instanceof RetryableError) throw err;
        // Anything else is a genuine unexpected error → let BullMQ retry too
        // (the pipeline's own fatal/retryable split already handles the
        // classified cases; reaching here means an unclassified throw, which
        // we treat as retryable-but-loud rather than silently completing).
        throw new RetryableError(`unhandled pipeline error: ${(err as Error).message}`, err);
      }
    }),
    {
      connection,
      // TASK-0102 §7: files pipeline is IO+CPU bound (scan/sharp/ffmpeg) →
      // its own lower concurrency, not the general WORKER_CONCURRENCY.
      concurrency: opts.concurrency ?? env.FILES_WORKER_CONCURRENCY
      // NOTE: the retry policy (3 attempts, exponential 2 s backoff,
      // removeOnComplete 500 / removeOnFail 200 — TASK-0102 §7) is applied at
      // ENQUEUE time by the API (uploads.service.enqueueProcessFile), not here.
      // BullMQ WorkerOptions has no per-worker defaultJobOptions; job options
      // ride on the enqueued job.
    }
  );
}

/**
 * Internal marker: a FATAL failure was already handled (row FAILED + audit)
 * and the job was discarded. Swallowed by BullMQ's failure path — it must
 * NOT be treated as a new retryable error. (BullMQ sees a throw; the row and
 * the discard already made the outcome final.)
 */
class FatalHandledMarker extends Error {
  constructor(code: string, step: string) {
    super(`fatal (handled) ${step} ${code}`);
    this.name = "FatalHandledMarker";
  }
}

/** Magic-byte re-sniff of the leading bytes (final authority; null = no signal). */
async function sniff(ctx: FilePipelineContext): Promise<string | null> {
  const { open } = await import("node:fs/promises");
  const fd = await open(ctx.temp.path, "r");
  try {
    const buf = Buffer.alloc(16 * 1024);
    const { bytesRead } = await fd.read(buf, 0, buf.length, 0);
    if (bytesRead === 0) return null;
    const det = detectMimeType(buf.subarray(0, bytesRead));
    return det.confidence === "unknown" ? null : det.mime;
  } finally {
    await fd.close();
  }
}

/** Mark the row FAILED with `metadata.pipeline.error` + audit (never throws). */
async function markFailed(
  ctx: FilePipelineContext,
  row: File,
  error: string,
  code: string
): Promise<void> {
  try {
    const cur = (await ctx.prisma.file.findUnique({ where: { id: row.id } })) as File | null;
    if (!cur) return;
    // A FAILED transition is legal from UPLOADING/SCANNING (and READY on
    // re-scan); assert so an unexpected state is visible, not silent.
    assertTransition(cur.status, "FAILED");
    const meta = (cur.metadata as Record<string, unknown> | null) ?? {};
    const pipeline = {
      scannedAt: new Date().toISOString(),
      sha256: ctx.sha256 || undefined,
      durationMs: 0,
      steps: [...ctx.stepTimings.entries()].map(([name, ms]) => ({ name, ms })),
      error,
      ...(ctx.warnings.length ? { warnings: [...ctx.warnings] } : {})
    };
    await ctx.prisma.file.update({
      where: { id: row.id },
      data: { status: "FAILED", metadata: asJson({ ...meta, pipeline }) }
    });
    await audit(ctx.prisma, {
      workspaceId: row.workspaceId,
      resourceType: "file",
      resourceId: row.id,
      action: "file.process_failed",
      metadata: { code, error, key: row.key }
    });
  } catch (err) {
    log.error({ fileId: row.id, err: (err as Error).message }, "markFailed failed (row may be inconsistent)");
  }
}

/** Stamp the total pipeline wall time into `metadata.pipeline.durationMs`. */
async function stampDuration(
  ctx: FilePipelineContext,
  row: File,
  totalMs: number
): Promise<void> {
  const cur = (await ctx.prisma.file.findUnique({ where: { id: row.id } })) as File | null;
  if (!cur) return;
  const meta = (cur.metadata as Record<string, unknown> | null) ?? {};
  const pipeline = (meta.pipeline as Record<string, unknown> | undefined) ?? {};
  await ctx.prisma.file.update({
    where: { id: row.id },
    data: { metadata: asJson({ ...meta, pipeline: { ...pipeline, durationMs: totalMs } }) }
  });
}

/** Re-export for tests (the status-machine error). */
export type { FileRowLike };
export { RetryableError, FatalError };