//! Getting bytes to S3.
//!
//! The one family here with real logic rather than transport boilerplate: a
//! file over [`MULTIPART_THRESHOLD_BYTES`] goes up in parts, which means a part
//! geometry from the server, a bounded set of presigned part URLs refreshed on
//! a [`PART_URL_WINDOW`] cadence, and a completion call that names every ETag.

use super::{MnwApiClient, json_response};
use serde::Deserialize;

/// Response from the presign-upload internal endpoint.
#[derive(Debug, Deserialize)]
#[allow(dead_code)]
pub(crate) struct PresignResponse {
    pub upload_url: String,
    pub s3_key: String,
    pub expires_in: u64,
    pub cache_control: Option<String>,
}

/// Above this size an upload goes through a multipart session instead of one
/// presigned PUT. The single-PUT path reads the whole file into memory, which is
/// fine for a small file and unacceptable for a multi-GB one; the multipart path
/// holds one part at a time. It is also the only path that can carry a file past
/// S3's 5 GiB single-PUT ceiling, which is what the tier limits allow for.
pub(crate) const MULTIPART_THRESHOLD_BYTES: u64 = 64 * 1024 * 1024;

/// How many presigned part URLs to request at a time. Must not exceed the
/// server's own window cap.
const PART_URL_WINDOW: u32 = 100;

/// An opened multipart upload session.
#[derive(Debug, Clone, Deserialize)]
pub(crate) struct MultipartStart {
    pub upload_id: String,
    pub s3_key: String,
    pub part_size: u64,
    pub part_count: u32,
    pub expires_in: u64,
}

/// One presigned part target, with the exact length the signature binds.
#[derive(Debug, Clone, Deserialize)]
pub(crate) struct MultipartPartUrl {
    pub part_number: i32,
    pub content_length: u64,
    pub url: String,
}

#[derive(Debug, Deserialize)]
struct MultipartPartsResponse {
    parts: Vec<MultipartPartUrl>,
}

impl MnwApiClient {
    /// Get a presigned S3 upload URL.
    pub(crate) async fn presign_upload(
        &self,
        user_id: &str,
        item_id: &str,
        file_type: &str,
        file_name: &str,
        content_type: &str,
    ) -> anyhow::Result<PresignResponse> {
        let url = format!("{}/api/internal/upload/presign", self.base_url);
        let resp = self
            .http
            .post(&url)
            .bearer_auth(&self.service_token)
            .header("X-MNW-Actor", self.actor_header())
            .json(&serde_json::json!({
                "user_id": user_id,
                "item_id": item_id,
                "file_type": file_type,
                "file_name": file_name,
                "content_type": content_type,
            }))
            .send()
            .await?;

        json_response(resp, "presign_upload").await
    }

    /// Confirm a completed S3 upload.
    pub(crate) async fn confirm_upload(
        &self,
        user_id: &str,
        item_id: &str,
        file_type: &str,
        s3_key: &str,
    ) -> anyhow::Result<bool> {
        let url = format!("{}/api/internal/upload/confirm", self.base_url);
        let resp = self
            .http
            .post(&url)
            .bearer_auth(&self.service_token)
            .header("X-MNW-Actor", self.actor_header())
            .json(&serde_json::json!({
                "user_id": user_id,
                "item_id": item_id,
                "file_type": file_type,
                "s3_key": s3_key,
            }))
            .send()
            .await?;

        #[derive(Deserialize)]
        struct Resp {
            success: bool,
        }
        let r: Resp = json_response(resp, "confirm_upload").await?;
        Ok(r.success)
    }

    /// Open a multipart upload session and get the part geometry.
    pub(crate) async fn multipart_start(
        &self,
        item_id: &str,
        file_type: &str,
        file_name: &str,
        content_type: &str,
        file_size_bytes: u64,
    ) -> anyhow::Result<MultipartStart> {
        let url = format!("{}/api/internal/upload/multipart/start", self.base_url);
        let resp = self
            .http
            .post(&url)
            .bearer_auth(&self.service_token)
            .header("X-MNW-Actor", self.actor_header())
            .json(&serde_json::json!({
                "item_id": item_id,
                "file_type": file_type,
                "file_name": file_name,
                "content_type": content_type,
                "file_size_bytes": file_size_bytes,
            }))
            .send()
            .await?;

        json_response(resp, "multipart_start").await
    }

