//! Database operations for file scan results.

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

use super::FileScanStatus;
use super::ItemId;
use super::UserId;
use super::VersionId;
use crate::scanning::ScanResult;

/// An item held for review, joined with creator info and latest scan layers.
#[derive(Debug, Clone, FromRow)]
pub struct HeldItemRow {
    pub item_id: ItemId,
    pub item_title: String,
    pub s3_key: Option<String>,
    pub creator_username: String,
    pub creator_id: UserId,
    pub upload_trusted: bool,
    pub held_at: DateTime<Utc>,
    /// Latest `file_scan_results.scan_layers` JSON for this entity's s3_key,
    /// or `null` if no scan has run yet. The dashboard renders this as chips.
    pub scan_layers: Option<serde_json::Value>,
}

/// A version held for review, joined with creator info and latest scan layers.
#[derive(Debug, Clone, FromRow)]
pub struct HeldVersionRow {
    pub item_id: ItemId,
    pub item_title: String,
    pub version_id: VersionId,
    pub version_number: String,
    pub s3_key: Option<String>,
    pub creator_username: String,
    pub creator_id: UserId,
    pub upload_trusted: bool,
    pub held_at: DateTime<Utc>,
    pub scan_layers: Option<serde_json::Value>,
}

/// Remove every CDN-served image reference to `s3_key`, across the image
/// surfaces that have no per-row scan gate: the gallery tables
/// (`item_images.s3_key`, `project_images.s3_key`, `content_insertions.storage_key`)
/// and the item *cover* columns on `items` (`cover_s3_key`/`cover_image_url`).
/// Returns the number of rows affected.
///
/// On quarantine of these kinds, removing the reference IS the primary
/// enforcement: it stops the app from ever rendering the (Cloudflare-served) URL
/// again, and, critically, makes the key non-live so the durable S3-deletion
/// queue will actually purge the object instead of parking it behind the
/// `is_s3_key_live` guard. `storage_used` counters self-heal on the weekly
/// `recalculate_all_storage_used` pass; we accept a transient over-count for a
/// malicious upload rather than join through three ownership paths here.
///
/// The item cover is a special case: the bad image lives in columns ON the
/// `items` row, so we NULL the three cover columns rather than DELETE the row,
/// deleting it would take the legitimate audio/video track down with the
/// thumbnail. Both cover references (`cover_s3_key` exact + `cover_image_url`
/// suffix) are registered in `S3_KEY_REFS`, so NULLing both makes the key dead.
#[tracing::instrument(skip_all)]
pub async fn purge_cdn_image_rows_by_key(db: &PgPool, s3_key: &str) -> Result<u64, sqlx::Error> {
    // Wrap all reference-clearing statements in one transaction: a mid-loop
    // failure must not leave a partially-purged malicious reference set (some
    // tables still pointing at a quarantined object while others were cleared).
    let mut tx = db.begin().await?;
    let mut removed = 0u64;
    for sql in [
        "DELETE FROM item_images WHERE s3_key = $1",
        "DELETE FROM project_images WHERE s3_key = $1",
        "DELETE FROM content_insertions WHERE storage_key = $1",
    ] {
        removed += sqlx::query(sql)
            .bind(s3_key)
            .execute(&mut *tx)
            .await?
            .rows_affected();
    }
    // Item + project covers: NULL the columns in place, keeping the row (a track
    // or a project must not be deleted along with its quarantined thumbnail).
    // Matches on the exact `cover_s3_key`; the cover_image_url suffix is cleared
    // in the same statement so neither reference keeps the object live. BOTH the
    // items and projects cover surfaces must be cleared, the read-side
    // counterpart `set_cdn_image_scan_status_by_key` stamps both, and
    // `promote_cdn_image_by_key` promotes both, so a purge that skipped projects
    // would leave a quarantined project cover rendered AND its (public-bucket)
    // object un-reapable behind the `is_s3_key_live` guard.
    for sql in [
        "UPDATE items SET cover_s3_key = NULL, cover_image_url = NULL, \
         cover_file_size_bytes = NULL, updated_at = NOW() WHERE cover_s3_key = $1",
        "UPDATE projects SET cover_s3_key = NULL, cover_image_url = NULL, \
         updated_at = NOW() WHERE cover_s3_key = $1",
    ] {
        removed += sqlx::query(sql)
            .bind(s3_key)
            .execute(&mut *tx)
            .await?
            .rows_affected();
    }
    tx.commit().await?;
    Ok(removed)
}

