//! Async scan-job queue for the malware pipeline.
//!
//! Upload routes call [`enqueue`] to register a scan job and return their
//! request to the client. A pool of scan workers (`crate::scanning::worker`)
//! drains the queue with `FOR UPDATE SKIP LOCKED` via [`claim_next`], runs the
//! pipeline against the S3 object, and finalizes the job with [`mark_done`] or
//! [`mark_failed`].
//!
//! The pipeline those workers run is `crate::scanning`; the admin review
//! surface over its verdicts is `crate::routes::admin::uploads`.

use chrono::{DateTime, Utc};
use sqlx::{FromRow, PgPool};
use uuid::Uuid;

use crate::storage::FileType;

use super::UserId;

/// The entity whose `scan_status` the worker should update when the scan
/// completes. `Item`, `Version`, and `Media` have `scan_status` columns;
/// `ProjectImage` and `ContentInsertion` do not, for those, the worker
/// still scans (recording results in `file_scan_results`) but only acts on
/// `Quarantined` by creating a WAM ticket for admin follow-up.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ScanTargetKind {
    Item,
    Version,
    Media,
    ProjectImage,
    ItemImage,
    GalleryImage,
    ContentInsertion,
    OtaArtifact,
}

impl ScanTargetKind {
    pub fn as_str(&self) -> &'static str {
        match self {
            ScanTargetKind::Item => "item",
            ScanTargetKind::Version => "version",
            ScanTargetKind::Media => "media",
            ScanTargetKind::ProjectImage => "project_image",
            ScanTargetKind::ItemImage => "item_image",
            ScanTargetKind::GalleryImage => "gallery_image",
            ScanTargetKind::ContentInsertion => "content_insertion",
            ScanTargetKind::OtaArtifact => "ota_artifact",
        }
    }

    /// Decode the DB `target_kind` text. Named `from_db_str` (not `from_str`) so
    /// it doesn't shadow the `FromStr` trait method, it returns `Option`, not the
    /// `Result` that trait requires, and pairs with `as_str` above.
    pub fn from_db_str(s: &str) -> Option<Self> {
        Some(match s {
            "item" => ScanTargetKind::Item,
            "version" => ScanTargetKind::Version,
            "media" => ScanTargetKind::Media,
            "project_image" => ScanTargetKind::ProjectImage,
            "item_image" => ScanTargetKind::ItemImage,
            "gallery_image" => ScanTargetKind::GalleryImage,
            "content_insertion" => ScanTargetKind::ContentInsertion,
            "ota_artifact" => ScanTargetKind::OtaArtifact,
            _ => return None,
        })
    }

    /// The S3 bucket this kind's objects live in. Single source of truth for
    /// the scan worker's client selection, quarantine delete, and durable-delete
    /// enqueue, all three MUST agree, or a quarantine deletes from the wrong
    /// bucket and idempotently "succeeds" while the malware persists (ultra-fuzz
    /// Run #24 Storage CRITICAL). OTA artifacts are the only synckit-bucket kind
    /// today; this mirrors the `S3_KEY_REFS` registry in `pending_s3_deletions`.
    pub fn storage_bucket(&self) -> crate::storage::S3Bucket {
        match self {
            ScanTargetKind::OtaArtifact => crate::storage::S3Bucket::Synckit,
            _ => crate::storage::S3Bucket::Main,
        }
    }

    /// Whether this kind carries no `scan_status` column and is rendered
    /// straight from `cdn.makenot.work/{key}`, so a `Quarantined` verdict can
    /// only be enforced by deleting the *DB row* (which stops the URL from
    /// rendering and makes the key non-live for the deletion queue).
    ///
    /// This governs the DB-row step only. The S3-object purge on quarantine is
    /// unconditional (see [`quarantine_purges_object`]); these kinds
    /// additionally need their row removed (or, for `ItemImage`, the cover
    /// columns NULLed) because they have no status column to gate on.
    ///
    /// `ItemImage` is here because the item *cover* (`items.cover_image_url`) is
    /// rendered straight from the CDN with no per-request gate, `items.scan_status`
    /// gates only the audio/video. On quarantine the purge NULLs the cover columns
    /// (keeping the track), it does NOT delete the item row.
    pub fn is_cdn_served_without_gate(&self) -> bool {
        matches!(
            self,
            ScanTargetKind::ItemImage
                | ScanTargetKind::ProjectImage
                | ScanTargetKind::GalleryImage
                | ScanTargetKind::ContentInsertion
        )
    }

    /// Whether this kind's *content* object (post-promote) is served UNSIGNED
    /// from the public CDN bucket, and so must be promoted cross-bucket into
    /// [`crate::storage::S3Bucket::Public`]. This is the three image cover/gallery
    /// kinds, the `is_cdn_served_without_gate` set MINUS `ContentInsertion`,
    /// which, despite carrying no per-request scan gate, is served *presigned*
    /// from the private bucket (see `promote_cdn_image_by_key`: it alone has no
    /// materialized public URL). The matching `S3_KEY_REFS` entries
    /// (`item_images`, `project_images`, `items.cover_s3_key`,
    /// `projects.cover_s3_key`) therefore live under bucket `public`, while
    /// `content_insertions.storage_key` stays `main`. Staging is ALWAYS private
    /// ([`storage_bucket`] is unchanged), only the promoted content object moves.
    pub fn content_served_from_public_bucket(&self) -> bool {
        matches!(
            self,
            ScanTargetKind::ItemImage | ScanTargetKind::ProjectImage | ScanTargetKind::GalleryImage
        )
    }

    /// Whether a `Quarantined` verdict purges the underlying S3 object.
    ///
    /// Always true: a confirmed-malicious object has no reason to remain in
    /// storage. Crucially, the per-request `scan_status` gate on
    /// `Item`/`Version`/`Media` is NOT sufficient on its own, free downloadable
    /// content is handed out as a *permanent* `cdn.makenot.work/{key}` URL with
    /// no per-request gate (see `routes/storage/downloads.rs::resolve_content_url`),
    /// so a leaked or edge-cached URL keeps serving the malware from origin until
    /// the object itself is gone. Removing the object closes that hole for
    /// downloadable content the same way it does for the gate-less image kinds.
    /// The decision lives here, beside `is_cdn_served_without_gate`, so the two
    /// halves of quarantine enforcement can't silently diverge per kind.
    pub fn quarantine_purges_object(&self) -> bool {
        true
    }
}