    /// Fetch a bounded window of presigned part URLs.
    async fn multipart_parts(
        &self,
        s3_key: &str,
        upload_id: &str,
        file_size_bytes: u64,
        first_part: u32,
        count: u32,
    ) -> anyhow::Result<Vec<MultipartPartUrl>> {
        let url = format!("{}/api/internal/upload/multipart/parts", self.base_url);
        let resp = self
            .http
            .post(&url)
            .bearer_auth(&self.service_token)
            .header("X-MNW-Actor", self.actor_header())
            .json(&serde_json::json!({
                "s3_key": s3_key,
                "upload_id": upload_id,
                "file_size_bytes": file_size_bytes,
                "first_part": first_part,
                "count": count,
            }))
            .send()
            .await?;

        let parts: MultipartPartsResponse = json_response(resp, "multipart_parts").await?;
        Ok(parts.parts)
    }

    /// Assemble the uploaded parts into the staging object.
    async fn multipart_complete(
        &self,
        s3_key: &str,
        upload_id: &str,
        parts: &[(i32, String)],
    ) -> anyhow::Result<()> {
        let url = format!("{}/api/internal/upload/multipart/complete", self.base_url);
        let parts: Vec<serde_json::Value> = parts
            .iter()
            .map(|(n, etag)| serde_json::json!({ "part_number": n, "etag": etag }))
            .collect();
        let resp = self
            .http
            .post(&url)
            .bearer_auth(&self.service_token)
            .header("X-MNW-Actor", self.actor_header())
            .json(&serde_json::json!({
                "s3_key": s3_key,
                "upload_id": upload_id,
                "parts": parts,
            }))
            .send()
            .await?;

        if !resp.status().is_success() {
            anyhow::bail!(
                "multipart_complete failed: HTTP {} {}",
                resp.status(),
                resp.text().await.unwrap_or_default()
            );
        }
        Ok(())
    }

    /// Release the parts of an abandoned session. Incomplete multipart uploads
    /// bill for their parts until aborted.
    pub(crate) async fn multipart_abort(
        &self,
        s3_key: &str,
        upload_id: &str,
    ) -> anyhow::Result<()> {
        let url = format!("{}/api/internal/upload/multipart/abort", self.base_url);
        let resp = self
            .http
            .post(&url)
            .bearer_auth(&self.service_token)
            .header("X-MNW-Actor", self.actor_header())
            .json(&serde_json::json!({ "s3_key": s3_key, "upload_id": upload_id }))
            .send()
            .await?;

        if !resp.status().is_success() {
            anyhow::bail!("multipart_abort failed: HTTP {}", resp.status());
        }
        Ok(())
    }

    /// Upload a file through a multipart session, holding one part in memory at
    /// a time. Returns the staging key to confirm against.
    ///
    /// `on_progress` is called with `(bytes_uploaded, total)` after each part.
    /// Any failure past the session opening aborts it, so a half-finished upload
    /// does not leave parts billing indefinitely — the single abort site means a
    /// future failure path added inside cannot forget to.
    #[allow(clippy::too_many_arguments)]
    pub(crate) async fn upload_file_multipart(
        &self,
        item_id: &str,
        file_type: &str,
        file_name: &str,
        content_type: &str,
        file_path: &std::path::Path,
        file_size: u64,
        mut on_progress: impl FnMut(u64, u64),
    ) -> anyhow::Result<String> {
        let start = self
            .multipart_start(item_id, file_type, file_name, content_type, file_size)
            .await?;

        tracing::info!(
            s3_key = %start.s3_key,
            part_count = start.part_count,
            part_size = start.part_size,
            expires_in = start.expires_in,
            file_size,
            "multipart upload session opened"
        );

        match self
            .run_multipart_upload(&start, file_path, file_size, &mut on_progress)
            .await
        {
            Ok(()) => Ok(start.s3_key),
            Err(e) => {
                if let Err(abort_err) = self.multipart_abort(&start.s3_key, &start.upload_id).await
                {
                    tracing::warn!(
                        error = %abort_err, s3_key = %start.s3_key,
                        "failed to abort multipart upload after a failed transfer"
                    );
                }
                Err(e)
            }
        }
    }

