//! Presigned upload and confirm handlers for version files.

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

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

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

/// JSON input for requesting a presigned version upload URL.
#[derive(Debug, Deserialize)]
pub(super) struct VersionPresignRequest {
    pub file_name: String,
    pub content_type: String,
    /// Optional declared size; when present it is signed into the presigned
    /// URL's `Content-Length` so S3 rejects oversized PUTs at the protocol
    /// level (version downloads are up to 500 MB).
    #[serde(default)]
    pub file_size_bytes: Option<i64>,
}

/// JSON input for confirming a completed version upload.
#[derive(Debug, Deserialize)]
pub(super) struct VersionConfirmRequest {
    pub s3_key: String,
}

/// Generate a presigned URL for uploading a version file to S3
///
/// POST /api/versions/{version_id}/upload/presign
///
/// Requires authentication. User must own the item (through version -> item -> project chain).
#[tracing::instrument(skip_all, name = "storage::version_presign_upload", fields(%version_id, user_id = %user.id))]
pub(super) async fn version_presign_upload(
    State(db): State<PgPool>,
    State(storage): State<AppStorage>,
    AuthUser(user): AuthUser,
    Path(version_id): Path<VersionId>,
    Json(req): Json<VersionPresignRequest>,
) -> Result<impl IntoResponse> {
    user.check_not_suspended()?;
    let s3 = storage.require_s3()?;

    let file_type = FileType::Download;

    // Validate content type and extension
    S3Client::validate_content_type(file_type, &req.content_type)?;
    S3Client::validate_extension(file_type, &req.file_name)?;

    // Fetch version and verify ownership through version -> item -> project chain
    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)?;

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

    // Early quota check
    db::creator_tiers::check_presign_allowed(&db, user.id, file_type).await?;

    let max_file_bytes =
        db::creator_tiers::get_effective_max_file_bytes(&db, user.id, file_type).await?;

    // Validate the declared size (if any) before signing it into Content-Length.
    super::validate_declared_upload_size(req.file_size_bytes, file_type, max_file_bytes)?;

    // Staging key (unserved); the scan worker promotes it to the content key on
    // a Clean verdict. The random staging uuid also guarantees two versions that
    // share a filename never collide onto one object (the property the old
    // version-id-woven key gave us, ultra-fuzz Run #1 Storage HIGH).
    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?;

    let expires_in = 3600;
    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 version file")?;

    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 a version file upload has completed and update the database
