//! Multipart (streaming) blob upload.

// ── Multipart blob upload (streaming) ──
//
// The transport for blobs above the server's one-shot PUT ceiling. What matters
// here is that the client never buffers the whole ciphertext yet still produces
// a byte-exact v3 blob: it seals 1 MiB chunks as it reads the file and cuts the
// sealed stream at the part boundaries the server signed, which it could only
// pick because `blob_encrypted_len` predicts the ciphertext size up front.
use crate::common::*;
use std::path::PathBuf;

const START_PATH: &str = "/api/v1/sync/blobs/multipart/start";
const PARTS_PATH: &str = "/api/v1/sync/blobs/multipart/parts";
const COMPLETE_PATH: &str = "/api/v1/sync/blobs/multipart/complete";
const ABORT_PATH: &str = "/api/v1/sync/blobs/multipart/abort";
const PART_PUT_PATH: &str = "/s3/part";

fn temp_blob(name: &str, contents: &[u8]) -> PathBuf {
    use std::sync::atomic::{AtomicU64, Ordering};
    static N: AtomicU64 = AtomicU64::new(0);
    let mut p = std::env::temp_dir();
    p.push(format!(
        "synckit_mp_{}_{}_{name}",
        std::process::id(),
        N.fetch_add(1, Ordering::Relaxed)
    ));
    std::fs::write(&p, contents).unwrap();
    p
}

/// Stands in for the server's part-URL minting: answers whatever window the
/// client asked for, rather than a fixed list, since the client requests one
/// part at a time (it can only checksum a part it has already sealed).
struct PartsResponder {
    cipher_len: usize,
    part_size: usize,
    base: String,
}

impl wiremock::Respond for PartsResponder {
    fn respond(&self, req: &wiremock::Request) -> ResponseTemplate {
        let body: serde_json::Value = serde_json::from_slice(&req.body).unwrap();
        let first = body["first_part"].as_u64().unwrap() as usize;
        let count = body["count"].as_u64().unwrap() as usize;
        let part_count = self.cipher_len.div_ceil(self.part_size);
        let last = (first + count - 1).min(part_count);

        let parts: Vec<serde_json::Value> = (first..=last)
            .map(|n| {
                let content_length = if n == part_count {
                    self.cipher_len - self.part_size * (part_count - 1)
                } else {
                    self.part_size
                };
                json!({
                    "part_number": n,
                    "content_length": content_length,
                    "url": format!("{}{PART_PUT_PATH}?partNumber={n}", self.base),
                })
            })
            .collect();
        ResponseTemplate::new(200).set_body_json(json!({ "parts": parts }))
    }
}

/// Mount the whole session: start (with the given plan), part-URL minting,
/// the PUT target, and complete.
async fn mount_session(kit: &MockKit, cipher_len: usize, part_size: usize) -> u32 {
    let part_count = mount_session_without_put(kit, cipher_len, part_size).await;
    kit.put(PART_PUT_PATH)
        .reply(ResponseTemplate::new(200).append_header("ETag", "\"part-etag\""))
        .await;
    part_count
}

/// The session without the part PUT, for a test that mounts its own (one that
/// fails part way, say).
async fn mount_session_without_put(kit: &MockKit, cipher_len: usize, part_size: usize) -> u32 {
    let part_count = cipher_len.div_ceil(part_size) as u32;

    kit.post(START_PATH)
        .json(json!({
            "upload_id": "test-upload-id",
            "part_size": part_size,
            "part_count": part_count,
            "already_exists": false,
        }))
        .await;
    kit.post(PARTS_PATH)
        .responder(PartsResponder {
            cipher_len,
            part_size,
            base: kit.uri(),
        })
        .await;
    kit.post(COMPLETE_PATH).code(204).empty().await;

    part_count
}