/// A queued or running scan job, as claimed by a worker. Most fields are
/// populated via `FromRow` from sqlx; fields not consumed by the worker
/// today are kept for the admin dashboard (Phase 2 of the audit).
#[allow(dead_code)]
#[derive(Debug, Clone, FromRow)]
pub struct ScanJob {
    pub id: Uuid,
    pub target_kind: String,
    pub target_id: Uuid,
    pub s3_key: String,
    pub file_type: String,
    pub user_id: UserId,
    pub file_size_bytes: i64,
    pub status: String,
    pub attempts: i32,
    pub enqueued_at: DateTime<Utc>,
    pub started_at: Option<DateTime<Utc>>,
    /// Last liveness beat from the worker running this job. Bumped on a cadence
    /// well under STUCK_JOB_SECS; the reaper keys off this (falling back to
    /// `started_at`) so a slow-but-progressing scan isn't mistaken for a
    /// crashed worker. See `bump_heartbeat` and migration 167.
    pub heartbeat_at: Option<DateTime<Utc>>,
    pub completed_at: Option<DateTime<Utc>>,
    pub last_error: Option<String>,
}

impl ScanJob {
    pub fn typed_kind(&self) -> Option<ScanTargetKind> {
        ScanTargetKind::from_db_str(&self.target_kind)
    }

    pub fn typed_file_type(&self) -> Option<FileType> {
        self.file_type.parse().ok()
    }
}

/// Enqueue a scan job. Returns the job id.
#[tracing::instrument(skip_all, fields(target_kind = target_kind.as_str(), %target_id, s3_key))]
pub async fn enqueue(
    db: &PgPool,
    target_kind: ScanTargetKind,
    target_id: Uuid,
    s3_key: &str,
    file_type: FileType,
    user_id: UserId,
    file_size_bytes: i64,
) -> Result<Uuid, sqlx::Error> {
    let id = sqlx::query_scalar::<_, Uuid>(
        r"
        INSERT INTO scan_jobs
            (target_kind, target_id, s3_key, file_type, user_id, file_size_bytes)
        VALUES ($1, $2, $3, $4, $5, $6)
        RETURNING id
        ",
    )
    .bind(target_kind.as_str())
    .bind(target_id)
    .bind(s3_key)
    .bind(file_type.as_str())
    .bind(user_id)
    .bind(file_size_bytes)
    .fetch_one(db)
    .await?;

    Ok(id)
}

