//! Content-addressed blob upload and download.
//!
//! Blobs are encrypted client-side and stored under the SHA-256 of their
//! plaintext. Uploads go through a presigned URL after a size/quota check;
//! downloads are decrypted and re-hashed on arrival and discarded on any
//! mismatch, so a server that swaps or rolls back a blob cannot slip it past
//! the client.

use std::path::Path;

use base64::Engine;
use bytes::{Buf, Bytes, BytesMut};
use sha2::{Digest, Sha256};
use tokio::io::AsyncReadExt;
use tokio_stream::StreamExt;
use tracing::instrument;

use crate::{
    crypto,
    error::{Result, SyncKitError},
    types::{
        BlobConfirmRequest, BlobDownloadUrlRequest, BlobDownloadUrlResponse,
        BlobMultipartAbortRequest, BlobMultipartCompleteRequest, BlobMultipartCompletedPart,
        BlobMultipartPartsRequest, BlobMultipartPartsResponse, BlobMultipartStartRequest,
        BlobMultipartStartResponse, BlobUploadUrlRequest, BlobUploadUrlResponse,
    },
};

use super::SyncKitClient;
use super::helpers::{Idempotency, check_response};

/// Upper bound on a single decrypted blob held in memory, guarding against a
/// hostile server returning an absurdly large body that would OOM the client.
/// Generous (4 GiB) so legitimate large media still flow.
const MAX_BLOB_BYTES: usize = 4 * 1024 * 1024 * 1024;

/// What [`SyncKitClient::blob_upload_streaming`] did.
///
/// The caller needs to tell the two apart: only an upload that sent bytes has
/// anything to confirm.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BlobUploadOutcome {
    /// The parts were uploaded and assembled. Call [`SyncKitClient::blob_confirm`]
    /// next to record the blob server-side.
    Uploaded,
    /// The server already held this content address, so no bytes were sent and
    /// there is nothing to confirm. Server-asserted, not verified, see
    /// [`BlobUploadUrlResponse::already_exists`] for what a wrong `true` costs.
    AlreadyPresent,
}

impl SyncKitClient {
    /// Request a presigned upload URL for a blob.
    /// Returns (upload_url, already_exists). If `already_exists` is true,
    /// the blob is already on the server and no upload is needed.
    ///
    /// `size_bytes` is the **plaintext** length, the size the caller sees on
    /// disk. What [`Self::blob_upload`] actually PUTs is the sealed ciphertext,
    /// which is longer by the v3 framing, and the server signs the declared
    /// size into the presigned URL as `Content-Length` (a SignedHeader), so
    /// declaring the plaintext length makes every PUT fail the signature. This
    /// method converts, keeping the wire-format arithmetic inside the SDK
    /// instead of asking three consumer apps to know it.
    ///
    /// `already_exists` is the server's claim, not a verified fact. See
    /// [`BlobUploadUrlResponse::already_exists`] for what a false `true` costs
    /// (a withheld blob, never a substituted one).
    #[instrument(skip(self))]
    pub async fn blob_upload_url(
        &self,
        hash: &str,
        size_bytes: i64,
    ) -> Result<BlobUploadUrlResponse> {
        let plaintext_len = usize::try_from(size_bytes)
            .map_err(|_| SyncKitError::InvalidArgument("size_bytes must be non-negative".into()))?;
        let token = self.require_token()?;

        let body = Bytes::from(serde_json::to_vec(&BlobUploadUrlRequest {
            hash: hash.to_string(),
            size_bytes: crypto::blob_encrypted_len(plaintext_len) as i64,
        })?);

        self.retry_request_json(Idempotency::ReadOnly, || {
            let req = self
                .http
                .post(&self.endpoints.blobs_upload)
                .bearer_auth(&token)
                .header("content-type", "application/json")
                .body(body.clone());
            async move { check_response(req.send().await?).await }
        })
        .await
    }