#[tokio::test]
async fn streaming_upload_tiles_the_parts_into_a_valid_blob() {
    let kit = MockKit::start().await;
    let (client, key) = kit.keyed();

    // Spans four 1 MiB chunks (three full plus a remainder), so sealed
    // chunks straddle part boundaries rather than lining up with them.
    let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 3 + 7))
        .map(|i| i as u8)
        .collect();
    let hash = hex::encode(sha2::Sha256::digest(&plaintext));
    let file = temp_blob("big.bin", &plaintext);

    let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
    let part_size = 1024 * 1024;
    let part_count = mount_session(&kit, cipher_len, part_size).await;
    assert!(part_count > 1, "the fixture must actually be multipart");

    client.blob_upload_streaming(&hash, &file).await.unwrap();

    // The session was sized in ciphertext, predicted from the plaintext.
    let start = kit.body(START_PATH).await;
    assert_eq!(start["size_bytes"].as_u64().unwrap(), cipher_len as u64);
    assert_eq!(start["hash"].as_str().unwrap(), hash);

    // Every part carried exactly the length the server signed for it.
    let puts = kit.requests_to(PART_PUT_PATH).await;
    assert_eq!(puts.len() as u32, part_count, "one PUT per planned part");
    for (i, put) in puts.iter().enumerate() {
        let expected = if i as u32 == part_count - 1 {
            cipher_len - part_size * (part_count as usize - 1)
        } else {
            part_size
        };
        assert_eq!(put.body.len(), expected, "part {} length", i + 1);
    }

    // Each part was requested with the SHA-256 of exactly the bytes that
    // part then carried, which is what S3 rehashes against at write time.
    // The pairing is what matters: a checksum bound to the wrong part is
    // worse than none, since it would reject a correct upload.
    let part_reqs = kit.bodies("POST", PARTS_PATH).await;
    assert_eq!(
        part_reqs.len() as u32,
        part_count,
        "one URL request per part: a digest exists only once the part is sealed"
    );
    for (i, body) in part_reqs.iter().enumerate() {
        assert_eq!(body["first_part"].as_u64().unwrap(), i as u64 + 1);
        assert_eq!(body["count"].as_u64().unwrap(), 1);
        let declared = body["checksums"][0].as_str().unwrap();
        let expected =
            base64::engine::general_purpose::STANDARD.encode(sha2::Sha256::digest(&puts[i].body));
        assert_eq!(
            declared,
            expected,
            "part {} checksum must match its bytes",
            i + 1
        );
        // And the client must actually send it: it is a signed header, so
        // dropping it would fail SigV4 at S3.
        assert_eq!(
            puts[i]
                .headers
                .get("x-amz-checksum-sha256")
                .expect("the PUT must carry the checksum header")
                .to_str()
                .unwrap(),
            declared
        );
    }

    // The concatenated parts are a valid v3 blob for this content address:
    // proof that streaming produced the same wire format as the in-memory
    // encrypt, boundaries and all.
    let assembled: Vec<u8> = puts.iter().flat_map(|r| r.body.clone()).collect();
    assert_eq!(assembled.len(), cipher_len);
    let decrypted = synckit_client::crypto::decrypt_blob_chunked(&assembled, &key, &hash).unwrap();
    assert_eq!(decrypted, plaintext, "streamed blob must round-trip");

    // Complete named every part, in order, with the ETag S3 returned.
    let complete = kit.body(COMPLETE_PATH).await;
    let named = complete["parts"].as_array().unwrap();
    assert_eq!(named.len() as u32, part_count);
    for (i, part) in named.iter().enumerate() {
        assert_eq!(part["part_number"].as_u64().unwrap(), i as u64 + 1);
        assert_eq!(part["etag"].as_str().unwrap(), "\"part-etag\"");
    }
    assert_eq!(
        kit.hits(ABORT_PATH).await,
        0,
        "a clean upload must not abort"
    );

    std::fs::remove_file(&file).ok();
}

#[tokio::test]
async fn streaming_upload_rejects_a_server_part_plan_that_lies_about_geometry() {
    let kit = MockKit::start().await;
    let (client, _key) = kit.keyed();

    // A blob that genuinely spans several 1 MiB parts.
    let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 3 + 7))
        .map(|i| i as u8)
        .collect();
    let hash = hex::encode(sha2::Sha256::digest(&plaintext));
    let file = temp_blob("liar.bin", &plaintext);

    // Hostile server: claims the whole multi-part blob fits in ONE part.
    // Trusting it would defeat the one-part-in-memory bound, so the client
    // must refuse before minting or PUTting anything.
    kit.post(START_PATH)
        .json(json!({
            "upload_id": "test-upload-id",
            "part_size": 1024 * 1024,
            "part_count": 1,
            "already_exists": false,
        }))
        .await;

    let err = client
        .blob_upload_streaming(&hash, &file)
        .await
        .unwrap_err();
    assert!(
        matches!(err, SyncKitError::Internal(ref m) if m.contains("does not match")),
        "expected a geometry-mismatch rejection, got {err:?}"
    );
    assert_eq!(
        kit.hits(PART_PUT_PATH).await,
        0,
        "no part may be uploaded once the plan is rejected"
    );

    std::fs::remove_file(&file).ok();
}

#[tokio::test]
async fn streaming_upload_handles_an_empty_file() {
    let kit = MockKit::start().await;
    let (client, key) = kit.keyed();

    let hash = hex::encode(sha2::Sha256::digest(b""));
    let file = temp_blob("empty.bin", b"");
    let cipher_len = synckit_client::crypto::blob_encrypted_len(0);
    mount_session(&kit, cipher_len, 1024 * 1024).await;

    client.blob_upload_streaming(&hash, &file).await.unwrap();

    let put = kit.raw_body(PART_PUT_PATH).await;
    assert_eq!(put.len(), cipher_len, "one part carries the whole blob");
    assert_eq!(
        synckit_client::crypto::decrypt_blob_chunked(&put, &key, &hash).unwrap(),
        Vec::<u8>::new(),
        "an empty blob is still an authenticated single chunk"
    );

    std::fs::remove_file(&file).ok();
}

#[tokio::test]
async fn streaming_upload_skips_when_the_server_already_has_the_content() {
    let kit = MockKit::start().await;
    let (client, _key) = kit.keyed();

    kit.post(START_PATH)
        .json(json!({
            "upload_id": "",
            "part_size": 0,
            "part_count": 0,
            "already_exists": true,
        }))
        .await;

    let plaintext = b"content the server already holds";
    let hash = hex::encode(sha2::Sha256::digest(plaintext));
    let file = temp_blob("dedup.bin", plaintext);

    client.blob_upload_streaming(&hash, &file).await.unwrap();

    // Dedup must cost nothing: no file bytes read out to the wire, no
    // session to clean up.
    assert_eq!(
        kit.hits(PART_PUT_PATH).await,
        0,
        "dedup must not upload parts"
    );
    assert_eq!(kit.hits(COMPLETE_PATH).await, 0);
    assert_eq!(kit.hits(ABORT_PATH).await, 0);

    std::fs::remove_file(&file).ok();
}

