//! Storage API routes for S3 file uploads and streaming

mod downloads;
mod gallery;
mod images;
pub(crate) mod media;
mod uploads;
mod versions;

use std::sync::Arc;

use axum::routing::get;
use serde::Serialize;
use sqlx::PgPool;
use tower_governor::GovernorLayer;
use uuid::Uuid;

use crate::{
    AppState, AppStorage,
    config::Config,
    constants,
    csrf::{CsrfRouter, delete_csrf, post_csrf},
    db,
    db::scan_jobs::ScanTargetKind,
    error::{AppError, Result},
    scanning::ScanPipeline,
    storage::FileType,
};

/// Validate a client-declared upload size before it is signed into the
/// presigned URL's `Content-Length`. Checks the static per-type cap and the
/// tier-effective cap. Centralized so every presign handler (item, version,
/// media, gallery) enforces the same bounds instead of re-deriving them, the
/// item path used to be the only one that bound the cap, leaving the largest
/// file classes (version downloads, media video) unenforced at the protocol
/// level. `None` size = the client didn't declare one; the confirm-time HEAD
/// still bounds it.
pub(crate) fn validate_declared_upload_size(
    size: Option<i64>,
    file_type: FileType,
    max_file_bytes: Option<u64>,
) -> Result<()> {
    validate_declared_upload_size_limits(size, file_type, max_file_bytes)?;
    let Some(size) = size else {
        return Ok(());
    };
    // Browser transport ceiling: every caller of THIS function issues a single
    // presigned PUT, which a tab cannot resume. The tier cap can be much higher
    // (BigFiles/Everything allow 20 GB), those files upload through the
    // CLI/desktop clients, which chunk and resume, so point a too-big browser
    // upload there instead of handing out a presigned URL for a transfer that
    // will strand the person if it drops.
    // Deliberately NOT in `validate_declared_upload_size_limits`: it is a
    // property of the one-shot transport, not of the file, and the multipart
    // path is exactly what this message points people toward.
    if size as u64 > constants::BROWSER_UPLOAD_MAX_BYTES {
        let limit_gb = constants::BROWSER_UPLOAD_MAX_BYTES / (1024 * 1024 * 1024);
        return Err(AppError::FileTooLarge(format!(
            "Files larger than {limit_gb} GB must be uploaded with the makenot.work CLI or desktop app."
        )));
    }
    Ok(())
}

/// The size limits that hold for *any* transport: positive, the per-file-type
/// cap, and the tier's per-file cap. The single-PUT ceiling is excluded, so the
/// multipart (CLI/desktop) path can accept files above it while still enforcing
/// every limit that describes the file rather than how it is carried.
pub(crate) fn validate_declared_upload_size_limits(
    size: Option<i64>,
    file_type: FileType,
    max_file_bytes: Option<u64>,
) -> Result<()> {
    let Some(size) = size else {
        return Ok(());
    };
    if size <= 0 {
        return Err(AppError::BadRequest(
            "file_size_bytes must be positive".to_string(),
        ));
    }
    if size as u64 > file_type.max_size() {
        let limit_mb = file_type.max_size() / (1024 * 1024);
        let file_mb = size as u64 / (1024 * 1024);
        return Err(AppError::FileTooLarge(format!(
            "File is {} MB but the maximum for {} files is {} MB.",
            file_mb,
            file_type.as_str(),
            limit_mb
        )));
    }
    if let Some(tier_cap) = max_file_bytes
        && (size as u64) > tier_cap
    {
        let limit_mb = tier_cap / (1024 * 1024);
        return Err(AppError::FileTooLarge(format!(
            "File exceeds your tier's per-file limit of {limit_mb} MB."
        )));
    }
    Ok(())
}