    /// Upload blob data directly to S3 via a presigned PUT URL.
    ///
    /// Encrypts the data with the master key before uploading, binding the
    /// content `hash` as AEAD associated data so the ciphertext is tied to its
    /// content address. The plaintext is never sent to the server, preserving
    /// the E2E encryption guarantee.
    ///
    /// This is the in-memory path: it buffers the whole ciphertext, so the input
    /// is capped at [`MAX_BLOB_BYTES`] rather than risking an unbounded
    /// allocation. For anything file-backed, prefer
    /// [`Self::blob_upload_streaming`], which holds one part at a time and is
    /// the only path the server accepts above its one-shot PUT ceiling.
    #[instrument(skip(self, presigned_url, data))]
    pub async fn blob_upload(&self, hash: &str, presigned_url: &str, data: Vec<u8>) -> Result<()> {
        if data.len() > MAX_BLOB_BYTES {
            return Err(SyncKitError::InvalidArgument(format!(
                "blob is {} bytes, over the {MAX_BLOB_BYTES}-byte in-memory cap; use blob_upload_streaming for a file this size",
                data.len()
            )));
        }
        let master_key = self.require_master_key()?;
        // v3 chunked format: each chunk sealed and bound to (hash, index, count),
        // so the reader verifies and decrypts one chunk at a time.
        let encrypted = Bytes::from(crypto::encrypt_blob_chunked(&data, &master_key, hash)?);
        // Release the plaintext before the (potentially long, multi-GB) network
        // send so the sustained peak during transfer is ~1x the blob, not ~2x.
        // The per-attempt `encrypted.clone()` below is a cheap `Bytes` refcount
        // bump, not a copy. Never materializing the full ciphertext at all is
        // what `blob_upload_streaming` does, for callers holding a file.
        drop(data);

        self.retry_request(
            Idempotency::IdempotentWrite {
                on: "content hash (S3 key is content-addressed)",
            },
            || {
                let req = self
                    .http_stream
                    .put(presigned_url)
                    .header("content-type", "application/octet-stream")
                    .body(encrypted.clone());
                async move { check_response(req.send().await?).await }
            },
        )
        .await?;

        Ok(())
    }

