//! Integration tests using wiremock to simulate the MNW SyncKit server.
//!
//! These tests verify the full HTTP round-trip including retry behavior,
//! encryption/decryption, and error classification.

use std::sync::Arc;

use base64::Engine;
use chrono::Utc;
use serde_json::json;
use sha2::Digest;
use uuid::Uuid;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};

use std::time::Duration;
use synckit_client::{
    AppId, ChangeEntry, ChangeOp, DeviceId, Hlc, SyncKitClient, SyncKitConfig, SyncKitError, UserId,
};

// ── Helpers ──

fn fake_jwt(exp: i64) -> String {
    let header =
        base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(r#"{"alg":"HS256","typ":"JWT"}"#);
    let payload = json!({
        "sub": "550e8400-e29b-41d4-a716-446655440000",
        "app": "6ba7b810-9dad-11d1-80b4-00c04fd430c8",
        "exp": exp,
        "iat": exp - 3600,
    });
    let payload_b64 =
        base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(payload.to_string().as_bytes());
    let sig = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(b"fake-signature");
    format!("{header}.{payload_b64}.{sig}")
}

fn fresh_token() -> String {
    fake_jwt(Utc::now().timestamp() + 3600)
}

fn test_ids() -> (UserId, AppId) {
    (
        UserId::new(Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap()),
        AppId::new(Uuid::parse_str("6ba7b810-9dad-11d1-80b4-00c04fd430c8").unwrap()),
    )
}

/// Install the rustls crypto provider once. reqwest is built `rustls-no-provider`,
/// so a real consumer app installs one at startup (audiofiles installs ring); these
/// tests have no such app, so they install ring themselves before building a client.
fn ensure_crypto_provider() {
    static PROVIDER: std::sync::Once = std::sync::Once::new();
    PROVIDER.call_once(|| {
        let _ = rustls::crypto::ring::default_provider().install_default();
    });
}

fn client_for(server: &MockServer) -> SyncKitClient {
    ensure_crypto_provider();
    SyncKitClient::new(SyncKitConfig {
        server_url: server.uri(),
        api_key: "test-api-key".to_string(),
    })
}

fn authed_client(server: &MockServer) -> SyncKitClient {
    let client = client_for(server);
    let (user_id, app_id) = test_ids();
    client.restore_session(&fresh_token(), user_id, app_id);
    client
}

fn auth_response_json() -> serde_json::Value {
    let (user_id, app_id) = test_ids();
    json!({
        "token": fresh_token(),
        "user_id": user_id,
        "app_id": app_id,
    })
}

fn device_json() -> serde_json::Value {
    let (user_id, app_id) = test_ids();
    json!({
        "id": Uuid::new_v4(),
        "app_id": app_id,
        "user_id": user_id,
        "device_name": "Test Device",
        "platform": "test",
        "last_seen_at": "2025-01-01T00:00:00Z",
        "created_at": "2025-01-01T00:00:00Z",
    })
}

// ── Response body cap (DoS) ──

#[tokio::test]
async fn oversized_response_body_is_capped_not_buffered() {
    // A hostile/buggy server streams a control-plane body far larger than the
    // 8 MiB cap. The client must reject it (the cap fast-rejects on the honest
    // Content-Length) instead of buffering it into memory and OOMing.
    let server = MockServer::start().await;
    let huge = vec![b'x'; 9 * 1024 * 1024];
    Mock::given(method("GET"))
        .and(path("/api/v1/sync/status"))
        .respond_with(ResponseTemplate::new(200).set_body_bytes(huge))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let err = client.status().await.unwrap_err();
    assert!(
        matches!(err, SyncKitError::Internal(ref m) if m.contains("cap")),
        "expected the body cap to reject the oversized response, got: {err:?}"
    );
}

// ── Auth flow ──

#[tokio::test]
async fn authenticate_success_stores_session() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/auth"))
        .respond_with(ResponseTemplate::new(200).set_body_json(auth_response_json()))
        .mount(&server)
        .await;

    let client = client_for(&server);
    let (user_id, app_id) = client
        .authenticate("user@test.com", "password", "test-key")
        .await
        .unwrap();

    let info = client.session_info().expect("session stored");
    assert_eq!(info.user_id, user_id);
    assert_eq!(info.app_id, app_id);
}

#[tokio::test]
async fn authenticate_wrong_password_no_retry() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/auth"))
        .respond_with(ResponseTemplate::new(401).set_body_string("Unauthorized"))
        .expect(1) // Must be called exactly once (no retry)
        .mount(&server)
        .await;

    let client = client_for(&server);
    let err = client
        .authenticate("user@test.com", "wrong", "test-key")
        .await
        .unwrap_err();

    assert!(
        matches!(err, SyncKitError::Server { status: 401, .. }),
        "Expected 401 error, got: {err:?}"
    );
}

#[tokio::test]
async fn authenticate_retries_on_503() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/auth"))
        .respond_with(ResponseTemplate::new(503).set_body_string("Service Unavailable"))
        .up_to_n_times(1)
        .mount(&server)
        .await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/auth"))
        .respond_with(ResponseTemplate::new(200).set_body_json(auth_response_json()))
        .mount(&server)
        .await;

    let client = client_for(&server);
    let result = client
        .authenticate("user@test.com", "password", "test-key")
        .await;
    assert!(result.is_ok(), "Should succeed after retry: {result:?}");
}

#[tokio::test]
async fn authenticate_with_code_success() {
    let server = MockServer::start().await;

    let (user_id, app_id) = test_ids();
    Mock::given(method("POST"))
        .and(path("/oauth/token"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "access_token": fresh_token(),
            "token_type": "Bearer",
            "expires_in": 3600,
            "user_id": user_id,
            "app_id": app_id,
        })))
        .mount(&server)
        .await;

    let client = client_for(&server);
    let (uid, aid) = client
        .authenticate_with_code("auth-code", "verifier", 8080, "test-key")
        .await
        .unwrap();

    assert_eq!(uid, user_id);
    assert_eq!(aid, app_id);
    assert!(client.session_info().is_some());
}

// ── Device management ──

#[tokio::test]
async fn register_device_success() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/devices"))
        .respond_with(ResponseTemplate::new(200).set_body_json(device_json()))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let device = client.register_device("MacBook", "macos").await.unwrap();
    assert_eq!(device.device_name, "Test Device");
}

#[tokio::test]
async fn register_device_retries_on_transient() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/devices"))
        .respond_with(ResponseTemplate::new(502).set_body_string("Bad Gateway"))
        .up_to_n_times(1)
        .mount(&server)
        .await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/devices"))
        .respond_with(ResponseTemplate::new(200).set_body_json(device_json()))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let result = client.register_device("MacBook", "macos").await;
    assert!(result.is_ok(), "Should succeed after retry: {result:?}");
}

#[tokio::test]
async fn list_devices_success() {
    let server = MockServer::start().await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/devices"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!([device_json()])))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let devices = client.list_devices().await.unwrap();
    assert_eq!(devices.len(), 1);
    assert_eq!(devices[0].device_name, "Test Device");
}

// ── Push / Pull with encryption ──

#[tokio::test]
async fn push_encrypts_data() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/push"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1})))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let device_id = DeviceId::new(Uuid::new_v4());
    let cursor = client
        .push(
            device_id,
            vec![ChangeEntry {
                table: "tasks".into(),
                op: ChangeOp::Insert,
                row_id: "row-1".into(),
                timestamp: Utc::now(),
                hlc: Hlc::zero(DeviceId::nil()),
                data: Some(json!({"title": "Secret task"})),
                extra: serde_json::Map::default(),
            }],
        )
        .await
        .unwrap();

    assert_eq!(cursor, 1);

    // Verify the request body was sent with encrypted data (not plaintext)
    let requests = server.received_requests().await.unwrap();
    let push_req = requests
        .iter()
        .find(|r| r.url.path() == "/api/v1/sync/push")
        .unwrap();
    let body: serde_json::Value = serde_json::from_slice(&push_req.body).unwrap();
    let wire_data = body["changes"][0]["data"].as_str().unwrap();
    assert!(
        !wire_data.contains("Secret task"),
        "Plaintext should not appear on the wire"
    );
}

#[tokio::test]
async fn pull_decrypts_data() {
    let server = MockServer::start().await;

    let client = authed_client(&server);
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    // Encrypt a value to simulate what the server would return
    let plaintext = json!({"title": "Decrypted task"});
    let encrypted = synckit_client::crypto::encrypt_json(&plaintext, &key).unwrap();

    let device_id = DeviceId::new(Uuid::new_v4());
    Mock::given(method("POST"))
        .and(path("/api/v1/sync/pull"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "changes": [{
                "seq": 1,
                "device_id": device_id,
                "table": "tasks",
                "op": "INSERT",
                "row_id": "row-1",
                "timestamp": "2025-06-01T12:00:00Z",
                "data": encrypted,
            }],
            "cursor": 1,
            "has_more": false,
        })))
        .mount(&server)
        .await;

    let (changes, cursor, has_more) = client.pull(device_id, 0).await.unwrap();
    assert_eq!(changes.len(), 1);
    assert_eq!(cursor, 1);
    assert!(!has_more);
    assert_eq!(changes[0].data.as_ref().unwrap(), &plaintext);
}

#[tokio::test]
async fn push_retries_on_503() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/push"))
        .respond_with(ResponseTemplate::new(503))
        .up_to_n_times(1)
        .mount(&server)
        .await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/push"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 5})))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let cursor = client
        .push(DeviceId::new(Uuid::new_v4()), vec![])
        .await
        .unwrap();
    assert_eq!(cursor, 5);
}

#[tokio::test]
async fn push_fails_immediately_on_401() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/push"))
        .respond_with(ResponseTemplate::new(401).set_body_string("Unauthorized"))
        .expect(1)
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let err = client
        .push(DeviceId::new(Uuid::new_v4()), vec![])
        .await
        .unwrap_err();
    assert!(matches!(err, SyncKitError::Server { status: 401, .. }));
}

#[tokio::test]
async fn pull_with_has_more_pagination() {
    let server = MockServer::start().await;

    let client = authed_client(&server);
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let device_id = DeviceId::new(Uuid::new_v4());

    // First pull: has_more = true
    Mock::given(method("POST"))
        .and(path("/api/v1/sync/pull"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "changes": [],
            "cursor": 50,
            "has_more": true,
        })))
        .up_to_n_times(1)
        .mount(&server)
        .await;

    let (changes, cursor, has_more) = client.pull(device_id, 0).await.unwrap();
    assert!(changes.is_empty());
    assert_eq!(cursor, 50);
    assert!(has_more);

    // Second pull from cursor 50: has_more = false
    Mock::given(method("POST"))
        .and(path("/api/v1/sync/pull"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "changes": [],
            "cursor": 100,
            "has_more": false,
        })))
        .mount(&server)
        .await;

    let (_, cursor2, has_more2) = client.pull(device_id, 50).await.unwrap();
    assert_eq!(cursor2, 100);
    assert!(!has_more2);
}

// ── Blob operations ──

#[tokio::test]
async fn blob_upload_url_success() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/blobs/upload"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "upload_url": "https://s3.example.com/put",
            "already_exists": false,
        })))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let resp = client.blob_upload_url("sha256-abc", 1024).await.unwrap();
    assert_eq!(resp.upload_url, "https://s3.example.com/put");
    assert!(!resp.already_exists);
}