#[tokio::test]
async fn streaming_upload_aborts_when_the_file_no_longer_matches_its_hash() {
    // The caller hashed the file in an earlier pass. If it changed since,
    // storing it under the stale content address would poison the address:
    // every later download would re-hash and reject it. Fail here instead,
    // and release the parts.
    let kit = MockKit::start().await;
    let (client, _key) = kit.keyed();

    let plaintext = b"the bytes actually on disk";
    let stale_hash = hex::encode(sha2::Sha256::digest(b"what the caller hashed earlier"));
    let file = temp_blob("changed.bin", plaintext);

    let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
    mount_session(&kit, cipher_len, 1024 * 1024).await;
    kit.post(ABORT_PATH).code(204).empty().await;

    let err = client
        .blob_upload_streaming(&stale_hash, &file)
        .await
        .expect_err("a hash mismatch must not be uploaded");
    assert!(
        matches!(err, SyncKitError::IntegrityFailed { .. }),
        "expected IntegrityFailed, got {err:?}"
    );

    assert_eq!(
        kit.hits(COMPLETE_PATH).await,
        0,
        "a mismatched blob must not be assembled"
    );
    assert_eq!(
        kit.hits(ABORT_PATH).await,
        1,
        "the session must be released"
    );

    std::fs::remove_file(&file).ok();
}

#[tokio::test]
async fn streaming_upload_aborts_when_a_part_upload_fails() {
    // Parts already sent are billed until the session is aborted, so any
    // failure past `start` has to release it.
    let kit = MockKit::start().await;
    let (client, _key) = kit.keyed();

    let plaintext = b"a blob whose part upload will fail";
    let hash = hex::encode(sha2::Sha256::digest(plaintext));
    let file = temp_blob("failing.bin", plaintext);
    let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());

    kit.post(START_PATH)
        .json(json!({
            "upload_id": "test-upload-id",
            "part_size": cipher_len,
            "part_count": 1,
            "already_exists": false,
        }))
        .await;
    kit.post(PARTS_PATH)
        .json(json!({
            "parts": [{
                "part_number": 1,
                "content_length": cipher_len,
                "url": kit.url(PART_PUT_PATH),
            }]
        }))
        .await;
    kit.put(PART_PUT_PATH).code(403).empty().await;
    kit.post(ABORT_PATH).code(204).empty().await;

    let err = client
        .blob_upload_streaming(&hash, &file)
        .await
        .unwrap_err();
    assert!(
        matches!(err, SyncKitError::Server { status: 403, .. }),
        "got {err:?}"
    );

    assert_eq!(kit.hits(COMPLETE_PATH).await, 0);
    assert_eq!(
        kit.hits(ABORT_PATH).await,
        1,
        "a failed transfer must release its parts"
    );

    std::fs::remove_file(&file).ok();
}

// ── Resuming an interrupted session ──
//
// A large blob is a long transfer, and a process killed part way used to throw
// all of it away: the parts were still at S3, but nothing on this side
// remembered the session. With a resume store installed the next attempt takes
// the session over and sends only what is missing.
//
// The hard part is not the bookkeeping, it is the crypto. Part boundaries come
// from the server and have nothing to do with the 1 MiB sealed-chunk geometry,
// so a resume almost always restarts inside a chunk whose leading bytes are
// already uploaded. Sealing draws a random nonce per chunk, so re-sealing that
// chunk with a new one would splice two keystreams together and the assembled
// object would never open. These tests are about that boundary.

use synckit_client::client::resume::BlobResumeStore;

/// PUTs that succeed for the first `ok` parts and then refuse, standing in for
/// a transfer that dies part way. 403 rather than 500 so the client treats it
/// as permanent and the test does not sit through the retry backoff.
struct DiesAfter {
    ok: usize,
    seen: std::sync::atomic::AtomicUsize,
}

impl wiremock::Respond for DiesAfter {
    fn respond(&self, _req: &wiremock::Request) -> ResponseTemplate {
        let n = self.seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        if n < self.ok {
            ResponseTemplate::new(200).append_header("ETag", format!("\"etag-{}\"", n + 1))
        } else {
            ResponseTemplate::new(403)
        }
    }
}

/// A resume store on its own scratch database, as the engine would install.
fn resume_store(name: &str) -> Arc<dyn BlobResumeStore> {
    use std::sync::atomic::{AtomicU64, Ordering};
    static N: AtomicU64 = AtomicU64::new(0);
    let mut p = std::env::temp_dir();
    p.push(format!(
        "synckit_resume_{}_{}_{name}",
        std::process::id(),
        N.fetch_add(1, Ordering::Relaxed)
    ));
    std::fs::create_dir_all(&p).unwrap();
    synckit_client::store::SqliteResumeStore::shared(synckit_client::store::DbSource::path(
        p.join("app.db"),
    ))
}

async fn put_bodies(kit: &MockKit) -> Vec<Vec<u8>> {
    kit.requests_to(PART_PUT_PATH)
        .await
        .into_iter()
        .map(|r| r.body)
        .collect()
}