    /// Upload a file-backed blob through the multipart session, never holding
    /// more than one part in memory.
    ///
    /// This is the bounded-memory counterpart to [`Self::blob_upload`], and the
    /// only path for blobs above the server's one-shot PUT ceiling. It works at
    /// any size (a small blob is a single part), so it is the better default
    /// whenever the blob is already a file on disk.
    ///
    /// The v3 chunk geometry is what makes it possible. A presigned PUT needs a
    /// signed `Content-Length` up front, and `blob_encrypted_len` gives the
    /// exact ciphertext length from the plaintext length alone, so the server
    /// signs every part boundary before a single byte has been read, and this
    /// method fills those boundaries by sealing 1 MiB chunks as it streams the
    /// file. Peak memory is one part plus one chunk, not the whole blob.
    ///
    /// `hash` must be the SHA-256 of the file's plaintext, as elsewhere. It is
    /// recomputed while streaming and the upload is abandoned on mismatch: the
    /// caller hashed the file in an earlier pass, so a file edited in between
    /// would otherwise be stored under an address that does not describe it.
    ///
    /// Returns [`BlobUploadOutcome`]: on `Uploaded` the caller calls
    /// [`Self::blob_confirm`] next, exactly as with the one-shot path (this
    /// method replaces the transport only); on `AlreadyPresent` the server
    /// already held the content, nothing was read or sent, and there is nothing
    /// to confirm. The dedup check happens before any file read or sealing, so
    /// a re-sync of content the server already has costs one round trip.
    #[instrument(skip(self, path), fields(path = %path.display()))]
    pub async fn blob_upload_streaming(
        &self,
        hash: &str,
        path: &Path,
    ) -> Result<BlobUploadOutcome> {
        let master_key = self.require_master_key()?;

        let meta = tokio::fs::metadata(path).await.map_err(|e| {
            SyncKitError::Internal(format!("stat blob file {}: {e}", path.display()))
        })?;
        let plaintext_len = usize::try_from(meta.len())
            .map_err(|_| SyncKitError::InvalidArgument("blob file is larger than usize".into()))?;
        if plaintext_len > MAX_BLOB_BYTES {
            return Err(SyncKitError::InvalidArgument(format!(
                "blob is {plaintext_len} bytes, over the {MAX_BLOB_BYTES}-byte client cap"
            )));
        }
        // The session is sized in ciphertext, which is knowable from the
        // plaintext length alone, that is the whole trick.
        let size_bytes = i64::try_from(crypto::blob_encrypted_len(plaintext_len))
            .map_err(|_| SyncKitError::InvalidArgument("blob ciphertext exceeds i64".into()))?;

        let token = self.require_token()?;
        let start_body = Bytes::from(serde_json::to_vec(&BlobMultipartStartRequest {
            hash: hash.to_string(),
            size_bytes,
        })?);
        let start: BlobMultipartStartResponse = self
            .retry_request_json(Idempotency::ReadOnly, || {
                let req = self
                    .http
                    .post(&self.endpoints.blobs_multipart_start)
                    .bearer_auth(&token)
                    .header("content-type", "application/json")
                    .body(start_body.clone());
                async move { check_response(req.send().await?).await }
            })
            .await?;

        if start.already_exists {
            return Ok(BlobUploadOutcome::AlreadyPresent);
        }

        // Every failure past this point leaves uploaded parts that S3 bills for
        // until the session is aborted, so the abort lives in exactly one place:
        // here, around the whole transfer. The server-side orphan reaper is the
        // backstop for a client that dies outright.
        let transfer = self
            .stream_blob_parts(hash, &start, path, plaintext_len, size_bytes, &master_key)
            .await;
        let parts = match transfer {
            Ok(parts) => parts,
            Err(e) => {
                if let Err(abort_err) = self.blob_multipart_abort(hash, &start.upload_id).await {
                    tracing::warn!(
                        error = %abort_err,
                        "failed to abort multipart blob session; the server reaper will release it"
                    );
                }
                return Err(e);
            }
        };

        let complete_body = Bytes::from(serde_json::to_vec(&BlobMultipartCompleteRequest {
            hash: hash.to_string(),
            upload_id: start.upload_id.clone(),
            parts,
        })?);
        self.retry_request(
            Idempotency::IdempotentWrite {
                on: "upload_id (completing twice assembles the same object)",
            },
            || {
                let req = self
                    .http
                    .post(&self.endpoints.blobs_multipart_complete)
                    .bearer_auth(&token)
                    .header("content-type", "application/json")
                    .body(complete_body.clone());
                async move { check_response(req.send().await?).await }
            },
        )
        .await?;

        Ok(BlobUploadOutcome::Uploaded)
    }

