//! Tests for [`super`].

use super::*;
use crate::types::*;

mod resume {
    use super::super::*;
    use std::sync::Mutex;

    /// A store that answers with whatever the test put in it.
    #[derive(Default)]
    struct Fake {
        record: Mutex<Option<ResumeRecord>>,
        cleared: Mutex<bool>,
    }
    impl BlobResumeStore for Fake {
        fn load(&self, _hash: &str) -> Result<Option<ResumeRecord>> {
            Ok(self.record.lock().unwrap().clone())
        }
        fn begin(&self, _hash: &str, _session: &ResumeSession) -> Result<()> {
            Ok(())
        }
        fn record_part(
            &self,
            _hash: &str,
            _part: &ResumePart,
            _chunks: &[ResumeChunk],
        ) -> Result<()> {
            Ok(())
        }
        fn clear(&self, _hash: &str) -> Result<()> {
            *self.cleared.lock().unwrap() = true;
            Ok(())
        }
    }

    /// A plausible session: 3 parts of 8 bytes over a 24-byte ciphertext,
    /// with the first part done.
    fn fake(age_secs: i64) -> Fake {
        Fake {
            record: Mutex::new(Some(ResumeRecord {
                session: ResumeSession {
                    upload_id: "u".into(),
                    part_size: 8,
                    part_count: 3,
                    size_bytes: 24,
                },
                age_secs,
                parts: vec![ResumePart {
                    part_number: 1,
                    etag: "e".into(),
                }],
                chunks: vec![],
            })),
            cleared: Mutex::new(false),
        }
    }

    #[test]
    fn a_fresh_matching_record_is_taken() {
        let store = fake(60);
        assert!(SyncKitClient::load_resume(&store, "h", 24).is_some());
        assert!(!*store.cleared.lock().unwrap());
    }

    #[test]
    fn a_session_past_the_reaper_window_is_dropped() {
        // The server aborts abandoned sessions at 24h, so an older record
        // names an upload_id that no longer exists. Resuming into it would
        // cost a doomed transfer before failing.
        let store = fake(RESUME_MAX_AGE_SECS + 1);
        assert!(SyncKitClient::load_resume(&store, "h", 24).is_none());
        assert!(
            *store.cleared.lock().unwrap(),
            "a dead record must be forgotten, not re-read next pass"
        );
    }

    #[test]
    fn a_store_failure_is_reported_and_swallowed() {
        // `best_effort` is the whole of the rule that nothing about the
        // resume store may fail an upload: it takes the error, says so, and
        // returns. Both halves matter and neither is a return value, so a
        // body replaced by `()` would behave identically to any caller. The
        // log is where the difference lives.
        let noisy = crate::test_support::events_from(|| {
            best_effort(
                "record_part",
                Err(SyncKitError::Internal("disk full".into())),
            );
        });
        let line = noisy
            .iter()
            .find(|e| {
                e.message
                    .as_deref()
                    .is_some_and(|m| m.contains("record_part") && m.contains("disk full"))
            })
            .expect("a store failure must name the operation and the cause");
        assert!(
            line.message
                .as_deref()
                .is_some_and(|m| m.contains("will not resume")),
            "the line must say what the failure costs, which is a restart from zero"
        );

        let quiet = crate::test_support::events_from(|| {
            best_effort("record_part", Ok(()));
        });
        assert!(
            quiet.is_empty(),
            "a store that worked has nothing to report"
        );
    }

    /// Half the server's 24h orphan-reaper window, in seconds, written out
    /// rather than read from [`RESUME_MAX_AGE_SECS`]. The point of the two
    /// tests below is to pin that constant's value as well as the
    /// comparison against it, and taking the number from the code under
    /// test would make them agree with whatever it happened to hold.
    const TWELVE_HOURS: i64 = 43_200;

    #[test]
    fn a_record_on_the_twelve_hour_boundary_is_still_usable() {
        // The comparison is `>`, not `>=`: the window is chosen to leave a
        // slow transfer room to finish inside it, and a record that has just
        // reached the boundary still names a session the server holds.
        let store = fake(TWELVE_HOURS);
        assert!(
            SyncKitClient::load_resume(&store, "h", 24).is_some(),
            "a record exactly at the limit is inside the window, not past it"
        );
        assert!(!*store.cleared.lock().unwrap());
    }