/// Maximum times a scan job may be claimed before it is given up as `failed`.
/// Bounds the crash-loop retry budget: a job that reliably wedges the OS process
/// (so it's reaped from `running` rather than reaching `mark_failed`) can only
/// be re-attempted this many times before `reap_stuck` retires it.
pub const MAX_SCAN_ATTEMPTS: i32 = 5;

/// Atomically claim the next queued scan job for processing.
///
/// Uses `FOR UPDATE SKIP LOCKED` so multiple workers can drain the queue in
/// parallel without contention. Sets `status='running'`, increments `attempts`,
/// and stamps `started_at`. Skips jobs that have already hit `MAX_SCAN_ATTEMPTS`
/// (those are retired to `failed` by `reap_stuck`). Returns `Ok(None)` if the
/// queue is empty.
#[tracing::instrument(skip_all)]
pub async fn claim_next(db: &PgPool) -> Result<Option<ScanJob>, sqlx::Error> {
    let job = sqlx::query_as::<_, ScanJob>(
        r"
        WITH next AS (
            SELECT id FROM scan_jobs
            WHERE status = 'queued' AND attempts < $1
            ORDER BY enqueued_at ASC
            FOR UPDATE SKIP LOCKED
            LIMIT 1
        )
        UPDATE scan_jobs
        SET status = 'running',
            attempts = attempts + 1,
            started_at = NOW(),
            heartbeat_at = NOW()
        WHERE id = (SELECT id FROM next)
        RETURNING *
        ",
    )
    .bind(MAX_SCAN_ATTEMPTS)
    .fetch_optional(db)
    .await?;

    Ok(job)
}

/// Mark a job as completed successfully.
#[tracing::instrument(skip_all, fields(%job_id))]
pub async fn mark_done(db: &PgPool, job_id: Uuid) -> Result<(), sqlx::Error> {
    sqlx::query(
        r"
        UPDATE scan_jobs
        SET status = 'done', completed_at = NOW(), last_error = NULL
        WHERE id = $1
        ",
    )
    .bind(job_id)
    .execute(db)
    .await?;
    Ok(())
}

/// Mark a job as failed (worker exception, S3 fetch failure, etc.).
///
/// Records `last_error` for admin inspection. The job stays in `failed`
/// status; an admin or operator can manually re-enqueue via the dashboard
/// (Phase 2) by inserting a fresh row.
#[tracing::instrument(skip_all, fields(%job_id))]
pub async fn mark_failed(db: &PgPool, job_id: Uuid, err: &str) -> Result<(), sqlx::Error> {
    sqlx::query(
        r"
        UPDATE scan_jobs
        SET status = 'failed', completed_at = NOW(), last_error = $1
        WHERE id = $2
        ",
    )
    .bind(err)
    .bind(job_id)
    .execute(db)
    .await?;
    Ok(())
}

/// Refresh a running job's liveness heartbeat.
///
/// Called periodically by the worker that owns the job so `reap_stuck` (and the
/// PoM stuck-count) can distinguish a slow-but-progressing scan (fresh beat)
/// from a crashed or hung worker (stale beat). The `status = 'running'` guard
/// makes a late beat a no-op once the job has already finished or been reaped,
/// it never resurrects a terminal row's timestamp.
#[tracing::instrument(skip_all, fields(%job_id))]
pub async fn bump_heartbeat(db: &PgPool, job_id: Uuid) -> Result<(), sqlx::Error> {
    sqlx::query("UPDATE scan_jobs SET heartbeat_at = NOW() WHERE id = $1 AND status = 'running'")
        .bind(job_id)
        .execute(db)
        .await?;
    Ok(())
}

