//! Presigned upload and confirm handlers for item content.

use axum::{Json, extract::State, response::IntoResponse};
use serde::Deserialize;
use sqlx::PgPool;
use std::str::FromStr;

use crate::{
    AppStorage, Scanning,
    auth::AuthUser,
    db::{self, ItemId},
    error::{AppError, Result, ResultExt},
    storage::{CACHE_CONTROL_IMMUTABLE, FileType, S3Client},
};

use super::{CommitTarget, ConfirmUploadResponse, PresignUploadResponse, commit_upload};

/// JSON input for requesting a presigned S3 upload URL.
///
/// `file_size_bytes` is optional for compatibility with older clients that
/// don't know the size ahead of time, but when supplied it is signed into
/// the presigned URL's `Content-Length` and S3 will reject any PUT whose
/// actual body length differs, protocol-level enforcement of the per-file
/// cap, no bandwidth wasted on oversized uploads that the confirm step
/// would have rejected anyway.
#[derive(Debug, Deserialize)]
pub(super) struct PresignUploadRequest {
    pub item_id: ItemId,
    pub file_type: String,
    pub file_name: String,
    pub content_type: String,
    #[serde(default)]
    pub file_size_bytes: Option<i64>,
}

/// JSON input for confirming a completed S3 upload.
#[derive(Debug, Deserialize)]
pub(super) struct ConfirmUploadRequest {
    pub item_id: ItemId,
    pub file_type: String,
    pub s3_key: String,
}

/// Generate a presigned URL for uploading a file to S3
///
/// POST /api/upload/presign
///
/// Requires authentication. User must own the item.
#[tracing::instrument(skip_all, name = "storage::presign_upload", fields(user_id = %user.id))]
pub(super) async fn presign_upload(
    State(db): State<PgPool>,
    State(storage): State<AppStorage>,
    AuthUser(user): AuthUser,
    Json(req): Json<PresignUploadRequest>,
) -> Result<impl IntoResponse> {
    user.check_not_suspended()?;
    // Check if S3 is configured
    let s3 = storage.require_s3()?;

    // Parse file type
    let file_type = FileType::from_str(&req.file_type)
        .map_err(|_| AppError::BadRequest(format!("Invalid file type: {}", req.file_type)))?;

    S3Client::validate_content_type(file_type, &req.content_type)?;

    S3Client::validate_extension(file_type, &req.file_name)?;

    // Verify user owns the item
    let owner = db::items::get_item_owner(&db, req.item_id)
        .await?
        .ok_or(AppError::NotFound)?;

    if owner != user.id {
        return Err(AppError::Forbidden);
    }

    // Early quota check (reject before generating presigned URL)
    db::creator_tiers::check_presign_allowed(&db, user.id, file_type).await?;

    // Get the effective max file size for client-side pre-validation
    let max_file_bytes =
        db::creator_tiers::get_effective_max_file_bytes(&db, user.id, file_type).await?;

    // If the client declared the file size, validate it before signing, both
    // against the static per-type cap and the tier-effective cap. The size is
    // bound as Content-Length below so S3 rejects oversized PUTs at the protocol
    // level. Clients omitting `file_size_bytes` fall back to the old behavior
    // (no protocol-level enforcement; confirm step still validates).
    super::validate_declared_upload_size(req.file_size_bytes, file_type, max_file_bytes)?;

    // Presign to an unserved staging key, never the served key. After a Clean
    // scan the worker copies the object to its content-addressed served key and
    // deletes this staging object, so a re-PUT to this presigned URL after the
    // scan can't change the bytes a buyer is served (C1).
    let s3_key = S3Client::generate_staging_key(&req.file_name);

    // Track the pending upload so the reaper can clean it up if never confirmed
    db::pending_uploads::record_pending_upload(&db, user.id, &s3_key, "main").await?;

    // Generate presigned upload URL with immutable cache headers
    let expires_in = 3600; // 1 hour
    let upload_url = s3
        .presign_upload(
            &s3_key,
            &req.content_type,
            Some(expires_in),
            Some(CACHE_CONTROL_IMMUTABLE),
            req.file_size_bytes,
        )
        .await
        .context("presign upload for item content")?;

    Ok(Json(PresignUploadResponse {
        upload_url,
        s3_key: s3_key.into_string(),
        expires_in,
        cache_control: Some(CACHE_CONTROL_IMMUTABLE.to_string()),
        max_file_bytes,
    }))
}

