//! Async scan worker.
//!
//! Spawned at startup from `main.rs`. Drains `scan_jobs` via
//! `db::scan_jobs::claim_next` and runs each job through the pipeline. On
//! completion, updates the target entity's `scan_status` (for entities that
//! have one) and creates a WAM ticket on `Quarantined`.
//!
//! See `docs/scan-pipeline-audit.md` § 4.4 for the architecture.

use std::sync::Arc;
use std::time::Duration;

use sqlx::PgPool;
use tokio::sync::Semaphore;
use uuid::Uuid;

use crate::constants;
use crate::db::{
    self, FileScanStatus, ItemId, VersionId,
    scan_jobs::{ScanJob, ScanTargetKind},
};
use crate::storage::{FileType, StorageBackend};
use crate::wam_client::WamClient;

use super::{LayerResult, LayerVerdict, ScanPipeline, ScanResult};

/// Worker poll interval when the queue is empty.
const IDLE_POLL_INTERVAL: Duration = Duration::from_millis(500);

/// How long a `running` job can go without a heartbeat before the reaper resets
/// it. This is measured from the worker's last liveness beat (see
/// [`HEARTBEAT_INTERVAL`]), not from claim time, so a genuinely slow-but-
/// progressing scan is never reaped, only a crashed or hung worker crosses it.
const STUCK_JOB_SECS: i64 = 300;

/// Cadence at which the worker running a job refreshes its `heartbeat_at`. Must
/// be comfortably smaller than [`STUCK_JOB_SECS`] (here ~10x) so a live job that
/// is merely slow keeps beating well inside the reaper's window.
const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(30);

/// Cadence at which any worker tries to reap stuck jobs.
const REAPER_INTERVAL: Duration = Duration::from_mins(1);

/// Aborts a spawned task when dropped, so a job's heartbeat companion never
/// outlives the job, including when `process_job` returns early with an error.
struct AbortOnDrop(tokio::task::JoinHandle<()>);

impl Drop for AbortOnDrop {
    fn drop(&mut self) {
        self.0.abort();
    }
}

/// Run `process_job` while a companion task keeps the job's `heartbeat_at`
/// fresh. The reaper distinguishes a slow-but-alive scan (recent beat) from a
/// crashed worker (stale beat) purely by this heartbeat, so a large scan that
/// runs past `STUCK_JOB_SECS` is no longer reclaimed and double-processed. The
/// companion is aborted the moment the job finishes (via `AbortOnDrop`),
/// error or not, so it cannot bump a job that has already left `running`.
async fn process_job_with_heartbeat(
    ctx: &WorkerContext,
    job: ScanJob,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let job_id = job.id;
    let hb_db = ctx.db.clone();
    let heartbeat = tokio::spawn(async move {
        let mut ticker = tokio::time::interval(HEARTBEAT_INTERVAL);
        // The claim already stamped heartbeat_at, so skip the immediate first
        // tick and beat one interval from now.
        ticker.tick().await;
        loop {
            ticker.tick().await;
            if let Err(e) = db::scan_jobs::bump_heartbeat(&hb_db, job_id).await {
                tracing::warn!(%job_id, error = %e, "scan heartbeat bump failed");
            }
        }
    });
    let _guard = AbortOnDrop(heartbeat);
    process_job(ctx, job).await
}

/// Shared dependencies the worker pool needs.
pub struct WorkerContext {
    pub db: PgPool,
    pub s3: Arc<dyn StorageBackend>,
    pub pipeline: Arc<ScanPipeline>,
    pub scan_semaphore: Arc<Semaphore>,
    pub wam: Option<WamClient>,
    /// Bounded, shutdown-drained background pool. Quarantine side-effects (WAM
    /// ticket, CF purge) go through this rather than a raw `tokio::spawn` so they
    /// can't accumulate unbounded or be dropped on shutdown (Run 9).
    pub bg: crate::background::BackgroundTx,
    /// Cloudflare edge-cache purger, `None` when `CF_API_TOKEN`/`CF_ZONE_ID`
    /// aren't configured. On quarantine we delete the origin object and, if a
    /// purger is present, evict its `cdn_base_url`-prefixed URL from the edge so
    /// an already-cached malicious copy stops serving before its TTL lapses.
    pub cloudflare: Option<crate::cloudflare::CloudflarePurger>,
    /// CDN base URL (e.g. "https://cdn.makenot.work"), used to build the purge
    /// target for a quarantined object's public URL.
    pub cdn_base_url: Arc<str>,
    /// SyncKit-bucket backend. OTA artifacts live here (not the main `s3`), so a
    /// `ScanTargetKind::OtaArtifact` job downloads from this backend instead.
    /// `None` when SyncKit storage isn't configured (OTA scans then fail closed).
    pub synckit_s3: Option<Arc<dyn StorageBackend>>,
    /// Public, CDN-served bucket backend. Staging + scanning always read from
    /// `s3`/`synckit_s3` (private); this backend is used ONLY to promote a Clean
    /// object of an `is_cdn_served_without_gate()` image kind cross-bucket into
    /// the public bucket. `None` when the public bucket isn't configured (image
    /// promotes then fail closed, leaving the entity on its unserved staging key).
    pub public_s3: Option<Arc<dyn StorageBackend>>,
}