    /// Seal the file chunk by chunk, cutting the ciphertext at the server's
    /// signed part boundaries and PUTting each part. Returns the completed
    /// `(part_number, etag)` pairs.
    ///
    /// Peak memory is one part plus one chunk: the staging buffer never holds
    /// more than a part's worth, because a full part is drained and sent the
    /// moment it is complete.
    async fn stream_blob_parts(
        &self,
        hash: &str,
        start: &BlobMultipartStartResponse,
        path: &Path,
        plaintext_len: usize,
        size_bytes: i64,
        master_key: &[u8; 32],
    ) -> Result<Vec<BlobMultipartCompletedPart>> {
        let part_size = usize::try_from(start.part_size).map_err(|_| {
            SyncKitError::Internal(format!(
                "server part_size {} exceeds usize",
                start.part_size
            ))
        })?;
        if part_size == 0 || start.part_count == 0 {
            return Err(SyncKitError::Internal(
                "server returned an empty multipart plan".into(),
            ));
        }

        // The part geometry is server-supplied, so bound it before it drives any
        // allocation. Without this a hostile part_count aborts the client on an
        // over-large `Vec`/`BytesMut` reservation, and a part_count/part_size that
        // does not match the blob defeats the one-part-plus-one-chunk memory bound.
        // The plan must describe exactly the ciphertext we are about to upload
        // (the server derives it from this same `size_bytes`), so recompute and
        // compare rather than trust it.
        const MAX_PART_SIZE: usize = 1 << 30; // 1 GiB; real plan parts are well under this
        const MAX_PART_COUNT: u32 = 10_000; // S3's hard multipart-part ceiling
        if part_size > MAX_PART_SIZE {
            return Err(SyncKitError::Internal(format!(
                "server part_size {part_size} exceeds the {MAX_PART_SIZE}-byte ceiling"
            )));
        }
        if start.part_count > MAX_PART_COUNT {
            return Err(SyncKitError::Internal(format!(
                "server part_count {} exceeds the {MAX_PART_COUNT}-part ceiling",
                start.part_count
            )));
        }
        let total = usize::try_from(size_bytes).map_err(|_| {
            SyncKitError::Internal(format!("blob size {size_bytes} is not a valid length"))
        })?;
        let expected_parts = total.div_ceil(part_size);
        if start.part_count as usize != expected_parts {
            return Err(SyncKitError::Internal(format!(
                "server multipart plan (part_count {}, part_size {part_size}) does not match the \
                 {total}-byte blob (expected {expected_parts} parts)",
                start.part_count
            )));
        }

        let mut file = tokio::fs::File::open(path).await.map_err(|e| {
            SyncKitError::Internal(format!("open blob file {}: {e}", path.display()))
        })?;

        let chunk_count = crypto::blob_chunk_count_for(plaintext_len);
        let mut hasher = Sha256::new();
        // Plaintext scratch: zeroized on drop so a decrypted chunk does not linger.
        let mut plain = zeroize::Zeroizing::new(vec![0u8; crypto::BLOB_CHUNK_SIZE]);
        let mut staged = BytesMut::with_capacity(part_size + crypto::BLOB_CHUNK_SIZE);
        staged.extend_from_slice(&crypto::blob_header_bytes(plaintext_len));

        let mut completed = Vec::with_capacity(start.part_count as usize);
        let mut next_part = 1u32;

        for index in 0..chunk_count {
            let offset = index as usize * crypto::BLOB_CHUNK_SIZE;
            let want = (plaintext_len - offset).min(crypto::BLOB_CHUNK_SIZE);
            file.read_exact(&mut plain[..want]).await.map_err(|e| {
                SyncKitError::Internal(format!(
                    "read blob file {} at offset {offset}: {e}",
                    path.display()
                ))
            })?;
            hasher.update(&plain[..want]);
            staged.extend_from_slice(&crypto::seal_blob_chunk(
                &plain[..want],
                master_key,
                hash,
                index,
                chunk_count,
            )?);

            // Every part but the last is exactly `part_size`; the remainder is
            // the final part, so it is never sent from inside this loop.
            while staged.len() >= part_size && next_part < start.part_count {
                let body = staged.split_to(part_size).freeze();
                let etag = self
                    .upload_blob_part(hash, start, size_bytes, next_part, body)
                    .await?;
                completed.push(BlobMultipartCompletedPart {
                    part_number: next_part as i32,
                    etag,
                });
                next_part += 1;
            }
        }

        // Verify the content address before the object can be assembled. A
        // mismatch here aborts the session, so a file that changed after the
        // caller hashed it is never stored under the stale address.
        let actual = hex::encode(hasher.finalize());
        if actual != hash {
            return Err(SyncKitError::IntegrityFailed {
                expected: hash.to_string(),
                actual,
            });
        }

        if next_part != start.part_count {
            return Err(SyncKitError::Internal(format!(
                "sealed stream produced {} parts, server planned {}",
                next_part, start.part_count
            )));
        }
        let body = staged.split().freeze();
        let etag = self
            .upload_blob_part(hash, start, size_bytes, next_part, body)
            .await?;
        completed.push(BlobMultipartCompletedPart {
            part_number: next_part as i32,
            etag,
        });

        Ok(completed)
    }