/// Of the given S3 keys, the subset whose authoritative scan verdict is
/// `Quarantined` (confirmed malicious).
///
/// `file_scan_results` is the per-object scan record the quarantine worker
/// writes; keying on it covers every content surface (audio/cover/video,
/// versions, insertions) uniformly. Used to exclude quarantined objects from
/// bulk key reads that ultimately serve content to a user, notably content
/// export, which collects raw keys and would otherwise hand a creator back their
/// own quarantined object that every download path already refuses (Run 20
/// Security). Absent rows (never scanned) are not quarantined and are not
/// returned.
#[tracing::instrument(skip_all)]
pub async fn quarantined_s3_keys(
    pool: &PgPool,
    keys: &[String],
) -> Result<std::collections::HashSet<String>, sqlx::Error> {
    if keys.is_empty() {
        return Ok(std::collections::HashSet::new());
    }
    let rows: Vec<(String,)> = sqlx::query_as(
        "SELECT DISTINCT s3_key FROM file_scan_results \
         WHERE s3_key = ANY($1) AND scan_status = 'quarantined'",
    )
    .bind(keys)
    .fetch_all(pool)
    .await?;
    Ok(rows.into_iter().map(|(k,)| k).collect())
}

/// Set the per-row `scan_status` on every CDN-served image surface keyed by
/// `s3_key`, in one transaction. The symmetric read-side counterpart of
/// [`purge_cdn_image_rows_by_key`]: where the purge DELETEs (or NULLs) a
/// quarantined reference, this stamps a non-quarantine terminal verdict,
/// `clean` (renders) or `held` (stays hidden), so the fail-closed gate can
/// distinguish "not yet scanned" (`pending`) from "scanned and cleared".
///
/// Touches the same surfaces as the purge: the gallery tables
/// (`item_images.s3_key`, `project_images.s3_key`,
/// `content_insertions.storage_key`) and the cover columns on `items` /
/// `projects` (`cover_s3_key`). Wrapped in a single transaction so a mid-loop
/// failure can't leave the reference set half-stamped (some surfaces cleared to
/// `clean` while others stay `pending`). Returns total rows affected.
#[tracing::instrument(skip_all)]
pub async fn set_cdn_image_scan_status_by_key(
    db: &PgPool,
    s3_key: &str,
    status: FileScanStatus,
) -> Result<u64, sqlx::Error> {
    let mut tx = db.begin().await?;
    let mut affected = 0u64;
    for sql in [
        "UPDATE item_images SET scan_status = $1 WHERE s3_key = $2",
        "UPDATE project_images SET scan_status = $1 WHERE s3_key = $2",
        "UPDATE content_insertions SET scan_status = $1 WHERE storage_key = $2",
        "UPDATE items SET cover_scan_status = $1 WHERE cover_s3_key = $2",
        "UPDATE projects SET cover_scan_status = $1 WHERE cover_s3_key = $2",
    ] {
        affected += sqlx::query(sql)
            .bind(status)
            .bind(s3_key)
            .execute(&mut *tx)
            .await?
            .rows_affected();
    }
    tx.commit().await?;
    Ok(affected)
}

/// Insert a scan result record for audit trail.
#[tracing::instrument(skip_all)]
pub async fn insert_scan_result(
    db: &PgPool,
    s3_key: &str,
    result: &ScanResult,
) -> Result<Uuid, sqlx::Error> {
    let layers_json =
        serde_json::to_value(&result.layers).unwrap_or_else(|_| serde_json::Value::Array(vec![]));

    let id = sqlx::query_scalar::<_, Uuid>(
        r"
        INSERT INTO file_scan_results (s3_key, scan_status, scan_layers, sha256, file_size_bytes)
        VALUES ($1, $2, $3, $4, $5)
        RETURNING id
        ",
    )
    .bind(s3_key)
    .bind(result.status)
    .bind(&layers_json)
    .bind(&result.sha256)
    .bind(result.file_size as i64)
    .fetch_one(db)
    .await?;

    Ok(id)
}

/// Update the scan_status column on an item.
#[tracing::instrument(skip_all)]
pub async fn update_item_scan_status(
    db: &PgPool,
    item_id: ItemId,
    status: FileScanStatus,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        r"
        UPDATE items SET scan_status = $1, updated_at = NOW() WHERE id = $2
        ",
    )
    .bind(status)
    .bind(item_id)
    .execute(db)
    .await?;

    Ok(())
}