#[tokio::test]
async fn blob_upload_url_declares_the_length_the_put_will_carry() {
    // The server signs the declared size into the presigned URL as
    // Content-Length, a SignedHeader, so declaring anything other than the
    // exact ciphertext length makes the PUT fail SigV4. The caller passes the
    // plaintext size it sees on disk; the SDK converts. This test pins the two
    // halves together, which is the only place the mismatch would show up:
    // wiremock does not verify signatures, and the server's own tests use an
    // in-memory backend that does not sign at all.
    let server = MockServer::start().await;

    let upload_path = "/s3/sized-upload";
    Mock::given(method("POST"))
        .and(path("/api/v1/sync/blobs/upload"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "upload_url": format!("{}{upload_path}", server.uri()),
            "already_exists": false,
        })))
        .mount(&server)
        .await;
    Mock::given(method("PUT"))
        .and(path(upload_path))
        .respond_with(ResponseTemplate::new(200))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    client.set_master_key_raw(synckit_client::crypto::generate_master_key());

    // Spans two chunks, so the framing overhead is more than a single chunk's.
    let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE + 500))
        .map(|i| i as u8)
        .collect();
    let hash = hex::encode(sha2::Sha256::digest(&plaintext));

    let resp = client
        .blob_upload_url(&hash, plaintext.len() as i64)
        .await
        .unwrap();
    client
        .blob_upload(&hash, &resp.upload_url, plaintext.clone())
        .await
        .unwrap();

    let reqs = server.received_requests().await.unwrap();
    let declared: serde_json::Value = reqs
        .iter()
        .find(|r| r.url.path() == "/api/v1/sync/blobs/upload")
        .unwrap()
        .body_json()
        .unwrap();
    let put_len = reqs
        .iter()
        .find(|r| r.url.path() == upload_path)
        .unwrap()
        .body
        .len();

    assert_eq!(
        declared["size_bytes"].as_u64().unwrap(),
        put_len as u64,
        "the declared size must equal the bytes actually PUT, or the signature fails"
    );
    assert!(
        put_len > plaintext.len(),
        "the PUT carries ciphertext, which is longer than the plaintext"
    );
}

#[tokio::test]
async fn blob_upload_encrypts_data() {
    let server = MockServer::start().await;

    let upload_path = "/s3/upload";
    Mock::given(method("PUT"))
        .and(path(upload_path))
        .respond_with(ResponseTemplate::new(200))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let plaintext = b"hello blob data";
    let presigned = format!("{}{}", server.uri(), upload_path);
    client
        .blob_upload("sha256-test", &presigned, plaintext.to_vec())
        .await
        .unwrap();

    // Verify uploaded body is encrypted (not plaintext)
    let requests = server.received_requests().await.unwrap();
    let upload_req = requests
        .iter()
        .find(|r| r.url.path() == upload_path)
        .unwrap();
    assert!(
        !upload_req
            .body
            .windows(plaintext.len())
            .any(|w| w == plaintext),
        "Plaintext should not appear in uploaded body"
    );
    // Encrypted blob should be larger due to nonce + tag overhead
    assert!(upload_req.body.len() > plaintext.len());
}

#[tokio::test]
async fn blob_download_decrypts_data() {
    let server = MockServer::start().await;

    let client = authed_client(&server);
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    // Encrypt data to simulate what S3 would return. A legacy (untagged) blob
    // still decrypts through the AAD-aware reader and must pass the hash check.
    let plaintext = b"decrypted blob content";
    let hash = hex::encode(sha2::Sha256::digest(plaintext));
    let encrypted = synckit_client::crypto::encrypt_bytes(plaintext, &key).unwrap();

    let download_path = "/s3/download";
    Mock::given(method("GET"))
        .and(path(download_path))
        .respond_with(ResponseTemplate::new(200).set_body_bytes(encrypted))
        .mount(&server)
        .await;

    let presigned = format!("{}{}", server.uri(), download_path);
    let result = client.blob_download(&hash, &presigned).await.unwrap();
    assert_eq!(result, plaintext);
}

#[tokio::test]
async fn blob_upload_retries_on_503() {
    let server = MockServer::start().await;

    let upload_path = "/s3/retry-upload";
    Mock::given(method("PUT"))
        .and(path(upload_path))
        .respond_with(ResponseTemplate::new(503))
        .up_to_n_times(1)
        .mount(&server)
        .await;

    Mock::given(method("PUT"))
        .and(path(upload_path))
        .respond_with(ResponseTemplate::new(200))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let presigned = format!("{}{}", server.uri(), upload_path);
    let result = client
        .blob_upload("sha256-x", &presigned, b"data".to_vec())
        .await;
    assert!(result.is_ok(), "Should succeed after retry: {result:?}");
}

// ── Token handling ──

#[tokio::test]
async fn expired_jwt_returns_token_expired() {
    let server = MockServer::start().await;
    let client = client_for(&server);
    let (user_id, app_id) = test_ids();

    let expired = fake_jwt(Utc::now().timestamp() - 3600);
    client.restore_session(&expired, user_id, app_id);

    let err = client.status().await.unwrap_err();
    assert!(
        matches!(err, SyncKitError::TokenExpired),
        "Expected TokenExpired, got: {err:?}"
    );
}

#[tokio::test]
async fn near_expiry_jwt_returns_token_expired() {
    let server = MockServer::start().await;
    let client = client_for(&server);
    let (user_id, app_id) = test_ids();

    // Token expires in 10 seconds (within 30-second buffer)
    let near_expiry = fake_jwt(Utc::now().timestamp() + 10);
    client.restore_session(&near_expiry, user_id, app_id);

    let err = client.status().await.unwrap_err();
    assert!(
        matches!(err, SyncKitError::TokenExpired),
        "Expected TokenExpired, got: {err:?}"
    );
}

// ── Session management ──

#[tokio::test]
async fn restore_then_clear_session() {
    let server = MockServer::start().await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/status"))
        .respond_with(
            ResponseTemplate::new(200)
                .set_body_json(json!({"total_changes": 0, "latest_cursor": null})),
        )
        .mount(&server)
        .await;

    let client = authed_client(&server);

    // Should work while authenticated
    let status = client.status().await.unwrap();
    assert_eq!(status.total_changes, 0);

    // Clear session
    client.clear_session();

    // Now should fail
    let err = client.status().await.unwrap_err();
    assert!(matches!(err, SyncKitError::NotAuthenticated));
}

#[tokio::test]
async fn status_without_auth_returns_not_authenticated() {
    let server = MockServer::start().await;
    let client = client_for(&server);

    let err = client.status().await.unwrap_err();
    assert!(matches!(err, SyncKitError::NotAuthenticated));
}

#[tokio::test]
async fn push_without_auth_returns_not_authenticated() {
    let server = MockServer::start().await;
    let client = client_for(&server);

    let err = client
        .push(DeviceId::new(Uuid::new_v4()), vec![])
        .await
        .unwrap_err();
    assert!(matches!(err, SyncKitError::NotAuthenticated));
}

// ── Key management ──

#[tokio::test]
async fn has_server_key_true_on_200() {
    let server = MockServer::start().await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/keys"))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": "envelope-data"})),
        )
        .mount(&server)
        .await;

    let client = authed_client(&server);
    assert!(client.has_server_key().await.unwrap());
}

#[tokio::test]
async fn has_server_key_false_on_404() {
    let server = MockServer::start().await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/keys"))
        .respond_with(ResponseTemplate::new(404))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    assert!(!client.has_server_key().await.unwrap());
}

#[tokio::test]
async fn has_server_key_retries_on_500() {
    let server = MockServer::start().await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/keys"))
        .respond_with(ResponseTemplate::new(500).set_body_string("Internal Server Error"))
        .up_to_n_times(1)
        .mount(&server)
        .await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/keys"))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": "envelope"})),
        )
        .mount(&server)
        .await;

    let client = authed_client(&server);
    assert!(client.has_server_key().await.unwrap());
}

// ── Concurrent access ──

#[tokio::test]
async fn concurrent_push_pull_no_panics() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/push"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1})))
        .mount(&server)
        .await;

    let device_id = DeviceId::new(Uuid::new_v4());
    Mock::given(method("POST"))
        .and(path("/api/v1/sync/pull"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "changes": [],
            "cursor": 0,
            "has_more": false,
        })))
        .mount(&server)
        .await;

    let client = Arc::new(authed_client(&server));
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let mut handles = Vec::new();
    for _ in 0..4 {
        let c = Arc::clone(&client);
        let did = device_id;
        handles.push(tokio::spawn(async move {
            let _ = c.push(did, vec![]).await;
            let _ = c.pull(did, 0).await;
        }));
    }

    for h in handles {
        h.await.unwrap(); // No panics
    }
}

// ── Error classification ──

#[tokio::test]
async fn error_429_is_retried() {
    let server = MockServer::start().await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/status"))
        .respond_with(ResponseTemplate::new(429).set_body_string("Too Many Requests"))
        .up_to_n_times(1)
        .mount(&server)
        .await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/status"))
        .respond_with(
            ResponseTemplate::new(200)
                .set_body_json(json!({"total_changes": 10, "latest_cursor": 5})),
        )
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let status = client.status().await.unwrap();
    assert_eq!(status.total_changes, 10);
}

#[tokio::test]
async fn error_400_not_retried() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/devices"))
        .respond_with(ResponseTemplate::new(400).set_body_string("Bad Request"))
        .expect(1)
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let err = client.register_device("Device", "test").await.unwrap_err();
    assert!(matches!(err, SyncKitError::Server { status: 400, .. }));
}

// ── Status endpoint ──

#[tokio::test]
async fn status_success() {
    let server = MockServer::start().await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/status"))
        .respond_with(
            ResponseTemplate::new(200)
                .set_body_json(json!({"total_changes": 42, "latest_cursor": 100})),
        )
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let status = client.status().await.unwrap();
    assert_eq!(status.total_changes, 42);
    assert_eq!(status.latest_cursor, Some(100));
}

#[tokio::test]
async fn status_retries_on_transient() {
    let server = MockServer::start().await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/status"))
        .respond_with(ResponseTemplate::new(504).set_body_string("Gateway Timeout"))
        .up_to_n_times(1)
        .mount(&server)
        .await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/status"))
        .respond_with(
            ResponseTemplate::new(200)
                .set_body_json(json!({"total_changes": 0, "latest_cursor": null})),
        )
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let status = client.status().await.unwrap();
    assert_eq!(status.total_changes, 0);
}

// ── Blob confirm ──

#[tokio::test]
async fn blob_confirm_success() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/blobs/confirm"))
        .respond_with(ResponseTemplate::new(200))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    client.blob_confirm("sha256-abc", 1024).await.unwrap();
}

// ── Blob download URL ──

#[tokio::test]
async fn blob_download_url_success() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/blobs/download"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "download_url": "https://s3.example.com/get",
        })))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let url = client.blob_download_url("sha256-abc").await.unwrap();
    assert_eq!(url, "https://s3.example.com/get");
}

// ── change_password: CRITICAL bug fix tests ──