    /// PUT one part to its presigned URL and return the ETag S3 assigned it.
    ///
    /// The URL is requested for this part alone, and carries the part's SHA-256
    /// bound as `x-amz-checksum-sha256`, so S3 rehashes what it receives and
    /// rejects a corrupted part at write time instead of letting it sit until a
    /// download fails to decrypt. That is the reason parts are not requested in
    /// windows: a digest only exists once the part has been sealed, and only one
    /// part is ever in memory. The cost is one small round trip per part.
    async fn upload_blob_part(
        &self,
        hash: &str,
        start: &BlobMultipartStartResponse,
        size_bytes: i64,
        part_number: u32,
        body: Bytes,
    ) -> Result<String> {
        let checksum = base64::engine::general_purpose::STANDARD.encode(Sha256::digest(&body));

        let token = self.require_token()?;
        let req_body = Bytes::from(serde_json::to_vec(&BlobMultipartPartsRequest {
            hash: hash.to_string(),
            upload_id: start.upload_id.clone(),
            size_bytes,
            first_part: part_number,
            count: 1,
            checksums: vec![checksum.clone()],
        })?);
        let window: BlobMultipartPartsResponse = self
            .retry_request_json(Idempotency::ReadOnly, || {
                let req = self
                    .http
                    .post(&self.endpoints.blobs_multipart_parts)
                    .bearer_auth(&token)
                    .header("content-type", "application/json")
                    .body(req_body.clone());
                async move { check_response(req.send().await?).await }
            })
            .await?;

        let part = window.parts.into_iter().next().ok_or_else(|| {
            SyncKitError::Internal(format!("server returned no URL for part {part_number}"))
        })?;
        // The two sides derive the plan independently, so a disagreement means
        // one of them is wrong about the geometry. Sending anyway would fail the
        // signed Content-Length at S3 with a far less legible error.
        if part.part_number != part_number as i32 || part.content_length != body.len() as u64 {
            return Err(SyncKitError::Internal(format!(
                "part geometry mismatch: server signed part {} for {} bytes, client has part {part_number} of {} bytes",
                part.part_number,
                part.content_length,
                body.len()
            )));
        }

        let resp = self
            .retry_request(
                Idempotency::IdempotentWrite {
                    on: "(upload_id, part_number), re-PUTting a part replaces it",
                },
                || {
                    let req = self
                        .http_stream
                        .put(&part.url)
                        .header("content-type", "application/octet-stream")
                        // Mandatory, not optional: the checksum is a signed
                        // header, so omitting it fails the signature.
                        .header("x-amz-checksum-sha256", &checksum)
                        .body(body.clone());
                    async move { check_response(req.send().await?).await }
                },
            )
            .await?;

        // S3 identifies each part by the ETag it returns; complete cannot be
        // assembled without it.
        resp.headers()
            .get(reqwest::header::ETAG)
            .and_then(|v| v.to_str().ok())
            .map(std::string::ToString::to_string)
            .ok_or_else(|| {
                SyncKitError::Internal(format!("part {part_number} upload returned no ETag"))
            })
    }

    /// Release an abandoned multipart session and the parts it holds.
    async fn blob_multipart_abort(&self, hash: &str, upload_id: &str) -> Result<()> {
        let token = self.require_token()?;
        let body = Bytes::from(serde_json::to_vec(&BlobMultipartAbortRequest {
            hash: hash.to_string(),
            upload_id: upload_id.to_string(),
        })?);
        self.retry_request(
            Idempotency::IdempotentWrite {
                on: "upload_id (aborting twice is a no-op)",
            },
            || {
                let req = self
                    .http
                    .post(&self.endpoints.blobs_multipart_abort)
                    .bearer_auth(&token)
                    .header("content-type", "application/json")
                    .body(body.clone());
                async move { check_response(req.send().await?).await }
            },
        )
        .await?;
        Ok(())
    }

