//! Internal upload pipeline: presigned URL generation, upload confirmation, and storage usage.

use crate::auth::InternalActor;
use axum::{
    Json,
    extract::{Query, State},
    response::IntoResponse,
};
use serde::{Deserialize, Serialize};
use std::str::FromStr;

use sqlx::PgPool;

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

// ── Presign upload (for CLI upload pipeline) ──

#[derive(Deserialize)]
pub(super) struct InternalPresignRequest {
    item_id: ItemId,
    file_type: String,
    file_name: String,
    content_type: String,
}

#[derive(Serialize)]
struct InternalPresignResponse {
    upload_url: String,
    s3_key: String,
    expires_in: u64,
    #[serde(skip_serializing_if = "Option::is_none")]
    cache_control: Option<String>,
}

/// POST /api/internal/upload/presign
///
/// Generate a presigned S3 upload URL. Used by the CLI upload pipeline.
#[tracing::instrument(skip_all, name = "internal::presign_upload")]
pub(super) async fn presign_upload(
    State(db): State<PgPool>,
    State(storage): State<AppStorage>,
    actor: InternalActor,
    _auth: ServiceAuth,
    Json(req): Json<InternalPresignRequest>,
) -> Result<impl IntoResponse> {
    let s3 = storage.require_s3()?;

    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 != actor.user_id() {
        return Err(AppError::Forbidden);
    }

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

    // Staging key (unserved); the scan worker promotes it to the content key on a
    // Clean verdict. See routes/storage/uploads.rs for the C1 rationale.
    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, actor.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?;

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

// ── Multipart upload session (CLI / desktop, large files) ──
//
// The chunked counterpart to `presign_upload`, and deliberately only on the
// internal (CLI/desktop) surface: a browser keeps the one-shot presigned PUT and
// its 2 GiB ceiling, because a tab cannot resume a multi-hour transfer. These
// three endpoints replace the *transport* only, the client finishes by calling
// the existing `/api/internal/upload/confirm`, which reads the authoritative
// object size from S3 and does all the size/tier/scan/DB work unchanged.

/// Largest window of presigned part URLs one `parts` call will mint. A 20 GB
/// object is ~1250 parts at the auto-chosen part size; handing out every URL at
/// once would mint thousands of credentials with a 1-hour life for an upload
/// that may never happen, so the client pulls them in windows as it progresses.
const MULTIPART_PART_URL_WINDOW: u32 = 100;

#[derive(Deserialize)]
pub(super) struct MultipartStartRequest {
    item_id: ItemId,
    file_type: String,
    file_name: String,
    content_type: String,
    file_size_bytes: i64,
}

#[derive(Serialize)]
struct MultipartStartResponse {
    upload_id: String,
    s3_key: String,
    part_size: usize,
    part_count: u32,
    expires_in: u64,
}

/// POST /api/internal/upload/multipart/start
///
/// Open a multipart upload session and return the part geometry the client
/// uploads against. Mirrors `presign_upload`'s pre-checks (ownership, type,
/// quota) and additionally validates the declared size, since a multipart
/// session stages real S3 state that a rejected upload would orphan.
#[tracing::instrument(skip_all, name = "internal::multipart_start")]
pub(super) async fn multipart_start(
    State(db): State<PgPool>,
    State(storage): State<AppStorage>,
    actor: InternalActor,
    _auth: ServiceAuth,
    Json(req): Json<MultipartStartRequest>,
) -> Result<impl IntoResponse> {
    let s3 = storage.require_s3()?;

    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)?;

    let owner = db::items::get_item_owner(&db, req.item_id)
        .await?
        .ok_or(AppError::NotFound)?;
    if owner != actor.user_id() {
        return Err(AppError::Forbidden);
    }

    db::creator_tiers::check_presign_allowed(&db, actor.user_id(), file_type).await?;

    // Every size limit that describes the file, but NOT the single-PUT ceiling,
    // chunking is precisely what lets this path exceed it. The tier's per-file
    // cap is enforced here so an over-cap file is refused before it stages parts;
    // confirm re-checks against the real object size regardless.
    let tier_cap = db::creator_tiers::get_active_creator_tier(&db, actor.user_id())
        .await?
        .map(|t| t.max_file_bytes() as u64);
    crate::routes::storage::validate_declared_upload_size_limits(
        Some(req.file_size_bytes),
        file_type,
        tier_cap,
    )?;

    // Part geometry is pure arithmetic over the declared size, so the client can
    // derive identical boundaries without a round trip.
    let plan = s3_storage::MultipartPlan::auto(req.file_size_bytes.max(0) as u64)
        .map_err(AppError::BadRequest)?;

    let s3_key = S3Client::generate_staging_key(&req.file_name);
    // Persist the tier-checked declared size so `multipart_parts` binds the part
    // geometry to it instead of trusting its own request body (deepaudit F1).
    db::pending_uploads::record_pending_multipart_upload(
        &db,
        actor.user_id(),
        &s3_key,
        "main",
        req.file_size_bytes,
    )
    .await?;

    let upload_id = s3
        .create_multipart_upload(&s3_key, &req.content_type)
        .await?;

    tracing::info!(
        user = %actor.user_id(), item = %req.item_id, s3_key = %s3_key,
        size = req.file_size_bytes, parts = plan.part_count,
        "CLI multipart upload started"
    );

    Ok(Json(MultipartStartResponse {
        upload_id,
        s3_key: s3_key.into_string(),
        part_size: plan.part_size,
        part_count: plan.part_count,
        expires_in: 3600,
    }))
}