    #[test]
    fn a_record_one_second_past_twelve_hours_is_dropped() {
        let store = fake(TWELVE_HOURS + 1);
        assert!(SyncKitClient::load_resume(&store, "h", 24).is_none());
        assert!(*store.cleared.lock().unwrap());
    }

    /// The line `load_resume` writes when it throws a record away.
    const DISCARD_LINE: &str = "discarding an unusable blob resume record";

    #[test]
    fn only_a_faulty_record_is_reported_as_discarded() {
        // The three ways a record is dropped are not one event. A stale
        // session and a plan that does not tile the blob are faults, and an
        // operator wondering why an upload restarted wants to see them. A
        // record with no completed parts is the ordinary case of a run that
        // died before its first part landed; logging that would put a line
        // in front of somebody on every such retry, and it says nothing.
        //
        // The guard that draws that distinction returns nothing and changes
        // nothing, so the log is the only place it is observable at all.

        let empty = fake(60);
        empty.record.lock().unwrap().as_mut().unwrap().parts.clear();
        let quiet = crate::test_support::events_from(|| {
            assert!(SyncKitClient::load_resume(&empty, "h", 24).is_none());
        });
        assert!(
            quiet
                .iter()
                .all(|e| e.message.as_deref() != Some(DISCARD_LINE)),
            "a record that simply has nothing to save is not a fault to report"
        );

        let stale = fake(TWELVE_HOURS + 1);
        let logged = crate::test_support::events_from(|| {
            assert!(SyncKitClient::load_resume(&stale, "h", 24).is_none());
        });
        let line = logged
            .iter()
            .find(|e| e.message.as_deref() == Some(DISCARD_LINE))
            .expect("a stale session is a fault and must be reported");
        assert_eq!(line.field("stale"), Some("true"));
        assert_eq!(line.field("fits"), Some("true"), "it fits, it is just dead");

        let misfit = fake(60);
        let logged = crate::test_support::events_from(|| {
            assert!(SyncKitClient::load_resume(&misfit, "h", 999).is_none());
        });
        let line = logged
            .iter()
            .find(|e| e.message.as_deref() == Some(DISCARD_LINE))
            .expect("a record that cannot describe this upload must be reported");
        assert_eq!(line.field("stale"), Some("false"));
        assert_eq!(line.field("fits"), Some("false"));
    }

    #[test]
    fn a_record_for_a_different_length_is_dropped() {
        // Same content hash, different ciphertext length is a contradiction:
        // whatever it describes, it is not this upload.
        let store = fake(60);
        assert!(SyncKitClient::load_resume(&store, "h", 999).is_none());
        assert!(*store.cleared.lock().unwrap());
    }

    #[test]
    fn a_plan_that_does_not_tile_the_blob_is_dropped() {
        let store = fake(60);
        store
            .record
            .lock()
            .unwrap()
            .as_mut()
            .unwrap()
            .session
            .part_count = 7;
        assert!(SyncKitClient::load_resume(&store, "h", 24).is_none());
    }

    #[test]
    fn a_plan_with_a_zero_part_size_is_dropped() {
        // part_size 0 is the hostile case the `> 0` guard exists for: it is
        // also the divisor of the tiling check below it, so a guard that let
        // it through would divide by zero rather than merely mis-resume.
        let store = fake(60);
        store
            .record
            .lock()
            .unwrap()
            .as_mut()
            .unwrap()
            .session
            .part_size = 0;
        assert!(SyncKitClient::load_resume(&store, "h", 24).is_none());
        assert!(*store.cleared.lock().unwrap());
    }

    #[test]
    fn a_plan_with_no_parts_is_dropped_even_where_the_tiling_check_would_agree() {
        // part_count 0 over a 0-byte session: 0.div_ceil(8) == 0, so the
        // tiling check is satisfied and the `part_count > 0` guard is the
        // only thing rejecting it. A record naming a completed part in a
        // zero-part plan describes nothing.
        let store = fake(60);
        {
            let mut held = store.record.lock().unwrap();
            let session = &mut held.as_mut().unwrap().session;
            session.part_count = 0;
            session.size_bytes = 0;
        }
        assert!(SyncKitClient::load_resume(&store, "h", 0).is_none());
    }

    #[test]
    fn a_record_with_no_completed_parts_saves_nothing() {
        // Not an error: the upload starts at part 1 either way. Dropping it
        // means the session recorded is the one actually being used.
        let store = fake(60);
        store.record.lock().unwrap().as_mut().unwrap().parts.clear();
        assert!(SyncKitClient::load_resume(&store, "h", 24).is_none());
    }