/// Helper: set up a server that serves a wrapped master key envelope.
/// Returns (client, master_key, envelope_json).
fn setup_change_password_test(
    server: &MockServer,
    password: &str,
) -> (SyncKitClient, [u8; 32], String) {
    let client = authed_client(server);
    let master_key = synckit_client::crypto::generate_master_key();
    let envelope = synckit_client::crypto::wrap_master_key(&master_key, password).unwrap();

    // Cache the master key in the client (simulating normal logged-in state)
    client.set_master_key_raw(master_key);

    (client, master_key, envelope)
}

#[tokio::test]
async fn change_password_wrong_old_password_with_cached_key_fails() {
    let server = MockServer::start().await;
    let (client, _master_key, envelope) = setup_change_password_test(&server, "correct-old-pass");

    // Server returns the envelope on GET
    Mock::given(method("GET"))
        .and(path("/api/v1/sync/keys"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
        .mount(&server)
        .await;

    // Attempt to change password with wrong old password.
    // The key IS cached, but the old password must still be validated.
    let result = client.change_password("wrong-old-pass", "new-pass").await;

    assert!(
        result.is_err(),
        "change_password must fail when old_password is wrong, even with cached key"
    );
    assert!(
        matches!(result.unwrap_err(), SyncKitError::DecryptionFailed),
        "Should get DecryptionFailed for wrong old password"
    );
}

#[tokio::test]
async fn change_password_correct_old_password_with_cached_key_succeeds() {
    let server = MockServer::start().await;
    let (client, master_key, envelope) = setup_change_password_test(&server, "correct-old-pass");

    // Server returns the envelope on GET
    Mock::given(method("GET"))
        .and(path("/api/v1/sync/keys"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
        .mount(&server)
        .await;

    // Server accepts the new envelope on PUT
    Mock::given(method("PUT"))
        .and(path("/api/v1/sync/keys"))
        .respond_with(ResponseTemplate::new(200))
        .mount(&server)
        .await;

    let result = client.change_password("correct-old-pass", "new-pass").await;
    assert!(
        result.is_ok(),
        "change_password should succeed with correct old password"
    );

    // Verify the PUT request was made (new envelope was uploaded)
    let requests = server.received_requests().await.unwrap();
    let put_requests: Vec<_> = requests
        .iter()
        .filter(|r| r.method.as_str() == "PUT" && r.url.path() == "/api/v1/sync/keys")
        .collect();
    assert_eq!(put_requests.len(), 1, "Should have sent exactly one PUT");

    // Verify the new envelope can be unwrapped with the new password
    let put_body: serde_json::Value = serde_json::from_slice(&put_requests[0].body).unwrap();
    let new_envelope = put_body["encrypted_key"].as_str().unwrap();
    let recovered = synckit_client::crypto::unwrap_master_key(new_envelope, "new-pass").unwrap();
    assert_eq!(
        recovered, master_key,
        "New envelope should unwrap to the same master key"
    );
}

#[tokio::test]
async fn change_password_wrong_old_password_without_cached_key_fails() {
    let server = MockServer::start().await;
    let master_key = synckit_client::crypto::generate_master_key();
    let envelope =
        synckit_client::crypto::wrap_master_key(&master_key, "correct-old-pass").unwrap();

    let client = authed_client(&server);
    // Deliberately NOT setting master key: no cached key

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/keys"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
        .mount(&server)
        .await;

    let result = client.change_password("wrong-old-pass", "new-pass").await;

    assert!(
        result.is_err(),
        "change_password must fail with wrong old password even without cached key"
    );
    assert!(matches!(
        result.unwrap_err(),
        SyncKitError::DecryptionFailed
    ));
}

#[tokio::test]
async fn change_password_old_envelope_invalid_with_new_password() {
    let server = MockServer::start().await;
    let (client, _master_key, envelope) = setup_change_password_test(&server, "old-pass");

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/keys"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
        .mount(&server)
        .await;

    Mock::given(method("PUT"))
        .and(path("/api/v1/sync/keys"))
        .respond_with(ResponseTemplate::new(200))
        .mount(&server)
        .await;

    client
        .change_password("old-pass", "new-pass")
        .await
        .unwrap();

    // Old password should NOT work on the new envelope
    let requests = server.received_requests().await.unwrap();
    let put_req = requests
        .iter()
        .find(|r| r.method.as_str() == "PUT" && r.url.path() == "/api/v1/sync/keys")
        .unwrap();
    let body: serde_json::Value = serde_json::from_slice(&put_req.body).unwrap();
    let new_envelope = body["encrypted_key"].as_str().unwrap();

    let result = synckit_client::crypto::unwrap_master_key(new_envelope, "old-pass");
    assert!(
        result.is_err(),
        "Old password must not work on the new envelope"
    );
}

// ── Empty changelog push ──

#[tokio::test]
async fn push_empty_changes_succeeds() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/push"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 0})))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let cursor = client
        .push(DeviceId::new(Uuid::new_v4()), vec![])
        .await
        .unwrap();
    assert_eq!(cursor, 0);
}

// ── Malformed server responses ──

#[tokio::test]
async fn push_malformed_json_response_handled() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/push"))
        .respond_with(ResponseTemplate::new(200).set_body_string("not valid json at all"))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let result = client.push(DeviceId::new(Uuid::new_v4()), vec![]).await;
    assert!(result.is_err(), "Malformed JSON should produce an error");
    // Should be a JSON parse error, not a panic
    let err = result.unwrap_err();
    assert!(
        matches!(err, SyncKitError::Http(_) | SyncKitError::Json(_)),
        "Expected Http or Json error for malformed response, got: {err:?}"
    );
}

#[tokio::test]
async fn pull_malformed_json_response_handled() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/pull"))
        .respond_with(ResponseTemplate::new(200).set_body_string("{invalid json}"))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let result = client.pull(DeviceId::new(Uuid::new_v4()), 0).await;
    assert!(result.is_err(), "Malformed JSON should produce an error");
}

#[tokio::test]
async fn status_malformed_json_response_handled() {
    let server = MockServer::start().await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/status"))
        .respond_with(ResponseTemplate::new(200).set_body_string("this is not json"))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let result = client.status().await;
    assert!(result.is_err());
}

// ── Server error messages preserved ──

#[tokio::test]
async fn server_error_message_preserved() {
    let server = MockServer::start().await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/status"))
        .respond_with(
            ResponseTemplate::new(422).set_body_string("Validation failed: missing field"),
        )
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let err = client.status().await.unwrap_err();
    match err {
        SyncKitError::Server {
            status, message, ..
        } => {
            assert_eq!(status, 422);
            assert!(
                message.contains("Validation failed"),
                "Error message should be preserved: {message}"
            );
        }
        other => panic!("Expected Server error, got: {other:?}"),
    }
}

// ── Concurrent operations ──

#[tokio::test]
async fn concurrent_push_operations_no_data_corruption() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/push"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1})))
        .mount(&server)
        .await;

    let client = Arc::new(authed_client(&server));
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let mut handles = Vec::new();
    for i in 0..8 {
        let c = Arc::clone(&client);
        handles.push(tokio::spawn(async move {
            let device_id = DeviceId::new(Uuid::new_v4());
            let entry = ChangeEntry {
                table: format!("table_{i}"),
                op: ChangeOp::Insert,
                row_id: format!("row_{i}"),
                timestamp: Utc::now(),
                hlc: Hlc::zero(DeviceId::nil()),
                data: Some(json!({"index": i})),
                extra: serde_json::Map::default(),
            };
            c.push(device_id, vec![entry]).await
        }));
    }

    for h in handles {
        let result = h.await.unwrap();
        assert!(result.is_ok(), "Concurrent push should succeed: {result:?}");
    }
}

#[tokio::test]
async fn concurrent_push_and_pull_interleaved() {
    let server = MockServer::start().await;
    let device_id = DeviceId::new(Uuid::new_v4());

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/push"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 10})))
        .mount(&server)
        .await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/pull"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "changes": [],
            "cursor": 10,
            "has_more": false,
        })))
        .mount(&server)
        .await;

    let client = Arc::new(authed_client(&server));
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let mut handles = Vec::new();
    for i in 0..4 {
        let c = Arc::clone(&client);
        let did = device_id;
        handles.push(tokio::spawn(async move {
            // Alternate push and pull
            if i % 2 == 0 {
                c.push(did, vec![]).await.map(|_| ())
            } else {
                c.pull(did, 0).await.map(|_| ())
            }
        }));
    }

    for h in handles {
        let result = h.await.unwrap();
        assert!(
            result.is_ok(),
            "Interleaved push/pull should succeed: {result:?}"
        );
    }
}

// ── Large payload handling ──

#[tokio::test]
async fn push_many_changes_succeeds() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/push"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1000})))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    // Create 1000+ change entries
    let changes: Vec<ChangeEntry> = (0..1100)
        .map(|i| ChangeEntry {
            table: "bulk_table".into(),
            op: ChangeOp::Insert,
            row_id: format!("row-{i}"),
            timestamp: Utc::now(),
            hlc: Hlc::zero(DeviceId::nil()),
            data: Some(json!({"index": i, "value": format!("data-{i}")})),
            extra: serde_json::Map::default(),
        })
        .collect();

    let cursor = client
        .push(DeviceId::new(Uuid::new_v4()), changes)
        .await
        .unwrap();
    assert_eq!(cursor, 1000);
}

// ── Session expiry handling ──

#[tokio::test]
async fn expired_token_detected_before_push() {
    let server = MockServer::start().await;
    let client = client_for(&server);
    let (user_id, app_id) = test_ids();

    let expired = fake_jwt(Utc::now().timestamp() - 100);
    client.restore_session(&expired, user_id, app_id);
    client.set_master_key_raw(synckit_client::crypto::generate_master_key());

    let err = client
        .push(DeviceId::new(Uuid::new_v4()), vec![])
        .await
        .unwrap_err();
    assert!(
        matches!(err, SyncKitError::TokenExpired),
        "Expired token should be detected pre-flight, got: {err:?}"
    );
}

#[tokio::test]
async fn expired_token_detected_before_pull() {
    let server = MockServer::start().await;
    let client = client_for(&server);
    let (user_id, app_id) = test_ids();

    let expired = fake_jwt(Utc::now().timestamp() - 100);
    client.restore_session(&expired, user_id, app_id);
    client.set_master_key_raw(synckit_client::crypto::generate_master_key());

    let err = client
        .pull(DeviceId::new(Uuid::new_v4()), 0)
        .await
        .unwrap_err();
    assert!(matches!(err, SyncKitError::TokenExpired));
}

#[tokio::test]
async fn expired_token_detected_before_register_device() {
    let server = MockServer::start().await;
    let client = client_for(&server);
    let (user_id, app_id) = test_ids();

    let expired = fake_jwt(Utc::now().timestamp() - 100);
    client.restore_session(&expired, user_id, app_id);

    let err = client.register_device("Test", "test").await.unwrap_err();
    assert!(matches!(err, SyncKitError::TokenExpired));
}

#[tokio::test]
async fn expired_token_detected_before_list_devices() {
    let server = MockServer::start().await;
    let client = client_for(&server);
    let (user_id, app_id) = test_ids();

    let expired = fake_jwt(Utc::now().timestamp() - 100);
    client.restore_session(&expired, user_id, app_id);

    let err = client.list_devices().await.unwrap_err();
    assert!(matches!(err, SyncKitError::TokenExpired));
}

// ── Blob edge cases ──