#[derive(Deserialize)]
pub(super) struct MultipartPartsRequest {
    s3_key: String,
    upload_id: String,
    /// Declared total size, so each part URL can be signed with its exact
    /// `Content-Length`. The plan is deterministic in this value, so it must
    /// match the one passed to `start` or the signed lengths will not line up.
    file_size_bytes: i64,
    first_part: u32,
    count: u32,
}

#[derive(Serialize)]
struct MultipartPartUrl {
    part_number: i32,
    content_length: u64,
    url: String,
}

#[derive(Serialize)]
struct MultipartPartsResponse {
    parts: Vec<MultipartPartUrl>,
    expires_in: u64,
}

/// POST /api/internal/upload/multipart/parts
///
/// Mint a bounded window of presigned `UploadPart` URLs. Each carries its exact
/// signed `Content-Length`, the same defense-in-depth the single-PUT presign
/// applies.
#[tracing::instrument(skip_all, name = "internal::multipart_parts")]
pub(super) async fn multipart_parts(
    State(db): State<PgPool>,
    State(storage): State<AppStorage>,
    actor: InternalActor,
    _auth: ServiceAuth,
    Json(req): Json<MultipartPartsRequest>,
) -> Result<impl IntoResponse> {
    let s3 = storage.require_s3()?;
    let s3_key = authorize_multipart_key(&db, actor.user_id(), &req.s3_key).await?;

    // F1: the part geometry must come from the size `start` validated against the
    // tier cap, not from this request body. Read it back and bind the request to
    // it, so a session opened for 1 GB cannot mint 5 TiB of part URLs.
    let declared = db::pending_uploads::declared_size(&db, actor.user_id(), &req.s3_key, "main")
        .await?
        .ok_or_else(|| {
            AppError::BadRequest("no multipart session was started for this key".to_string())
        })?;
    if req.file_size_bytes != declared {
        return Err(AppError::BadRequest(format!(
            "file_size_bytes {} does not match the size declared at start ({declared})",
            req.file_size_bytes
        )));
    }
    // F4: this session is actively receiving parts, so refresh its liveness and
    // keep the 24h orphan reaper from aborting a legitimate slow transfer.
    db::pending_uploads::touch_pending_upload(&db, actor.user_id(), &req.s3_key, "main").await?;

    let plan =
        s3_storage::MultipartPlan::auto(declared.max(0) as u64).map_err(AppError::BadRequest)?;

    if req.count == 0 || req.count > MULTIPART_PART_URL_WINDOW {
        return Err(AppError::BadRequest(format!(
            "count must be between 1 and {MULTIPART_PART_URL_WINDOW}"
        )));
    }
    if req.first_part == 0 || req.first_part > plan.part_count {
        return Err(AppError::BadRequest(format!(
            "first_part must be between 1 and {}",
            plan.part_count
        )));
    }

    let expires_in = 3600u64;
    let last = (req.first_part + req.count - 1).min(plan.part_count);
    let mut parts = Vec::with_capacity((last - req.first_part + 1) as usize);
    for part_number in req.first_part..=last {
        let content_length = plan.part_len(part_number);
        let url = s3
            .presign_upload_part(
                &s3_key,
                &req.upload_id,
                part_number as i32,
                Some(expires_in),
                Some(content_length as i64),
                // No checksum: the CLI streams plain file bytes and asks for
                // part URLs ahead of reading them, so it has no digest to bind
                // yet. The synckit blob path, which seals a part before asking
                // for its URL, does bind one.
                None,
            )
            .await?;
        parts.push(MultipartPartUrl {
            part_number: part_number as i32,
            content_length,
            url,
        });
    }

    Ok(Json(MultipartPartsResponse { parts, expires_in }))
}