    #[test]
    fn a_store_that_errors_costs_a_restart_and_nothing_else() {
        struct Broken;
        impl BlobResumeStore for Broken {
            fn load(&self, _: &str) -> Result<Option<ResumeRecord>> {
                Err(SyncKitError::Internal("disk gone".into()))
            }
            fn begin(&self, _: &str, _: &ResumeSession) -> Result<()> {
                Ok(())
            }
            fn record_part(&self, _: &str, _: &ResumePart, _: &[ResumeChunk]) -> Result<()> {
                Ok(())
            }
            fn clear(&self, _: &str) -> Result<()> {
                Ok(())
            }
        }
        assert!(SyncKitClient::load_resume(&Broken, "h", 24).is_none());
    }

    #[test]
    fn only_a_recurring_failure_gives_up_the_session() {
        assert!(is_resumable_failure(&SyncKitError::Server {
            status: 503,
            message: String::new(),
            retry_after_secs: None,
        }));
        assert!(!is_resumable_failure(&SyncKitError::IntegrityFailed {
            expected: "a".into(),
            actual: "b".into(),
        }));
        assert!(!is_resumable_failure(&SyncKitError::Internal(
            "geometry".into()
        )));
    }

    /// The header plus every sealed chunk, which is what the boundary
    /// arithmetic walks.
    fn header_len() -> usize {
        crypto::blob_header_bytes(0).len()
    }

    #[test]
    fn a_fresh_upload_starts_at_the_top() {
        assert_eq!(resume_boundary(4096, header_len(), 0), (0, 0));
    }

    #[test]
    fn a_boundary_inside_the_first_chunk_reports_its_offset() {
        let h = header_len();
        // 1000 bytes into chunk 0's sealed bytes.
        assert_eq!(resume_boundary(4 << 20, h, h + 1000), (0, 1000));
    }

    #[test]
    fn a_boundary_past_a_whole_chunk_lands_in_the_next() {
        let h = header_len();
        let c0 = crypto::sealed_blob_chunk_len(4 << 20, 0);
        assert_eq!(resume_boundary(4 << 20, h, h + c0), (1, 0));
        assert_eq!(resume_boundary(4 << 20, h, h + c0 + 5), (1, 5));
    }

    #[test]
    fn a_boundary_past_the_last_chunk_means_nothing_is_left_to_send() {
        let len = 4 << 20;
        let cipher = crypto::blob_encrypted_len(len);
        assert_eq!(
            resume_boundary(len, header_len(), cipher),
            (crypto::blob_chunk_count_for(len), 0)
        );
    }

    #[test]
    fn every_boundary_of_a_real_blob_maps_back_to_the_bytes_it_names() {
        // The invariant the resume depends on: skipping to a part boundary
        // and re-emitting from `within` into the boundary chunk reproduces
        // the ciphertext tail exactly. Checked against a real sealed blob.
        let key = [7u8; 32];
        let plaintext: Vec<u8> = (0..(crypto::BLOB_CHUNK_SIZE * 2 + 511))
            .map(|i| i as u8)
            .collect();
        let hash = "a".repeat(64);
        let whole = crypto::encrypt_blob_chunked(&plaintext, &key, &hash).unwrap();
        let h = crypto::blob_header_bytes(plaintext.len()).len();
        let count = crypto::blob_chunk_count_for(plaintext.len());

        for skip in [h + 1, h + 700 * 1024, h + 1_400_000, whole.len() - 3] {
            let (index, within) = resume_boundary(plaintext.len(), h, skip);
            assert!(index < count, "skip {skip} fell off the end");
            // Where that chunk starts in the ciphertext.
            let start: usize = h
                + (0..index)
                    .map(|i| crypto::sealed_blob_chunk_len(plaintext.len(), i))
                    .sum::<usize>();
            assert_eq!(start + within, skip, "boundary {skip} must be exact");
            // And re-sealing it under its own nonce reproduces those bytes.
            let sealed =
                &whole[start..start + crypto::sealed_blob_chunk_len(plaintext.len(), index)];
            let from = index as usize * crypto::BLOB_CHUNK_SIZE;
            let to = (from + crypto::BLOB_CHUNK_SIZE).min(plaintext.len());
            let again = crypto::reseal_blob_chunk(
                &plaintext[from..to],
                &key,
                &hash,
                index,
                count,
                &crypto::blob_chunk_nonce(sealed).unwrap(),
            )
            .unwrap();
            assert_eq!(again, sealed, "chunk {index} must re-seal byte-identically");
        }
    }
}