#[tokio::test]
async fn blob_upload_zero_byte_data() {
    let server = MockServer::start().await;

    let upload_path = "/s3/zero-byte";
    Mock::given(method("PUT"))
        .and(path(upload_path))
        .respond_with(ResponseTemplate::new(200))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let presigned = format!("{}{}", server.uri(), upload_path);
    let result = client.blob_upload("sha256-empty", &presigned, vec![]).await;
    assert!(result.is_ok(), "Zero-byte blob upload should succeed");

    // Verify the uploaded data is the v3 chunked framing over an empty blob.
    let requests = server.received_requests().await.unwrap();
    let req = requests
        .iter()
        .find(|r| r.url.path() == upload_path)
        .unwrap();
    assert_eq!(
        req.body.len(),
        synckit_client::crypto::chunked_blob_overhead(0),
        "Empty plaintext should produce exactly the chunked overhead bytes"
    );
}

#[tokio::test]
async fn blob_upload_download_roundtrip() {
    let server = MockServer::start().await;

    let client = authed_client(&server);
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let plaintext = b"roundtrip blob data with special bytes \x00\xFF\x01";
    let hash = hex::encode(sha2::Sha256::digest(plaintext));

    // Upload
    let upload_path = "/s3/roundtrip-upload";
    Mock::given(method("PUT"))
        .and(path(upload_path))
        .respond_with(ResponseTemplate::new(200))
        .mount(&server)
        .await;

    client
        .blob_upload(
            &hash,
            &format!("{}{}", server.uri(), upload_path),
            plaintext.to_vec(),
        )
        .await
        .unwrap();

    // Capture what was uploaded
    let requests = server.received_requests().await.unwrap();
    let uploaded_body = &requests
        .iter()
        .find(|r| r.url.path() == upload_path)
        .unwrap()
        .body;

    // Serve that exact encrypted data back for download
    let download_path = "/s3/roundtrip-download";
    Mock::given(method("GET"))
        .and(path(download_path))
        .respond_with(ResponseTemplate::new(200).set_body_bytes(uploaded_body.clone()))
        .mount(&server)
        .await;

    let downloaded = client
        .blob_download(&hash, &format!("{}{}", server.uri(), download_path))
        .await
        .unwrap();

    assert_eq!(downloaded, plaintext, "Blob roundtrip must preserve data");
}

// ── Push without master key ──

#[tokio::test]
async fn push_with_data_fails_without_master_key() {
    let server = MockServer::start().await;
    let client = authed_client(&server);
    // No master key set

    let changes = vec![ChangeEntry {
        table: "tasks".into(),
        op: ChangeOp::Insert,
        row_id: "r1".into(),
        timestamp: Utc::now(),
        hlc: Hlc::zero(DeviceId::nil()),
        data: Some(json!({"title": "test"})),
        extra: serde_json::Map::default(),
    }];

    let err = client
        .push(DeviceId::new(Uuid::new_v4()), changes)
        .await
        .unwrap_err();
    assert!(
        matches!(err, SyncKitError::NoMasterKey),
        "Push with data should fail without master key: {err:?}"
    );
}

#[tokio::test]
async fn push_delete_requires_master_key() {
    // Deletes used to push without a key (no payload to encrypt). With HLC, a
    // Delete now seals its clock into an encrypted envelope, so the master key is
    // required for every push, Deletes included.
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/push"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1})))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    // No master key loaded.

    let changes = vec![ChangeEntry {
        table: "tasks".into(),
        op: ChangeOp::Delete,
        row_id: "r1".into(),
        timestamp: Utc::now(),
        hlc: Hlc::zero(DeviceId::nil()),
        data: None,
        extra: serde_json::Map::default(),
    }];

    let err = client
        .push(DeviceId::new(Uuid::new_v4()), changes)
        .await
        .unwrap_err();
    assert!(
        matches!(err, SyncKitError::NoMasterKey),
        "Delete now seals an HLC envelope and needs the key: {err:?}"
    );
}

// ── Blob operations require auth ──

#[tokio::test]
async fn blob_upload_url_without_auth_fails() {
    let server = MockServer::start().await;
    let client = client_for(&server);

    let result = client.blob_upload_url("hash", 100).await;
    match result {
        Err(SyncKitError::NotAuthenticated) => {} // expected
        Err(other) => panic!("Expected NotAuthenticated, got: {other:?}"),
        Ok(_) => panic!("Expected NotAuthenticated error, got Ok"),
    }
}

#[tokio::test]
async fn blob_confirm_without_auth_fails() {
    let server = MockServer::start().await;
    let client = client_for(&server);

    let err = client.blob_confirm("hash", 100).await.unwrap_err();
    assert!(matches!(err, SyncKitError::NotAuthenticated));
}

#[tokio::test]
async fn blob_download_url_without_auth_fails() {
    let server = MockServer::start().await;
    let client = client_for(&server);

    let err = client.blob_download_url("hash").await.unwrap_err();
    assert!(matches!(err, SyncKitError::NotAuthenticated));
}

// ── Double-push same data ──

#[tokio::test]
async fn double_push_same_data_both_succeed() {
    let server = MockServer::start().await;

    // Server returns incrementing cursors
    Mock::given(method("POST"))
        .and(path("/api/v1/sync/push"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1})))
        .up_to_n_times(1)
        .mount(&server)
        .await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/push"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 2})))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let entry = ChangeEntry {
        table: "tasks".into(),
        op: ChangeOp::Insert,
        row_id: "same-row".into(),
        timestamp: Utc::now(),
        hlc: Hlc::zero(DeviceId::nil()),
        data: Some(json!({"title": "duplicate push test"})),
        extra: serde_json::Map::default(),
    };

    let cursor1 = client
        .push(DeviceId::new(Uuid::new_v4()), vec![entry.clone()])
        .await
        .unwrap();
    let cursor2 = client
        .push(DeviceId::new(Uuid::new_v4()), vec![entry])
        .await
        .unwrap();

    assert_eq!(cursor1, 1);
    assert_eq!(cursor2, 2);
}

// ── Encryption setup ──

#[tokio::test]
async fn setup_encryption_new_stores_key_and_uploads_envelope() {
    let server = MockServer::start().await;

    Mock::given(method("PUT"))
        .and(path("/api/v1/sync/keys"))
        .respond_with(ResponseTemplate::new(200))
        .expect(1)
        .mount(&server)
        .await;

    let client = authed_client(&server);
    assert!(!client.has_master_key());

    client.setup_encryption_new("test-password").await.unwrap();

    // Master key should now be in memory
    assert!(client.has_master_key());

    // Verify the PUT body contains a valid envelope unwrappable with the same password
    let requests = server.received_requests().await.unwrap();
    let put_req = requests
        .iter()
        .find(|r| r.method.as_str() == "PUT" && r.url.path() == "/api/v1/sync/keys")
        .unwrap();
    let body: serde_json::Value = serde_json::from_slice(&put_req.body).unwrap();
    let envelope_str = body["encrypted_key"].as_str().unwrap();
    let recovered =
        synckit_client::crypto::unwrap_master_key(envelope_str, "test-password").unwrap();
    assert_eq!(recovered.len(), 32);
}

#[tokio::test]
async fn setup_encryption_new_without_auth_fails() {
    let server = MockServer::start().await;
    let client = client_for(&server);

    let err = client.setup_encryption_new("password").await.unwrap_err();
    assert!(matches!(err, SyncKitError::NotAuthenticated));
}

#[tokio::test]
async fn setup_encryption_new_retries_on_server_error() {
    let server = MockServer::start().await;

    Mock::given(method("PUT"))
        .and(path("/api/v1/sync/keys"))
        .respond_with(ResponseTemplate::new(500).set_body_string("Internal Server Error"))
        .up_to_n_times(1)
        .mount(&server)
        .await;

    Mock::given(method("PUT"))
        .and(path("/api/v1/sync/keys"))
        .respond_with(ResponseTemplate::new(200))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let result = client.setup_encryption_new("password").await;
    assert!(result.is_ok(), "Should succeed after retry: {result:?}");
    assert!(client.has_master_key());
}

#[tokio::test]
async fn setup_encryption_existing_recovers_key() {
    let server = MockServer::start().await;

    let master_key = synckit_client::crypto::generate_master_key();
    let envelope = synckit_client::crypto::wrap_master_key(&master_key, "my-password").unwrap();

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/keys"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    assert!(!client.has_master_key());

    client
        .setup_encryption_existing("my-password")
        .await
        .unwrap();

    assert!(client.has_master_key());
}

#[tokio::test]
async fn setup_encryption_existing_wrong_password_fails() {
    let server = MockServer::start().await;

    let master_key = synckit_client::crypto::generate_master_key();
    let envelope =
        synckit_client::crypto::wrap_master_key(&master_key, "correct-password").unwrap();

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/keys"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let err = client
        .setup_encryption_existing("wrong-password")
        .await
        .unwrap_err();
    assert!(
        matches!(err, SyncKitError::DecryptionFailed),
        "Wrong password should produce DecryptionFailed: {err:?}"
    );
    assert!(!client.has_master_key());
}

#[tokio::test]
async fn setup_encryption_existing_without_auth_fails() {
    let server = MockServer::start().await;
    let client = client_for(&server);

    let err = client
        .setup_encryption_existing("password")
        .await
        .unwrap_err();
    assert!(matches!(err, SyncKitError::NotAuthenticated));
}

#[tokio::test]
async fn setup_encryption_existing_retries_on_server_error() {
    let server = MockServer::start().await;

    let master_key = synckit_client::crypto::generate_master_key();
    let envelope = synckit_client::crypto::wrap_master_key(&master_key, "password").unwrap();

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/keys"))
        .respond_with(ResponseTemplate::new(502).set_body_string("Bad Gateway"))
        .up_to_n_times(1)
        .mount(&server)
        .await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/keys"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let result = client.setup_encryption_existing("password").await;
    assert!(result.is_ok(), "Should succeed after retry: {result:?}");
    assert!(client.has_master_key());
}

#[tokio::test]
async fn setup_encryption_existing_no_server_key_returns_error() {
    let server = MockServer::start().await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/keys"))
        .respond_with(ResponseTemplate::new(404).set_body_string("Not Found"))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let err = client
        .setup_encryption_existing("password")
        .await
        .unwrap_err();
    assert!(
        matches!(err, SyncKitError::Server { status: 404, .. }),
        "Missing server key should produce 404 error: {err:?}"
    );
}