#[derive(Deserialize)]
pub(super) struct MultipartCompletedPart {
    part_number: i32,
    etag: String,
}

#[derive(Deserialize)]
pub(super) struct MultipartCompleteRequest {
    s3_key: String,
    upload_id: String,
    parts: Vec<MultipartCompletedPart>,
}

#[derive(Serialize)]
struct MultipartCompleteResponse {
    success: bool,
    s3_key: String,
}

/// POST /api/internal/upload/multipart/complete
///
/// Assemble the uploaded parts into the staging object. This finalizes the
/// transport only; the client then calls `/api/internal/upload/confirm`, which
/// applies every size/tier/scan/commit rule against the real object.
#[tracing::instrument(skip_all, name = "internal::multipart_complete")]
pub(super) async fn multipart_complete(
    State(db): State<PgPool>,
    State(storage): State<AppStorage>,
    actor: InternalActor,
    _auth: ServiceAuth,
    Json(req): Json<MultipartCompleteRequest>,
) -> Result<impl IntoResponse> {
    let s3 = storage.require_s3()?;
    let s3_key = authorize_multipart_key(&db, actor.user_id(), &req.s3_key).await?;

    let parts: Vec<(i32, String)> = req
        .parts
        .into_iter()
        .map(|p| (p.part_number, p.etag))
        .collect();

    s3.complete_multipart_upload(&s3_key, &req.upload_id, &parts)
        .await?;

    tracing::info!(
        user = %actor.user_id(), s3_key = %s3_key, parts = parts.len(),
        "CLI multipart upload completed"
    );

    Ok(Json(MultipartCompleteResponse {
        success: true,
        s3_key: s3_key.into_string(),
    }))
}

#[derive(Deserialize)]
pub(super) struct MultipartAbortRequest {
    s3_key: String,
    upload_id: String,
}

/// POST /api/internal/upload/multipart/abort
///
/// Release the parts of an abandoned session (client cancel). Incomplete
/// multipart uploads bill for their parts until aborted, so the client cleaning
/// up on cancel is the cheapest fix; the pending-upload reaper is the backstop
/// for clients that vanish.
#[tracing::instrument(skip_all, name = "internal::multipart_abort")]
pub(super) async fn multipart_abort(
    State(db): State<PgPool>,
    State(storage): State<AppStorage>,
    actor: InternalActor,
    _auth: ServiceAuth,
    Json(req): Json<MultipartAbortRequest>,
) -> Result<impl IntoResponse> {
    let s3 = storage.require_s3()?;
    let s3_key = authorize_multipart_key(&db, actor.user_id(), &req.s3_key).await?;

    s3.abort_multipart_upload(&s3_key, &req.upload_id).await?;

    tracing::info!(user = %actor.user_id(), s3_key = %s3_key, "CLI multipart upload aborted");
    Ok(Json(InternalConfirmResponse { success: true }))
}