/// Decide the final status for a non-quarantined scan.
///
/// First, the pipeline's own status is authoritative when it already says
/// `HeldForReview`: that means a `FailClosed` layer errored (a structural/YARA
/// panic, an oversize-to-spool file, or a reachable-but-incomplete clamav scan),
/// i.e. the file was *not fully scanned*. `ErrorPolicy::FailClosed` means "hold
/// for admin review" unconditionally, including for trusted uploaders, so we
/// never downgrade it here. (CHRONIC S1's fix for trusted uploaders depends on
/// this: the `clamav_incomplete` hold would otherwise be silently turned back
/// into `Clean` below.)
///
/// Otherwise the pipeline returned `Clean`: a trusted uploader normally passes;
/// an untrusted one always routes to admin review. The ClamAV degraded-mode
/// overlay adds one exception, the clamav layer is `FailOpen` (a transport
/// error makes `final_status` skip it → Clean), but accepting a trusted upload
/// on zero AV coverage is the one fail-open we refuse, so ANY clamav error holds
/// the file for admin review rather than passing it (Run 9 Sec-S1: fail closed on
/// the first error instead of waiting for a runtime probe to observe a sustained
/// outage, the probe-lag window is gone).
fn resolve_pass_status(
    pipeline_status: FileScanStatus,
    is_trusted: bool,
    layers: &[LayerResult],
) -> FileScanStatus {
    if pipeline_status == FileScanStatus::HeldForReview {
        return FileScanStatus::HeldForReview;
    }
    if is_trusted && !clamav_layer_errored(layers) {
        FileScanStatus::Clean
    } else {
        FileScanStatus::HeldForReview
    }
}

/// True if the clamav layer reported a transport/scan error on this file. Under
/// the layer's `FailOpen` policy `final_status` skips such an error, so the trust
/// overlay re-reads it to fail closed for trusted uploads.
fn clamav_layer_errored(layers: &[LayerResult]) -> bool {
    layers
        .iter()
        .any(|l| l.layer == "clamav" && l.verdict == LayerVerdict::Error)
}

/// Spawn `n` scan workers on the current tokio runtime. Each worker drains
/// `scan_jobs` independently with FOR UPDATE SKIP LOCKED. A single reaper
/// task per pool resets jobs that get stuck in `running`.
///
/// All tasks observe `shutdown_rx`: when the sender is dropped (or the value
/// changes), they exit on their next idle cycle.
pub fn spawn_pool(
    n: usize,
    ctx: &Arc<WorkerContext>,
    shutdown_rx: tokio::sync::watch::Receiver<()>,
) {
    for worker_id in 0..n {
        let ctx = Arc::clone(ctx);
        let mut shutdown_rx = shutdown_rx.clone();
        tokio::spawn(async move {
            tracing::info!(worker_id, "scan worker started");
            loop {
                match db::scan_jobs::claim_next(&ctx.db).await {
                    Ok(Some(job)) => {
                        let job_id = job.id;
                        if let Err(e) = process_job_with_heartbeat(&ctx, job).await {
                            tracing::error!(worker_id, %job_id, error = %e, "scan job failed");
                            if let Err(e2) =
                                db::scan_jobs::mark_failed(&ctx.db, job_id, &e.to_string()).await
                            {
                                tracing::error!(worker_id, %job_id, error = %e2, "failed to mark job failed");
                            }
                        }
                    }
                    Ok(None) => {
                        tokio::select! {
                            () = tokio::time::sleep(IDLE_POLL_INTERVAL) => {}
                            res = shutdown_rx.changed() => {
                                if res.is_err() {
                                    tracing::info!(worker_id, "scan worker shutting down");
                                    break;
                                }
                            }
                        }
                    }
                    Err(e) => {
                        tracing::error!(worker_id, error = %e, "claim_next failed; backing off");
                        tokio::select! {
                            () = tokio::time::sleep(Duration::from_secs(5)) => {}
                            res = shutdown_rx.changed() => {
                                if res.is_err() {
                                    break;
                                }
                            }
                        }
                    }
                }
            }
        });
    }

    let ctx_reaper = Arc::clone(ctx);
    let mut shutdown_rx = shutdown_rx;
    tokio::spawn(async move {
        loop {
            match db::scan_jobs::reap_stuck(&ctx_reaper.db, STUCK_JOB_SECS).await {
                Ok(n) if n > 0 => {
                    tracing::warn!(
                        reset = n,
                        max_age_secs = STUCK_JOB_SECS,
                        "reset stuck scan jobs"
                    );
                }
                Ok(_) => {}
                Err(e) => tracing::error!(error = %e, "scan job reaper failed"),
            }
            tokio::select! {
                () = tokio::time::sleep(REAPER_INTERVAL) => {}
                res = shutdown_rx.changed() => {
                    if res.is_err() {
                        break;
                    }
                }
            }
        }
    });
}