///
/// POST /api/versions/{version_id}/upload/confirm
///
/// Requires authentication. User must own the item.
#[tracing::instrument(skip_all, name = "storage::version_confirm_upload", fields(%version_id, user_id = %user.id))]
pub(super) async fn version_confirm_upload(
    State(db): State<PgPool>,
    State(storage): State<AppStorage>,
    State(scanning): State<Scanning>,
    AuthUser(user): AuthUser,
    Path(version_id): Path<VersionId>,
    Json(req): Json<VersionConfirmRequest>,
) -> Result<impl IntoResponse> {
    user.check_not_suspended()?;
    let s3 = storage.require_s3()?;

    // Fetch version and verify ownership
    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)?;

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

    // Idempotent re-confirm: the version already references this exact key.
    // Handled BEFORE the ownership gate and the scan enqueue, re-confirming an
    // already-Clean version must not knock it back to Pending, and the pending
    // row was consumed by the first confirm so the gate below would reject it.
    if version.s3_key.as_deref() == Some(&req.s3_key) {
        // Still clear pending_uploads, orphan reaper would otherwise delete the
        // live S3 object 24h later (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,
        }));
    }

    // Authorize the staging key for a fresh confirm. A `staging/{uuid}` key has
    // no user/item/version in its path, so ownership is proved via the
    // `pending_uploads` row recorded at presign, not a prefix check. Placed
    // before the size/tier reject paths so an unowned (at most another user's
    // in-flight) staging object is never enqueued for deletion.
    if !db::pending_uploads::is_owned(&db, user.id, &req.s3_key, "main").await? {
        return Err(AppError::BadRequest("Invalid upload key".to_string()));
    }

    // A single HEAD: `object_size` returns None when the object isn't there, so
    // it doubles as the existence check (no separate object_exists round-trip).
    // Versions are always downloads, so enforce that size limit.
    let file_size_bytes = s3.object_size(&req.s3_key).await?.ok_or_else(|| {
        AppError::BadRequest("Upload not found. Please try uploading again.".to_string())
    })?;
    if file_size_bytes as u64 > FileType::Download.max_size() {
        super::enqueue_s3_orphan(
            &db,
            &req.s3_key,
            crate::storage::S3Bucket::Main,
            "version_upload_rejected",
        )
        .await;
        let limit_mb = FileType::Download.max_size() / (1024 * 1024);
        let file_mb = file_size_bytes as u64 / (1024 * 1024);
        return Err(AppError::FileTooLarge(format!(
            "File is {file_mb} MB but the maximum for download files is {limit_mb} MB."
        )));
    }

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

    let old_s3_key = version.s3_key.clone();
    let old_size = version.file_size_bytes.unwrap_or(0);
    let is_replace = old_s3_key.is_some() && old_size > 0;

    // Extract file name from the s3_key (last path segment)
    let file_name = req
        .s3_key
        .rsplit('/')
        .next()
        .map(std::string::ToString::to_string);

    // Storage credit + version UPDATE in ONE transaction. The expected-old guard
    // (`s3_key IS NOT DISTINCT FROM`) returns no row if another confirm raced
    // ahead; we leave the tx uncommitted so the rollback undoes the storage
    // change with no compensating math (the previous swallowed-`.ok()` path).
    // `commit_upload` stays AFTER the commit (the blessed scan-ordering path).
    // `Ok(false)` = lost race (rolled back, nothing charged).
    let committed: Result<bool> = async {
        let mut tx = db.begin().await?;
        db::creator_tiers::try_apply_storage_on(
            &mut tx,
            user.id,
            is_replace.then_some(old_size),
            file_size_bytes,
            max_storage,
        )
        .await?;
        let updated = db::versions::update_version_file(
            &mut *tx,
            version_id,
            old_s3_key.as_deref(),
            &req.s3_key,
            Some(file_size_bytes),
            file_name.as_deref(),
        )
        .await?;
        if updated.is_none() {
            // Lost race, drop tx to roll back the storage change.
            return Ok(false);
        }
        // Enqueue the OLD key for deletion in the SAME tx as the row UPDATE, so
        // a crash between commit and a post-commit enqueue can't orphan it with
        // no durable record (ultra-fuzz Run #1 Storage LOW; mirrors the in-tx
        // ordering delete_version uses). 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 anything 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())],
                "version_replace",
            )
            .await?;
        }
        tx.commit().await?;
        Ok(true)
    }
    .await;

    match committed {
        Err(e) => {
            // The tx rolled back, so nothing this request wrote references the
            // key, but a concurrent double-confirm of this same key could have
            // committed it onto the row before our `try_apply_storage_on`
            // errored (e.g. the storage cap filled in between). A blind delete
            // would then destroy the live object the winning confirm points at.
            // Route through the orphan queue, whose `is_s3_key_live` check skips
            // any key a row still references.
            super::enqueue_s3_orphan(
                &db,
                &req.s3_key,
                crate::storage::S3Bucket::Main,
                "version_confirm_failed",
            )
            .await;
            return Err(e);
        }
        Ok(false) => {
            // Lost the CAS race: a concurrent confirm swapped the version's
            // s3_key out from under us. If it committed THIS key, a direct
            // delete would 404 every fan download of the version the winner
            // just published. The orphan queue's liveness check is the guard.
            super::enqueue_s3_orphan(
                &db,
                &req.s3_key,
                crate::storage::S3Bucket::Main,
                "version_confirm_lost_race",
            )
            .await;
            return Err(AppError::BadRequest(
                "Version was modified concurrently. Please try uploading again.".to_string(),
            ));
        }
        Ok(true) => {}
    }

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

    let scan_status = commit_upload(
        &db,
        scanning.scanner.as_ref(),
        CommitTarget::Version(version_id),
        &req.s3_key,
        FileType::Download,
        user.id,
        file_size_bytes,
    )
    .await?;

    // (The old S3 key was enqueued for deletion inside the commit tx above.)

    tracing::info!(
        key = %req.s3_key,
        size = file_size_bytes,
        is_replace,
        ?scan_status,
        "version upload confirmed"
    );

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