#[tokio::test]
async fn a_killed_upload_resumes_and_the_assembled_blob_still_opens() {
    let kit = MockKit::start().await;
    let key = synckit_client::crypto::generate_master_key();
    let store = resume_store("kill");

    // Four 1 MiB chunks against 700 KiB parts: no part boundary can land on a
    // chunk boundary, so the resume is guaranteed to restart mid-chunk. That is
    // the case the stored nonces exist for.
    let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 3 + 7))
        .map(|i| i as u8)
        .collect();
    let hash = hex::encode(sha2::Sha256::digest(&plaintext));
    let file = temp_blob("resume.bin", &plaintext);
    let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
    let part_size = 700 * 1024;
    let part_count = cipher_len.div_ceil(part_size);
    assert!(part_count > 3, "the fixture must have parts to resume from");

    // ── First attempt: dies after two parts ──
    let client = kit.authed();
    client.set_master_key_raw(key);
    client.set_resume_store(Arc::clone(&store));

    mount_session_without_put(&kit, cipher_len, part_size).await;
    kit.put(PART_PUT_PATH)
        .responder(DiesAfter {
            ok: 2,
            seen: std::sync::atomic::AtomicUsize::new(0),
        })
        .await;
    kit.post(ABORT_PATH).code(204).empty().await;

    let err = client
        .blob_upload_streaming(&hash, &file)
        .await
        .unwrap_err();
    assert!(
        matches!(err, SyncKitError::Server { status: 403, .. }),
        "got {err:?}"
    );
    // The session is the asset now: aborting it would throw away exactly what
    // the next attempt is going to reuse.
    assert_eq!(
        kit.hits(ABORT_PATH).await,
        0,
        "a resumable failure must keep the session"
    );
    let first_two: Vec<Vec<u8>> = put_bodies(&kit).await.into_iter().take(2).collect();

    let record = store.load(&hash).unwrap().expect("a session was recorded");
    assert_eq!(record.usable_parts().len(), 2);
    assert_eq!(record.session.upload_id, "test-upload-id");

    // ── Second attempt: a fresh client, as a restarted process would have ──
    kit.reset().await;
    mount_session(&kit, cipher_len, part_size).await;
    kit.post(ABORT_PATH).code(204).empty().await;

    let restarted = kit.authed();
    restarted.set_master_key_raw(key);
    restarted.set_resume_store(Arc::clone(&store));
    restarted.blob_upload_streaming(&hash, &file).await.unwrap();

    let resumed = put_bodies(&kit).await;
    assert_eq!(
        resumed.len(),
        part_count - 2,
        "a resume must not re-send the parts already at S3"
    );
    // `start` is unconditional (it carries the dedup answer), so the redundant
    // session it opens has to be released rather than left to the reaper.
    assert_eq!(kit.hits(ABORT_PATH).await, 1);

    // The whole point: the two runs' parts concatenate into one valid v3 blob,
    // which means the chunk straddling the boundary came back byte-identical.
    let assembled: Vec<u8> = first_two
        .iter()
        .chain(resumed.iter())
        .flat_map(Clone::clone)
        .collect();
    assert_eq!(assembled.len(), cipher_len);
    assert_eq!(
        synckit_client::crypto::decrypt_blob_chunked(&assembled, &key, &hash).unwrap(),
        plaintext,
        "a resumed blob must decrypt to the original"
    );

    // Complete named every part, and the kept ones carry the first run's ETags.
    let complete = kit.body(COMPLETE_PATH).await;
    let named = complete["parts"].as_array().unwrap();
    assert_eq!(named.len(), part_count);
    assert_eq!(named[0]["etag"].as_str().unwrap(), "\"etag-1\"");
    assert_eq!(named[1]["etag"].as_str().unwrap(), "\"etag-2\"");
    assert_eq!(complete["upload_id"].as_str().unwrap(), "test-upload-id");

    // Assembled means the record describes nothing.
    assert!(store.load(&hash).unwrap().is_none());

    std::fs::remove_file(&file).ok();
}

#[tokio::test]
async fn a_resume_that_fails_again_gives_up_the_session_rather_than_wedging() {
    // A session the server has already reaped would fail identically on every
    // future pass. One resume attempt, then a clean slate.
    let kit = MockKit::start().await;
    let key = synckit_client::crypto::generate_master_key();
    let store = resume_store("wedge");

    let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 2 + 3))
        .map(|i| i as u8)
        .collect();
    let hash = hex::encode(sha2::Sha256::digest(&plaintext));
    let file = temp_blob("wedge.bin", &plaintext);
    let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
    let part_size = 700 * 1024;

    let client = kit.authed();
    client.set_master_key_raw(key);
    client.set_resume_store(Arc::clone(&store));

    // Attempt one: two parts land, then the transfer dies. The record survives.
    mount_session_without_put(&kit, cipher_len, part_size).await;
    kit.put(PART_PUT_PATH)
        .responder(DiesAfter {
            ok: 2,
            seen: std::sync::atomic::AtomicUsize::new(0),
        })
        .await;
    kit.post(ABORT_PATH).code(204).empty().await;
    client
        .blob_upload_streaming(&hash, &file)
        .await
        .unwrap_err();
    assert!(store.load(&hash).unwrap().is_some());

    // Attempt two resumes into a session that refuses everything.
    kit.reset().await;
    mount_session_without_put(&kit, cipher_len, part_size).await;
    kit.put(PART_PUT_PATH).code(403).empty().await;
    kit.post(ABORT_PATH).code(204).empty().await;
    client
        .blob_upload_streaming(&hash, &file)
        .await
        .unwrap_err();

    assert!(
        store.load(&hash).unwrap().is_none(),
        "a failed resume must drop the record so the next pass starts clean"
    );
    assert!(
        kit.hits(ABORT_PATH).await >= 1,
        "and release the parts it is giving up on"
    );

    std::fs::remove_file(&file).ok();
}