/// Two-device roundtrip: device 1 generates key via setup_encryption_new,
/// device 2 recovers it via setup_encryption_existing. Data encrypted by
/// device 1 must be decryptable by device 2.
#[tokio::test]
async fn encryption_setup_cross_device_roundtrip() {
    let server = MockServer::start().await;

    // Device 1: setup_encryption_new
    Mock::given(method("PUT"))
        .and(path("/api/v1/sync/keys"))
        .respond_with(ResponseTemplate::new(200))
        .mount(&server)
        .await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/push"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1})))
        .mount(&server)
        .await;

    let client1 = authed_client(&server);
    client1
        .setup_encryption_new("shared-password")
        .await
        .unwrap();

    // Push encrypted data from device 1
    let device_id = DeviceId::new(Uuid::new_v4());
    let original_data = json!({"title": "cross-device test", "secret": true});
    client1
        .push(
            device_id,
            vec![ChangeEntry {
                table: "tasks".into(),
                op: ChangeOp::Insert,
                row_id: "cross-r1".into(),
                timestamp: Utc::now(),
                hlc: Hlc::zero(DeviceId::nil()),
                data: Some(original_data.clone()),
                extra: serde_json::Map::default(),
            }],
        )
        .await
        .unwrap();

    // Capture the envelope and encrypted data
    let requests = server.received_requests().await.unwrap();
    let put_req = requests
        .iter()
        .find(|r| r.method.as_str() == "PUT" && r.url.path() == "/api/v1/sync/keys")
        .unwrap();
    let put_body: serde_json::Value = serde_json::from_slice(&put_req.body).unwrap();
    let envelope = put_body["encrypted_key"].as_str().unwrap().to_string();

    let push_req = requests
        .iter()
        .find(|r| r.url.path() == "/api/v1/sync/push")
        .unwrap();
    let push_body: serde_json::Value = serde_json::from_slice(&push_req.body).unwrap();
    let encrypted_data = push_body["changes"][0]["data"].clone();

    // Device 2: setup_encryption_existing with same password
    server.reset().await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/keys"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
        .mount(&server)
        .await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/pull"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "changes": [{
                "seq": 1,
                "device_id": device_id,
                "table": "tasks",
                "op": "INSERT",
                "row_id": "cross-r1",
                "timestamp": "2025-06-01T12:00:00Z",
                "data": encrypted_data,
            }],
            "cursor": 1,
            "has_more": false,
        })))
        .mount(&server)
        .await;

    let client2 = authed_client(&server);
    client2
        .setup_encryption_existing("shared-password")
        .await
        .unwrap();

    // Pull and decrypt with device 2's recovered key
    let (changes, _, _) = client2.pull(device_id, 0).await.unwrap();
    assert_eq!(changes.len(), 1);
    assert_eq!(
        changes[0].data.as_ref().unwrap(),
        &original_data,
        "Data encrypted by device 1 must be decryptable by device 2"
    );
}

// ── Config persistence (SyncKitConfig serialization) ──

#[tokio::test]
async fn config_serialization_roundtrip() {
    let config = SyncKitConfig {
        server_url: "https://makenot.work".to_string(),
        api_key: "ak_test_12345".to_string(),
    };

    // SyncKitConfig derives Clone and Debug; verify round trip through Debug
    let debug = format!("{config:?}");
    assert!(debug.contains("makenot.work"));
    assert!(debug.contains("ak_test_12345"));

    let cloned = config.clone();
    assert_eq!(cloned.server_url, config.server_url);
    assert_eq!(cloned.api_key, config.api_key);
}

// ── API error mapping ──

#[tokio::test]
async fn all_4xx_error_codes_mapped() {
    let server = MockServer::start().await;

    for status_code in [400, 401, 403, 404, 409, 422] {
        // Reset mocks
        server.reset().await;

        Mock::given(method("GET"))
            .and(path("/api/v1/sync/status"))
            .respond_with(
                ResponseTemplate::new(status_code).set_body_string(format!("Error {status_code}")),
            )
            .mount(&server)
            .await;

        let client = authed_client(&server);
        let err = client.status().await.unwrap_err();

        match err {
            SyncKitError::Server {
                status, message, ..
            } => {
                assert_eq!(status, status_code);
                assert!(message.contains(&format!("Error {status_code}")));
            }
            other => panic!("Status {status_code} should map to Server error, got: {other:?}"),
        }
    }
}

#[tokio::test]
async fn all_5xx_error_codes_retried() {
    for status_code in [500, 502, 503, 504] {
        let server = MockServer::start().await;

        // First request fails with 5xx
        Mock::given(method("GET"))
            .and(path("/api/v1/sync/status"))
            .respond_with(ResponseTemplate::new(status_code).set_body_string("Server Error"))
            .up_to_n_times(1)
            .mount(&server)
            .await;

        // Second request succeeds
        Mock::given(method("GET"))
            .and(path("/api/v1/sync/status"))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_body_json(json!({"total_changes": 0, "latest_cursor": null})),
            )
            .mount(&server)
            .await;

        let client = authed_client(&server);
        let result = client.status().await;
        assert!(
            result.is_ok(),
            "Status {status_code} should be retried and succeed: {result:?}"
        );
    }
}

// ── Blob download with wrong key ──

#[tokio::test]
async fn blob_download_with_wrong_key_fails() {
    let server = MockServer::start().await;

    let key1 = synckit_client::crypto::generate_master_key();
    let key2 = synckit_client::crypto::generate_master_key();

    // Encrypt with key1
    let plaintext = b"encrypted with key1";
    let encrypted = synckit_client::crypto::encrypt_bytes(plaintext, &key1).unwrap();

    let download_path = "/s3/wrong-key";
    Mock::given(method("GET"))
        .and(path(download_path))
        .respond_with(ResponseTemplate::new(200).set_body_bytes(encrypted))
        .mount(&server)
        .await;

    // Client has key2 (wrong key)
    let client = authed_client(&server);
    client.set_master_key_raw(key2);

    let result = client
        .blob_download("sha256-x", &format!("{}{}", server.uri(), download_path))
        .await;

    assert!(
        result.is_err(),
        "Download with wrong key should fail: {result:?}"
    );
    assert!(matches!(
        result.unwrap_err(),
        SyncKitError::DecryptionFailed
    ));
}

// ── Pull without auth ──

#[tokio::test]
async fn pull_without_auth_returns_not_authenticated() {
    let server = MockServer::start().await;
    let client = client_for(&server);

    let err = client
        .pull(DeviceId::new(Uuid::new_v4()), 0)
        .await
        .unwrap_err();
    assert!(matches!(err, SyncKitError::NotAuthenticated));
}

// ── List devices without auth ──

#[tokio::test]
async fn list_devices_without_auth_returns_not_authenticated() {
    let server = MockServer::start().await;
    let client = client_for(&server);

    let err = client.list_devices().await.unwrap_err();
    assert!(matches!(err, SyncKitError::NotAuthenticated));
}

// ── has_server_key without auth ──

#[tokio::test]
async fn has_server_key_without_auth_returns_not_authenticated() {
    let server = MockServer::start().await;
    let client = client_for(&server);

    let err = client.has_server_key().await.unwrap_err();
    assert!(matches!(err, SyncKitError::NotAuthenticated));
}

// ── Encryption roundtrip through push/pull (end-to-end) ──

#[tokio::test]
async fn end_to_end_push_pull_encryption_roundtrip() {
    let server = MockServer::start().await;

    let client = authed_client(&server);
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let device_id = DeviceId::new(Uuid::new_v4());
    let original_data = json!({
        "title": "End-to-end test",
        "tags": ["e2e", "encryption"],
        "nested": {"key": "value"},
        "count": 42
    });

    // Capture push request to feed back through pull
    Mock::given(method("POST"))
        .and(path("/api/v1/sync/push"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1})))
        .mount(&server)
        .await;

    client
        .push(
            device_id,
            vec![ChangeEntry {
                table: "tasks".into(),
                op: ChangeOp::Insert,
                row_id: "e2e-row".into(),
                timestamp: Utc::now(),
                hlc: Hlc::zero(DeviceId::nil()),
                data: Some(original_data.clone()),
                extra: serde_json::Map::default(),
            }],
        )
        .await
        .unwrap();

    // Extract the encrypted data that was sent to the server
    let requests = server.received_requests().await.unwrap();
    let push_body: serde_json::Value = serde_json::from_slice(
        &requests
            .iter()
            .find(|r| r.url.path() == "/api/v1/sync/push")
            .unwrap()
            .body,
    )
    .unwrap();

    let wire_entry = &push_body["changes"][0];

    // Feed encrypted data back through pull
    Mock::given(method("POST"))
        .and(path("/api/v1/sync/pull"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "changes": [{
                "seq": 1,
                "device_id": device_id,
                "table": wire_entry["table"],
                "op": wire_entry["op"],
                "row_id": wire_entry["row_id"],
                "timestamp": wire_entry["timestamp"],
                "data": wire_entry["data"],
            }],
            "cursor": 1,
            "has_more": false,
        })))
        .mount(&server)
        .await;

    let (changes, _, _) = client.pull(device_id, 0).await.unwrap();
    assert_eq!(changes.len(), 1);
    assert_eq!(
        changes[0].data.as_ref().unwrap(),
        &original_data,
        "Data must survive push encryption + pull decryption"
    );
}

// ── Retry count verification ──

#[tokio::test]
async fn retry_exhausts_all_attempts_on_persistent_503() {
    let server = MockServer::start().await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/status"))
        .respond_with(ResponseTemplate::new(503).set_body_string("Service Unavailable"))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let err = client.status().await.unwrap_err();
    assert!(matches!(err, SyncKitError::Server { status: 503, .. }));

    // Should have made exactly 4 requests (1 initial + 3 retries)
    let requests = server.received_requests().await.unwrap();
    let status_requests: Vec<_> = requests
        .iter()
        .filter(|r| r.url.path() == "/api/v1/sync/status")
        .collect();
    assert_eq!(
        status_requests.len(),
        4,
        "Expected 4 total requests (1 + MAX_RETRIES=3), got {}",
        status_requests.len()
    );
}

#[tokio::test]
async fn unsafe_op_makes_exactly_one_attempt_on_transient_error() {
    // create_subscription_checkout is Idempotency::Unsafe (a retry could mint a
    // second Stripe session), so a transient 503 must NOT be retried.
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/subscription/checkout"))
        .respond_with(ResponseTemplate::new(503).set_body_string("Service Unavailable"))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let err = client
        .create_subscription_checkout(1_000_000_000, synckit_client::BillingInterval::Monthly)
        .await
        .unwrap_err();
    assert!(matches!(err, SyncKitError::Server { status: 503, .. }));

    let requests = server.received_requests().await.unwrap();
    let checkout_requests = requests
        .iter()
        .filter(|r| r.url.path() == "/api/v1/sync/subscription/checkout")
        .count();
    assert_eq!(
        checkout_requests, 1,
        "Unsafe op must be attempted exactly once, got {checkout_requests}"
    );
}

#[tokio::test]
async fn retry_not_attempted_on_404() {
    let server = MockServer::start().await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/status"))
        .respond_with(ResponseTemplate::new(404).set_body_string("Not Found"))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let err = client.status().await.unwrap_err();
    assert!(matches!(err, SyncKitError::Server { status: 404, .. }));

    let requests = server.received_requests().await.unwrap();
    let count = requests
        .iter()
        .filter(|r| r.url.path() == "/api/v1/sync/status")
        .count();
    assert_eq!(count, 1, "404 should not be retried");
}

#[tokio::test]
async fn retry_succeeds_on_third_attempt() {
    let server = MockServer::start().await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/status"))
        .respond_with(ResponseTemplate::new(503).set_body_string("Service Unavailable"))
        .up_to_n_times(2)
        .mount(&server)
        .await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/status"))
        .respond_with(
            ResponseTemplate::new(200)
                .set_body_json(json!({"total_changes": 7, "latest_cursor": 3})),
        )
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let status = client.status().await.unwrap();
    assert_eq!(status.total_changes, 7);

    let requests = server.received_requests().await.unwrap();
    let count = requests
        .iter()
        .filter(|r| r.url.path() == "/api/v1/sync/status")
        .count();
    assert_eq!(count, 3, "Should succeed on 3rd attempt");
}

