//! Presigned upload and confirm handlers for project and item images.

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

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

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

/// JSON input for requesting a presigned project image upload URL.
#[derive(Debug, Deserialize)]
pub(super) struct ProjectImagePresignRequest {
    pub project_id: ProjectId,
    pub file_name: String,
    pub content_type: String,
}

/// JSON input for confirming a completed project image upload.
#[derive(Debug, Deserialize)]
pub(super) struct ProjectImageConfirmRequest {
    pub project_id: ProjectId,
    pub s3_key: String,
}

/// JSON response from a successful project image confirm.
#[derive(Debug, Serialize)]
pub(super) struct ProjectImageConfirmResponse {
    pub success: bool,
    pub image_url: String,
}

/// JSON input for requesting a presigned item image upload URL.
#[derive(Debug, Deserialize)]
pub(super) struct ItemImagePresignRequest {
    pub item_id: ItemId,
    pub file_name: String,
    pub content_type: String,
}

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

/// Generate a presigned URL for uploading a project image
///
/// POST /api/projects/image/presign
///
/// Requires authentication. User must own the project.
#[tracing::instrument(skip_all, name = "storage::project_image_presign", fields(user_id = %user.id))]
pub(super) async fn project_image_presign(
    State(db): State<PgPool>,
    State(storage): State<AppStorage>,
    AuthUser(user): AuthUser,
    Json(req): Json<ProjectImagePresignRequest>,
) -> Result<impl IntoResponse> {
    user.check_not_suspended()?;
    let s3 = storage.require_s3()?;

    let file_type = FileType::Cover;
    S3Client::validate_content_type(file_type, &req.content_type)?;
    S3Client::validate_extension(file_type, &req.file_name)?;

    // Verify user owns the project
    let project = db::projects::get_project_by_id(&db, req.project_id)
        .await?
        .ok_or(AppError::NotFound)?;

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

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

    // Staging key (unserved); the scan worker promotes it to the content key and
    // rebuilds the public `cover_image_url` on a Clean verdict (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?;

    let expires_in = 3600;
    let upload_url = s3
        .presign_upload(
            &s3_key,
            &req.content_type,
            Some(expires_in),
            Some(CACHE_CONTROL_IMMUTABLE),
            None,
        )
        .await
        .context("presign upload for project image")?;

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

/// Confirm a project image upload, scan, store URL
///
/// POST /api/projects/image/confirm
///
/// Requires authentication. User must own the project.
#[tracing::instrument(skip_all, name = "storage::project_image_confirm", fields(user_id = %user.id))]
pub(super) async fn project_image_confirm(
    State(db): State<PgPool>,
    State(storage): State<AppStorage>,
    State(config): State<Config>,
    State(scanning): State<Scanning>,
    AuthUser(user): AuthUser,
    Json(req): Json<ProjectImageConfirmRequest>,
) -> Result<impl IntoResponse> {
    user.check_not_suspended()?;
    let s3 = storage.require_s3()?;

    // Verify user owns the project
    let project = db::projects::get_project_by_id(&db, req.project_id)
        .await?
        .ok_or(AppError::NotFound)?;

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

    // Authorize the staging key: a `staging/{uuid}` key has no project in its
    // path, so ownership is proved via the `pending_uploads` row recorded at
    // presign, not a prefix check. Gate before the size-reject path 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).
    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::Cover.max_size() {
        super::enqueue_s3_orphan(
            &db,
            &req.s3_key,
            crate::storage::S3Bucket::Main,
            "image_upload_rejected",
        )
        .await;
        return Err(AppError::BadRequest(format!(
            "File exceeds maximum size of {} MB",
            FileType::Cover.max_size() / (1024 * 1024)
        )));
    }

    // Idempotency: if the project already references this same s3_key, return success.
    // The Run #6 audit caught a silent-data-loss bug here: without this check, a benign
    // retry would queue `req.s3_key` (== current `cover_image_url`) for deletion.
    if let Some(ref cur_url) = project.cover_image_url
        && let Some(cur_key) = storage::extract_s3_key_from_url(
            cur_url,
            &config.cdn_base_url,
            Some(s3.bucket()),
            config.storage.as_ref().map(|c| c.endpoint.as_str()),
        )
        && cur_key == 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(ProjectImageConfirmResponse {
            success: true,
            image_url: cur_url.clone(),
        }));
    }

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

    // Probe the old S3 object's size FIRST (async, before any tx). If the row
    // references an old image we MUST determine its size or the storage counter
    // drifts on every replacement; treating Err/Ok(None) as "no old image" would
    // silently over-count. This probe stays outside the transaction so we never
    // hold a DB connection across an S3 round-trip.
    let mut replace_old_size: Option<i64> = None;
    let old_key_to_delete: Option<String> = if let Some(ref old_url) = project.cover_image_url
        && let Some(old_key) = storage::extract_s3_key_from_url(
            old_url,
            &config.cdn_base_url,
            Some(s3.bucket()),
            config.storage.as_ref().map(|c| c.endpoint.as_str()),
        ) {
        match s3.object_size(&old_key).await {
            Ok(Some(old_size)) if old_size > 0 => {
                replace_old_size = Some(old_size);
                Some(old_key)
            }
            Ok(Some(_) | None) => {
                // Old URL parsed but the object is gone (or zero-sized): treat as a
                // fresh upload, nothing to refund. Still queue the old key for
                // deletion in case of S3 eventual-consistency.
                Some(old_key)
            }
            Err(e) => {
                // S3 probe failed (transient). Refuse to write, letting a probe failure
                // silently over-count storage on every replace is the bug this branch exists to prevent.
                super::enqueue_s3_orphan(
                    &db,
                    &req.s3_key,
                    crate::storage::S3Bucket::Main,
                    "image_upload_rejected",
                )
                .await;
                tracing::warn!(key = %old_key, error = ?e, "S3 probe failed during project image replace");
                return Err(AppError::ServiceUnavailable(
                    "Could not verify previous image. Please try again.".to_string(),
                ));
            }
        }
    } else {
        None
    };

    let image_url = storage::build_project_image_url(&config.cdn_base_url, &req.s3_key);

    // Storage credit + project image URL UPDATE in ONE transaction. A rollback
    // restores the counter, so the previous compensating `rollback_and_orphan`
    // math (with a second S3 probe and swallowed `.ok()`s) is gone. `commit_upload`
    // stays AFTER the commit. `Ok(false)` = ownership filter no-matched (project
    // deleted/transferred mid-flight); the tx 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,
            replace_old_size,
            file_size_bytes,
            max_storage,
        )
        .await?;
        let ok = db::projects::update_project_cover_cas(
            &mut *tx,
            req.project_id,
            user.id,
            project.cover_image_url.as_deref(),
            &image_url,
            &req.s3_key,
            file_size_bytes,
        )
        .await?;
        if !ok {
            return Ok(false);
        }
        // Enqueue the OLD object for deletion in the SAME tx as the row UPDATE,
        // closing the crash-between-commit-and-enqueue orphan window (ultra-fuzz
        // Run #1 Storage LOW). is_s3_key_live guards the worker if a row still
        // references it.
        if let Some(old_key) = old_key_to_delete.as_deref() {
            db::pending_s3_deletions::enqueue_deletions(
                &mut *tx,
                // A cover key is a private staging key until promote repoints it to
                // the public content key, so the old cover may be in EITHER bucket.
                // Enqueue both; the reaper no-ops the bucket the object isn't in
                // (content keys are unique to one bucket) and `is_s3_key_live`
                // still guards each bucket against a live reference.
                &crate::storage::both_bucket_delete(old_key),
                "project_image_replace",
            )
            .await?;
        }
        tx.commit().await?;
        Ok(true)
    }
    .await;

    match committed {
        Err(e) => {
            // tx rolled back, counter unchanged. Orphan-queue the new key for cleanup.
            super::enqueue_s3_orphan(
                &db,
                &req.s3_key,
                crate::storage::S3Bucket::Main,
                "project_image_update_failed",
            )
            .await;
            return Err(e);
        }
        Ok(false) => {
            super::enqueue_s3_orphan(
                &db,
                &req.s3_key,
                crate::storage::S3Bucket::Main,
                "project_image_update_failed",
            )
            .await;
            return Err(AppError::BadRequest(
                "Project 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?;

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

    // Scan enqueue AFTER the DB write commits (Phase 5 chronic fix, the same
    // ordering rule that uploads/versions/media follow via `commit_upload`).
    commit_upload(
        &db,
        scanning.scanner.as_ref(),
        CommitTarget::ProjectImage(req.project_id),
        &req.s3_key,
        FileType::Cover,
        user.id,
        file_size_bytes,
    )
    .await?;

    db::projects::bump_cache_generation(&db, req.project_id).await?;

    tracing::info!(
        "Project image confirmed: project={}, key={}, size={}",
        req.project_id,
        req.s3_key,
        file_size_bytes
    );

    Ok(Json(ProjectImageConfirmResponse {
        success: true,
        image_url,
    }))
}

/// Generate a presigned URL for uploading an item image (logo/cover)
///
/// POST /api/items/image/presign
///
/// Requires authentication. User must own the item.
#[tracing::instrument(skip_all, name = "storage::item_image_presign", fields(user_id = %user.id))]
pub(super) async fn item_image_presign(
    State(db): State<PgPool>,
    State(storage): State<AppStorage>,
    AuthUser(user): AuthUser,
    Json(req): Json<ItemImagePresignRequest>,
) -> Result<impl IntoResponse> {
    user.check_not_suspended()?;
    let s3 = storage.require_s3()?;

    let file_type = FileType::Cover;
    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
    db::creator_tiers::check_presign_allowed(&db, user.id, file_type).await?;

    // Staging key (unserved); the scan worker promotes it to the content key and
    // rebuilds the public `cover_image_url` on a Clean verdict (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?;

    let expires_in = 3600;
    let upload_url = s3
        .presign_upload(
            &s3_key,
            &req.content_type,
            Some(expires_in),
            Some(CACHE_CONTROL_IMMUTABLE),
            None,
        )
        .await
        .context("presign upload for item image")?;

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

/// Confirm an item image upload, scan, store URL
///
/// POST /api/items/image/confirm
///
/// Requires authentication. User must own the item.
#[tracing::instrument(skip_all, name = "storage::item_image_confirm", fields(user_id = %user.id))]
pub(super) async fn item_image_confirm(
    State(db): State<PgPool>,
    State(storage): State<AppStorage>,
    State(config): State<Config>,
    State(scanning): State<Scanning>,
    AuthUser(user): AuthUser,
    Json(req): Json<ItemImageConfirmRequest>,
) -> Result<impl IntoResponse> {
    user.check_not_suspended()?;
    let s3 = storage.require_s3()?;

    // 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);
    }

    // Authorize the staging key: a `staging/{uuid}` key has no user/item in its
    // path (and no `cover/` segment to pin), so ownership, and the fact that it
    // was minted for THIS user's upload flow, is proved via the `pending_uploads`
    // row recorded at presign, not a prefix check. Gate before the size-reject
    // path so an unowned 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).
    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::Cover.max_size() {
        super::enqueue_s3_orphan(
            &db,
            &req.s3_key,
            crate::storage::S3Bucket::Main,
            "image_upload_rejected",
        )
        .await;
        return Err(AppError::BadRequest(format!(
            "File exceeds maximum size of {} MB",
            FileType::Cover.max_size() / (1024 * 1024)
        )));
    }

    // Idempotency: if cover_s3_key already matches, return success (no-op) BEFORE
    // the tier/cap check, a creator at their storage cap must still be able to
    // re-confirm a cover they already own (Run 9; matches project_image_confirm's
    // ordering). Otherwise capture the existing cover for atomic replacement below.
    let existing_item = db::items::get_item_by_id(&db, req.item_id).await?;
    if let Some(ref item) = existing_item
        && item.cover_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(super::images::ProjectImageConfirmResponse {
            success: true,
            image_url: item.cover_image_url.clone().unwrap_or_default(),
        }));
    }

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

    // The key the CAS must still see on the row for this confirm to win, the
    // raw current cover key, independent of its recorded size (a zero/NULL-size
    // old cover is still a key the column holds, and the CAS must match it or a
    // legitimate first replace would be misread as a lost race).
    let expected_old_key: Option<String> =
        existing_item.as_ref().and_then(|i| i.cover_s3_key.clone());

    // Old cover key+size come straight from the already-loaded item row (no S3
    // probe needed). We refund/clean up only when the old object has a real
    // size; split that decision once into the replace-size and the key to delete.
    let (replace_old_size, old_key_to_delete): (Option<i64>, Option<String>) =
        match existing_item.as_ref().and_then(|i| {
            Some((
                i.cover_s3_key.clone()?,
                i.cover_file_size_bytes.unwrap_or(0),
            ))
        }) {
            Some((key, size)) if size > 0 => (Some(size), Some(key)),
            _ => (None, None),
        };

    let image_url = storage::build_project_image_url(&config.cdn_base_url, &req.s3_key);

    // Storage credit + item cover UPDATE in ONE transaction (Run #7 HIGH-2 made
    // these atomic via compensating actions; this makes them atomic via a real
    // tx, a rollback restores the counter with no swallowed-`.ok()` math).
    // `commit_upload` stays AFTER the commit. `Ok(false)` = ownership filter
    // no-matched (item deleted/moved mid-flight); the tx 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,
            replace_old_size,
            file_size_bytes,
            max_storage,
        )
        .await?;
        let ok = db::items::update_item_cover(
            &mut *tx,
            req.item_id,
            user.id,
            expected_old_key.as_deref(),
            &image_url,
            &req.s3_key,
            file_size_bytes,
        )
        .await?;
        if !ok {
            return Ok(false);
        }
        // Enqueue the OLD cover for deletion in the SAME tx as the row UPDATE
        // (ultra-fuzz Run #1 Storage LOW; matches the in-tx ordering the delete
        // paths use). is_s3_key_live guards the worker against a live reference.
        if let Some(old_key) = old_key_to_delete.as_deref() {
            db::pending_s3_deletions::enqueue_deletions(
                &mut *tx,
                // Old cover may be a private staging key or a public content key;
                // enqueue both (see `both_bucket_delete`).
                &crate::storage::both_bucket_delete(old_key),
                "item_image_replace",
            )
            .await?;
        }
        tx.commit().await?;
        Ok(true)
    }
    .await;

    match committed {
        Err(e) => {
            super::enqueue_s3_orphan(
                &db,
                &req.s3_key,
                crate::storage::S3Bucket::Main,
                "item_image_update_failed",
            )
            .await;
            return Err(e);
        }
        Ok(false) => {
            super::enqueue_s3_orphan(
                &db,
                &req.s3_key,
                crate::storage::S3Bucket::Main,
                "item_image_update_failed",
            )
            .await;
            return Err(AppError::BadRequest(
                "Item was modified concurrently. Please try uploading again.".to_string(),
            ));
        }
        Ok(true) => {}
    }

    db::pending_uploads::remove_pending_upload(&db, user.id, &req.s3_key, "main").await?;

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

    // Scan enqueue + scan_status flip AFTER the DB write, same ordering rule
    // as uploads/versions/media. The Run #6 audit caught this same bug here.
    commit_upload(
        &db,
        scanning.scanner.as_ref(),
        CommitTarget::ItemImage(req.item_id),
        &req.s3_key,
        FileType::Cover,
        user.id,
        file_size_bytes,
    )
    .await?;

    // Bump project cache
    if let Some(item) = db::items::get_item_by_id(&db, req.item_id).await?
        && let Err(e) = db::projects::bump_cache_generation(&db, item.project_id).await
    {
        tracing::warn!(project_id = %item.project_id, error = ?e, "failed to bump cache generation after image upload");
    }

    tracing::info!(
        "Item image confirmed: item={}, key={}, size={}",
        req.item_id,
        req.s3_key,
        file_size_bytes
    );

    Ok(Json(ProjectImageConfirmResponse {
        success: true,
        image_url,
    }))
}