/// Enqueue an orphaned S3 key for the pending-deletion worker.
///
/// This is the ONLY way a route handler may cause an S3 object to be deleted.
/// Direct deletion is sealed off, the `StorageBackend` delete methods require
/// an [`S3DeleteAuthority`](crate::storage::S3DeleteAuthority) that handlers
/// cannot mint (Run #18 CHRONIC B′), so every handler-side delete, whether a
/// post-credit failure (storage credited / row inserted / old object being
/// replaced) or a pre-credit rejection (size cap, type-mismatch, tier check),
/// routes through here. The queue worker applies the `is_s3_key_live` guard
/// before deleting, so enqueuing a key a live row still references is safe (it
/// is skipped), and a transient failure is retried rather than leaking.
pub(crate) async fn enqueue_s3_orphan(
    pool: &sqlx::PgPool,
    s3_key: &str,
    bucket: crate::storage::S3Bucket,
    source: &'static str,
) {
    if let Err(e) = db::pending_s3_deletions::enqueue_deletions(
        pool,
        &[(s3_key.to_string(), bucket.as_str().to_string())],
        source,
    )
    .await
    {
        tracing::warn!(error = ?e, key = %s3_key, bucket = %bucket.as_str(), source = %source, "failed to enqueue orphan S3 key");
    }
}

/// Register S3 upload and streaming routes.
///
/// Upload routes (presign + confirm) are rate limited per IP (see `constants::UPLOAD_RATE_LIMIT_*`).
/// Stream/download endpoints are unlimited (presigned URLs already expire in 1 hour).
pub fn storage_routes() -> CsrfRouter<AppState> {
    let upload_rate_limit = crate::helpers::rate_limiter_ms(
        constants::UPLOAD_RATE_LIMIT_MS,
        constants::UPLOAD_RATE_LIMIT_BURST,
    );

    let upload_routes = CsrfRouter::new()
        .route("/api/upload/presign", post_csrf(uploads::presign_upload))
        .route("/api/upload/confirm", post_csrf(uploads::confirm_upload))
        .route(
            "/api/versions/{version_id}/upload/presign",
            post_csrf(versions::version_presign_upload),
        )
        .route(
            "/api/versions/{version_id}/upload/confirm",
            post_csrf(versions::version_confirm_upload),
        )
        .route(
            "/api/projects/image/presign",
            post_csrf(images::project_image_presign),
        )
        .route(
            "/api/projects/image/confirm",
            post_csrf(images::project_image_confirm),
        )
        .route(
            "/api/items/image/presign",
            post_csrf(images::item_image_presign),
        )
        .route(
            "/api/items/image/confirm",
            post_csrf(images::item_image_confirm),
        )
        .route("/api/gallery/presign", post_csrf(gallery::gallery_presign))
        .route("/api/gallery/confirm", post_csrf(gallery::gallery_confirm))
        .route("/api/gallery/reorder", post_csrf(gallery::gallery_reorder))
        .route_get(
            "/api/gallery/list/{target_type}/{target_id}",
            get(gallery::gallery_list),
        )
        .route(
            "/api/gallery/image/{target_type}/{image_id}",
            delete_csrf(gallery::gallery_delete),
        )
        .route("/api/media/presign", post_csrf(media::media_presign))
        .route("/api/media/confirm", post_csrf(media::media_confirm))
        .route_get("/api/media", get(media::media_list))
        .route_get("/api/media/folders", get(media::media_folders))
        .route("/api/media/{id}", delete_csrf(media::media_delete))
        .route_layer(GovernorLayer::new(upload_rate_limit));

    let stream_rate_limit = crate::helpers::rate_limiter_ms(
        constants::STREAM_RATE_LIMIT_MS,
        constants::STREAM_RATE_LIMIT_BURST,
    );

    let stream_routes = CsrfRouter::new()
        .route_get("/api/stream/{item_id}", get(downloads::stream_url))
        .route_get(
            "/api/versions/{version_id}/download",
            get(downloads::version_download),
        )
        .route_layer(GovernorLayer::new(stream_rate_limit));

    upload_routes.merge(stream_routes)
}

// Shared Request/Response Types

/// JSON response containing the presigned upload URL and S3 key.
#[derive(Debug, Serialize)]
pub struct PresignUploadResponse {
    pub upload_url: String,
    pub s3_key: String,
    pub expires_in: u64,
    /// Cache-Control header the client must send with the S3 PUT (part of the presigned signature).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub cache_control: Option<String>,
    /// Maximum file size in bytes for this upload (for client-side pre-validation).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub max_file_bytes: Option<u64>,
}