/// Update the scan_status column on a version.
#[tracing::instrument(skip_all)]
pub async fn update_version_scan_status(
    db: &PgPool,
    version_id: VersionId,
    status: FileScanStatus,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        r"
        UPDATE versions SET scan_status = $1 WHERE id = $2
        ",
    )
    .bind(status)
    .bind(version_id)
    .execute(db)
    .await?;

    Ok(())
}

/// Update the scan_status column on a media file.
#[tracing::instrument(skip_all)]
pub async fn update_media_file_scan_status(
    db: &PgPool,
    media_file_id: crate::db::MediaFileId,
    status: FileScanStatus,
) -> Result<(), sqlx::Error> {
    let id: uuid::Uuid = media_file_id.into();
    sqlx::query("UPDATE media_files SET scan_status = $1 WHERE id = $2")
        .bind(status)
        .bind(id)
        .execute(db)
        .await?;
    Ok(())
}

/// Most recent content hash recorded for a key, from the scan-result audit
/// trail. The scan worker computes and stores the sha256 in `file_scan_results`
/// at scan time; the admin-approve promote path (which flips a held file to
/// Clean and must copy it to its content-addressed key) has only the staging
/// key in scope, so it reads the hash back here. Empty hashes, recorded by a
/// degraded/held scan that never fully hashed the object, are skipped, so a
/// content key is never derived from a blank digest.
#[tracing::instrument(skip_all)]
pub async fn latest_sha256_by_key(
    db: &PgPool,
    s3_key: &str,
) -> Result<Option<String>, sqlx::Error> {
    sqlx::query_scalar::<_, String>(
        "SELECT sha256 FROM file_scan_results \
         WHERE s3_key = $1 AND sha256 <> '' \
         ORDER BY scanned_at DESC LIMIT 1",
    )
    .bind(s3_key)
    .fetch_optional(db)
    .await
}

/// Repoint a GATED entity (Item audio/video, Version, Media, OTA artifact) from
/// its staging key to the immutable content key and mark it Clean in one write.
///
/// This is the scan-then-promote closing move for the kinds that carry their own
/// `scan_status` gate. The `(table, key column)` pair is a compile-time constant
/// selected by `(kind, file_type)`, never user input, so formatting it into the
/// statement is safe. Uses the runtime `query` (not the `!` macro) so extending
/// the promote set needs no offline-cache regeneration. Accepts any executor so
/// the caller can run it inside the same transaction as the staging-key delete
/// enqueue.
pub async fn promote_gated<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    kind: crate::db::scan_jobs::ScanTargetKind,
    file_type: crate::storage::FileType,
    target_id: Uuid,
    content_key: &str,
) -> Result<(), sqlx::Error> {
    use crate::db::scan_jobs::ScanTargetKind as K;
    use crate::storage::FileType as F;
    let sql: &'static str = match (kind, file_type) {
        (K::Item, F::Audio) => {
            "UPDATE items SET audio_s3_key = $1, scan_status = 'clean', updated_at = NOW() WHERE id = $2"
        }
        (K::Item, F::Video) => {
            "UPDATE items SET video_s3_key = $1, scan_status = 'clean', updated_at = NOW() WHERE id = $2"
        }
        (K::Version, _) => "UPDATE versions SET s3_key = $1, scan_status = 'clean' WHERE id = $2",
        (K::Media, _) => "UPDATE media_files SET s3_key = $1, scan_status = 'clean' WHERE id = $2",
        (K::OtaArtifact, _) => {
            "UPDATE ota_artifacts SET s3_key = $1, scan_status = 'clean' WHERE id = $2"
        }
        (other_kind, other_ft) => {
            // A CDN-image kind (or an Item file role with no key column) must go
            // through `promote_cdn_image_by_key` instead; reaching here is a wiring
            // bug, not a data condition.
            return Err(sqlx::Error::Protocol(format!(
                "promote_gated: {other_kind:?}/{other_ft:?} is not a gated promote target"
            )));
        }
    };
    sqlx::query(sql)
        .bind(content_key)
        .bind(target_id)
        .execute(executor)
        .await?;
    Ok(())
}