/// Confirm that an upload has completed and update the database
///
/// POST /api/upload/confirm
///
/// Requires authentication. User must own the item.
#[tracing::instrument(skip_all, name = "storage::confirm_upload", fields(user_id = %user.id))]
pub(super) async fn confirm_upload(
    State(db): State<PgPool>,
    State(storage): State<AppStorage>,
    State(scanning): State<Scanning>,
    AuthUser(user): AuthUser,
    Json(req): Json<ConfirmUploadRequest>,
) -> Result<impl IntoResponse> {
    user.check_not_suspended()?;
    // Check if S3 is configured
    let s3 = storage.require_s3()?;

    // Parse file type
    let file_type = FileType::from_str(&req.file_type)
        .map_err(|_| AppError::BadRequest(format!("Invalid file type: {}", req.file_type)))?;

    // Verify user owns the item
    let owner = db::items::get_item_owner(&db, req.item_id)
        .await?
        .ok_or(AppError::NotFound)?;

    if owner != user.id {
        return Err(AppError::Forbidden);
    }

    // Ownership of the staging key is proved below (after the idempotent
    // re-confirm short-circuit) via `pending_uploads`, since a `staging/{uuid}`
    // key carries no user/item in its path for a prefix check to bind.

    // Verify the object exists in S3
    if !s3.object_exists(&req.s3_key).await? {
        return Err(AppError::BadRequest(
            "Upload not found. Please try uploading again.".to_string(),
        ));
    }

    // Enforce file size limit (static per-type limit)
    let file_size_bytes = s3.object_size(&req.s3_key).await?.ok_or_else(|| {
        AppError::BadRequest(
            "Could not determine file size. Please try uploading again.".to_string(),
        )
    })?;
    if file_size_bytes as u64 > file_type.max_size() {
        super::enqueue_s3_orphan(
            &db,
            &req.s3_key,
            crate::storage::S3Bucket::Main,
            "item_upload_rejected",
        )
        .await;
        let limit_mb = file_type.max_size() / (1024 * 1024);
        let file_mb = file_size_bytes 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
        )));
    }

    // Enforce tier-based limits (per-file + storage cap)
    let max_storage =
        match db::creator_tiers::check_upload_allowed(&db, user.id, file_type, file_size_bytes)
            .await
        {
            Ok(max) => max,
            Err(e) => {
                super::enqueue_s3_orphan(
                    &db,
                    &req.s3_key,
                    crate::storage::S3Bucket::Main,
                    "item_upload_rejected",
                )
                .await;
                return Err(e);
            }
        };

    // Resolve, from the single exhaustive declaration on `FileType`, how this
    // type is confirmed on an `items` row, and reject types that belong to a
    // dedicated route BEFORE any scan enqueue or scan_status flip. (A misrouted
    // but valid item_id would otherwise flip scan_status to Pending, block every
    // fan's download, and leak a scan_jobs row for an S3 key we're about to
    // delete.) Cover is rejected here now: it also needs `cover_image_url`,
    // which only /api/items/image/confirm writes, the old generic two-column
    // path left it NULL and rendered an invisible cover (Run #13 SERIOUS).
    if let crate::storage::GenericItemConfirm::UseRoute(route) = file_type.generic_item_confirm() {
        super::enqueue_s3_orphan(
            &db,
            &req.s3_key,
            crate::storage::S3Bucket::Main,
            "item_upload_rejected",
        )
        .await;
        return Err(AppError::BadRequest(format!(
            "This file type isn't confirmed here. Use {route}."
        )));
    }

    // Idempotency + replace detection. We read the item ONCE here and reuse its
    // project_id for the cache bump below (previously a second get_item_by_id).
    let mut old_s3_key: Option<String> = None;
    let mut replaced_old_size: i64 = 0;
    let mut item_project_id: Option<db::ProjectId> = None;
    if let Some(item) = db::items::get_item_by_id(&db, req.item_id).await? {
        item_project_id = Some(item.project_id);
        let existing_key = match file_type {
            FileType::Audio => item.audio_s3_key.as_deref(),
            FileType::Cover => item.cover_s3_key.as_deref(),
            FileType::Video => item.video_s3_key.as_deref(),
            _ => None,
        };
        if existing_key == Some(&req.s3_key) {
            // Idempotent re-confirm: the entity already references this s3_key.
            // Still clear the pending_uploads row, otherwise the orphan reaper
            // will fire 24h later and delete the live S3 object out from under
            // a perfectly happy DB row (Run #7 HIGH-1).
            if let Err(e) =
                db::pending_uploads::remove_pending_upload(&db, user.id, &req.s3_key, "main").await
            {
                tracing::warn!(error = ?e, key = %req.s3_key, "remove_pending_upload failed on idempotent re-confirm");
            }
            return Ok(Json(ConfirmUploadResponse {
                success: true,
                pending_review: None,
            }));
        }
        if let Some(old_key) = existing_key {
            old_s3_key = Some(old_key.to_string());
            replaced_old_size = match file_type {
                FileType::Audio => item.audio_file_size_bytes.unwrap_or(0),
                FileType::Cover => item.cover_file_size_bytes.unwrap_or(0),
                FileType::Video => item.video_file_size_bytes.unwrap_or(0),
                _ => 0,
            };
        }
    }
    let is_replace = old_s3_key.is_some();

    // Authorize the staging key for a *fresh* confirm: the caller must have
    // presigned it (recorded against them in `pending_uploads`). The idempotent
    // re-confirm above already returned, it consumed the pending row on the
    // first confirm and is authorized by the item ownership check + the entity
    // already referencing the key, so this gate only guards new writes. A
    // `staging/{uuid}` key has no owner in its path, so this lookup (not a prefix
    // check) is what prevents confirming another user's staging object.
    // Do NOT enqueue this key for orphan deletion on failure: an unowned staging
    // key is (at most) another user's in-flight upload, and the deletion queue
    // would treat the not-yet-referenced staging object as dead and delete it,
    // a cross-user griefing delete. Just reject; the real owner's confirm or the
    // pending-upload reaper handles their key.
    if !db::pending_uploads::is_owned(&db, user.id, &req.s3_key, "main").await? {
        return Err(AppError::BadRequest("Invalid upload key".to_string()));
    }

    // Storage credit + item UPDATE in ONE transaction. A rollback restores both,
    // so a mid-write failure can't leave the storage counter inflated against a
    // row that never got the key (the previous compensating-action path with
    // swallowed `.ok()` errors). The write goes through the sealed
    // `update_item_file_cas`, whose `IS NOT DISTINCT FROM old_key` guard makes a
    // concurrent double-confirm lose the race (`LostRace`) instead of
    // double-crediting storage or clobbering the live object. `commit_upload`
    // (scan enqueue + scan_status flip) stays AFTER the commit, that ordering
    // is the blessed path; see `routes/storage/mod.rs::commit_upload`. `Ok(0)`
    // signals the item vanished or lost the CAS race (tx rolled back, nothing
    // charged).
    let tx_result: Result<u64> = async {
        let mut tx = db.begin().await?;
        db::creator_tiers::try_apply_storage_on(
            &mut tx,
            user.id,
            is_replace.then_some(replaced_old_size),
            file_size_bytes,
            max_storage,
        )
        .await?;
        match db::items::update_item_file_cas(
            &mut *tx,
            req.item_id,
            user.id,
            file_type,
            old_s3_key.as_deref(),
            &req.s3_key,
            file_size_bytes,
        )
        .await?
        {
            db::items::FileConfirmOutcome::LostRace => {
                // Leave the tx uncommitted, drop rolls it back, undoing the
                // storage change with no manual compensation.
                Ok(0)
            }
            db::items::FileConfirmOutcome::Committed => {
                // Enqueue the OLD object for deletion in the SAME tx as the row
                // swap, closing the crash-between-commit-and-enqueue orphan
                // window (ultra-fuzz Run #1 Storage LOW; mirrors the project-
                // image replace path in images.rs). After commit the row points
                // at the new key, so the old key is non-live; the worker's
                // is_s3_key_live check is the backstop if a row still references
                // it.
                if let Some(old_key) = old_s3_key.as_deref() {
                    db::pending_s3_deletions::enqueue_deletions(
                        &mut *tx,
                        &[(old_key.to_string(), "main".to_string())],
                        "item_upload_replace",
                    )
                    .await?;
                }
                tx.commit().await?;
                Ok(1)
            }
        }
    }
    .await;

    match tx_result {
        Err(e) => {
            // tx rolled back, storage counter unchanged. A concurrent
            // double-confirm of this same key could have committed it onto the
            // item row before our `try_apply_storage_on` errored (storage cap
            // filling in between), so a blind delete could destroy the live
            // object the winner points at. Route through the orphan queue; its
            // `is_s3_key_live` check skips any key a row still references.
            super::enqueue_s3_orphan(
                &db,
                &req.s3_key,
                crate::storage::S3Bucket::Main,
                "item_confirm_failed",
            )
            .await;
            return Err(e);
        }
        Ok(0) => {
            // CAS matched zero rows: the item was deleted/transferred out from
            // under the ownership filter, OR a concurrent confirm won the race
            // and swapped the target column (so it no longer holds the key we
            // observed). Either way the tx rolled back, nothing was charged;
            // route the now-unreferenced object through the orphan queue so the
            // reaper still cleans it.
            super::enqueue_s3_orphan(
                &db,
                &req.s3_key,
                crate::storage::S3Bucket::Main,
                "item_upload_target_missing",
            )
            .await;
            return Err(AppError::BadRequest(
                "Item was modified concurrently. Please try uploading again.".to_string(),
            ));
        }
        Ok(_) => {}
    }

    // Scan enqueue + scan_status flip happens AFTER the DB UPDATE commits via
    // the shared `commit_upload` helper, which is the only blessed path for
    // this ordering. See `routes/storage/mod.rs::commit_upload` for the bug
    // shapes this prevents.
    let scan_status = commit_upload(
        &db,
        scanning.scanner.as_ref(),
        CommitTarget::Item(req.item_id),
        &req.s3_key,
        file_type,
        user.id,
        file_size_bytes,
    )
    .await?;

    // (The old S3 object on a replace was enqueued for deletion inside the
    // confirm tx above, so a crash here can't orphan it.)

    // Clear the pending upload record now that the upload is confirmed
    db::pending_uploads::remove_pending_upload(&db, user.id, &req.s3_key, "main").await?;

    // Bump project cache generation so dashboard tabs reflect the new upload.
    // Reuses the project_id read during idempotency above (no second fetch).
    if let Some(project_id) = item_project_id
        && let Err(e) = db::projects::bump_cache_generation(&db, project_id).await
    {
        tracing::warn!(%project_id, error = ?e, "failed to bump cache generation after upload");
    }

    tracing::info!(
        "Upload confirmed: item={}, type={:?}, key={}, size={}",
        req.item_id,
        file_type,
        req.s3_key,
        file_size_bytes
    );

    let pending_review = if scan_status == db::FileScanStatus::HeldForReview {
        Some(true)
    } else {
        None
    };
    Ok(Json(ConfirmUploadResponse {
        success: true,
        pending_review,
    }))
}