/// Reset jobs whose worker has gone silent for longer than `max_age_secs`.
///
/// Run on worker startup and on a timer to recover from a previous-process crash
/// mid-scan: the row would otherwise stay `running` forever and never be
/// re-claimed. Liveness is measured from the job's last heartbeat
/// (`COALESCE(heartbeat_at, started_at)`), NOT from `started_at` alone, a
/// legitimately slow scan (large object over a slow S3 link) can run past
/// `max_age_secs` while still making progress, and reaping *that* let a second
/// worker double-process the same object and inflate `attempts` toward
/// MAX_SCAN_ATTEMPTS until a valid file was force-retired (audit Run 22). The
/// running worker bumps `heartbeat_at` on a cadence well under `max_age_secs`,
/// so only a crashed/hung process (no fresh beat) crosses the threshold. A job
/// whose `attempts` has already reached `MAX_SCAN_ATTEMPTS` is retired to
/// `failed`; otherwise it returns to `queued` for another attempt.
#[tracing::instrument(skip_all)]
pub async fn reap_stuck(db: &PgPool, max_age_secs: i64) -> Result<u64, sqlx::Error> {
    let affected = sqlx::query(
        r"
        UPDATE scan_jobs
        SET status = CASE WHEN attempts >= $2 THEN 'failed' ELSE 'queued' END,
            started_at = NULL,
            heartbeat_at = NULL,
            completed_at = CASE WHEN attempts >= $2 THEN NOW() ELSE completed_at END,
            last_error = CASE WHEN attempts >= $2
                THEN 'exceeded max scan attempts (reaped from running)'
                ELSE last_error END
        WHERE status = 'running'
          AND COALESCE(heartbeat_at, started_at) < NOW() - ($1 || ' seconds')::interval
        ",
    )
    .bind(max_age_secs.to_string())
    .bind(MAX_SCAN_ATTEMPTS)
    .execute(db)
    .await?
    .rows_affected();
    Ok(affected)
}

/// Delete terminal-state rows older than `older_than`. Returns the count.
///
/// Only touches `done`/`failed` rows, operational state (`queued`,
/// `running`) is owned by the worker loop and `reap_stuck`. The verdict
/// (Clean / Quarantined / HeldForReview) lives on the entity's
/// `scan_status` column, not here, so dropping a `done` row loses queue
/// history only, not malware-detection state.
///
/// No supporting index today: at soft-launch volume Postgres seq-scans the
/// table fine. Revisit once `EXPLAIN ANALYZE` shows it as a bottleneck.
#[tracing::instrument(skip_all)]
pub async fn purge_old_terminal(
    db: &PgPool,
    older_than: chrono::Duration,
) -> Result<u64, sqlx::Error> {
    let cutoff = chrono::Utc::now() - older_than;
    let n = sqlx::query(
        r"
        DELETE FROM scan_jobs
        WHERE status IN ('done', 'failed')
          AND COALESCE(completed_at, started_at, enqueued_at) < $1
        ",
    )
    .bind(cutoff)
    .execute(db)
    .await?
    .rows_affected();
    Ok(n)
}

/// Count of currently-queued jobs. Used by the admin dashboard health panel
/// (Phase 2 of the audit). Allowed dead code until that route lands.
#[allow(dead_code)]
pub async fn queued_count(db: &PgPool) -> Result<i64, sqlx::Error> {
    sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM scan_jobs WHERE status = 'queued'")
        .fetch_one(db)
        .await
}

/// Count of currently-running jobs. Phase 2 dashboard consumer.
#[allow(dead_code)]
pub async fn running_count(db: &PgPool) -> Result<i64, sqlx::Error> {
    sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM scan_jobs WHERE status = 'running'")
        .fetch_one(db)
        .await
}

/// Count of running jobs whose worker has gone silent longer than
/// `max_age_secs`. Used by PoM to alert on stuck workers. Keyed off the
/// heartbeat (falling back to `started_at`) so a slow-but-alive scan isn't
/// counted as stuck, matching `reap_stuck`'s liveness definition.
pub async fn stuck_count(db: &PgPool, max_age_secs: i64) -> Result<i64, sqlx::Error> {
    sqlx::query_scalar::<_, i64>(
        "SELECT COUNT(*) FROM scan_jobs WHERE status = 'running' AND COALESCE(heartbeat_at, started_at) < NOW() - ($1 || ' seconds')::interval",
    )
    .bind(max_age_secs.to_string())
    .fetch_one(db)
    .await
}

/// A held version with enough context to re-enqueue it for scanning.
#[allow(dead_code)]
#[derive(Debug, Clone, FromRow)]
pub struct RescanCandidateVersion {
    pub version_id: Uuid,
    pub s3_key: String,
    pub file_size_bytes: i64,
    pub user_id: UserId,
}

/// A held item (audio or cover) with re-enqueue context.
#[allow(dead_code)]
#[derive(Debug, Clone, FromRow)]
pub struct RescanCandidateItem {
    pub item_id: Uuid,
    pub s3_key: String,
    pub file_size_bytes: i64,
    pub user_id: UserId,
    /// Audio / cover, selected by the query.
    pub file_type: String,
}