    /// Confirm that a blob upload completed successfully.
    /// The server verifies the object exists in S3 and records it.
    #[instrument(skip(self))]
    pub async fn blob_confirm(&self, hash: &str, size_bytes: i64) -> Result<()> {
        if size_bytes < 0 {
            return Err(SyncKitError::InvalidArgument(
                "size_bytes must be non-negative".into(),
            ));
        }
        let token = self.require_token()?;

        let body = Bytes::from(serde_json::to_vec(&BlobConfirmRequest {
            hash: hash.to_string(),
            size_bytes,
        })?);

        self.retry_request(Idempotency::IdempotentWrite { on: "blob hash" }, || {
            let req = self
                .http
                .post(&self.endpoints.blobs_confirm)
                .bearer_auth(&token)
                .header("content-type", "application/json")
                .body(body.clone());
            async move { check_response(req.send().await?).await }
        })
        .await?;
        Ok(())
    }

    /// Get a presigned download URL for a blob by hash.
    #[instrument(skip(self))]
    pub async fn blob_download_url(&self, hash: &str) -> Result<String> {
        let token = self.require_token()?;

        let body = Bytes::from(serde_json::to_vec(&BlobDownloadUrlRequest {
            hash: hash.to_string(),
        })?);

        let download: BlobDownloadUrlResponse = self
            .retry_request_json(Idempotency::ReadOnly, || {
                let req = self
                    .http
                    .post(&self.endpoints.blobs_download)
                    .bearer_auth(&token)
                    .header("content-type", "application/json")
                    .body(body.clone());
                async move { check_response(req.send().await?).await }
            })
            .await?;
        Ok(download.download_url)
    }