/// Test/dev helper: claim and process at most one queued scan job synchronously.
/// Returns `Ok(true)` when a job ran, `Ok(false)` when the queue was empty.
/// Mirrors `spawn_pool`'s per-iteration logic without spawning a background
/// task, so integration tests can deterministically drain the queue between
/// upload-confirm and assertion.
pub async fn process_next_for_test(
    ctx: &WorkerContext,
) -> Result<bool, Box<dyn std::error::Error + Send + Sync>> {
    match db::scan_jobs::claim_next(&ctx.db).await? {
        Some(job) => {
            let job_id = job.id;
            if let Err(e) = process_job(ctx, job).await {
                db::scan_jobs::mark_failed(&ctx.db, job_id, &e.to_string()).await?;
                return Err(e);
            }
            Ok(true)
        }
        None => Ok(false),
    }
}

/// Run a single scan job end-to-end. On success the job is marked done; the
/// caller marks failed if this returns an error.
///
/// On pipeline error (e.g. S3 download failure), reset the entity from
/// Scanning back to HeldForReview before bubbling the error up. Otherwise
/// the entity stays stuck at Scanning forever, a real regression we hit
/// in production with stale s3_keys.
#[tracing::instrument(skip_all, fields(%job_id = job.id, target_kind = %job.target_kind, %target_id = job.target_id, attempts = job.attempts))]
async fn process_job(
    ctx: &WorkerContext,
    job: ScanJob,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let job_id = job.id;
    let kind = job
        .typed_kind()
        .ok_or_else(|| format!("unknown target_kind: {}", job.target_kind))?;
    let file_type = job
        .typed_file_type()
        .ok_or_else(|| format!("unknown file_type: {}", job.file_type))?;
    let target_id = job.target_id;
    let started = std::time::Instant::now();

    // Mark target as Scanning while the worker is running (only entities with
    // a scan_status column). This is a visible signal in the admin dashboard
    // queue panel.
    update_entity_status(&ctx.db, kind, target_id, FileScanStatus::Scanning)
        .await
        .ok();

    let entity_status = match run_pipeline_and_decide(ctx, &job, kind, file_type).await {
        Ok(s) => s,
        Err(e) => {
            // Pipeline blew up, most often a stale s3_key. Reset entity to
            // HeldForReview so admins see it on the dashboard and decide
            // whether to delete the orphan record.
            update_entity_status(&ctx.db, kind, target_id, FileScanStatus::HeldForReview)
                .await
                .ok();
            crate::metrics::record_scan_verdict("error");
            crate::metrics::record_scan_duration(started.elapsed().as_secs_f64());
            return Err(e);
        }
    };
    // Stamp the entity's terminal scan_status. For a Clean staging upload the
    // promote inside `run_pipeline_and_decide` already set the key column AND
    // `scan_status = 'clean'` in one transaction; this re-stamp is then a
    // harmless idempotent write. It is load-bearing, though, for a Clean file
    // that was uploaded server-side to a NON-staging key (the build runner's OTA
    // artifacts), there is nothing to promote, so this is the only place its
    // status is cleared. (For the gate-less image kinds this is a no-op; their
    // row was stamped by the promote or the held-image branch above.)
    update_entity_status(&ctx.db, kind, target_id, entity_status).await?;

    // Verdict + duration metrics (Run 20 Observability): quarantine/hold/error
    // rates and scan latency are now graphable at /metrics.
    let verdict_label = match entity_status {
        FileScanStatus::Clean => "clean",
        FileScanStatus::Quarantined => "quarantined",
        FileScanStatus::HeldForReview => "held_for_review",
        FileScanStatus::Pending => "pending",
        FileScanStatus::Scanning => "scanning",
        FileScanStatus::Error => "error",
    };
    crate::metrics::record_scan_verdict(verdict_label);
    crate::metrics::record_scan_duration(started.elapsed().as_secs_f64());

    db::scan_jobs::mark_done(&ctx.db, job_id).await?;
    Ok(())
}