// ── Malformed / unexpected responses ──

#[tokio::test]
async fn authenticate_html_response_returns_error() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/auth"))
        .respond_with(
            ResponseTemplate::new(200)
                .insert_header("content-type", "text/html")
                .set_body_string("<html><body>Not JSON</body></html>"),
        )
        .mount(&server)
        .await;

    let client = client_for(&server);
    let err = client
        .authenticate("user@test.com", "pass", "test-key")
        .await
        .unwrap_err();
    // reqwest .json() fails when body isn't valid JSON
    assert!(
        matches!(err, SyncKitError::Http(_) | SyncKitError::Json(_)),
        "HTML response should produce Http or Json error, got: {err:?}"
    );
}

#[tokio::test]
async fn push_empty_response_body_returns_error() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/push"))
        .respond_with(ResponseTemplate::new(200).set_body_string(""))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let err = client
        .push(DeviceId::new(Uuid::new_v4()), vec![])
        .await
        .unwrap_err();
    assert!(
        matches!(err, SyncKitError::Http(_) | SyncKitError::Json(_)),
        "Empty body should produce parse error, got: {err:?}"
    );
}

#[tokio::test]
async fn pull_response_missing_has_more_returns_error() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/pull"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "changes": [],
            "cursor": 0
            // missing "has_more"
        })))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let err = client
        .pull(DeviceId::new(Uuid::new_v4()), 0)
        .await
        .unwrap_err();
    assert!(
        matches!(err, SyncKitError::Http(_) | SyncKitError::Json(_)),
        "Missing has_more should produce parse error, got: {err:?}"
    );
}

#[tokio::test]
async fn status_response_cursor_wrong_type_returns_error() {
    let server = MockServer::start().await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/status"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "total_changes": 10,
            "latest_cursor": "not-a-number"
        })))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let err = client.status().await.unwrap_err();
    assert!(
        matches!(err, SyncKitError::Http(_) | SyncKitError::Json(_)),
        "Wrong type for cursor should produce parse error, got: {err:?}"
    );
}

#[tokio::test]
async fn authenticate_response_missing_app_id_returns_error() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/auth"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "token": fresh_token(),
            "user_id": "550e8400-e29b-41d4-a716-446655440000"
            // missing "app_id"
        })))
        .mount(&server)
        .await;

    let client = client_for(&server);
    let err = client
        .authenticate("user@test.com", "pass", "test-key")
        .await
        .unwrap_err();
    assert!(
        matches!(err, SyncKitError::Http(_) | SyncKitError::Json(_)),
        "Missing app_id should produce parse error, got: {err:?}"
    );
}

#[tokio::test]
async fn register_device_extra_fields_ignored() {
    let server = MockServer::start().await;

    let (user_id, app_id) = test_ids();
    let response_with_extras = json!({
        "id": Uuid::new_v4(),
        "app_id": app_id,
        "user_id": user_id,
        "device_name": "Test Device",
        "platform": "test",
        "last_seen_at": "2025-01-01T00:00:00Z",
        "created_at": "2025-01-01T00:00:00Z",
        "extra_field": "should be ignored",
        "unknown_number": 42
    });

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/devices"))
        .respond_with(ResponseTemplate::new(200).set_body_json(response_with_extras))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let device = client.register_device("Test", "test").await.unwrap();
    assert_eq!(device.device_name, "Test Device");
}

#[tokio::test]
async fn blob_upload_url_response_missing_already_exists_returns_error() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/blobs/upload"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "upload_url": "https://s3.example.com/put"
            // missing "already_exists"
        })))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let result = client.blob_upload_url("hash", 100).await;
    match result {
        Err(err) => assert!(
            matches!(err, SyncKitError::Http(_) | SyncKitError::Json(_)),
            "Missing already_exists should produce error, got: {err:?}"
        ),
        Ok(_) => panic!("Expected error for missing already_exists field"),
    }
}

#[tokio::test]
async fn server_returns_413_request_entity_too_large() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/push"))
        .respond_with(ResponseTemplate::new(413).set_body_string("Request entity too large"))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let err = client
        .push(DeviceId::new(Uuid::new_v4()), vec![])
        .await
        .unwrap_err();
    match err {
        SyncKitError::Server {
            status, message, ..
        } => {
            assert_eq!(status, 413);
            assert!(message.contains("too large"));
        }
        other => panic!("Expected Server error, got: {other:?}"),
    }
}

// ── Session edge cases ──

#[tokio::test]
async fn double_authenticate_overwrites_session() {
    let server = MockServer::start().await;

    let (user_id, app_id) = test_ids();
    let second_user_id = UserId::new(Uuid::new_v4());

    // First auth response
    Mock::given(method("POST"))
        .and(path("/api/v1/sync/auth"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "token": fresh_token(),
            "user_id": user_id,
            "app_id": app_id,
        })))
        .up_to_n_times(1)
        .mount(&server)
        .await;

    // Second auth response with different user_id
    Mock::given(method("POST"))
        .and(path("/api/v1/sync/auth"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "token": fresh_token(),
            "user_id": second_user_id,
            "app_id": app_id,
        })))
        .mount(&server)
        .await;

    let client = client_for(&server);
    let (uid1, _) = client
        .authenticate("user1@test.com", "pass1", "test-key")
        .await
        .unwrap();
    assert_eq!(uid1, user_id);

    let (uid2, _) = client
        .authenticate("user2@test.com", "pass2", "test-key")
        .await
        .unwrap();
    assert_eq!(uid2, second_user_id);

    // Session should now reflect the second auth
    let info = client.session_info().unwrap();
    assert_eq!(info.user_id, second_user_id);
}

#[tokio::test]
async fn clear_session_then_authenticate_succeeds() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/auth"))
        .respond_with(ResponseTemplate::new(200).set_body_json(auth_response_json()))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    assert!(client.session_info().is_some());

    client.clear_session();
    assert!(client.session_info().is_none());

    // Re-authenticate should work
    let result = client
        .authenticate("user@test.com", "pass", "test-key")
        .await;
    assert!(
        result.is_ok(),
        "Should be able to re-authenticate after clear: {result:?}"
    );
    assert!(client.session_info().is_some());
}

#[tokio::test]
async fn restore_session_with_expired_token_then_push_returns_token_expired() {
    let server = MockServer::start().await;
    let client = client_for(&server);
    let (user_id, app_id) = test_ids();

    let expired = fake_jwt(Utc::now().timestamp() - 3600);
    client.restore_session(&expired, user_id, app_id);
    client.set_master_key_raw(synckit_client::crypto::generate_master_key());

    let err = client
        .push(DeviceId::new(Uuid::new_v4()), vec![])
        .await
        .unwrap_err();
    assert!(
        matches!(err, SyncKitError::TokenExpired),
        "Restored expired token should return TokenExpired, got: {err:?}"
    );
}

// ── Encryption state edge cases ──

#[tokio::test]
async fn setup_encryption_new_twice_overwrites() {
    let server = MockServer::start().await;

    Mock::given(method("PUT"))
        .and(path("/api/v1/sync/keys"))
        .respond_with(ResponseTemplate::new(200))
        .mount(&server)
        .await;

    let client = authed_client(&server);

    client.setup_encryption_new("pass1").await.unwrap();
    assert!(client.has_master_key());

    // Second call overwrites
    client.setup_encryption_new("pass2").await.unwrap();
    assert!(client.has_master_key());

    // Verify the second PUT used a different envelope
    let requests = server.received_requests().await.unwrap();
    let puts: Vec<_> = requests
        .iter()
        .filter(|r| r.method.as_str() == "PUT" && r.url.path() == "/api/v1/sync/keys")
        .collect();
    assert_eq!(puts.len(), 2);
    // Different envelopes (different random keys)
    assert_ne!(puts[0].body, puts[1].body);
}

// ── Blob edge cases ──

#[tokio::test]
async fn blob_confirm_retries_on_503() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/blobs/confirm"))
        .respond_with(ResponseTemplate::new(503).set_body_string("Service Unavailable"))
        .up_to_n_times(1)
        .mount(&server)
        .await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/blobs/confirm"))
        .respond_with(ResponseTemplate::new(200))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let result = client.blob_confirm("sha256-retry", 512).await;
    assert!(result.is_ok(), "Should succeed after retry: {result:?}");
}

#[tokio::test]
async fn blob_download_retries_on_503() {
    let server = MockServer::start().await;

    let client = authed_client(&server);
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let plaintext = b"retry download test";
    let hash = hex::encode(sha2::Sha256::digest(plaintext));
    let encrypted = synckit_client::crypto::encrypt_bytes(plaintext, &key).unwrap();

    let download_path = "/s3/retry-download";
    Mock::given(method("GET"))
        .and(path(download_path))
        .respond_with(ResponseTemplate::new(503))
        .up_to_n_times(1)
        .mount(&server)
        .await;

    Mock::given(method("GET"))
        .and(path(download_path))
        .respond_with(ResponseTemplate::new(200).set_body_bytes(encrypted))
        .mount(&server)
        .await;

    let presigned = format!("{}{}", server.uri(), download_path);
    let result = client.blob_download(&hash, &presigned).await.unwrap();
    assert_eq!(result, plaintext);
}

#[tokio::test]
async fn blob_upload_1mb_with_correct_overhead() {
    let server = MockServer::start().await;

    let upload_path = "/s3/1mb-upload";
    Mock::given(method("PUT"))
        .and(path(upload_path))
        .respond_with(ResponseTemplate::new(200))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let plaintext: Vec<u8> = (0..1_048_576u32).map(|i| (i % 256) as u8).collect();
    let presigned = format!("{}{}", server.uri(), upload_path);
    client
        .blob_upload("sha256-1mb", &presigned, plaintext.clone())
        .await
        .unwrap();

    let requests = server.received_requests().await.unwrap();
    let upload_req = requests
        .iter()
        .find(|r| r.url.path() == upload_path)
        .unwrap();
    assert_eq!(
        upload_req.body.len(),
        plaintext.len() + synckit_client::crypto::chunked_blob_overhead(plaintext.len()),
        "1MB upload should add exactly the v3 chunked overhead"
    );
}

// ── Device management edge cases ──

#[tokio::test]
async fn register_device_with_empty_name() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/devices"))
        .respond_with(ResponseTemplate::new(200).set_body_json(device_json()))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    // Empty name should not panic; server may accept or reject
    let result = client.register_device("", "macos").await;
    assert!(result.is_ok(), "Empty device name should not panic");
}

#[tokio::test]
async fn register_device_with_unicode_name() {
    let server = MockServer::start().await;

    let (user_id, app_id) = test_ids();
    Mock::given(method("POST"))
        .and(path("/api/v1/sync/devices"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "id": Uuid::new_v4(),
            "app_id": app_id,
            "user_id": user_id,
            "device_name": "\u{30DE}\u{30C3}\u{30AF}\u{30D6}\u{30C3}\u{30AF}",
            "platform": "macos",
            "last_seen_at": "2025-01-01T00:00:00Z",
            "created_at": "2025-01-01T00:00:00Z",
        })))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let device = client
        .register_device("\u{30DE}\u{30C3}\u{30AF}\u{30D6}\u{30C3}\u{30AF}", "macos")
        .await
        .unwrap();
    assert_eq!(
        device.device_name,
        "\u{30DE}\u{30C3}\u{30AF}\u{30D6}\u{30C3}\u{30AF}"
    );
}