#[tokio::test]
async fn a_file_that_changed_under_the_session_is_refused_rather_than_re_sealed() {
    // The nonce is the danger. Re-sealing different plaintext under a nonce this
    // key has already used would leak the XOR of the two chunks, so the resume
    // path checks a recorded plaintext digest before it re-uses one. A file
    // edited between attempts must stop the upload, not quietly seal.
    let kit = MockKit::start().await;
    let key = synckit_client::crypto::generate_master_key();
    let store = resume_store("changed");

    let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 3 + 7))
        .map(|i| i as u8)
        .collect();
    let hash = hex::encode(sha2::Sha256::digest(&plaintext));
    let file = temp_blob("changed.bin", &plaintext);
    let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
    let part_size = 700 * 1024;

    let client = kit.authed();
    client.set_master_key_raw(key);
    client.set_resume_store(Arc::clone(&store));
    mount_session_without_put(&kit, cipher_len, part_size).await;
    kit.put(PART_PUT_PATH)
        .responder(DiesAfter {
            ok: 2,
            seen: std::sync::atomic::AtomicUsize::new(0),
        })
        .await;
    kit.post(ABORT_PATH).code(204).empty().await;
    client
        .blob_upload_streaming(&hash, &file)
        .await
        .unwrap_err();
    assert_eq!(store.load(&hash).unwrap().unwrap().usable_parts().len(), 2);

    // Same length, different bytes, and the byte is inside the chunk the resume
    // has to re-seal under the stored nonce (two 700 KiB parts land the boundary
    // in chunk 1). Every length check still passes, so the digest is the only
    // thing between this and a nonce re-use.
    let mut edited = plaintext.clone();
    edited[synckit_client::crypto::BLOB_CHUNK_SIZE + 5] ^= 0xff;
    std::fs::write(&file, &edited).unwrap();

    kit.reset().await;
    mount_session(&kit, cipher_len, part_size).await;
    kit.post(ABORT_PATH).code(204).empty().await;
    let err = client
        .blob_upload_streaming(&hash, &file)
        .await
        .unwrap_err();
    assert!(
        matches!(err, SyncKitError::Internal(ref m) if m.contains("changed under an in-flight upload")),
        "got {err:?}"
    );
    assert_eq!(
        kit.hits(COMPLETE_PATH).await,
        0,
        "nothing may be assembled from two different files"
    );

    std::fs::remove_file(&file).ok();
}

/// A previous attempt that got every part to S3 and died on the assemble call
/// must assemble on the next attempt without re-sending a byte.
///
/// This is the one resume shape where the streaming loop sends nothing at all:
/// the recorded parts cover the whole ciphertext, so the boundary lands past
/// the last chunk and every chunk is read for the content-address check and
/// then skipped. What the client owes the server is the part list it already
/// has, and only `complete` is left to do.
///
/// It is reachable because a failed `complete` is the one failure that keeps
/// the session and the record: `stream_blob_parts` returned `Ok`, so the abort
/// and clear that guard a failed transfer are never run.
#[tokio::test]
async fn a_resume_that_already_holds_every_part_assembles_without_sending_one() {
    let kit = MockKit::start().await;
    let key = synckit_client::crypto::generate_master_key();
    let store = resume_store("complete-died");

    let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 2 + 11))
        .map(|i| i as u8)
        .collect();
    let hash = hex::encode(sha2::Sha256::digest(&plaintext));
    let file = temp_blob("complete-died.bin", &plaintext);
    let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
    let part_size = 700 * 1024;
    let part_count = cipher_len.div_ceil(part_size);
    assert!(
        part_count > 1,
        "the fixture must be a real multipart upload"
    );

    // ── First attempt: every part lands, the assemble call is refused ──
    let client = kit.authed();
    client.set_master_key_raw(key);
    client.set_resume_store(Arc::clone(&store));

    kit.post(START_PATH)
        .json(json!({
            "upload_id": "test-upload-id",
            "part_size": part_size,
            "part_count": part_count,
            "already_exists": false,
        }))
        .await;
    kit.post(PARTS_PATH)
        .responder(PartsResponder {
            cipher_len,
            part_size,
            base: kit.uri(),
        })
        .await;
    kit.put(PART_PUT_PATH)
        .responder(DiesAfter {
            // Never dies: this run is about what happens after the parts are up.
            ok: usize::MAX,
            seen: std::sync::atomic::AtomicUsize::new(0),
        })
        .await;
    // 403 rather than 500 so the client treats it as permanent and the test does
    // not sit through the retry backoff.
    kit.post(COMPLETE_PATH)
        .code(403)
        .json(json!({ "message": "assemble refused" }))
        .await;
    kit.post(ABORT_PATH).code(204).empty().await;

    let err = client
        .blob_upload_streaming(&hash, &file)
        .await
        .unwrap_err();
    assert!(
        matches!(err, SyncKitError::Server { status: 403, .. }),
        "got {err:?}"
    );
    let sent = put_bodies(&kit).await;
    assert_eq!(sent.len(), part_count, "the first attempt sent every part");

    let record = store
        .load(&hash)
        .unwrap()
        .expect("a failed complete keeps the session: it is what the retry needs");
    assert_eq!(
        record.usable_parts().len(),
        part_count,
        "every part must be recorded, or this is a different resume shape"
    );

    // ── Second attempt: nothing left to send ──
    kit.reset().await;
    mount_session(&kit, cipher_len, part_size).await;
    kit.post(ABORT_PATH).code(204).empty().await;

    let restarted = kit.authed();
    restarted.set_master_key_raw(key);
    restarted.set_resume_store(Arc::clone(&store));
    restarted.blob_upload_streaming(&hash, &file).await.unwrap();

    assert!(
        put_bodies(&kit).await.is_empty(),
        "a resume holding every part must not re-send one"
    );
    // The redundant session `start` opened is released rather than left to the reaper.
    assert_eq!(kit.hits(ABORT_PATH).await, 1);

    // What it did instead: named the parts the first run uploaded, with the
    // ETags that run was given.
    let complete = kit.body(COMPLETE_PATH).await;
    let named = complete["parts"].as_array().unwrap();
    assert_eq!(named.len(), part_count, "complete must name every part");
    for (i, part) in named.iter().enumerate() {
        assert_eq!(part["part_number"].as_u64(), Some(i as u64 + 1));
        assert_eq!(
            part["etag"].as_str(),
            Some(format!("\"etag-{}\"", i + 1)).as_deref(),
            "part {} lost the ETag the first run was given",
            i + 1
        );
    }
    assert_eq!(complete["upload_id"].as_str().unwrap(), "test-upload-id");

    // Guard against a vacuous pass: the bytes the first run sent really were the
    // whole blob, so assembling them is the right thing to have done.
    let assembled: Vec<u8> = sent.into_iter().flatten().collect();
    assert_eq!(assembled.len(), cipher_len);
    assert_eq!(
        synckit_client::crypto::decrypt_blob_chunked(&assembled, &key, &hash).unwrap(),
        plaintext
    );

    // Assembled means the record describes nothing.
    assert!(store.load(&hash).unwrap().is_none());

    std::fs::remove_file(&file).ok();
}