/// Repoint every CDN-served image surface that currently references `staging_key`
/// to the immutable `content_key` (and its rebuilt public `content_url`), marking
/// the row Clean. Mirrors [`set_cdn_image_scan_status_by_key`]'s surface list so
/// the promote and the status-stamp can't drift apart. Only the one surface whose
/// key matches is touched; the others no-op. `content_insertions` carries no URL
/// column (it is served presigned, not straight from the CDN) so only its key and
/// status move. Returns the number of rows repointed (expected: exactly 1).
pub async fn promote_cdn_image_by_key(
    conn: &mut sqlx::PgConnection,
    staging_key: &str,
    content_key: &str,
    content_url: &str,
) -> Result<u64, sqlx::Error> {
    let mut affected = 0u64;
    // (SQL, binds_content_url), the surfaces with a materialized public URL take
    // three binds (key, url, where-key); content_insertions takes two.
    for (sql, has_url) in [
        (
            "UPDATE item_images SET s3_key = $1, image_url = $2, scan_status = 'clean' WHERE s3_key = $3",
            true,
        ),
        (
            "UPDATE project_images SET s3_key = $1, image_url = $2, scan_status = 'clean' WHERE s3_key = $3",
            true,
        ),
        (
            "UPDATE items SET cover_s3_key = $1, cover_image_url = $2, cover_scan_status = 'clean' WHERE cover_s3_key = $3",
            true,
        ),
        (
            "UPDATE projects SET cover_s3_key = $1, cover_image_url = $2, cover_scan_status = 'clean' WHERE cover_s3_key = $3",
            true,
        ),
        (
            "UPDATE content_insertions SET storage_key = $1, scan_status = 'clean' WHERE storage_key = $2",
            false,
        ),
    ] {
        let q = if has_url {
            sqlx::query(sql)
                .bind(content_key)
                .bind(content_url)
                .bind(staging_key)
        } else {
            sqlx::query(sql).bind(content_key).bind(staging_key)
        };
        affected += q.execute(&mut *conn).await?.rows_affected();
    }
    Ok(affected)
}

/// Get items held for review, joined with creator info + latest scan layers.
/// Oldest first.
#[tracing::instrument(skip_all)]
pub async fn get_held_items(db: &PgPool) -> Result<Vec<HeldItemRow>, sqlx::Error> {
    let rows = sqlx::query_as::<_, HeldItemRow>(
        r"
        SELECT i.id AS item_id, i.title AS item_title,
               COALESCE(i.audio_s3_key, i.cover_s3_key) AS s3_key,
               u.username AS creator_username, u.id AS creator_id,
               u.upload_trusted, i.updated_at AS held_at,
               (
                   SELECT fsr.scan_layers FROM file_scan_results fsr
                   WHERE fsr.s3_key = COALESCE(i.audio_s3_key, i.cover_s3_key)
                   ORDER BY fsr.scanned_at DESC LIMIT 1
               ) AS scan_layers
        FROM items i
        JOIN projects p ON p.id = i.project_id
        JOIN users u ON u.id = p.user_id
        WHERE i.scan_status = 'held_for_review'
        ORDER BY i.updated_at ASC
        LIMIT 1000
        ",
    )
    .fetch_all(db)
    .await?;

    Ok(rows)
}

/// Get versions held for review, joined with creator info + latest scan layers.
/// Oldest first.
#[tracing::instrument(skip_all)]
pub async fn get_held_versions(db: &PgPool) -> Result<Vec<HeldVersionRow>, sqlx::Error> {
    let rows = sqlx::query_as::<_, HeldVersionRow>(
        r"
        SELECT i.id AS item_id, i.title AS item_title,
               v.id AS version_id, v.version_number,
               v.s3_key,
               u.username AS creator_username, u.id AS creator_id,
               u.upload_trusted, v.created_at AS held_at,
               (
                   SELECT fsr.scan_layers FROM file_scan_results fsr
                   WHERE fsr.s3_key = v.s3_key
                   ORDER BY fsr.scanned_at DESC LIMIT 1
               ) AS scan_layers
        FROM versions v
        JOIN items i ON i.id = v.item_id
        JOIN projects p ON p.id = i.project_id
        JOIN users u ON u.id = p.user_id
        WHERE v.scan_status = 'held_for_review'
        ORDER BY v.created_at ASC
        LIMIT 1000
        ",
    )
    .fetch_all(db)
    .await?;

    Ok(rows)
}