#[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);
}

// ── The in-memory and streaming size caps ──

#[test]
fn the_blob_cap_is_four_gibibytes_exactly() {
    // Pinned as a literal rather than as the same arithmetic the constant
    // uses, because that arithmetic is what can be wrong. The two readings a
    // single wrong operator produces here are 1_077_936_128 and 4_195_328:
    // both look like plausible caps, and either would refuse legitimate
    // media the SDK documents itself as carrying. Nothing else in the suite
    // can see the difference, since no fixture is anywhere near any of the
    // three values.
    assert_eq!(MAX_BLOB_BYTES, 4_294_967_296, "4 GiB");
}

/// A client holding a key but no session: every blob path gets past the
/// key check and stops at `require_token`, which is what makes
/// `NotAuthenticated` mean "the size check let this through".
fn keyed_but_unauthenticated() -> SyncKitClient {
    let client = SyncKitClient::new(crate::SyncKitConfig {
        server_url: "https://example.invalid".to_string(),
        api_key: "test-api-key".to_string(),
    });
    client.set_master_key_raw([9u8; 32]);
    client
}

/// A sparse file of `len` bytes: `set_len` allocates nothing, so the
/// multi-gigabyte sizes the cap is written in terms of cost no disk. The
/// cap is read off `metadata`, which is all these tests reach.
fn sparse_file(len: u64) -> std::path::PathBuf {
    use std::sync::atomic::{AtomicU64, Ordering};
    static N: AtomicU64 = AtomicU64::new(0);
    let mut p = std::env::temp_dir();
    p.push(format!(
        "synckit_cap_{}_{}",
        std::process::id(),
        N.fetch_add(1, Ordering::Relaxed)
    ));
    let f = std::fs::File::create(&p).unwrap();
    f.set_len(len).unwrap();
    p
}

#[tokio::test]
async fn the_streaming_cap_accepts_a_blob_of_exactly_the_cap_and_refuses_one_byte_more() {
    // Both sides of the bound. `>` differs from `>=` and from `==` only at
    // the cap itself, so a test that only uploads something small cannot
    // see any of them: at every reachable size all three agree.
    let client = keyed_but_unauthenticated();
    let hash = "b".repeat(64);

    let at_cap = sparse_file(MAX_BLOB_BYTES as u64);
    let err = client
        .blob_upload_streaming(&hash, &at_cap)
        .await
        .unwrap_err();
    assert!(
        matches!(err, SyncKitError::NotAuthenticated),
        "a blob of exactly {MAX_BLOB_BYTES} bytes is under the cap and must reach the session check, got {err:?}"
    );

    let over = sparse_file(MAX_BLOB_BYTES as u64 + 1);
    let err = client
        .blob_upload_streaming(&hash, &over)
        .await
        .unwrap_err();
    match err {
        SyncKitError::InvalidArgument(m) => {
            assert!(m.contains("client cap"), "wrong rejection: {m}");
        }
        other => panic!("one byte over the cap must be refused, got {other:?}"),
    }

    // And a small file is not refused by a cap that has been inverted.
    let small = sparse_file(1_000);
    let err = client
        .blob_upload_streaming(&hash, &small)
        .await
        .unwrap_err();
    assert!(
        matches!(err, SyncKitError::NotAuthenticated),
        "a 1000-byte blob must reach the session check, got {err:?}"
    );

    for p in [at_cap, over, small] {
        let _ = std::fs::remove_file(p);
    }
}

#[tokio::test]
async fn the_in_memory_cap_passes_an_ordinary_blob_through_to_the_put() {
    // The reachable half of the same bound: a blob far under the cap must
    // not be rejected by it, so the call fails at the transport instead.
    // (The unreachable half is the cap itself, which would need a 4 GiB
    // allocation to reach.) A relative URL is a reqwest builder error,
    // classified as permanent, so no network attempt is made.
    let client = keyed_but_unauthenticated();
    let err = client
        .blob_upload(&"c".repeat(64), "not-a-url", vec![7u8; 5_000])
        .await
        .unwrap_err();
    assert!(
        matches!(err, SyncKitError::Http(_)),
        "a 5000-byte blob is under the cap and must reach the PUT, got {err:?}"
    );
}