// ── Part-boundary arithmetic ──
//
// The part plan is arithmetic on the ciphertext length, and the cases that break
// it are the exact multiples and their two neighbours: a ciphertext of exactly N
// parts, one byte short of N parts, and one byte over. Every fixture above sits
// mid-part, so none of them separates `div_ceil` from a truncating divide, nor a
// final part sized `cipher_len - part_size * (n - 1)` from one sized `part_size`.

/// A plaintext length whose v3 ciphertext is exactly `n * part_size`, plus that
/// part size. `approx` grows by at most `n - 1` bytes to reach divisibility,
/// which cannot change the chunk count for a length that is not itself on a
/// chunk boundary.
fn exact_part_multiple(n: usize, approx: usize) -> (usize, usize) {
    let cipher = synckit_client::crypto::blob_encrypted_len(approx);
    let plaintext_len = approx + (n - cipher % n) % n;
    let cipher = synckit_client::crypto::blob_encrypted_len(plaintext_len);
    assert_eq!(cipher % n, 0, "the fixture must land on a part boundary");
    (plaintext_len, cipher / n)
}

/// Upload a blob whose ciphertext is `n * part_size + delta` bytes and check the
/// whole plan: how many parts were requested, how long each PUT was, and that
/// the parts concatenate back into a blob that opens.
async fn boundary_upload(n: usize, approx: usize, delta: isize) {
    let (exact_len, part_size) = exact_part_multiple(n, approx);
    let plaintext_len = exact_len.checked_add_signed(delta).unwrap();

    let plaintext: Vec<u8> = (0..plaintext_len).map(|i| i as u8).collect();
    let hash = hex::encode(sha2::Sha256::digest(&plaintext));
    let file = temp_blob("boundary.bin", &plaintext);
    let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext_len);

    // The three cases differ in exactly the way the arithmetic has to notice:
    // a byte over spills into an extra part carrying a single byte.
    let expected_parts = if delta > 0 { n + 1 } else { n };
    assert_eq!(
        cipher_len.div_ceil(part_size),
        expected_parts,
        "fixture geometry: n={n} delta={delta}"
    );

    let kit = MockKit::start().await;
    let (client, key) = kit.keyed();
    let planned = mount_session(&kit, cipher_len, part_size).await;
    assert_eq!(planned as usize, expected_parts);

    client.blob_upload_streaming(&hash, &file).await.unwrap();

    let puts = kit.requests_to(PART_PUT_PATH).await;
    assert_eq!(
        puts.len(),
        expected_parts,
        "one PUT per planned part: n={n} delta={delta}"
    );
    for (i, put) in puts.iter().enumerate() {
        let expected = if i + 1 == expected_parts {
            cipher_len - part_size * (expected_parts - 1)
        } else {
            part_size
        };
        assert_eq!(
            put.body.len(),
            expected,
            "n={n} delta={delta} part {} length",
            i + 1
        );
    }

    let assembled: Vec<u8> = puts.iter().flat_map(|r| r.body.clone()).collect();
    assert_eq!(assembled.len(), cipher_len, "n={n} delta={delta}");
    assert_eq!(
        synckit_client::crypto::decrypt_blob_chunked(&assembled, &key, &hash).unwrap(),
        plaintext,
        "n={n} delta={delta}: the parts must reassemble into the original"
    );

    std::fs::remove_file(&file).ok();
}

#[tokio::test]
async fn two_part_boundaries_are_planned_and_sent_exactly() {
    // One sealed chunk cut into two parts.
    for delta in [-1, 0, 1] {
        boundary_upload(2, 600_000, delta).await;
    }
}

#[tokio::test]
async fn three_part_boundaries_are_planned_and_sent_exactly() {
    // Three sealed chunks cut into three parts, so chunk and part boundaries
    // are near each other without coinciding.
    for delta in [-1, 0, 1] {
        boundary_upload(
            3,
            synckit_client::crypto::BLOB_CHUNK_SIZE * 2 + 500_000,
            delta,
        )
        .await;
    }
}

// ── Hostile part plans ──
//
// `part_size` and `part_count` come from the server and drive both the
// allocation and the cut points of the sealed stream, so every bound on them is
// arithmetic the client cannot get wrong quietly. Each case below is written as
// a pair: the value that must be accepted by a gate and the adjacent one that
// must not, told apart by which message came back. A test that only checked
// "is_err" would pass with any gate firing, including the wrong one.