/// JSON response confirming a successful upload.
#[derive(Debug, Serialize)]
pub struct ConfirmUploadResponse {
    pub success: bool,
    /// When true, the file was uploaded but is pending manual review before
    /// it becomes available to fans. The creator should see a "pending review"
    /// indicator instead of assuming the file is live.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub pending_review: Option<bool>,
}

// Helpers

/// Discriminates which entity an upload commit applies to, and carries the
/// per-target ID and the corresponding scan_status update.
///
/// Construct one of these in your handler AFTER the entity's DB write has
/// committed, then pass it to [`commit_upload`]. Order matters, see
/// [`commit_upload`] docs.
pub(crate) enum CommitTarget {
    /// An item (Audio/Cover/Video s3_key column on `items`).
    Item(db::ItemId),
    /// A version (`versions` table).
    Version(db::VersionId),
    /// A media library file (`media_files` table).
    Media(db::MediaFileId),
    /// A project cover image (`projects.cover_image_url`/`cover_s3_key`). Gated
    /// by the CDN-image `cover_scan_status` column (migration 162): the worker
    /// stamps it clean/held keyed on s3_key, and on quarantine purges the row.
    ProjectImage(db::ProjectId),
    /// An item cover image (`items.cover_s3_key`/`cover_image_url`), CDN-served
    /// with no per-request gate. It is gated by the separate `cover_scan_status`
    /// column (migration 162), NOT `items.scan_status` (which gates the
    /// audio/video track, flipping that would take the published track offline).
    /// The worker stamps `cover_scan_status` clean/held and, on quarantine, NULLs
    /// the cover columns.
    ItemImage(db::ItemId),
    /// A gallery image row (`item_images`/`project_images`), gated by the row's
    /// `scan_status` column (migration 162). The worker stamps it clean/held by
    /// s3_key and purges on quarantine. Carries the gallery row's own id (for log
    /// correlation only; unused by the worker, which acts on the s3_key).
    GalleryImage(Uuid),
    /// A content insertion clip (`content_insertions`), gated by its
    /// `scan_status` column (migration 162); worker stamps clean/held by key.
    ContentInsertion(db::ContentInsertionId),
}

impl CommitTarget {
    fn kind(&self) -> ScanTargetKind {
        match self {
            CommitTarget::Item(_) => ScanTargetKind::Item,
            CommitTarget::ItemImage(_) => ScanTargetKind::ItemImage,
            CommitTarget::Version(_) => ScanTargetKind::Version,
            CommitTarget::Media(_) => ScanTargetKind::Media,
            CommitTarget::ProjectImage(_) => ScanTargetKind::ProjectImage,
            CommitTarget::GalleryImage(_) => ScanTargetKind::GalleryImage,
            CommitTarget::ContentInsertion(_) => ScanTargetKind::ContentInsertion,
        }
    }

    fn target_uuid(&self) -> Uuid {
        match self {
            CommitTarget::Item(id) | CommitTarget::ItemImage(id) => (*id).into(),
            CommitTarget::Version(id) => (*id).into(),
            CommitTarget::Media(id) => (*id).into(),
            CommitTarget::ProjectImage(id) => (*id).into(),
            CommitTarget::GalleryImage(id) => *id,
            CommitTarget::ContentInsertion(id) => (*id).into(),
        }
    }
}