#[tokio::test]
async fn list_devices_empty_array() {
    let server = MockServer::start().await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/devices"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!([])))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let devices = client.list_devices().await.unwrap();
    assert!(devices.is_empty());
}

// ── Concurrency stress tests ──

#[tokio::test]
async fn concurrent_session_info_reads() {
    let server = MockServer::start().await;
    let client = Arc::new(authed_client(&server));

    let mut handles = Vec::new();
    for _ in 0..50 {
        let c = Arc::clone(&client);
        handles.push(tokio::spawn(async move { c.session_info() }));
    }

    for h in handles {
        let info = h.await.unwrap();
        assert!(
            info.is_some(),
            "All concurrent reads should see the session"
        );
    }
}

#[tokio::test]
async fn concurrent_has_master_key_reads() {
    let server = MockServer::start().await;
    let client = Arc::new(authed_client(&server));
    client.set_master_key_raw(synckit_client::crypto::generate_master_key());

    let mut handles = Vec::new();
    for _ in 0..50 {
        let c = Arc::clone(&client);
        handles.push(tokio::spawn(async move { c.has_master_key() }));
    }

    for h in handles {
        let has_key = h.await.unwrap();
        assert!(has_key, "All concurrent reads should see the master key");
    }
}

#[tokio::test]
async fn concurrent_status_checks() {
    let server = MockServer::start().await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/status"))
        .respond_with(
            ResponseTemplate::new(200)
                .set_body_json(json!({"total_changes": 5, "latest_cursor": 3})),
        )
        .mount(&server)
        .await;

    let client = Arc::new(authed_client(&server));

    let mut handles = Vec::new();
    for _ in 0..20 {
        let c = Arc::clone(&client);
        handles.push(tokio::spawn(async move { c.status().await }));
    }

    for h in handles {
        let result = h.await.unwrap();
        assert!(
            result.is_ok(),
            "All concurrent status checks should succeed: {result:?}"
        );
        assert_eq!(result.unwrap().total_changes, 5);
    }
}

#[tokio::test]
async fn concurrent_push_100_entries_each() {
    let server = MockServer::start().await;

    Mock::given(method("POST"))
        .and(path("/api/v1/sync/push"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1})))
        .mount(&server)
        .await;

    let client = Arc::new(authed_client(&server));
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let mut handles = Vec::new();
    for batch in 0..4 {
        let c = Arc::clone(&client);
        handles.push(tokio::spawn(async move {
            let changes: Vec<ChangeEntry> = (0..100)
                .map(|i| ChangeEntry {
                    table: format!("batch_{batch}"),
                    op: ChangeOp::Insert,
                    row_id: format!("row_{i}"),
                    timestamp: Utc::now(),
                    hlc: Hlc::zero(DeviceId::nil()),
                    data: Some(json!({"index": i})),
                    extra: serde_json::Map::default(),
                })
                .collect();
            c.push(DeviceId::new(Uuid::new_v4()), changes).await
        }));
    }

    for h in handles {
        let result = h.await.unwrap();
        assert!(
            result.is_ok(),
            "Concurrent 100-entry push should succeed: {result:?}"
        );
    }
}

// ── Timeout tests ──

fn short_timeout_client(server: &MockServer) -> SyncKitClient {
    ensure_crypto_provider();
    let http = reqwest::Client::builder()
        .timeout(Duration::from_millis(100))
        .connect_timeout(Duration::from_millis(100))
        .build()
        .unwrap();
    SyncKitClient::with_http_client(
        SyncKitConfig {
            server_url: server.uri(),
            api_key: "test-api-key".to_string(),
        },
        http,
    )
}

fn authed_short_timeout_client(server: &MockServer) -> SyncKitClient {
    let client = short_timeout_client(server);
    let (user_id, app_id) = test_ids();
    client.restore_session(&fresh_token(), user_id, app_id);
    client
}

#[tokio::test]
async fn status_times_out_on_slow_server() {
    let server = MockServer::start().await;

    Mock::given(method("GET"))
        .and(path("/api/v1/sync/status"))
        .respond_with(
            ResponseTemplate::new(200)
                .set_body_json(json!({"total_changes": 0, "latest_cursor": null}))
                .set_delay(Duration::from_secs(5)),
        )
        .mount(&server)
        .await;

    let client = authed_short_timeout_client(&server);
    let err = client.status().await.unwrap_err();
    // Timeout triggers Http error, which is transient, so it retries and eventually exhausts
    assert!(
        matches!(err, SyncKitError::Http(_)),
        "Slow server should produce Http (timeout) error, got: {err:?}"
    );
}

#[tokio::test]
async fn push_retries_on_timeout_then_succeeds() {
    let server = MockServer::start().await;

    // First request: slow (will timeout)
    Mock::given(method("POST"))
        .and(path("/api/v1/sync/push"))
        .respond_with(
            ResponseTemplate::new(200)
                .set_body_json(json!({"cursor": 1}))
                .set_delay(Duration::from_secs(5)),
        )
        .up_to_n_times(1)
        .mount(&server)
        .await;

    // Second request: fast (succeeds)
    Mock::given(method("POST"))
        .and(path("/api/v1/sync/push"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 42})))
        .mount(&server)
        .await;

    let client = authed_short_timeout_client(&server);
    let key = synckit_client::crypto::generate_master_key();
    client.set_master_key_raw(key);

    let cursor = client
        .push(DeviceId::new(Uuid::new_v4()), vec![])
        .await
        .unwrap();
    assert_eq!(cursor, 42);
}

// ── SSE subscribe / reconnect state machine ──
//
// These drive the real `subscribe()` -> `SyncNotifyStream::next_change()` path
// against wiremock so the reconnect state machine (subscribe.rs) is exercised
// end to end: a live notification, a transparent reconnect across a stream
// drop, the fatal auth-rejection exit, and the give-up-after-N-failures cap.

const SUBSCRIBE_PATH: &str = "/api/v1/sync/subscribe";

/// One complete SSE "changed" block, body-terminated so the client sees a full
/// event and then end-of-stream.
fn sse_changed_block() -> &'static str {
    "event: changed\n\n"
}

#[tokio::test]
async fn subscribe_yields_changed_event() {
    let server = MockServer::start().await;
    Mock::given(method("GET"))
        .and(path(SUBSCRIBE_PATH))
        .respond_with(ResponseTemplate::new(200).set_body_string(sse_changed_block()))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let mut stream = client.subscribe().await.expect("subscribe should succeed");

    // The first block is delivered inside the initial response body, so this
    // returns without any reconnect.
    assert_eq!(stream.next_change().await, Some(()));
}

#[tokio::test]
async fn subscribe_reconnects_transparently_after_stream_drop() {
    let server = MockServer::start().await;
    // Every connection serves one block then ends (Content-Length terminates the
    // body). Consuming the first event, then reading past it, drops the stream
    // and forces a reconnect that must transparently yield the next event.
    Mock::given(method("GET"))
        .and(path(SUBSCRIBE_PATH))
        .respond_with(ResponseTemplate::new(200).set_body_string(sse_changed_block()))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let mut stream = client.subscribe().await.expect("subscribe should succeed");

    // #1 comes from the initial connection; #2 can only arrive after the stream
    // drops (body EOF) and reconnect() re-establishes it.
    assert_eq!(stream.next_change().await, Some(()));
    assert_eq!(stream.next_change().await, Some(()));

    // subscribe() opened one connection; the second event required at least one
    // reconnect, so the server saw two or more subscribe requests.
    let hits = server
        .received_requests()
        .await
        .unwrap()
        .into_iter()
        .filter(|r| r.url.path() == SUBSCRIBE_PATH)
        .count();
    assert!(
        hits >= 2,
        "expected a reconnect (>=2 subscribe requests), got {hits}"
    );
}

#[tokio::test]
async fn subscribe_stream_closes_on_auth_rejection() {
    let server = MockServer::start().await;
    // First connection opens cleanly but carries no event and ends immediately,
    // forcing a reconnect. The reconnect is rejected for auth -> fatal, so the
    // stream ends with `None` rather than retrying.
    Mock::given(method("GET"))
        .and(path(SUBSCRIBE_PATH))
        .respond_with(ResponseTemplate::new(200).set_body_string(""))
        .up_to_n_times(1)
        .mount(&server)
        .await;
    Mock::given(method("GET"))
        .and(path(SUBSCRIBE_PATH))
        .respond_with(ResponseTemplate::new(401).set_body_string("unauthorized"))
        .mount(&server)
        .await;

    let client = authed_client(&server);
    let mut stream = client.subscribe().await.expect("initial subscribe is 200");

    // Empty body -> EOF -> reconnect -> 401 -> fatal -> None.
    assert_eq!(stream.next_change().await, None);
}

// Note: the "give up after MAX_RECONNECT_ATTEMPTS consecutive failures" path is
// deliberately not covered end-to-end here. Exercising it against a live mock
// server would incur the real exponential backoff (growing to a 60s cap, minutes
// of wall-clock), and `tokio::time::pause()` cannot collapse it: the wiremock
// server runs on its own runtime, so a paused clock auto-advances past the
// cross-thread request to the nearest timer and the reconnect never completes.
// The two pieces of that path are covered separately: the backoff schedule by
// `reconnect_delay_grows_then_caps` (subscribe.rs), and the fatal-exit mechanics
// (returning `None` and clearing state) by the auth-rejection test above.

// ── End-to-end key-rotation orchestration ──
//
// These drive `rotate_key()` through the full server protocol against wiremock:
// fetch key -> begin -> re-encrypt loop -> complete, plus the straggler retry on
// a 409. They are gated `#[cfg(not(feature = "keychain"))]` because `rotate_key`
// finishes by caching the new key with `keystore::store_key`, which hits the OS
// secret-service under the `keychain` feature and is unavailable on a headless
// test host. With keychain off, `store_key` is the no-op stub, so the test
// exercises the orchestration hermetically. Run with:
//   cargo test --no-default-features --features store,testing
#[cfg(not(feature = "keychain"))]
mod rotation_e2e {
    use super::*;

    const KEYS_PATH: &str = "/api/v1/sync/keys";
    const ROTATE_PATH: &str = "/api/v1/sync/keys/rotate";
    const ENTRIES_PATH: &str = "/api/v1/sync/keys/rotate/entries";
    const BATCH_PATH: &str = "/api/v1/sync/keys/rotate/batch";
    const COMPLETE_PATH: &str = "/api/v1/sync/keys/rotate/complete";

    const ROTATE_PW: &str = "rotate-password";

    /// A `GET /keys` body wrapping `old_key` under [`ROTATE_PW`], with no rotation
    /// in progress, so `rotate_key` verifies the password and mints a fresh key.
    fn get_keys_body(old_key: &[u8; 32]) -> serde_json::Value {
        let envelope = synckit_client::crypto::wrap_master_key(old_key, ROTATE_PW).unwrap();
        json!({ "encrypted_key": envelope, "key_version": 1, "key_id": 1 })
    }