/// Mount a start response carrying an arbitrary plan, plus the abort the client
/// makes on its way out. No part-URL route is mounted: every case here must be
/// refused before a part is requested.
async fn mount_hostile_plan(kit: &MockKit, part_size: u64, part_count: u32) {
    kit.post(START_PATH)
        .json(json!({
            "upload_id": "hostile-upload-id",
            "part_size": part_size,
            "part_count": part_count,
            "already_exists": false,
        }))
        .await;
    kit.post(ABORT_PATH).code(204).empty().await;
}

/// Run a streaming upload of a 5000-byte blob against `(part_size, part_count)`
/// and return the internal-error message it was refused with.
async fn plan_rejection(part_size: u64, part_count: u32) -> String {
    let kit = MockKit::start().await;
    let (client, _key) = kit.keyed();
    let plaintext: Vec<u8> = (0..5_000u32).map(|i| i as u8).collect();
    let hash = hex::encode(sha2::Sha256::digest(&plaintext));
    let file = temp_blob("hostile.bin", &plaintext);
    mount_hostile_plan(&kit, part_size, part_count).await;

    let err = client
        .blob_upload_streaming(&hash, &file)
        .await
        .unwrap_err();
    std::fs::remove_file(&file).ok();
    assert_eq!(
        kit.hits(PARTS_PATH).await,
        0,
        "a plan refused up front must not mint a single part URL"
    );
    match err {
        SyncKitError::Internal(message) => message,
        other => panic!("expected an Internal rejection, got {other:?}"),
    }
}

#[tokio::test]
async fn a_plan_with_no_bytes_per_part_or_no_parts_is_refused_as_empty() {
    // part_size 0 is also the divisor of the tiling check below the guard, so a
    // guard that let it through would divide by zero rather than mis-upload.
    assert!(
        plan_rejection(0, 3).await.contains("empty multipart plan"),
        "part_size 0 must be refused as an empty plan"
    );
    // part_count 0 is the other half of the same `||`: with an `&&` in its
    // place, a plan that is empty in only one of the two ways gets through.
    assert!(
        plan_rejection(1024 * 1024, 0)
            .await
            .contains("empty multipart plan"),
        "part_count 0 must be refused as an empty plan"
    );
}

#[tokio::test]
async fn the_part_size_ceiling_admits_exactly_one_gibibyte_and_refuses_one_byte_more() {
    // At the ceiling the plan is legal geometry and is judged on whether it
    // tiles the blob (it does not: 5000 bytes is one part, not two). One byte
    // over is refused by the ceiling itself. The two messages name which gate
    // fired, which is the only thing that separates `>` from `>=` and `==`.
    let at = plan_rejection(1 << 30, 2).await;
    assert!(
        at.contains("does not match"),
        "a part_size of exactly 1 GiB is under the ceiling: {at}"
    );
    let over = plan_rejection((1 << 30) + 1, 2).await;
    assert!(
        over.contains("exceeds the 1073741824-byte ceiling"),
        "one byte over the ceiling must be refused by it: {over}"
    );
}

#[tokio::test]
async fn the_part_count_ceiling_admits_exactly_ten_thousand_and_refuses_one_more() {
    // S3's own hard limit, so 10_000 parts is a legal plan and must reach the
    // tiling check; 10_001 is not.
    let at = plan_rejection(1024 * 1024, 10_000).await;
    assert!(
        at.contains("does not match"),
        "a part_count of exactly 10000 is under the ceiling: {at}"
    );
    let over = plan_rejection(1024 * 1024, 10_001).await;
    assert!(
        over.contains("exceeds the 10000-part ceiling"),
        "one part over the ceiling must be refused by it: {over}"
    );
}

/// Mints one part URL per request with a caller-chosen `part_number` and
/// `content_length`, so a test can make the server's signed geometry disagree
/// with the bytes the client holds.
struct LyingPartsResponder {
    part_number: i64,
    content_length: u64,
    base: String,
}

impl wiremock::Respond for LyingPartsResponder {
    fn respond(&self, _req: &wiremock::Request) -> ResponseTemplate {
        ResponseTemplate::new(200).set_body_json(json!({
            "parts": [{
                "part_number": self.part_number,
                "content_length": self.content_length,
                "url": format!("{}{PART_PUT_PATH}?partNumber={}", self.base, self.part_number),
            }],
        }))
    }
}

/// A single-part session whose minted URL carries the given geometry. Returns
/// the error the upload was refused with, or `None` if it went through.
async fn minted_part_rejection(part_number: i64, content_length_delta: i64) -> Option<String> {
    let kit = MockKit::start().await;
    let (client, _key) = kit.keyed();
    let plaintext: Vec<u8> = (0..5_000u32).map(|i| i as u8).collect();
    let hash = hex::encode(sha2::Sha256::digest(&plaintext));
    let file = temp_blob("mismatched-part.bin", &plaintext);
    let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());

    // One part holding the whole blob, so the plan itself is beyond reproach and
    // only the minted URL disagrees.
    kit.post(START_PATH)
        .json(json!({
            "upload_id": "mismatch-upload-id",
            "part_size": cipher_len,
            "part_count": 1,
            "already_exists": false,
        }))
        .await;
    kit.post(PARTS_PATH)
        .responder(LyingPartsResponder {
            part_number,
            content_length: (cipher_len as i64 + content_length_delta) as u64,
            base: kit.uri(),
        })
        .await;
    kit.put(PART_PUT_PATH)
        .reply(ResponseTemplate::new(200).append_header("ETag", "\"part-etag\""))
        .await;
    kit.post(COMPLETE_PATH).code(204).empty().await;
    kit.post(ABORT_PATH).code(204).empty().await;

    let outcome = client.blob_upload_streaming(&hash, &file).await;
    std::fs::remove_file(&file).ok();
    match outcome {
        Ok(_) => {
            assert_eq!(kit.hits(PART_PUT_PATH).await, 1, "an accepted plan is PUT");
            None
        }
        Err(SyncKitError::Internal(message)) => {
            assert_eq!(
                kit.hits(PART_PUT_PATH).await,
                0,
                "a part whose geometry is disputed must not be sent anyway"
            );
            Some(message)
        }
        Err(other) => panic!("expected an Internal rejection, got {other:?}"),
    }
}