/// Enqueue a scan job and write the resulting status onto the target entity.
///
/// **Call this AFTER the DB write that commits the upload has succeeded.**
/// Calling it earlier produces three known bug shapes, chronic across four
/// audit runs, which is why the lower-level pieces (`enqueue_scan_for`,
/// `update_*_scan_status`) are gated behind this single entry point:
///
/// 1. A handler that early-returns (idempotent re-confirm, route mismatch,
///    quota rejection) leaks a `scan_jobs` row and flips a Clean status back
///    to Pending, blocking every fan's download until a rescan.
/// 2. A failed DB write leaves a dangling scan_jobs row pointing at an S3
///    key that's about to be deleted.
/// 3. The worker can race the still-uncommitted entity row.
///
/// Use [`CommitTarget`] to bind the target id + per-target status updater.
/// The function returns the `FileScanStatus` that was written (callers use
/// this to populate the `pending_review` field of `ConfirmUploadResponse`).
#[tracing::instrument(
    skip_all,
    name = "storage::commit_upload",
    fields(kind = ?target.kind(), target_id = %target.target_uuid(), %user_id, file_size_bytes, scan_status = tracing::field::Empty)
)]
pub(crate) async fn commit_upload(
    db: &PgPool,
    scanner: Option<&Arc<ScanPipeline>>,
    target: CommitTarget,
    s3_key: &str,
    file_type: FileType,
    user_id: db::UserId,
    file_size_bytes: i64,
) -> Result<db::FileScanStatus> {
    let scan_status = enqueue_scan_for(
        db,
        scanner,
        target.kind(),
        target.target_uuid(),
        s3_key,
        file_type,
        user_id,
        file_size_bytes,
    )
    .await?;
    match target {
        CommitTarget::Item(id) => {
            db::scanning::update_item_scan_status(db, id, scan_status).await?;
        }
        CommitTarget::Version(id) => {
            db::scanning::update_version_scan_status(db, id, scan_status).await?;
        }
        CommitTarget::Media(id) => {
            db::scanning::update_media_file_scan_status(db, id, scan_status).await?;
        }
        CommitTarget::ItemImage(_)
        | CommitTarget::ProjectImage(_)
        | CommitTarget::GalleryImage(_)
        | CommitTarget::ContentInsertion(_) => {
            // Nothing to flip here at commit time: these CDN-served image kinds
            // carry their own per-row scan gate (`cover_scan_status` on
            // items/projects; `scan_status` on item_images/project_images/
            // content_insertions, migration 162), which defaults to 'pending' so
            // the row stays hidden until scanned. The worker stamps it clean/held
            // keyed on s3_key when the scan completes (fail-closed held gate), and
            // on quarantine purges the row / NULLs the cover columns. We must NOT
            // flip `items.scan_status` here, that gates the audio/video track, and
            // a cover re-scan would take the published track offline (Run #20).
        }
    }
    tracing::Span::current().record("scan_status", tracing::field::debug(&scan_status));
    Ok(scan_status)
}

/// Admin-rescan entry point. The entity already exists; we just need to
/// re-run the scan pipeline against its existing `s3_key`. Enqueues the
/// scan job then flips the per-row `scan_status` to Pending in the same
/// order `commit_upload` uses for first-scan, so admin handlers can't
/// invert it (the chronic disease the seal was built to prevent).
#[tracing::instrument(
    skip_all,
    name = "storage::commit_rescan",
    fields(kind = ?target.kind(), target_id = %target.target_uuid(), %user_id, file_size_bytes)
)]
pub(crate) async fn commit_rescan(
    db: &PgPool,
    scanner: Option<&Arc<ScanPipeline>>,
    target: CommitTarget,
    s3_key: &str,
    file_type: FileType,
    user_id: db::UserId,
    file_size_bytes: i64,
) -> Result<db::FileScanStatus> {
    enqueue_scan_for(
        db,
        scanner,
        target.kind(),
        target.target_uuid(),
        s3_key,
        file_type,
        user_id,
        file_size_bytes,
    )
    .await?;
    let pending = db::FileScanStatus::Pending;
    match target {
        CommitTarget::Item(id) => {
            db::scanning::update_item_scan_status(db, id, pending).await?;
        }
        CommitTarget::Version(id) => {
            db::scanning::update_version_scan_status(db, id, pending).await?;
        }
        CommitTarget::Media(id) => {
            db::scanning::update_media_file_scan_status(db, id, pending).await?;
        }
        CommitTarget::ItemImage(_)
        | CommitTarget::ProjectImage(_)
        | CommitTarget::GalleryImage(_)
        | CommitTarget::ContentInsertion(_) => {
            // No per-row scan_status column. An `ItemImage` rescan must not flip
            // `items.scan_status`, that gates the audio/video, not the cover.
        }
    }
    Ok(pending)
}