/// Bind a caller-supplied staging key to the caller.
///
/// A `staging/{uuid}` key carries no user in its path, so ownership comes from
/// the `pending_uploads` row `start` recorded, the same proof `confirm_upload`
/// uses. Without this, any authenticated creator could drive parts into another
/// creator's in-flight session.
async fn authorize_multipart_key(
    db: &PgPool,
    user_id: crate::db::UserId,
    s3_key: &str,
) -> Result<crate::storage::S3Key> {
    if !db::pending_uploads::is_owned(db, user_id, s3_key, "main").await? {
        return Err(AppError::BadRequest("Invalid upload key".to_string()));
    }
    Ok(crate::storage::S3Key::from_stored(s3_key))
}

// ── Confirm upload (for CLI upload pipeline) ──

#[derive(Deserialize)]
pub(super) struct InternalConfirmRequest {
    item_id: ItemId,
    file_type: String,
    s3_key: String,
}

#[derive(Serialize)]
struct InternalConfirmResponse {
    success: bool,
}

/// POST /api/internal/upload/confirm
///
/// Confirm a completed S3 upload: verify, scan, update DB. Used by the CLI upload pipeline.
#[tracing::instrument(skip_all, name = "internal::confirm_upload")]
pub(super) async fn confirm_upload(
    State(db): State<PgPool>,
    State(storage): State<AppStorage>,
    State(scanning): State<Scanning>,
    actor: InternalActor,
    _auth: ServiceAuth,
    Json(req): Json<InternalConfirmRequest>,
) -> Result<impl IntoResponse> {
    let s3 = storage.require_s3()?;

    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 != actor.user_id() {
        return Err(AppError::Forbidden);
    }

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

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

    // Idempotent replay: the S3 key is deterministic (`{user}/{item}/...`), and this
    // handler is non-atomic (storage increment + item/version writes). A retried
    // confirm for a key already committed must not re-charge storage or create a
    // duplicate version, detect the committed state and return success without
    // repeating the side effects (ultra-fuzz Run 12 Storage: confirm idempotency).
    let already_committed = match file_type {
        FileType::Audio | FileType::Video => {
            match db::items::get_item_by_id(&db, req.item_id).await? {
                Some(item) if file_type == FileType::Audio => {
                    item.audio_s3_key.as_deref() == Some(req.s3_key.as_str())
                }
                Some(item) => item.video_s3_key.as_deref() == Some(req.s3_key.as_str()),
                None => false,
            }
        }
        FileType::Download => db::versions::get_versions_by_item(&db, req.item_id)
            .await?
            .iter()
            .any(|v| v.s3_key.as_deref() == Some(req.s3_key.as_str())),
        _ => false,
    };
    if already_committed {
        tracing::info!(
            user = %actor.user_id(), item = %req.item_id, s3_key = %req.s3_key,
            "CLI upload confirm replay, already committed, skipping duplicate side effects"
        );
        return Ok(Json(InternalConfirmResponse { success: true }));
    }

    // Authorize the staging key for a fresh confirm: the caller must have
    // presigned it (recorded against them in `pending_uploads`). Placed after the
    // replay short-circuit, which consumed the pending row on the first confirm,
    // and before any reject path that enqueues the object for deletion, so an
    // unowned (at most another user's in-flight) staging object is never touched.
    if !db::pending_uploads::is_owned(&db, actor.user_id(), &req.s3_key, "main").await? {
        return Err(AppError::BadRequest("Invalid upload key".to_string()));
    }

    // Enforce file size 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() {
        crate::routes::storage::enqueue_s3_orphan(
            &db,
            &req.s3_key,
            crate::storage::S3Bucket::Main,
            "cli_upload_rejected",
        )
        .await;
        return Err(AppError::BadRequest(format!(
            "File exceeds maximum size of {} MB",
            file_type.max_size() / (1024 * 1024)
        )));
    }

    // Enforce tier-based limits
    let max_storage = match db::creator_tiers::check_upload_allowed(
        &db,
        actor.user_id(),
        file_type,
        file_size_bytes,
    )
    .await
    {
        Ok(max) => max,
        Err(e) => {
            crate::routes::storage::enqueue_s3_orphan(
                &db,
                &req.s3_key,
                crate::storage::S3Bucket::Main,
                "cli_upload_rejected",
            )
            .await;
            return Err(e);
        }
    };

    // Reject unsupported file types BEFORE any side effect, same ordering rule
    // as the web upload handlers (see routes/storage/mod.rs::commit_upload).
    if !matches!(
        file_type,
        FileType::Audio | FileType::Download | FileType::Video
    ) {
        crate::routes::storage::enqueue_s3_orphan(
            &db,
            &req.s3_key,
            crate::storage::S3Bucket::Main,
            "cli_upload_rejected",
        )
        .await;
        return Err(AppError::BadRequest(
            "CLI upload only supports audio, video, and download file types".to_string(),
        ));
    }

    // Increment storage BEFORE writing the DB record (if quota exceeded, the
    // S3 object is cleaned up and no DB record is created).
    if let Err(e) =
        db::creator_tiers::try_increment_storage(&db, actor.user_id(), file_size_bytes, max_storage)
            .await
    {
        crate::routes::storage::enqueue_s3_orphan(
            &db,
            &req.s3_key,
            crate::storage::S3Bucket::Main,
            "cli_upload_rejected",
        )
        .await;
        return Err(e);
    }

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

    // Update the database with S3 key and file size.
    let file_name = req
        .s3_key
        .rsplit('/')
        .next()
        .map(std::string::ToString::to_string);
    let commit_target = match file_type {
        FileType::Audio => {
            db::items::update_item_audio_s3_key(&db, req.item_id, actor.user_id(), &req.s3_key)
                .await?;
            db::items::update_item_audio_file_size(
                &db,
                req.item_id,
                actor.user_id(),
                file_size_bytes,
            )
            .await?;
            crate::routes::storage::CommitTarget::Item(req.item_id)
        }
        FileType::Download => {
            let version = db::versions::create_version(
                &db,
                req.item_id,
                "1.0",
                None,
                Some(&req.s3_key),
                Some(file_size_bytes),
                file_name.as_deref(),
                None,
            )
            .await?;
            crate::routes::storage::CommitTarget::Version(version.id)
        }
        FileType::Video => {
            db::items::update_item_video_s3_key(&db, req.item_id, actor.user_id(), &req.s3_key)
                .await?;
            db::items::update_item_video_file_size(
                &db,
                req.item_id,
                actor.user_id(),
                file_size_bytes,
            )
            .await?;
            crate::routes::storage::CommitTarget::Item(req.item_id)
        }
        _ => unreachable!("guarded above"),
    };

    // Scan enqueue + scan_status flip AFTER the DB writes commit, chronic
    // ordering invariant enforced via the shared commit_upload helper.
    let _status = crate::routes::storage::commit_upload(
        &db,
        scanning.scanner.as_ref(),
        commit_target,
        &req.s3_key,
        file_type,
        actor.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 upload");
    }

    tracing::info!(
        user = %actor.user_id(),
        item = %req.item_id,
        file_type = ?file_type,
        s3_key = %req.s3_key,
        size = file_size_bytes,
        "CLI upload confirmed"
    );

    Ok(Json(InternalConfirmResponse { success: true }))
}

// ── Storage info ──

#[derive(Deserialize)]
pub(super) struct UserIdQuery {}

#[derive(Serialize)]
struct StorageInfoResponse {
    storage_used_bytes: i64,
    max_storage_bytes: i64,
    allows_file_uploads: bool,
}

/// GET /api/internal/creator/storage?user_id={uuid}
///
/// Get storage usage and limits for a creator.
#[tracing::instrument(skip_all, name = "internal::creator_storage")]
pub(super) async fn creator_storage(
    State(db): State<PgPool>,
    actor: InternalActor,
    _auth: ServiceAuth,
    Query(_query): Query<UserIdQuery>,
) -> Result<impl IntoResponse> {
    let used = db::creator_tiers::get_storage_used(&db, actor.user_id()).await?;

    // Resolve effective tier
    let tier = db::creator_tiers::get_active_creator_tier(&db, actor.user_id()).await?;

    let (max_storage, allows_uploads) = match tier {
        Some(t) => (t.max_storage_bytes(), t.allows_file_uploads()),
        None => (0, false),
    };

    Ok(Json(StorageInfoResponse {
        storage_used_bytes: used,
        max_storage_bytes: max_storage,
        allows_file_uploads: allows_uploads,
    }))
}