    /// One rotation entry: `plaintext` sealed under `old_key` with the same
    /// `(table, row_id)` AAD the client rebinds during re-encryption.
    fn rotation_entry(old_key: &[u8; 32], table: &str, row_id: &str) -> serde_json::Value {
        let ctx = synckit_client::crypto::AeadContext::entry(table, row_id);
        let sealed =
            synckit_client::crypto::encrypt_json_aad(&json!({"title": "secret"}), old_key, &ctx)
                .unwrap();
        json!({ "seq": 1, "table": table, "row_id": row_id, "data": sealed })
    }

    fn hits(reqs: &[wiremock::Request], p: &str) -> usize {
        reqs.iter().filter(|r| r.url.path() == p).count()
    }

    #[tokio::test]
    async fn rotate_key_drives_full_orchestration() {
        let server = MockServer::start().await;
        let old_key = synckit_client::crypto::generate_master_key();

        Mock::given(method("GET"))
            .and(path(KEYS_PATH))
            .respond_with(ResponseTemplate::new(200).set_body_json(get_keys_body(&old_key)))
            .mount(&server)
            .await;
        Mock::given(method("POST"))
            .and(path(ROTATE_PATH))
            .respond_with(ResponseTemplate::new(200).set_body_json(
                json!({ "rotation_id": Uuid::new_v4(), "target_seq": 1, "new_key_id": 2 }),
            ))
            .mount(&server)
            .await;
        // One batch of work, then drained (has_more = false ends the re-encrypt loop).
        Mock::given(method("POST"))
            .and(path(ENTRIES_PATH))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "entries": [rotation_entry(&old_key, "tasks", "r1")],
                "has_more": false
            })))
            .mount(&server)
            .await;
        Mock::given(method("POST"))
            .and(path(BATCH_PATH))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "updated_count": 1 })))
            .mount(&server)
            .await;
        Mock::given(method("POST"))
            .and(path(COMPLETE_PATH))
            .respond_with(ResponseTemplate::new(200))
            .mount(&server)
            .await;

        let client = authed_client(&server);
        client
            .rotate_key(DeviceId::new(Uuid::new_v4()), ROTATE_PW)
            .await
            .expect("full rotation should complete");

        // Every stage of the protocol was driven, in the right shape.
        let reqs = server.received_requests().await.unwrap();
        assert_eq!(hits(&reqs, KEYS_PATH), 1, "fetched key state once");
        assert_eq!(hits(&reqs, ROTATE_PATH), 1, "began rotation once");
        assert!(
            hits(&reqs, ENTRIES_PATH) >= 1,
            "pulled entries to re-encrypt"
        );
        assert_eq!(hits(&reqs, BATCH_PATH), 1, "pushed one re-encrypted batch");
        assert_eq!(
            hits(&reqs, COMPLETE_PATH),
            1,
            "completed once (no stragglers)"
        );
    }

    #[tokio::test]
    async fn rotate_key_retries_reencrypt_on_straggler_conflict() {
        let server = MockServer::start().await;
        let old_key = synckit_client::crypto::generate_master_key();

        Mock::given(method("GET"))
            .and(path(KEYS_PATH))
            .respond_with(ResponseTemplate::new(200).set_body_json(get_keys_body(&old_key)))
            .mount(&server)
            .await;
        Mock::given(method("POST"))
            .and(path(ROTATE_PATH))
            .respond_with(ResponseTemplate::new(200).set_body_json(
                json!({ "rotation_id": Uuid::new_v4(), "target_seq": 1, "new_key_id": 2 }),
            ))
            .mount(&server)
            .await;
        // First entries pull returns work; every later pull is drained. Mounted in
        // this order so the up_to_n_times(1) mock wins the first call, then the
        // empty-set fallback serves the straggler round's re-pull.
        Mock::given(method("POST"))
            .and(path(ENTRIES_PATH))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "entries": [rotation_entry(&old_key, "tasks", "r1")],
                "has_more": false
            })))
            .up_to_n_times(1)
            .mount(&server)
            .await;
        Mock::given(method("POST"))
            .and(path(ENTRIES_PATH))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_body_json(json!({ "entries": [], "has_more": false })),
            )
            .mount(&server)
            .await;
        Mock::given(method("POST"))
            .and(path(BATCH_PATH))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "updated_count": 1 })))
            .mount(&server)
            .await;
        // First completion reports a straggler (409); the retry then succeeds.
        Mock::given(method("POST"))
            .and(path(COMPLETE_PATH))
            .respond_with(
                ResponseTemplate::new(409).set_body_json(json!({ "message": "stragglers" })),
            )
            .up_to_n_times(1)
            .mount(&server)
            .await;
        Mock::given(method("POST"))
            .and(path(COMPLETE_PATH))
            .respond_with(ResponseTemplate::new(200))
            .mount(&server)
            .await;

        let client = authed_client(&server);
        client
            .rotate_key(DeviceId::new(Uuid::new_v4()), ROTATE_PW)
            .await
            .expect("rotation should converge after the straggler retry");

        // The 409 forced a second completion attempt, and the straggler round
        // re-ran the re-encrypt loop (a second entries pull).
        let reqs = server.received_requests().await.unwrap();
        assert_eq!(
            hits(&reqs, COMPLETE_PATH),
            2,
            "completed twice: 409 then 200"
        );
        assert!(
            hits(&reqs, ENTRIES_PATH) >= 2,
            "straggler round re-pulled entries"
        );
    }
}

// ── 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.
mod blob_multipart {
    use super::*;
    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(server: &MockServer, cipher_len: usize, part_size: usize) -> u32 {
        let part_count = cipher_len.div_ceil(part_size) as u32;

        Mock::given(method("POST"))
            .and(path(START_PATH))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "upload_id": "test-upload-id",
                "part_size": part_size,
                "part_count": part_count,
                "already_exists": false,
            })))
            .mount(server)
            .await;
        Mock::given(method("POST"))
            .and(path(PARTS_PATH))
            .respond_with(PartsResponder {
                cipher_len,
                part_size,
                base: server.uri(),
            })
            .mount(server)
            .await;
        Mock::given(method("PUT"))
            .and(path(PART_PUT_PATH))
            .respond_with(ResponseTemplate::new(200).append_header("ETag", "\"part-etag\""))
            .mount(server)
            .await;
        Mock::given(method("POST"))
            .and(path(COMPLETE_PATH))
            .respond_with(ResponseTemplate::new(204))
            .mount(server)
            .await;

        part_count
    }

    fn hits(reqs: &[wiremock::Request], p: &str) -> usize {
        reqs.iter().filter(|r| r.url.path() == p).count()
    }

    #[tokio::test]
    async fn streaming_upload_tiles_the_parts_into_a_valid_blob() {
        let server = MockServer::start().await;
        let client = authed_client(&server);
        let key = synckit_client::crypto::generate_master_key();
        client.set_master_key_raw(key);

        // 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(&server, cipher_len, part_size).await;
        assert!(part_count > 1, "the fixture must actually be multipart");

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

        let reqs = server.received_requests().await.unwrap();

        // The session was sized in ciphertext, predicted from the plaintext.
        let start: serde_json::Value = reqs
            .iter()
            .find(|r| r.url.path() == START_PATH)
            .unwrap()
            .body_json()
            .unwrap();
        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: Vec<&wiremock::Request> = reqs
            .iter()
            .filter(|r| r.url.path() == PART_PUT_PATH)
            .collect();
        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: Vec<&wiremock::Request> =
            reqs.iter().filter(|r| r.url.path() == PARTS_PATH).collect();
        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, req) in part_reqs.iter().enumerate() {
            let body: serde_json::Value = req.body_json().unwrap();
            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: serde_json::Value = reqs
            .iter()
            .find(|r| r.url.path() == COMPLETE_PATH)
            .unwrap()
            .body_json()
            .unwrap();
        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!(hits(&reqs, ABORT_PATH), 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 server = MockServer::start().await;
        let client = authed_client(&server);
        let key = synckit_client::crypto::generate_master_key();
        client.set_master_key_raw(key);

        // 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.
        Mock::given(method("POST"))
            .and(path(START_PATH))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "upload_id": "test-upload-id",
                "part_size": 1024 * 1024,
                "part_count": 1,
                "already_exists": false,
            })))
            .mount(&server)
            .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:?}"
        );
        let reqs = server.received_requests().await.unwrap();
        assert_eq!(
            hits(&reqs, PART_PUT_PATH),
            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 server = MockServer::start().await;
        let client = authed_client(&server);
        let key = synckit_client::crypto::generate_master_key();
        client.set_master_key_raw(key);

        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(&server, cipher_len, 1024 * 1024).await;

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

        let reqs = server.received_requests().await.unwrap();
        let put = reqs.iter().find(|r| r.url.path() == PART_PUT_PATH).unwrap();
        assert_eq!(
            put.body.len(),
            cipher_len,
            "one part carries the whole blob"
        );
        assert_eq!(
            synckit_client::crypto::decrypt_blob_chunked(&put.body, &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 server = MockServer::start().await;
        let client = authed_client(&server);
        client.set_master_key_raw(synckit_client::crypto::generate_master_key());

        Mock::given(method("POST"))
            .and(path(START_PATH))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "upload_id": "",
                "part_size": 0,
                "part_count": 0,
                "already_exists": true,
            })))
            .mount(&server)
            .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.
        let reqs = server.received_requests().await.unwrap();
        assert_eq!(hits(&reqs, PART_PUT_PATH), 0, "dedup must not upload parts");
        assert_eq!(hits(&reqs, COMPLETE_PATH), 0);
        assert_eq!(hits(&reqs, ABORT_PATH), 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 server = MockServer::start().await;
        let client = authed_client(&server);
        client.set_master_key_raw(synckit_client::crypto::generate_master_key());

        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(&server, cipher_len, 1024 * 1024).await;
        Mock::given(method("POST"))
            .and(path(ABORT_PATH))
            .respond_with(ResponseTemplate::new(204))
            .mount(&server)
            .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:?}"
        );

        let reqs = server.received_requests().await.unwrap();
        assert_eq!(
            hits(&reqs, COMPLETE_PATH),
            0,
            "a mismatched blob must not be assembled"
        );
        assert_eq!(hits(&reqs, ABORT_PATH), 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 server = MockServer::start().await;
        let client = authed_client(&server);
        client.set_master_key_raw(synckit_client::crypto::generate_master_key());

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

        Mock::given(method("POST"))
            .and(path(START_PATH))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "upload_id": "test-upload-id",
                "part_size": cipher_len,
                "part_count": 1,
                "already_exists": false,
            })))
            .mount(&server)
            .await;
        Mock::given(method("POST"))
            .and(path(PARTS_PATH))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "parts": [{
                    "part_number": 1,
                    "content_length": cipher_len,
                    "url": format!("{}{PART_PUT_PATH}", server.uri()),
                }]
            })))
            .mount(&server)
            .await;
        Mock::given(method("PUT"))
            .and(path(PART_PUT_PATH))
            .respond_with(ResponseTemplate::new(403))
            .mount(&server)
            .await;
        Mock::given(method("POST"))
            .and(path(ABORT_PATH))
            .respond_with(ResponseTemplate::new(204))
            .mount(&server)
            .await;

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

        let reqs = server.received_requests().await.unwrap();
        assert_eq!(hits(&reqs, COMPLETE_PATH), 0);
        assert_eq!(
            hits(&reqs, ABORT_PATH),
            1,
            "a failed transfer must release its parts"
        );

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