    /// Read and upload every part, then complete. Returns `Err` without
    /// aborting; the caller owns the single abort.
    async fn run_multipart_upload(
        &self,
        start: &MultipartStart,
        file_path: &std::path::Path,
        file_size: u64,
        on_progress: &mut impl FnMut(u64, u64),
    ) -> anyhow::Result<()> {
        use tokio::io::AsyncReadExt;

        let mut file = tokio::fs::File::open(file_path)
            .await
            .map_err(|e| anyhow::anyhow!("opening {} for upload: {e}", file_path.display()))?;

        let mut completed: Vec<(i32, String)> = Vec::with_capacity(start.part_count as usize);
        let mut uploaded: u64 = 0;
        let mut next: u32 = 1;

        while next <= start.part_count {
            let count = PART_URL_WINDOW.min(start.part_count - next + 1);
            let urls = self
                .multipart_parts(&start.s3_key, &start.upload_id, file_size, next, count)
                .await?;
            if urls.is_empty() {
                anyhow::bail!("server returned no part URLs for part {next}");
            }

            for part in urls {
                // One part resident at a time — this is the whole point of the
                // multipart path over the single-PUT one.
                let mut buf = vec![0u8; part.content_length as usize];
                file.read_exact(&mut buf).await.map_err(|e| {
                    anyhow::anyhow!(
                        "reading part {} ({} bytes) from {}: {e}",
                        part.part_number,
                        part.content_length,
                        file_path.display()
                    )
                })?;

                let etag = self.put_part(&part, buf).await?;
                completed.push((part.part_number, etag));
                uploaded += part.content_length;
                on_progress(uploaded, file_size);
                next += 1;
            }
        }

        self.multipart_complete(&start.s3_key, &start.upload_id, &completed)
            .await
    }

    /// PUT one part to its presigned URL, returning the ETag the completion call
    /// needs. Retries transient failures — losing a part to a network blip
    /// should not discard the whole transfer.
    async fn put_part(&self, part: &MultipartPartUrl, body: Vec<u8>) -> anyhow::Result<String> {
        // `Bytes` so a retry clones a refcount rather than re-copying the part.
        let body = bytes::Bytes::from(body);
        let mut attempt: u32 = 0;
        loop {
            attempt += 1;
            let sent = self
                .http
                .put(&part.url)
                .header(reqwest::header::CONTENT_LENGTH, part.content_length)
                .body(body.clone())
                .send()
                .await;

            let retriable = match sent {
                Ok(resp) if resp.status().is_success() => {
                    let etag = resp
                        .headers()
                        .get(reqwest::header::ETAG)
                        .and_then(|v| v.to_str().ok())
                        .unwrap_or_default()
                        .to_string();
                    if etag.is_empty() {
                        anyhow::bail!(
                            "S3 returned no ETag for part {}; cannot complete the upload",
                            part.part_number
                        );
                    }
                    return Ok(etag);
                }
                Ok(resp) => {
                    let status = resp.status();
                    if attempt >= 3 {
                        anyhow::bail!(
                            "part {} failed after {attempt} attempts: HTTP {status}",
                            part.part_number
                        );
                    }
                    format!("HTTP {status}")
                }
                Err(e) => {
                    if attempt >= 3 {
                        return Err(anyhow::Error::new(e).context(format!(
                            "part {} failed after {attempt} attempts",
                            part.part_number
                        )));
                    }
                    e.to_string()
                }
            };

            let delay_ms = 200u64 * (1u64 << ((attempt - 1) * 2));
            tracing::warn!(
                part_number = part.part_number, attempt, delay_ms, error = %retriable,
                "part upload transient failure, retrying"
            );
            tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
        }
    }

    /// Upload a file to S3 using a presigned URL.
    pub(crate) async fn upload_to_s3(
        &self,
        presigned_url: &str,
        file_path: &std::path::Path,
        content_type: &str,
        cache_control: Option<&str>,
    ) -> anyhow::Result<()> {
        let data = tokio::fs::read(file_path).await?;
        let mut req = self
            .http
            .put(presigned_url)
            .header("content-type", content_type)
            .body(data);

        if let Some(cc) = cache_control {
            req = req.header("cache-control", cc);
        }

        let resp = req.send().await?;

        if !resp.status().is_success() {
            anyhow::bail!("S3 upload failed: HTTP {}", resp.status());
        }

        Ok(())
    }
}