/// Admin approve-held: promote an item's held gated files (audio/video) to their
/// content keys, then mark the item Clean. A held file sits at its unserved
/// staging key (the scan didn't promote it), so approving it must run the SAME
/// copy-then-repoint the scan worker's Clean path runs, otherwise the item is
/// marked Clean while still pointing at a staging object that is about to be
/// reaped. Factored here (beside `commit_rescan`) so the worker and admin promote
/// can't diverge (the anti-drift discipline the scan-ordering seal uses).
///
/// Idempotent: a file already at a `{owner}/c/...` content key is skipped, so a
/// bulk approve over a mixed set (some already promoted) is safe.
#[tracing::instrument(skip_all, name = "storage::commit_promote_item", fields(%item_id))]
pub(crate) async fn commit_promote_item(
    db: &PgPool,
    storage: &AppStorage,
    config: &Config,
    item_id: db::ItemId,
) -> Result<()> {
    let item = db::items::get_item_by_id(db, item_id)
        .await?
        .ok_or(AppError::NotFound)?;
    let owner = db::items::get_item_owner(db, item_id)
        .await?
        .ok_or(AppError::NotFound)?;
    // Only the files still at a staging key need a copy; skip a file that is
    // absent or already content-keyed (bulk approve over a mixed set). Scan
    // results are recorded against the staging key, so a hash lookup on a content
    // key would miss, hence the staging check gates everything, including the
    // `require_s3` below (a held row with nothing to promote, e.g. no S3 backend
    // configured, still marks Clean without needing storage).
    let staged: Vec<(String, FileType)> = [
        (item.audio_s3_key.clone(), FileType::Audio),
        (item.video_s3_key.clone(), FileType::Video),
    ]
    .into_iter()
    .filter_map(|(key, ft)| match key {
        Some(k) if k.starts_with("staging/") => Some((k, ft)),
        _ => None,
    })
    .collect();
    if !staged.is_empty() {
        let s3 = storage.require_s3()?;
        for (staging_key, file_type) in staged {
            let sha256 = db::scanning::latest_sha256_by_key(db, &staging_key)
                .await?
                .ok_or_else(|| {
                    AppError::Storage(format!(
                        "cannot approve item {item_id}: no scan hash recorded for {staging_key}"
                    ))
                })?;
            crate::scanning::promote_staging_to_content(
                db,
                s3.as_ref(),
                // Item audio is a gated Main-bucket kind; the public backend is
                // unused here but passed for signature consistency.
                storage.public_s3.as_deref(),
                &config.cdn_base_url,
                ScanTargetKind::Item,
                file_type,
                item_id.into(),
                owner,
                &staging_key,
                &sha256,
                crate::storage::S3Bucket::Main,
            )
            .await?;
        }
    }
    // `promote_gated` already set `scan_status = 'clean'` per promoted file; this
    // also covers the no-staged-file case (all already content-keyed) so the
    // admin action still lands the item Clean.
    db::scanning::update_item_scan_status(db, item_id, db::FileScanStatus::Clean).await?;
    Ok(())
}

/// Admin approve-held: promote a version's held download file to its content key,
/// then mark the version Clean. See [`commit_promote_item`] for why the copy must
/// happen at approve time and why it is factored here.
#[tracing::instrument(skip_all, name = "storage::commit_promote_version", fields(%version_id))]
pub(crate) async fn commit_promote_version(
    db: &PgPool,
    storage: &AppStorage,
    config: &Config,
    version_id: db::VersionId,
) -> Result<()> {
    let version = db::versions::get_version_by_id(db, version_id)
        .await?
        .ok_or(AppError::NotFound)?;
    let owner = db::items::get_item_owner(db, version.item_id)
        .await?
        .ok_or(AppError::NotFound)?;
    // `require_s3` only when there is a staging file to copy, a held row with no
    // staging download (or an already content-keyed one) still marks Clean with no
    // storage backend needed.
    if let Some(staging_key) = version.s3_key.clone()
        && staging_key.starts_with("staging/")
    {
        let s3 = storage.require_s3()?;
        let sha256 = db::scanning::latest_sha256_by_key(db, &staging_key)
            .await?
            .ok_or_else(|| {
                AppError::Storage(format!(
                    "cannot approve version {version_id}: no scan hash recorded for {staging_key}"
                ))
            })?;
        crate::scanning::promote_staging_to_content(
            db,
            s3.as_ref(),
            // Version downloads are a gated Main-bucket kind; public backend unused.
            storage.public_s3.as_deref(),
            &config.cdn_base_url,
            ScanTargetKind::Version,
            FileType::Download,
            version_id.into(),
            owner,
            &staging_key,
            &sha256,
            crate::storage::S3Bucket::Main,
        )
        .await?;
    }
    db::scanning::update_version_scan_status(db, version_id, db::FileScanStatus::Clean).await?;
    Ok(())
}