/// Per-layer aggregate stats over a window for the admin dashboard.
#[derive(Debug, Clone, FromRow)]
pub struct LayerHealthRow {
    pub layer: String,
    pub pass_count: i64,
    pub skip_count: i64,
    pub fail_count: i64,
    pub error_count: i64,
    pub last_pass_or_skip: Option<DateTime<Utc>>,
}

/// Compute per-layer health stats over the last N hours.
///
/// Reads `file_scan_results.scan_layers` JSONB and rolls up verdict counts
/// per layer. `last_pass_or_skip` is the most recent timestamp at which the
/// layer returned a non-error, non-fail verdict, the indicator the admin
/// panel uses to flag a layer as down.
#[tracing::instrument(skip_all)]
pub async fn layer_health_window(
    db: &PgPool,
    hours: i64,
) -> Result<Vec<LayerHealthRow>, sqlx::Error> {
    sqlx::query_as::<_, LayerHealthRow>(
        r"
        WITH expanded AS (
            SELECT fsr.scanned_at,
                   (l ->> 'layer')   AS layer,
                   (l ->> 'verdict') AS verdict
            FROM file_scan_results fsr,
                 jsonb_array_elements(fsr.scan_layers) AS l
            WHERE fsr.scanned_at > NOW() - ($1 || ' hours')::interval
        )
        SELECT layer,
               COUNT(*) FILTER (WHERE verdict = 'pass')  AS pass_count,
               COUNT(*) FILTER (WHERE verdict = 'skip')  AS skip_count,
               COUNT(*) FILTER (WHERE verdict = 'fail')  AS fail_count,
               COUNT(*) FILTER (WHERE verdict = 'error') AS error_count,
               MAX(scanned_at) FILTER (WHERE verdict IN ('pass', 'skip')) AS last_pass_or_skip
        FROM expanded
        GROUP BY layer
        ORDER BY layer
        ",
    )
    .bind(hours.to_string())
    .fetch_all(db)
    .await
}

/// A scan-history row for the dashboard's "Recent" grid.
#[derive(Debug, Clone, FromRow)]
pub struct ScanHistoryRow {
    pub scanned_at: DateTime<Utc>,
    pub s3_key: String,
    pub scan_status: String,
    pub sha256: Option<String>,
    pub file_size_bytes: Option<i64>,
    pub scan_layers: serde_json::Value,
}

/// Aggregate counts of entities currently in non-clean states. Used by the
/// PoM health endpoint to alert on growing review backlogs.
#[derive(Debug, Clone)]
pub struct HeldCounts {
    pub held_versions: i64,
    pub held_items: i64,
    pub held_media: i64,
}

#[tracing::instrument(skip_all)]
pub async fn held_counts(db: &PgPool) -> Result<HeldCounts, sqlx::Error> {
    let held_versions: i64 =
        sqlx::query_scalar("SELECT COUNT(*) FROM versions WHERE scan_status = 'held_for_review'")
            .fetch_one(db)
            .await?;
    let held_items: i64 =
        sqlx::query_scalar("SELECT COUNT(*) FROM items WHERE scan_status = 'held_for_review'")
            .fetch_one(db)
            .await?;
    let held_media: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM media_files WHERE scan_status = 'held_for_review'",
    )
    .fetch_one(db)
    .await?;
    Ok(HeldCounts {
        held_versions,
        held_items,
        held_media,
    })
}

/// Recent scan results across all entities. Newest first, capped at `limit`.
/// Used by the Recent History collapsible section. `since_hours` bounds the
/// window so the grid renders fast.
#[tracing::instrument(skip_all)]
pub async fn recent_history(
    db: &PgPool,
    since_hours: i64,
    limit: i64,
) -> Result<Vec<ScanHistoryRow>, sqlx::Error> {
    sqlx::query_as::<_, ScanHistoryRow>(
        r"
        SELECT fsr.scanned_at, fsr.s3_key, fsr.scan_status,
               fsr.sha256, fsr.file_size_bytes, fsr.scan_layers
        FROM file_scan_results fsr
        WHERE fsr.scanned_at > NOW() - ($1 || ' hours')::interval
        ORDER BY fsr.scanned_at DESC
        LIMIT $2
        ",
    )
    .bind(since_hours.to_string())
    .bind(limit)
    .fetch_all(db)
    .await
}