/// Find currently-held versions with enough context to be re-scanned.
pub async fn rescan_candidates_versions(
    db: &PgPool,
) -> Result<Vec<RescanCandidateVersion>, sqlx::Error> {
    sqlx::query_as::<_, RescanCandidateVersion>(
        r"
        SELECT v.id AS version_id,
               v.s3_key,
               COALESCE(v.file_size_bytes, 0) AS file_size_bytes,
               p.user_id
        FROM versions v
        JOIN items i ON i.id = v.item_id
        JOIN projects p ON p.id = i.project_id
        WHERE v.scan_status = 'held_for_review'
          AND v.s3_key IS NOT NULL
        ",
    )
    .fetch_all(db)
    .await
}

/// Find currently-held items (audio or cover) with re-enqueue context.
pub async fn rescan_candidates_items(db: &PgPool) -> Result<Vec<RescanCandidateItem>, sqlx::Error> {
    sqlx::query_as::<_, RescanCandidateItem>(
        r"
        SELECT i.id AS item_id,
               COALESCE(i.audio_s3_key, i.cover_s3_key) AS s3_key,
               COALESCE(i.audio_file_size_bytes, i.cover_file_size_bytes, 0) AS file_size_bytes,
               p.user_id,
               CASE WHEN i.audio_s3_key IS NOT NULL THEN 'audio' ELSE 'cover' END AS file_type
        FROM items i
        JOIN projects p ON p.id = i.project_id
        WHERE i.scan_status = 'held_for_review'
          AND COALESCE(i.audio_s3_key, i.cover_s3_key) IS NOT NULL
        ",
    )
    .fetch_all(db)
    .await
}

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

    #[test]
    fn target_kind_round_trip() {
        for kind in [
            ScanTargetKind::Item,
            ScanTargetKind::Version,
            ScanTargetKind::Media,
            ScanTargetKind::ProjectImage,
            ScanTargetKind::ItemImage,
            ScanTargetKind::GalleryImage,
            ScanTargetKind::ContentInsertion,
        ] {
            assert_eq!(ScanTargetKind::from_db_str(kind.as_str()), Some(kind));
        }
        assert_eq!(ScanTargetKind::from_db_str("bogus"), None);
    }

    #[test]
    fn all_kinds_purge_object_on_quarantine() {
        // The S3-object purge is unconditional, a confirmed-malicious object
        // must never survive in storage, including downloadable Item/Version/
        // Media content reachable via a permanent CDN URL.
        for kind in [
            ScanTargetKind::Item,
            ScanTargetKind::Version,
            ScanTargetKind::Media,
            ScanTargetKind::ProjectImage,
            ScanTargetKind::ItemImage,
            ScanTargetKind::GalleryImage,
            ScanTargetKind::ContentInsertion,
            ScanTargetKind::OtaArtifact,
        ] {
            assert!(
                kind.quarantine_purges_object(),
                "{} must purge object",
                kind.as_str()
            );
        }
    }

    #[test]
    fn storage_bucket_matches_download_and_delete_path() {
        // Regression for ultra-fuzz Run #24 Storage CRITICAL: the quarantine
        // worker deleted OTA objects from the main bucket (idempotent success)
        // while the object persisted in the synckit bucket. `storage_bucket()`
        // is now the one source of truth the worker uses for the download
        // client, the delete, and the durable-delete enqueue, pin it here so a
        // new synckit-bucket kind can't silently default to Main.
        use crate::storage::S3Bucket;
        use ScanTargetKind::*;
        assert_eq!(OtaArtifact.storage_bucket(), S3Bucket::Synckit);
        for kind in [
            Item,
            Version,
            Media,
            ProjectImage,
            ItemImage,
            GalleryImage,
            ContentInsertion,
        ] {
            assert_eq!(
                kind.storage_bucket(),
                S3Bucket::Main,
                "{} is a main-bucket kind",
                kind.as_str()
            );
        }
    }

    #[test]
    fn only_gateless_image_kinds_delete_their_row() {
        // The DB-row deletion is the gate-less-image-only half of enforcement.
        use ScanTargetKind::*;
        // ItemImage is gate-less: the item *cover* is CDN-served with no
        // per-request gate; `items.scan_status` gates the audio/video, not the
        // cover (Run #20 Storage SERIOUS).
        for kind in [ItemImage, ProjectImage, GalleryImage, ContentInsertion] {
            assert!(
                kind.is_cdn_served_without_gate(),
                "{} is gate-less",
                kind.as_str()
            );
        }
        for kind in [Item, Version, Media] {
            assert!(
                !kind.is_cdn_served_without_gate(),
                "{} has a gate/status",
                kind.as_str()
            );
        }
    }
}