/// Enqueue an async scan job for an uploaded file and return the initial
/// `scan_status` to write onto the target entity.
///
/// **Storage handlers should not call this directly**, use [`commit_upload`]
/// so the ordering invariant (scan-after-DB-commit) cannot be inverted by a
/// future sibling handler. This function remains `pub(super)`-equivalent for
/// the `commit_upload` implementation and for the worker / admin tooling
/// that legitimately needs the lower-level op.
#[allow(clippy::too_many_arguments)]
#[tracing::instrument(
    skip_all,
    name = "storage::enqueue_scan_for",
    fields(kind = ?target_kind, %target_id, %user_id, file_size_bytes)
)]
async fn enqueue_scan_for(
    db: &PgPool,
    scanner: Option<&Arc<ScanPipeline>>,
    target_kind: ScanTargetKind,
    target_id: Uuid,
    s3_key: &str,
    file_type: FileType,
    user_id: db::UserId,
    file_size_bytes: i64,
) -> Result<db::FileScanStatus> {
    if scanner.is_none() {
        let is_trusted = db::users::is_upload_trusted(db, user_id).await?;
        let status = if is_trusted {
            db::FileScanStatus::Clean
        } else {
            db::FileScanStatus::HeldForReview
        };
        tracing::info!(
            scanner = "disabled",
            is_trusted,
            ?status,
            "scanner unavailable; status assigned without enqueue"
        );
        return Ok(status);
    }

    db::scan_jobs::enqueue(
        db,
        target_kind,
        target_id,
        s3_key,
        file_type,
        user_id,
        file_size_bytes,
    )
    .await?;

    Ok(db::FileScanStatus::Pending)
}

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

    const GIB: i64 = 1024 * 1024 * 1024;

    #[test]
    fn browser_validator_refuses_above_the_browser_ceiling() {
        // 10 GiB video: within the 20 GB per-type cap, but a browser issues one
        // unresumable presigned PUT, so it is refused above 2 GiB.
        let err = validate_declared_upload_size(Some(10 * GIB), FileType::Video, None)
            .expect_err("browser upload above the browser ceiling must be refused");
        assert!(
            matches!(err, AppError::FileTooLarge(ref m) if m.contains("CLI or desktop app")),
            "expected a pointer to the CLI, got: {err:?}"
        );
    }

    #[test]
    fn multipart_validator_allows_above_the_browser_ceiling() {
        // The same file over the chunked CLI path is fine, exceeding the
        // one-shot ceiling is the entire point of multipart. If this ever starts
        // failing, the 2 GiB-to-20 GB tier band is unreachable again.
        validate_declared_upload_size_limits(Some(10 * GIB), FileType::Video, None)
            .expect("multipart upload above the browser ceiling must be allowed");
    }

    #[test]
    fn multipart_validator_still_enforces_the_per_type_cap() {
        // Skipping the transport ceiling must not skip the limits that describe
        // the file itself: 25 GiB is past FileType::Video's 20 GB cap.
        assert!(
            validate_declared_upload_size_limits(Some(25 * GIB), FileType::Video, None).is_err()
        );
    }

    #[test]
    fn multipart_validator_still_enforces_the_tier_cap() {
        // A 10 GiB file under a 6 GiB tier cap is refused before it stages parts.
        assert!(
            validate_declared_upload_size_limits(
                Some(10 * GIB),
                FileType::Video,
                Some(6 * GIB as u64)
            )
            .is_err()
        );
    }

    #[test]
    fn both_validators_reject_non_positive_sizes() {
        assert!(validate_declared_upload_size(Some(0), FileType::Video, None).is_err());
        assert!(validate_declared_upload_size_limits(Some(0), FileType::Video, None).is_err());
        assert!(validate_declared_upload_size_limits(Some(-1), FileType::Video, None).is_err());
    }

    #[test]
    fn absent_size_is_accepted_by_both() {
        // No declared size: the confirm-time HEAD is the bound.
        validate_declared_upload_size(None, FileType::Video, None).unwrap();
        validate_declared_upload_size_limits(None, FileType::Video, None).unwrap();
    }
}