#[tokio::test]
async fn a_minted_part_url_that_disagrees_with_the_bytes_in_hand_is_refused() {
    // The agreeing case first, so the two disagreements below are known to be
    // the only difference: part 1, exactly the bytes the client sealed.
    assert!(
        minted_part_rejection(1, 0).await.is_none(),
        "a URL signed for the part the client actually holds must be used"
    );

    // Signed for a different part: PUTting anyway would store the bytes at the
    // wrong index and assemble a scrambled object.
    let wrong_number = minted_part_rejection(2, 0)
        .await
        .expect("a URL signed for part 2 must not be used for part 1");
    assert!(
        wrong_number.contains("part geometry mismatch"),
        "wrong part_number: {wrong_number}"
    );

    // Signed for a different length: Content-Length is a signed header, so this
    // fails SigV4 at S3 with a far less legible error if it is sent.
    let wrong_length = minted_part_rejection(1, -1)
        .await
        .expect("a URL signed for one byte less must not be used");
    assert!(
        wrong_length.contains("part geometry mismatch"),
        "wrong content_length: {wrong_length}"
    );
}

#[tokio::test]
async fn a_resume_that_lands_exactly_on_a_chunk_boundary_seals_the_chunk_afresh() {
    // The other resume test picks part boundaries that can never coincide with
    // a chunk boundary, which is the common case but only one side of the
    // question. Here the first part is exactly the header plus chunk 0, so the
    // resume restarts with `within == 0`: nothing of the boundary chunk is at
    // S3, and it must therefore be sealed from scratch. Demanding a recorded
    // nonce here would fail the upload outright, since no nonce was ever
    // recorded for a chunk that was never sent.
    let kit = MockKit::start().await;
    let key = synckit_client::crypto::generate_master_key();
    let store = resume_store("aligned");

    let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 2 + 4_321))
        .map(|i| i as u8)
        .collect();
    let hash = hex::encode(sha2::Sha256::digest(&plaintext));
    let file = temp_blob("aligned-resume.bin", &plaintext);
    let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
    // Header plus one whole sealed chunk: the one part size that puts the
    // second part's first byte on chunk 1's first byte.
    let part_size = synckit_client::crypto::blob_header_bytes(plaintext.len()).len()
        + synckit_client::crypto::sealed_blob_chunk_len(plaintext.len(), 0);
    let part_count = cipher_len.div_ceil(part_size);
    assert_eq!(part_count, 3, "three parts, resuming at the second");

    // ── First attempt: one part lands, then the transfer dies ──
    let client = kit.authed();
    client.set_master_key_raw(key);
    client.set_resume_store(Arc::clone(&store));

    mount_session_without_put(&kit, cipher_len, part_size).await;
    kit.put(PART_PUT_PATH)
        .responder(DiesAfter {
            ok: 1,
            seen: std::sync::atomic::AtomicUsize::new(0),
        })
        .await;
    kit.post(ABORT_PATH).code(204).empty().await;

    let err = client
        .blob_upload_streaming(&hash, &file)
        .await
        .unwrap_err();
    assert!(
        matches!(err, SyncKitError::Server { status: 403, .. }),
        "got {err:?}"
    );
    let first: Vec<Vec<u8>> = put_bodies(&kit).await.into_iter().take(1).collect();
    assert_eq!(
        first[0].len(),
        part_size,
        "the first part is the header and chunk 0 exactly"
    );

    let record = store.load(&hash).unwrap().expect("a session was recorded");
    assert_eq!(record.usable_parts().len(), 1);
    assert!(
        record.chunk(1).is_none(),
        "chunk 1 was never sent, so no nonce for it can have been recorded"
    );

    // ── Second attempt ──
    kit.reset().await;
    mount_session(&kit, cipher_len, part_size).await;
    kit.post(ABORT_PATH).code(204).empty().await;

    let restarted = kit.authed();
    restarted.set_master_key_raw(key);
    restarted.set_resume_store(Arc::clone(&store));
    restarted.blob_upload_streaming(&hash, &file).await.unwrap();

    let resumed = put_bodies(&kit).await;
    assert_eq!(
        resumed.len(),
        part_count - 1,
        "only the missing parts go up"
    );

    let assembled: Vec<u8> = first
        .iter()
        .chain(resumed.iter())
        .flat_map(Clone::clone)
        .collect();
    assert_eq!(assembled.len(), cipher_len);
    assert_eq!(
        synckit_client::crypto::decrypt_blob_chunked(&assembled, &key, &hash).unwrap(),
        plaintext,
        "a resume aligned to a chunk boundary must still assemble"
    );

    std::fs::remove_file(&file).ok();
}