/// Run the pipeline against the S3 object and return the entity status to
/// apply, honoring the size guard, trust gate, and WAM ticketing.
async fn run_pipeline_and_decide(
    ctx: &WorkerContext,
    job: &ScanJob,
    kind: ScanTargetKind,
    file_type: FileType,
) -> Result<FileScanStatus, Box<dyn std::error::Error + Send + Sync>> {
    // Two paths, gated on file size. Small files go through the original
    // buffered `Pipeline::scan(Vec<u8>)`: a single S3 GET into a heap
    // buffer, then layers walk the slice. Big files (>= SCAN_MAX_MEMORY_BYTES)
    // stream from S3 into a tempfile under SCAN_SPOOL_DIR, then layers
    // run against the spooled path (mmap or streamed). The buffered path
    // stays alive: it's the hot path for tip-jar avatars / small audio /
    // download files, and avoiding the tempfile syscall + write matters at
    // that scale. Both branches run S3 IO *outside* the scan_semaphore:
    // the permit bounds the CPU/clamd-heavy scan phase, not network IO.
    // Holding it across the GET serializes downloads at SCAN_MAX_CONCURRENT
    // and lets a scan backlog starve the DB pool.
    // OTA artifacts live in the SyncKit bucket; everything else in the main
    // bucket. `kind.storage_bucket()` is the single source of truth: the same
    // value drives the download client here, the quarantine delete, and the
    // durable-delete enqueue below, so they can't diverge (ultra-fuzz Run #24
    // Storage CRITICAL was a wrong-bucket quarantine delete from exactly that
    // divergence).
    let bucket = kind.storage_bucket();
    let backend: &Arc<dyn StorageBackend> = match bucket {
        crate::storage::S3Bucket::Synckit => ctx.synckit_s3.as_ref().ok_or_else(|| {
            Box::<dyn std::error::Error + Send + Sync>::from(
                "SyncKit storage not configured; cannot scan OTA artifact",
            )
        })?,
        // `storage_bucket()` is the STAGING/scan bucket and is never `Public`
        // (unscanned bytes stay private; only the promoted content object moves
        // to the public bucket, see `content_served_from_public_bucket`). Guard
        // it so a future routing change fails loudly rather than reading a
        // scan object from the wrong bucket.
        crate::storage::S3Bucket::Public => {
            return Err(Box::<dyn std::error::Error + Send + Sync>::from(
                "invariant: a staging/scan object must never live in the public bucket",
            ));
        }
        crate::storage::S3Bucket::Main => &ctx.s3,
    };

    let result: ScanResult = if job.file_size_bytes as u64 > constants::SCAN_SPOOL_MAX_BYTES {
        // Too large to spool for the CPU layers. Hold for admin review under an
        // explicit policy rather than letting `download_into_tempfile` return an
        // Err that marks the job failed, that left the file stuck `Pending`
        // (download-blocked but never resolved) and retried forever. See
        // `super::too_large_to_scan`.
        tracing::warn!(
            job_id = %job.id, size = job.file_size_bytes,
            cap = constants::SCAN_SPOOL_MAX_BYTES,
            "upload exceeds scan spool ceiling; holding for review (not auto-scanned)"
        );
        super::too_large_to_scan(job.file_size_bytes as u64)
    } else if (job.file_size_bytes as usize) < constants::SCAN_MAX_MEMORY_BYTES {
        // `download_object_buf_capped` returns the aggregated body as `Bytes` (no
        // `to_vec` copy); `scan` takes it directly (Run #2 Performance SERIOUS).
        // Bound the aggregation by the recorded size + slack, never above the
        // in-memory threshold: `file_size_bytes` is asserted at upload and could
        // under-report the real object, so cap the buffered read like the spool
        // path does rather than pulling an unbounded body into RAM (Run 22 Perf).
        let cap = (job.file_size_bytes as u64)
            .saturating_add(constants::SCAN_SPOOL_SLACK_BYTES)
            .min(constants::SCAN_MAX_MEMORY_BYTES as u64);
        let data = backend.download_object_buf_capped(&job.s3_key, cap).await?;
        let _permit = ctx.scan_semaphore.acquire().await?;
        Arc::clone(&ctx.pipeline).scan(data, file_type).await
    } else {
        let stream = backend.download_stream(&job.s3_key).await?;
        let spool = super::spool::download_into_tempfile(
            std::path::Path::new(constants::SCAN_SPOOL_DIR),
            &job.id.to_string(),
            &job.s3_key,
            job.file_size_bytes as u64,
            stream,
        )
        .await?;
        let _permit = ctx.scan_semaphore.acquire().await?;
        Arc::clone(&ctx.pipeline)
            .scan_stream(spool, file_type)
            .await
    };

    db::scanning::insert_scan_result(&ctx.db, &job.s3_key, &result).await?;

    if result.status == FileScanStatus::Quarantined {
        let failed_layers: Vec<&str> = result
            .layers
            .iter()
            .filter(|l| l.verdict == LayerVerdict::Fail)
            .map(|l| l.layer)
            .collect();
        if let Some(wam) = ctx.wam.clone() {
            // Fire-and-forget: the WAM call has a multi-second timeout, and the
            // verdict enforcement below (row purge + object delete) must not wait
            // on it, otherwise a WAM outage stalls every quarantine ~5s before
            // the malicious object is removed (Run #21 Performance MODERATE).
            // Matches the spawned-ticket pattern the checkout path already uses.
            let title = format!("File quarantined: {}", job.s3_key);
            let body = format!(
                "Upload by user {} flagged as malicious.\n\
                 Failed layers: {}\nFile type: {file_type:?}\nSize: {}",
                job.user_id,
                failed_layers.join(", "),
                job.file_size_bytes,
            );
            let s3_key = job.s3_key.clone();
            ctx.bg.spawn("malware quarantine ticket", async move {
                wam.create_ticket(
                    &title,
                    Some(&body),
                    "high",
                    "malware-quarantine",
                    Some(&s3_key),
                )
                .await;
            });
        }

        // Enforce the verdict by removing the malicious content. Two parts:
        //
        //   1. DB row (gate-less image kinds only). Project/gallery/content-insertion
        //      images carry no `scan_status` column and no app-proxied download
        //      route, so the only way to stop the (Cloudflare-served) URL from
        //      rendering is to delete the row. Doing it first also makes the
        //      s3_key non-live, so the durable-deletion queue won't park the
        //      object behind the `is_s3_key_live` guard.
        //   2. S3 object (every kind). The per-request `scan_status` gate on
        //      Item/Version/Media stops the *proxied* download, but free
        //      downloadable content is served as a PERMANENT cdn.makenot.work/{key}
        //      URL with no per-request gate, so a leaked or edge-cached URL keeps
        //      serving the malware from origin until the object is gone. Deleting
        //      the object closes that hole for downloadable content the same way
        //      it already does for images.
        //
        // Edge cache: Cloudflare caches objects immutably for a year, so an
        // already-edge-cached copy would survive origin deletion until the cache
        // TTL lapses. After the origin delete below we fire a Cloudflare
        // cache-purge for the object's URL (see the `ctx.cloudflare` block) when
        // `CF_API_TOKEN`/`CF_ZONE_ID` are configured; when they aren't, the purge
        // is a logged no-op and the WAM ticket above remains the manual trigger.
        // The verdict + failed layers stay in file_scan_results for admin review.
        let row_deleted = kind.is_cdn_served_without_gate();
        if row_deleted {
            match db::scanning::purge_cdn_image_rows_by_key(&ctx.db, &job.s3_key).await {
                Ok(n) => tracing::warn!(
                    s3_key = %job.s3_key, target_kind = %kind.as_str(), rows_removed = n,
                    "removed quarantined CDN-served image row(s); URL is no longer rendered"
                ),
                Err(e) => tracing::error!(
                    s3_key = %job.s3_key, target_kind = %kind.as_str(), error = %e,
                    "FAILED to remove quarantined image row(s); the URL may still render until manual removal"
                ),
            }
        }

        // Track whether the malicious object was actually removed (deleted now,
        // or durably enqueued for deletion). A kind that doesn't purge objects
        // has nothing to remove here, so it counts as removed. If removal fails
        // we must NOT report a successful quarantine (see the check below).
        let mut object_removed = !kind.quarantine_purges_object();
        if kind.quarantine_purges_object() {
            // Sanctioned direct delete: the quarantine worker must immediately
            // purge a confirmed-malicious object (the durable queue would park a
            // still-referenced gated key). Mint the authority the sealed delete
            // API requires.
            let auth = crate::storage::S3DeleteAuthority::new();
            match backend
                .delete_object(&auth, &crate::storage::S3Key::from_stored(&job.s3_key))
                .await
            {
                Ok(()) => {
                    tracing::warn!(
                        s3_key = %job.s3_key, target_kind = %kind.as_str(),
                        "purged quarantined object from storage"
                    );
                    object_removed = true;
                }
                Err(e) => {
                    tracing::error!(
                        s3_key = %job.s3_key, target_kind = %kind.as_str(), error = %e,
                        "immediate purge of quarantined object failed"
                    );
                    // The durable-deletion queue only deletes keys the
                    // `is_s3_key_live` guard considers dead. For gate-less image
                    // kinds we just deleted the row, so the key is dead and the
                    // queue will finish the job. For gated kinds (Item/Version/
                    // Media) the entity row still references the key, the queue
                    // would park it indefinitely, so enqueuing is futile and the
                    // job must be retried (below) to re-attempt the direct delete.
                    if row_deleted {
                        match db::pending_s3_deletions::enqueue_deletions(
                            &ctx.db,
                            &[(job.s3_key.clone(), bucket.as_str().to_string())],
                            "malware_quarantine",
                        )
                        .await
                        {
                            Ok(()) => {
                                tracing::warn!(
                                    s3_key = %job.s3_key,
                                    "quarantined object enqueued for durable deletion after direct purge failed"
                                );
                                object_removed = true;
                            }
                            Err(enqueue_err) => tracing::error!(
                                s3_key = %job.s3_key, error = %enqueue_err,
                                "FAILED to enqueue quarantined object for durable deletion; will retry the scan job"
                            ),
                        }
                    } else {
                        tracing::error!(
                            s3_key = %job.s3_key, target_kind = %kind.as_str(),
                            "quarantined object still referenced by its entity row; durable queue cannot delete it, will retry the scan job"
                        );
                    }
                }
            }
        }

        // Never report a successful quarantine while the malware is still live at
        // origin. If we could neither delete nor durably enqueue the object,
        // return an error: `process_job` leaves the job un-done (the scan-job
        // retry budget re-runs the quarantine, a transient S3 delete resolves on
        // retry) and holds the entity for review meanwhile; the WAM ticket above
        // is the manual-removal escalation if retries are exhausted (ultra-fuzz
        // Run 13 Storage: quarantine must not fail-to-success).
        if !object_removed {
            return Err(format!(
                "quarantine incomplete: malicious object {} could neither be purged nor enqueued for deletion",
                job.s3_key
            )
            .into());
        }

        // Evict any edge-cached copy of the now-deleted object. Origin deletion
        // above stops origin serving, but Cloudflare caches cdn.makenot.work/{key}
        // immutably for up to a year, so a previously-fetched malicious URL keeps
        // serving from the edge until its TTL lapses. Fire-and-forget (the CF API
        // has its own latency and must not stall enforcement); a no-op when the
        // purger isn't configured (the WAM ticket stays the fallback).
        // Purging a never-cached URL is harmless, so we don't gate on kind.
        if let Some(cf) = ctx.cloudflare.clone() {
            let url = format!("{}/{}", ctx.cdn_base_url.trim_end_matches('/'), job.s3_key);
            ctx.bg.spawn("malware quarantine cdn purge", async move {
                cf.purge_urls(vec![url]).await;
            });
        }

        return Ok(FileScanStatus::Quarantined);
    }

    // Pipeline returned Clean or HeldForReview. Apply the uploader-trust
    // overlay (untrusted users always route to admin review) plus the ClamAV
    // degraded-mode overlay (a trusted upload whose clamav layer errored is held
    // rather than passed on reduced AV coverage). See `resolve_pass_status`.
    let is_trusted = db::users::is_upload_trusted(&ctx.db, job.user_id).await?;
    let status = resolve_pass_status(result.status, is_trusted, &result.layers);

    // Visibility: an otherwise-clean trusted upload that we held *because* its
    // clamav layer errored. We now fail closed on the first such error (Run 9
    // Sec-S1), so this is no longer a silent acceptance, but a clamd outage that
    // starts holding trusted uploads is still operationally important, so surface
    // it via a metric (scraped) and a WAM ticket (active alert).
    if clamav_degraded_hold_occurred(result.status, is_trusted, &result.layers) {
        tracing::warn!(
            s3_key = %job.s3_key,
            user_id = %job.user_id,
            "clamav layer errored on a trusted upload; held for review on reduced AV coverage"
        );
        crate::metrics::record_clamav_degraded_hold();
        if let Some(wam) = ctx.wam.clone() {
            let body = format!(
                "A trusted upload was held for review because its clamav layer errored \
                 (reduced AV coverage, clamd may be unreachable).\n\n\
                 s3_key: {}\nuser_id: {}",
                job.s3_key, job.user_id
            );
            wam.create_ticket(
                "ClamAV degraded: trusted upload held on reduced AV coverage",
                Some(&body),
                "medium",
                "clamav-degraded-hold",
                Some(&job.s3_key),
            )
            .await;
        }
    }

    // Gate-less CDN-served image kinds (item/project covers, gallery carousels,
    // content-insertion clips) carry no entity `scan_status` column for
    // `update_entity_status` to flip, and they render straight from
    // cdn.makenot.work/{key} with no per-request gate. Stamp their per-row
    // scan_status here, keyed on s3_key (symmetric with the quarantine purge
    // above), so the fail-closed render gate can distinguish an unscanned
    // `pending` row from a cleared `clean` one. We stamp the trust-overlaid
    // `status`, not the raw pipeline `result.status`: an untrusted (or held)
    // upload's image stays hidden until an admin clears it, the same
    // fail-closed posture the gated Item/Version/Media kinds already get.
    // Quarantine never reaches here (it returned above with the row purged).
    // C1 scan-then-promote: a Clean verdict copies the object from its unserved
    // staging key to the immutable content key and repoints the entity (gated
    // kinds by id, CDN-image kinds by staging key, incl. rebuilding the public
    // URL) in one shared step. This is the ONLY place a served key comes into
    // existence, the presign handlers can only mint staging keys (the sealed
    // generators are `pub(crate)`, callable only from here via `content_key`).
    // Fail-safe: a copy/DB error bubbles up so `process_job` leaves the job
    // un-done and the entity keeps pointing at the (unserved) staging key; the
    // retry re-promotes.
    if status == FileScanStatus::Clean && job.s3_key.starts_with("staging/") {
        // Client-presigned upload: copy staging -> content key and repoint the
        // entity. A Clean file at a non-staging key was uploaded server-side to
        // its final key (build-runner OTA) and needs no promote, its status is
        // stamped by `update_entity_status` in `process_job`.
        super::promote_staging_to_content(
            &ctx.db,
            backend.as_ref(),
            ctx.public_s3.as_deref(),
            &ctx.cdn_base_url,
            kind,
            file_type,
            job.target_id,
            job.user_id,
            &job.s3_key,
            &result.sha256,
            bucket,
        )
        .await?;
    } else if status != FileScanStatus::Clean && kind.is_cdn_served_without_gate() {
        // Held/pending image (never promoted): stamp the row keyed on its staging
        // key so the fail-closed render gate hides it until an admin clears it.
        match db::scanning::set_cdn_image_scan_status_by_key(&ctx.db, &job.s3_key, status).await {
            Ok(n) => tracing::info!(
                s3_key = %job.s3_key, target_kind = %kind.as_str(), scan_status = %status, rows = n,
                "stamped CDN-served image scan_status (renders only when clean)"
            ),
            Err(e) => tracing::warn!(
                s3_key = %job.s3_key, target_kind = %kind.as_str(), error = %e,
                "failed to stamp CDN-served image scan_status; image stays hidden until re-scan"
            ),
        }
    }

    Ok(status)
}