    /// Download blob data from S3 via a presigned GET URL and verify it.
    ///
    /// Decrypts with the master key (binding the content `expected_hash` as AEAD
    /// associated data), then **re-verifies** that the plaintext hashes to
    /// `expected_hash`. The AEAD tag alone proves the bytes were not forged, but
    /// not that the untrusted server served the bytes for *this* address; the
    /// re-hash closes the substitution/rollback gap. Returns
    /// [`SyncKitError::IntegrityFailed`] on mismatch.
    ///
    /// The decrypted blob is returned in memory (the caller writes it to disk);
    /// the size is capped at [`MAX_BLOB_BYTES`]. A v3 chunked blob is decrypted
    /// chunk-by-chunk off the HTTP body, so peak memory is the plaintext plus one
    /// in-flight chunk rather than ciphertext *and* plaintext at once. Legacy
    /// `sk2:`/untagged blobs fall back to a whole-buffer decrypt (no flag-day).
    #[instrument(skip(self, presigned_url))]
    pub async fn blob_download(&self, expected_hash: &str, presigned_url: &str) -> Result<Vec<u8>> {
        let resp = self
            .retry_request(Idempotency::ReadOnly, || {
                let req = self.http_stream.get(presigned_url);
                async move { check_response(req.send().await?).await }
            })
            .await?;

        let master_key = self.require_master_key()?;
        let mut stream = resp.bytes_stream();
        let mut buf = BytesMut::new();
        let mut header: Option<crypto::BlobHeader> = None;
        let mut format_known = false;
        let mut chunked = false;
        let mut next_chunk: u32 = 0;
        let mut plaintext: Vec<u8> = Vec::new();
        let mut total_in: usize = 0;

        while let Some(part) = stream.next().await {
            let part = part.map_err(SyncKitError::Http)?;
            total_in = total_in.saturating_add(part.len());
            if total_in > MAX_BLOB_BYTES {
                return Err(SyncKitError::Internal(format!(
                    "blob exceeds {MAX_BLOB_BYTES}-byte limit",
                )));
            }
            buf.extend_from_slice(&part);

            // Pick the format once the 4-byte tag is in hand.
            if !format_known && buf.len() >= 4 {
                chunked = crypto::is_chunked_blob(&buf);
                format_known = true;
            }
            if !chunked {
                continue; // legacy: accumulate, decrypt whole-buffer at the end
            }
            // Parse the v3 header once available, consuming tag + header bytes.
            if header.is_none() {
                if buf.len() < 4 + 13 {
                    continue;
                }
                let (h, consumed) = crypto::parse_blob_header(&buf)?;
                buf.advance(consumed);
                header = Some(h);
            }
            // Peel and decrypt every complete chunk currently buffered.
            let h = header.unwrap();
            while next_chunk < h.chunk_count {
                let len = h.sealed_chunk_len(next_chunk);
                if buf.len() < len {
                    break;
                }
                let sealed = buf.split_to(len);
                let chunk = crypto::decrypt_blob_chunk(
                    &sealed,
                    &master_key,
                    expected_hash,
                    next_chunk,
                    h.chunk_count,
                )?;
                plaintext.extend_from_slice(&chunk);
                next_chunk += 1;
            }
        }

        if chunked {
            let h = header
                .ok_or_else(|| SyncKitError::Crypto("v3 blob ended before its header".into()))?;
            if next_chunk != h.chunk_count || buf.has_remaining() {
                return Err(SyncKitError::Crypto(
                    "v3 blob ended mid-chunk or had trailing bytes".into(),
                ));
            }
        } else {
            // Legacy (sk2: / untagged): decrypt the accumulated body in one shot.
            plaintext = crypto::decrypt_bytes_aad(
                &buf,
                &master_key,
                &crypto::AeadContext::blob(expected_hash),
            )?;
        }

        let actual = hex::encode(Sha256::digest(&plaintext));
        if actual != expected_hash {
            return Err(SyncKitError::IntegrityFailed {
                expected: expected_hash.to_string(),
                actual,
            });
        }
        Ok(plaintext)
    }
}

#[cfg(test)]
mod tests {
    use crate::types::*;

    #[test]
    fn blob_upload_url_response_deserialization() {
        let json = r#"{"upload_url": "https://s3.example.com/upload", "already_exists": false}"#;
        let resp: BlobUploadUrlResponse = serde_json::from_str(json).unwrap();
        assert_eq!(resp.upload_url, "https://s3.example.com/upload");
        assert!(!resp.already_exists);

        let json = r#"{"upload_url": "", "already_exists": true}"#;
        let resp: BlobUploadUrlResponse = serde_json::from_str(json).unwrap();
        assert!(resp.already_exists);
    }

    #[test]
    fn blob_upload_url_request_serialization() {
        let req = BlobUploadUrlRequest {
            hash: "sha256-abc123".to_string(),
            size_bytes: 1024,
        };

        let json = serde_json::to_string(&req).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["hash"], "sha256-abc123");
        assert_eq!(parsed["size_bytes"], 1024);
    }

    #[test]
    fn blob_content_hash_format_matches_consumer() {
        use sha2::{Digest, Sha256};
        // The integrity check in blob_download compares against this exact form:
        // lowercase hex of SHA-256, the same string consumers store as the blob
        // hash. If this drifts, every verified download would falsely reject.
        let h = hex::encode(Sha256::digest(b"hello blob"));
        assert_eq!(h.len(), 64);
        assert!(
            h.chars()
                .all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase())
        );
    }

    #[test]
    fn blob_confirm_request_serialization() {
        let req = BlobConfirmRequest {
            hash: "sha256-def456".to_string(),
            size_bytes: 2048,
        };

        let json = serde_json::to_string(&req).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["hash"], "sha256-def456");
        assert_eq!(parsed["size_bytes"], 2048);
    }
}