/// True when an otherwise-clean trusted upload was held *because* its clamav layer
/// errored (the file would have passed but for the AV-coverage gap). Extracted as
/// a pure predicate so the condition is unit-tested rather than living only inline.
fn clamav_degraded_hold_occurred(
    pipeline_status: FileScanStatus,
    is_trusted: bool,
    layers: &[crate::scanning::LayerResult],
) -> bool {
    pipeline_status == FileScanStatus::Clean && is_trusted && clamav_layer_errored(layers)
}

/// Update the per-entity `scan_status` column for the kinds that have one.
/// `ItemImage` / `ProjectImage` / `GalleryImage` / `ContentInsertion` don't
/// carry their own column, the worker still scanned the file and recorded
/// results, but there's no status to flip on those entities (the `ItemImage`
/// cover shares the `items` row but must NOT flip `items.scan_status`, which
/// gates the audio/video, not the cover).
async fn update_entity_status(
    db: &PgPool,
    kind: ScanTargetKind,
    target_id: Uuid,
    status: FileScanStatus,
) -> Result<(), sqlx::Error> {
    match kind {
        ScanTargetKind::Version => {
            db::scanning::update_version_scan_status(db, VersionId::from(target_id), status).await
        }
        ScanTargetKind::Item => {
            db::scanning::update_item_scan_status(db, ItemId::from(target_id), status).await
        }
        ScanTargetKind::Media => {
            db::scanning::update_media_file_scan_status(
                db,
                db::MediaFileId::from(target_id),
                status,
            )
            .await
        }
        ScanTargetKind::OtaArtifact => {
            db::ota::update_artifact_scan_status(db, db::OtaArtifactId::from(target_id), status)
                .await
        }
        ScanTargetKind::ItemImage
        | ScanTargetKind::ProjectImage
        | ScanTargetKind::GalleryImage
        | ScanTargetKind::ContentInsertion => Ok(()),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn layer(name: &'static str, verdict: LayerVerdict) -> LayerResult {
        LayerResult {
            layer: name,
            verdict,
            detail: None,
        }
    }

    // The first arg is the pipeline's own status (Clean unless a FailClosed
    // layer errored). The cases below exercise a Clean pipeline; the
    // *_not_downgraded tests cover a HeldForReview pipeline.
    const CLEAN: FileScanStatus = FileScanStatus::Clean;

    #[test]
    fn trusted_passes_when_clamav_clean() {
        let layers = [layer("clamav", LayerVerdict::Pass)];
        assert_eq!(
            resolve_pass_status(CLEAN, true, &layers),
            FileScanStatus::Clean
        );
    }

    #[test]
    fn trusted_held_on_first_clamav_error() {
        // Sec-S1 (Run 9): fail closed on the FIRST clamav error. The clamav layer
        // is FailOpen (final_status skipped the error → Clean), but a trusted
        // upload on zero AV coverage is held for review immediately, no waiting
        // for a runtime probe to observe a sustained outage.
        let layers = [layer("clamav", LayerVerdict::Error)];
        assert_eq!(
            resolve_pass_status(CLEAN, true, &layers),
            FileScanStatus::HeldForReview
        );
    }

    #[test]
    fn untrusted_always_held() {
        let layers = [layer("clamav", LayerVerdict::Pass)];
        assert_eq!(
            resolve_pass_status(CLEAN, false, &layers),
            FileScanStatus::HeldForReview
        );
    }

    #[test]
    fn trusted_held_when_pipeline_already_held_not_downgraded() {
        // CHRONIC S1: a reachable-but-incomplete clamav scan (or any FailClosed
        // layer error) makes final_status = HeldForReview. A trusted uploader must
        // NOT have that downgraded to Clean.
        let layers = [layer("clamav_incomplete", LayerVerdict::Error)];
        assert_eq!(
            resolve_pass_status(FileScanStatus::HeldForReview, true, &layers),
            FileScanStatus::HeldForReview
        );
    }

    #[test]
    fn pipeline_hold_survives_even_with_no_clamav_layer() {
        // The no-downgrade guard is layer-agnostic: any FailClosed hold (e.g. an
        // oversize-to-spool file or a structural panic) is honored for trusted
        // uploaders too.
        let layers = [layer("scan_size_limit", LayerVerdict::Error)];
        assert_eq!(
            resolve_pass_status(FileScanStatus::HeldForReview, true, &layers),
            FileScanStatus::HeldForReview
        );
    }

    // ── clamav_degraded_hold_occurred (observability predicate) ──

    #[test]
    fn degraded_hold_detected_for_trusted_clamav_error() {
        let layers = [layer("clamav", LayerVerdict::Error)];
        assert!(clamav_degraded_hold_occurred(CLEAN, true, &layers));
    }

    #[test]
    fn degraded_hold_not_for_untrusted_upload() {
        // Untrusted uploads are always held regardless of clamav; the degraded
        // metric tracks only the trusted-upload coverage-gap case.
        let layers = [layer("clamav", LayerVerdict::Error)];
        assert!(!clamav_degraded_hold_occurred(CLEAN, false, &layers));
    }

    #[test]
    fn degraded_hold_not_when_clamav_passed() {
        let layers = [layer("clamav", LayerVerdict::Pass)];
        assert!(!clamav_degraded_hold_occurred(CLEAN, true, &layers));
    }

    #[test]
    fn degraded_hold_not_when_pipeline_not_clean() {
        // A file the pipeline already held (FailClosed) wasn't held *because* of
        // the clamav coverage gap, so it is not a degraded-hold event.
        let layers = [layer("clamav", LayerVerdict::Error)];
        assert!(!clamav_degraded_hold_occurred(
            FileScanStatus::HeldForReview,
            true,
            &layers
        ));
    }
}