//! Tests for [`super`].

use super::*;
use crate::ids::DeviceId;
use crate::types::ChangeOp;
use base64::Engine;
use chrono::Utc;
use std::time::Duration;
use uuid::Uuid;

use super::super::TOKEN_EXPIRY_BUFFER_SECS;

fn test_config() -> super::super::SyncKitConfig {
    super::super::SyncKitConfig {
        server_url: "https://example.com".to_string(),
        api_key: "test-api-key-123".to_string(),
    }
}

/// Build a fake JWT with the given `exp` claim (no real signature).
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 = serde_json::json!({
        "sub": "550e8400-e29b-41d4-a716-446655440000",
        "app": "6ba7b810-9dad-11d1-80b4-00c04fd430c8",
        "exp": exp,
        "iat": exp - 3600,
    });
    let payload = base64::engine::general_purpose::URL_SAFE_NO_PAD
        .encode(payload_json.to_string().as_bytes());
    let signature = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(b"fake-signature");
    format!("{header}.{payload}.{signature}")
}

// ── wire-version envelope dispatch ──

#[test]
fn split_envelope_dispatches_on_explicit_version() {
    let node = DeviceId::new(Uuid::from_u128(1));
    let hlc = Hlc {
        wall_ms: 5,
        counter: 2,
        node,
    };

    // v2 envelope: explicit __skver, parsed by version.
    let v2 = serde_json::json!({ "__skver": 2, "__skhlc": hlc, "data": {"k": "v"} });
    let (got, data, _) = SyncKitClient::split_hlc_envelope(v2, node, 0).unwrap();
    assert_eq!(got, hlc);
    assert_eq!(data, Some(serde_json::json!({"k": "v"})));

    // gen-1 envelope: __skhlc present, no version tag.
    let gen1 = serde_json::json!({ "__skhlc": hlc, "data": null });
    let (got, data, _) = SyncKitClient::split_hlc_envelope(gen1, node, 0).unwrap();
    assert_eq!(got, hlc);
    assert_eq!(data, None);

    // Bare legacy row: HLC synthesized from node + timestamp.
    let bare = serde_json::json!({ "title": "buy milk" });
    let (got, data, _) = SyncKitClient::split_hlc_envelope(bare.clone(), node, 1234).unwrap();
    assert_eq!(got, Hlc::from_legacy(1234, node));
    assert_eq!(data, Some(bare));
}

#[test]
fn the_storage_stamp_rides_inside_the_sealed_envelope_and_survives_a_round_trip() {
    let node = DeviceId::new(Uuid::from_u128(1));
    let hlc = Hlc {
        wall_ms: 5,
        counter: 2,
        node,
    };

    let stamped = SyncKitClient::hlc_envelope(&hlc, Some(&serde_json::json!({"k": "v"})), Some(4));
    assert_eq!(stamped["__sksv"], serde_json::json!(4));
    let (got, data, version) = SyncKitClient::split_hlc_envelope(stamped, node, 0).unwrap();
    assert_eq!(got, hlc);
    assert_eq!(data, Some(serde_json::json!({"k": "v"})));
    assert_eq!(version, Some(4));

    // A Delete carries no row payload and still carries the stamp, which is
    // what lets the gate see a peer whose only pending change is a delete.
    let delete = SyncKitClient::hlc_envelope(&hlc, None, Some(4));
    let (_, data, version) = SyncKitClient::split_hlc_envelope(delete, node, 0).unwrap();
    assert_eq!(data, None);
    assert_eq!(version, Some(4));
}

#[test]
fn an_undeclared_version_leaves_the_sealed_bytes_exactly_as_they_were() {
    let node = DeviceId::new(Uuid::from_u128(1));
    let hlc = Hlc {
        wall_ms: 5,
        counter: 2,
        node,
    };
    let plain = SyncKitClient::hlc_envelope(&hlc, None, None);
    assert!(plain.get("__sksv").is_none());
    assert_eq!(
        plain,
        serde_json::json!({ "__skver": 2, "__skhlc": hlc, "data": null }),
        "an app that declares no version must push byte-identical envelopes"
    );
}

/// The stamp is added *within* v2 rather than as a new `__skver`, so a reader
/// that predates it addresses `__skhlc` and `data` by name and is unaffected.
#[test]
fn a_stamped_envelope_still_reads_as_an_ordinary_v2_envelope() {
    let node = DeviceId::new(Uuid::from_u128(1));
    let hlc = Hlc {
        wall_ms: 9,
        counter: 1,
        node,
    };
    let stamped = SyncKitClient::hlc_envelope(&hlc, Some(&serde_json::json!(7)), Some(12));
    assert_eq!(stamped["__skver"], serde_json::json!(2));
    assert_eq!(
        serde_json::from_value::<Hlc>(stamped["__skhlc"].clone()).unwrap(),
        hlc
    );
    assert_eq!(stamped["data"], serde_json::json!(7));
}

/// Optional by construction: a malformed stamp must not fail a change that is
/// otherwise fine.
#[test]
fn a_malformed_stamp_reads_as_no_stamp() {
    let node = DeviceId::new(Uuid::from_u128(1));
    let hlc = Hlc {
        wall_ms: 1,
        counter: 0,
        node,
    };
    for bad in [
        serde_json::json!("four"),
        serde_json::json!(-1),
        serde_json::json!(null),
        serde_json::json!(u64::from(u32::MAX) + 1),
    ] {
        let env = serde_json::json!({
            "__skver": 2, "__skhlc": hlc, "data": null, "__sksv": bad
        });
        let (_, _, version) = SyncKitClient::split_hlc_envelope(env, node, 0).unwrap();
        assert_eq!(version, None, "bad stamp {bad} should read as absent");
    }
}

#[test]
fn split_envelope_rejects_unknown_version_loudly() {
    // The X2 hazard: a future envelope version must error, not silently
    // fall back to a bare-row read (which would corrupt the clock).
    let node = DeviceId::new(Uuid::from_u128(1));
    let hlc = Hlc {
        wall_ms: 5,
        counter: 0,
        node,
    };
    let future = serde_json::json!({ "__skver": 3, "__skhlc": hlc, "data": null });
    let err = SyncKitClient::split_hlc_envelope(future, node, 0).unwrap_err();
    assert!(
        matches!(err, SyncKitError::Crypto(ref m) if m.contains("envelope version 3")),
        "unexpected error: {err:?}"
    );
}

// ── encrypt_change / decrypt_change ──

#[test]
fn delete_seals_hlc_envelope_and_roundtrips() {
    // A Delete carries no row payload, but its HLC must still travel, so it is
    // now encrypted into an envelope (data = Some), and decrypt restores the
    // op, a None payload, and the exact HLC.
    let client = SyncKitClient::new(test_config());
    let key = crypto::generate_master_key();
    *client.master_key.write() = Some(crypto::ZeroizeOnDrop(key));

    let device = DeviceId::new(Uuid::new_v4());
    let hlc = Hlc {
        wall_ms: 12_345,
        counter: 7,
        node: device,
    };
    let entry = ChangeEntry {
        table: "tasks".to_string(),
        op: ChangeOp::Delete,
        row_id: "row-1".to_string(),
        timestamp: Utc::now(),
        hlc,
        data: None,
        extra: serde_json::Map::default(),
    };

    let wire = client.encrypt_change(entry).unwrap();
    assert_eq!(wire.op, ChangeOp::Delete);
    assert!(
        wire.data.is_some(),
        "delete now seals an encrypted HLC envelope"
    );

    let pull_entry = PullChangeEntry {
        seq: 1,
        device_id: device,
        table: wire.table,
        op: wire.op,
        row_id: wire.row_id,
        timestamp: wire.timestamp,
        data: wire.data,
        key_id: None,
        gck_version: None,
    };
    let decrypted = client.decrypt_change(pull_entry).unwrap();
    assert_eq!(decrypted.op, ChangeOp::Delete);
    assert_eq!(decrypted.row_id, "row-1");
    assert!(
        decrypted.data.is_none(),
        "payload is still None after the envelope unwraps"
    );
    assert_eq!(decrypted.hlc, hlc, "HLC survives the round trip");
}

#[test]
fn encrypt_change_fails_without_master_key() {
    let client = SyncKitClient::new(test_config());
    let entry = ChangeEntry {
        table: "tasks".to_string(),
        op: ChangeOp::Insert,
        row_id: "row-1".to_string(),
        timestamp: Utc::now(),
        hlc: Hlc::zero(DeviceId::nil()),
        data: Some(serde_json::json!({"title": "test"})),
        extra: serde_json::Map::default(),
    };

    let err = client.encrypt_change(entry).unwrap_err();
    assert!(matches!(err, SyncKitError::NoMasterKey));
}

#[test]
fn encrypt_change_produces_encrypted_data() {
    let client = SyncKitClient::new(test_config());
    let key = crypto::generate_master_key();
    *client.master_key.write() = Some(crypto::ZeroizeOnDrop(key));

    let original_data = serde_json::json!({"title": "Buy milk", "priority": 3});
    let entry = ChangeEntry {
        table: "tasks".to_string(),
        op: ChangeOp::Insert,
        row_id: "row-1".to_string(),
        timestamp: Utc::now(),
        hlc: Hlc::zero(DeviceId::nil()),
        data: Some(original_data.clone()),
        extra: serde_json::Map::default(),
    };

    let wire = client.encrypt_change(entry).unwrap();
    assert!(wire.data.is_some());
    let encrypted = wire.data.unwrap();
    assert!(encrypted.is_string());
    assert_ne!(encrypted, original_data);
}

#[test]
fn encrypt_decrypt_roundtrip() {
    let client = SyncKitClient::new(test_config());
    let key = crypto::generate_master_key();
    *client.master_key.write() = Some(crypto::ZeroizeOnDrop(key));

    let original_data = serde_json::json!({
        "title": "Buy milk",
        "tags": ["groceries", "urgent"],
        "count": 42
    });
    let ts = Utc::now();
    let entry = ChangeEntry {
        table: "tasks".to_string(),
        op: ChangeOp::Update,
        row_id: "row-abc".to_string(),
        timestamp: ts,
        hlc: Hlc::zero(DeviceId::nil()),
        data: Some(original_data.clone()),
        extra: serde_json::Map::default(),
    };

    let wire = client.encrypt_change(entry).unwrap();
    let pull_entry = PullChangeEntry {
        seq: 1,
        device_id: crate::ids::DeviceId::new(uuid::Uuid::new_v4()),
        table: wire.table,
        op: wire.op,
        row_id: wire.row_id,
        timestamp: wire.timestamp,
        data: wire.data,
        key_id: None,
        gck_version: None,
    };

    let decrypted = client.decrypt_change(pull_entry).unwrap();
    assert_eq!(decrypted.table, "tasks");
    assert_eq!(decrypted.op, ChangeOp::Update);
    assert_eq!(decrypted.row_id, "row-abc");
    assert_eq!(decrypted.data.unwrap(), original_data);
}

#[test]
fn decrypt_change_with_no_data() {
    let client = SyncKitClient::new(test_config());
    let pull_entry = PullChangeEntry {
        seq: 5,
        device_id: crate::ids::DeviceId::new(uuid::Uuid::new_v4()),
        table: "events".to_string(),
        op: ChangeOp::Delete,
        row_id: "evt-1".to_string(),
        timestamp: Utc::now(),
        data: None,
        key_id: None,
        gck_version: None,
    };

    let decrypted = client.decrypt_change(pull_entry).unwrap();
    assert_eq!(decrypted.table, "events");
    assert_eq!(decrypted.op, ChangeOp::Delete);
    assert!(decrypted.data.is_none());
}

#[test]
fn decrypt_change_fails_without_master_key() {
    let client = SyncKitClient::new(test_config());
    let pull_entry = PullChangeEntry {
        seq: 1,
        device_id: crate::ids::DeviceId::new(uuid::Uuid::new_v4()),
        table: "tasks".to_string(),
        op: ChangeOp::Insert,
        row_id: "row-1".to_string(),
        timestamp: Utc::now(),
        data: Some(serde_json::json!("some-encrypted-string")),
        key_id: None,
        gck_version: None,
    };

    let err = client.decrypt_change(pull_entry).unwrap_err();
    assert!(matches!(err, SyncKitError::NoMasterKey));
}

// ── is_transient error classification ──

#[test]
fn is_transient_server_5xx() {
    let err = SyncKitError::Server {
        status: 500,
        message: "Internal Server Error".to_string(),
        retry_after_secs: None,
    };
    assert!(is_transient(&err));
    let err = SyncKitError::Server {
        status: 502,
        message: "Bad Gateway".to_string(),
        retry_after_secs: None,
    };
    assert!(is_transient(&err));
    let err = SyncKitError::Server {
        status: 503,
        message: "Service Unavailable".to_string(),
        retry_after_secs: None,
    };
    assert!(is_transient(&err));
    let err = SyncKitError::Server {
        status: 504,
        message: "Gateway Timeout".to_string(),
        retry_after_secs: None,
    };
    assert!(is_transient(&err));
}

#[test]
fn is_transient_rate_limited_429() {
    let err = SyncKitError::Server {
        status: 429,
        message: "Too Many Requests".to_string(),
        retry_after_secs: None,
    };
    assert!(is_transient(&err));
}

#[test]
fn is_not_transient_client_4xx() {
    let err = SyncKitError::Server {
        status: 400,
        message: "Bad Request".to_string(),
        retry_after_secs: None,
    };
    assert!(!is_transient(&err));
    let err = SyncKitError::Server {
        status: 401,
        message: "Unauthorized".to_string(),
        retry_after_secs: None,
    };
    assert!(!is_transient(&err));
    let err = SyncKitError::Server {
        status: 403,
        message: "Forbidden".to_string(),
        retry_after_secs: None,
    };
    assert!(!is_transient(&err));
    let err = SyncKitError::Server {
        status: 404,
        message: "Not Found".to_string(),
        retry_after_secs: None,
    };
    assert!(!is_transient(&err));
    let err = SyncKitError::Server {
        status: 409,
        message: "Conflict".to_string(),
        retry_after_secs: None,
    };
    assert!(!is_transient(&err));
    let err = SyncKitError::Server {
        status: 422,
        message: "Unprocessable Entity".to_string(),
        retry_after_secs: None,
    };
    assert!(!is_transient(&err));
}

#[test]
fn is_not_transient_not_authenticated() {
    assert!(!is_transient(&SyncKitError::NotAuthenticated));
}

#[test]
fn is_not_transient_no_master_key() {
    assert!(!is_transient(&SyncKitError::NoMasterKey));
}

#[test]
fn is_not_transient_decryption_failed() {
    assert!(!is_transient(&SyncKitError::DecryptionFailed));
}

#[test]
fn is_not_transient_invalid_envelope() {
    assert!(!is_transient(&SyncKitError::InvalidEnvelope(
        "bad version".to_string()
    )));
}

#[test]
fn is_not_transient_crypto() {
    assert!(!is_transient(&SyncKitError::Crypto(
        "encrypt failed".to_string()
    )));
}

#[test]
fn is_not_transient_json() {
    let err: SyncKitError = serde_json::from_str::<serde_json::Value>("not json")
        .unwrap_err()
        .into();
    assert!(!is_transient(&err));
}

#[test]
fn is_not_transient_base64() {
    let err: SyncKitError = base64::engine::general_purpose::STANDARD
        .decode("!!!invalid!!!")
        .unwrap_err()
        .into();
    assert!(!is_transient(&err));
}

#[test]
fn is_not_transient_token_expired() {
    assert!(!is_transient(&SyncKitError::TokenExpired));
}

#[test]
fn is_not_transient_internal() {
    assert!(!is_transient(&SyncKitError::Internal(
        "lock poisoned".to_string()
    )));
}

// ── Retry constants ──

#[test]
fn retry_constants_are_sensible() {
    assert_eq!(MAX_RETRIES, 3);
    assert_eq!(BASE_DELAY, Duration::from_secs(1));
}

#[test]
fn backoff_delays_are_exponential() {
    let delay_0 = BASE_DELAY * 2u32.pow(0);
    let delay_1 = BASE_DELAY * 2u32.pow(1);
    let delay_2 = BASE_DELAY * 2u32.pow(2);

    assert_eq!(delay_0, Duration::from_secs(1));
    assert_eq!(delay_1, Duration::from_secs(2));
    assert_eq!(delay_2, Duration::from_secs(4));
}

// ── is_transient boundary: 429 vs 428, 499 vs 500 ──

#[test]
fn is_transient_boundary_values() {
    assert!(!is_transient(&SyncKitError::Server {
        status: 428,
        message: String::new(),
        retry_after_secs: None
    }));
    assert!(is_transient(&SyncKitError::Server {
        status: 429,
        message: String::new(),
        retry_after_secs: None
    }));
    assert!(!is_transient(&SyncKitError::Server {
        status: 430,
        message: String::new(),
        retry_after_secs: None
    }));
    assert!(!is_transient(&SyncKitError::Server {
        status: 499,
        message: String::new(),
        retry_after_secs: None
    }));
    assert!(is_transient(&SyncKitError::Server {
        status: 500,
        message: String::new(),
        retry_after_secs: None
    }));
}

// ── Token expiry detection ──

#[test]
fn jwt_exp_extracts_expiry() {
    let exp = Utc::now().timestamp() + 3600;
    let token = fake_jwt(exp);
    assert_eq!(jwt_exp(&token), Some(exp));
}

#[test]
fn jwt_exp_returns_none_for_garbage() {
    assert_eq!(jwt_exp("not-a-jwt"), None);
    assert_eq!(jwt_exp("a.b.c"), None);
    assert_eq!(jwt_exp(""), None);
}

#[test]
fn token_is_expired_for_past_exp() {
    let token = fake_jwt(Utc::now().timestamp() - 3600);
    assert!(token_is_expired(&token));
}

#[test]
fn token_is_expired_within_buffer() {
    let token = fake_jwt(Utc::now().timestamp() + 10);
    assert!(token_is_expired(&token));
}

#[test]
fn token_is_not_expired_when_fresh() {
    let token = fake_jwt(Utc::now().timestamp() + 3600);
    assert!(!token_is_expired(&token));
}

#[test]
fn token_is_not_expired_for_garbage() {
    assert!(!token_is_expired("garbage"));
}

#[test]
fn token_expires_exactly_at_buffer_boundary() {
    let token = fake_jwt(Utc::now().timestamp() + TOKEN_EXPIRY_BUFFER_SECS);
    assert!(token_is_expired(&token));
}

#[test]
fn token_expires_just_past_buffer() {
    let token = fake_jwt(Utc::now().timestamp() + TOKEN_EXPIRY_BUFFER_SECS + 1);
    assert!(!token_is_expired(&token));
}

// ── encrypt_change preserves metadata ──

#[test]
fn encrypt_change_preserves_all_metadata() {
    let client = SyncKitClient::new(test_config());
    let key = crypto::generate_master_key();
    client.set_master_key_raw(key);

    let ts = Utc::now();
    let entry = ChangeEntry {
        table: "contacts".to_string(),
        op: ChangeOp::Update,
        row_id: "unique-row-id".to_string(),
        timestamp: ts,
        hlc: Hlc::zero(DeviceId::nil()),
        data: Some(serde_json::json!({"name": "Alice"})),
        extra: serde_json::Map::default(),
    };

    let wire = client.encrypt_change(entry).unwrap();
    assert_eq!(wire.table, "contacts");
    assert_eq!(wire.op, ChangeOp::Update);
    assert_eq!(wire.row_id, "unique-row-id");
    assert_eq!(wire.timestamp, ts);
}

// ── Multiple entries encrypt/decrypt ──

#[test]
fn multiple_entries_encrypt_decrypt_roundtrip() {
    let client = SyncKitClient::new(test_config());
    let key = crypto::generate_master_key();
    client.set_master_key_raw(key);

    let entries = [
        ChangeEntry {
            table: "tasks".to_string(),
            op: ChangeOp::Insert,
            row_id: "r1".to_string(),
            timestamp: Utc::now(),
            hlc: Hlc::zero(DeviceId::nil()),
            data: Some(serde_json::json!({"title": "Task 1"})),
            extra: serde_json::Map::default(),
        },
        ChangeEntry {
            table: "tasks".to_string(),
            op: ChangeOp::Update,
            row_id: "r2".to_string(),
            timestamp: Utc::now(),
            hlc: Hlc::zero(DeviceId::nil()),
            data: Some(serde_json::json!({"title": "Task 2", "done": true})),
            extra: serde_json::Map::default(),
        },
        ChangeEntry {
            table: "events".to_string(),
            op: ChangeOp::Delete,
            row_id: "r3".to_string(),
            timestamp: Utc::now(),
            hlc: Hlc::zero(DeviceId::nil()),
            data: None,
            extra: serde_json::Map::default(),
        },
    ];

    let wire_entries: Vec<_> = entries
        .iter()
        .cloned()
        .map(|e| client.encrypt_change(e).unwrap())
        .collect();

    assert_eq!(wire_entries.len(), 3);
    assert!(wire_entries[0].data.is_some());
    assert!(wire_entries[1].data.is_some());
    // The Delete now also seals an encrypted HLC envelope (was None before HLC).
    assert!(wire_entries[2].data.is_some());

    for (i, wire) in wire_entries.into_iter().enumerate() {
        let pull = PullChangeEntry {
            seq: i as i64,
            device_id: crate::ids::DeviceId::new(uuid::Uuid::new_v4()),
            table: wire.table,
            op: wire.op,
            row_id: wire.row_id,
            timestamp: wire.timestamp,
            data: wire.data,
            key_id: None,
            gck_version: None,
        };
        let decrypted = client.decrypt_change(pull).unwrap();
        assert_eq!(decrypted.table, entries[i].table);
        assert_eq!(decrypted.op, entries[i].op);
        assert_eq!(decrypted.data, entries[i].data);
    }
}

// ── Unicode and edge-case roundtrips ──

#[test]
fn encrypt_decrypt_roundtrip_unicode_table() {
    let client = SyncKitClient::new(test_config());
    let key = crypto::generate_master_key();
    client.set_master_key_raw(key);

    let entry = ChangeEntry {
        table: "\u{65E5}\u{672C}\u{8A9E}\u{30C6}\u{30FC}\u{30D6}\u{30EB}".into(),
        op: ChangeOp::Insert,
        row_id: "row-1".into(),
        timestamp: Utc::now(),
        hlc: Hlc::zero(DeviceId::nil()),
        data: Some(serde_json::json!({"name": "\u{30C6}\u{30B9}\u{30C8}"})),
        extra: serde_json::Map::default(),
    };

    let wire = client.encrypt_change(entry).unwrap();
    let pull = PullChangeEntry {
        seq: 1,
        device_id: crate::ids::DeviceId::new(uuid::Uuid::new_v4()),
        table: wire.table,
        op: wire.op,
        row_id: wire.row_id,
        timestamp: wire.timestamp,
        data: wire.data,
        key_id: None,
        gck_version: None,
    };
    let decrypted = client.decrypt_change(pull).unwrap();
    assert_eq!(
        decrypted.table,
        "\u{65E5}\u{672C}\u{8A9E}\u{30C6}\u{30FC}\u{30D6}\u{30EB}"
    );
}

#[test]
fn encrypt_decrypt_roundtrip_empty_row_id() {
    let client = SyncKitClient::new(test_config());
    let key = crypto::generate_master_key();
    client.set_master_key_raw(key);

    let entry = ChangeEntry {
        table: "t".into(),
        op: ChangeOp::Insert,
        row_id: String::new(),
        timestamp: Utc::now(),
        hlc: Hlc::zero(DeviceId::nil()),
        data: Some(serde_json::json!(42)),
        extra: serde_json::Map::default(),
    };

    let wire = client.encrypt_change(entry).unwrap();
    let pull = PullChangeEntry {
        seq: 1,
        device_id: crate::ids::DeviceId::new(uuid::Uuid::new_v4()),
        table: wire.table,
        op: wire.op,
        row_id: wire.row_id,
        timestamp: wire.timestamp,
        data: wire.data,
        key_id: None,
        gck_version: None,
    };
    let decrypted = client.decrypt_change(pull).unwrap();
    assert_eq!(decrypted.row_id, "");
    assert_eq!(decrypted.data.unwrap(), serde_json::json!(42));
}

// ── decrypt_change_multi_key key selection ──
//
// Pins the key-id boundary logic that picks between primary and pending
// master keys during a rotation window. The mutations targeted here:
//   * `effective_key_id == pending_key_id` (== ↔ !=)
//   * `effective_key_id == primary_key_id || effective_key_id <= 1`
//     (`||` ↔ `&&`, `<= 1` ↔ `< 1`/`<= 0`)
//   * `entry.key_id.unwrap_or(1)` default
//   * the Err-fallthrough that tries pending if primary fails

fn encrypt_with(key: &[u8; 32], value: &serde_json::Value) -> serde_json::Value {
    crypto::encrypt_json(value, key).unwrap()
}

fn pull_entry_with(encrypted: serde_json::Value, key_id: Option<i32>) -> PullChangeEntry {
    PullChangeEntry {
        seq: 1,
        device_id: crate::ids::DeviceId::new(uuid::Uuid::new_v4()),
        table: "tasks".to_string(),
        op: ChangeOp::Insert,
        row_id: "row-multikey".to_string(),
        timestamp: Utc::now(),
        data: Some(encrypted),
        key_id,
        gck_version: None,
    }
}

#[test]
fn multi_key_picks_pending_when_key_id_matches() {
    let primary = crypto::generate_master_key();
    let pending = crypto::generate_master_key();
    let plaintext = serde_json::json!({"v": "pending-payload"});
    let entry = pull_entry_with(encrypt_with(&pending, &plaintext), Some(7));

    let decrypted = SyncKitClient::decrypt_with_rotation_keys(
        entry,
        &primary,
        1,
        &pending,
        7,
        &SyncKitClient::decrypt_change_with_key,
    )
    .unwrap();
    assert_eq!(decrypted.data.unwrap(), plaintext);
}

#[test]
fn multi_key_picks_primary_when_key_id_matches_primary() {
    let primary = crypto::generate_master_key();
    let pending = crypto::generate_master_key();
    let plaintext = serde_json::json!({"v": "primary-payload"});
    let entry = pull_entry_with(encrypt_with(&primary, &plaintext), Some(3));

    let decrypted = SyncKitClient::decrypt_with_rotation_keys(
        entry,
        &primary,
        3,
        &pending,
        9,
        &SyncKitClient::decrypt_change_with_key,
    )
    .unwrap();
    assert_eq!(decrypted.data.unwrap(), plaintext);
}

#[test]
fn multi_key_treats_missing_key_id_as_primary() {
    // `entry.key_id.unwrap_or(1)` defaults to 1; `effective_key_id <= 1`
    // arm routes to primary. A mutation changing `unwrap_or(1)` to
    // `unwrap_or(99)` would route to "unknown" path and try primary anyway
    // via the fallback, but a mutation to `<= 1` → `< 1` would skip the
    // direct-primary branch and fall into the fallback path.
    let primary = crypto::generate_master_key();
    let pending = crypto::generate_master_key();
    let plaintext = serde_json::json!({"v": "legacy-no-key-id"});
    let entry = pull_entry_with(encrypt_with(&primary, &plaintext), None);

    let decrypted = SyncKitClient::decrypt_with_rotation_keys(
        entry,
        &primary,
        5,
        &pending,
        6,
        &SyncKitClient::decrypt_change_with_key,
    )
    .unwrap();
    assert_eq!(decrypted.data.unwrap(), plaintext);
}

#[test]
fn multi_key_unknown_key_id_falls_back_to_primary_first() {
    // effective_key_id (42) matches neither primary (5) nor pending (6),
    // and is > 1. The fallback first tries primary; here the data WAS
    // encrypted with primary, so the fallback succeeds.
    let primary = crypto::generate_master_key();
    let pending = crypto::generate_master_key();
    let plaintext = serde_json::json!({"v": "via-fallback-primary"});
    let entry = pull_entry_with(encrypt_with(&primary, &plaintext), Some(42));

    let decrypted = SyncKitClient::decrypt_with_rotation_keys(
        entry,
        &primary,
        5,
        &pending,
        6,
        &SyncKitClient::decrypt_change_with_key,
    )
    .unwrap();
    assert_eq!(decrypted.data.unwrap(), plaintext);
}

#[test]
fn multi_key_unknown_key_id_falls_back_to_pending_when_primary_fails() {
    // Unknown key_id and the data was encrypted with pending, fallback
    // must try primary first (fails), then pending (succeeds).
    let primary = crypto::generate_master_key();
    let pending = crypto::generate_master_key();
    let plaintext = serde_json::json!({"v": "via-fallback-pending"});
    let entry = pull_entry_with(encrypt_with(&pending, &plaintext), Some(42));

    let decrypted = SyncKitClient::decrypt_with_rotation_keys(
        entry,
        &primary,
        5,
        &pending,
        6,
        &SyncKitClient::decrypt_change_with_key,
    )
    .unwrap();
    assert_eq!(decrypted.data.unwrap(), plaintext);
}

// ── Backoff, jitter and Retry-After ──

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 server_err(status: u16, retry_after_secs: Option<u64>) -> SyncKitError {
    SyncKitError::Server {
        status,
        message: String::new(),
        retry_after_secs,
    }
}

#[test]
fn retry_delay_uses_retry_after_capped_at_sixty_seconds() {
    assert_eq!(
        retry_delay(&server_err(429, Some(1)), 0),
        Duration::from_secs(1)
    );
    assert_eq!(
        retry_delay(&server_err(429, Some(7)), 3),
        Duration::from_secs(7)
    );
    assert_eq!(
        retry_delay(&server_err(429, Some(59)), 0),
        Duration::from_secs(59)
    );
    assert_eq!(
        retry_delay(&server_err(429, Some(60)), 0),
        Duration::from_mins(1)
    );
    assert_eq!(
        retry_delay(&server_err(429, Some(61)), 0),
        Duration::from_mins(1)
    );
    assert_eq!(
        retry_delay(&server_err(503, Some(86_400)), 2),
        Duration::from_mins(1)
    );
}

#[test]
fn retry_delay_without_retry_after_is_jittered_exponential() {
    let errors = [
        server_err(503, None),
        server_err(429, None),
        SyncKitError::Internal("no retry-after on this variant".to_string()),
    ];
    for attempt in 0..4u32 {
        let base = BASE_DELAY * 2u32.pow(attempt);
        let low = base.mul_f64(0.8);
        let high = base.mul_f64(1.2);
        for err in &errors {
            for _ in 0..32 {
                let delay = retry_delay(err, attempt);
                assert!(
                    delay >= low && delay <= high,
                    "attempt {attempt}: {delay:?} outside [{low:?}, {high:?}]"
                );
            }
        }
    }
}

#[test]
fn retry_delay_grows_with_the_attempt() {
    // The windows for consecutive attempts do not overlap, so the ordering
    // holds for every draw rather than on average.
    let err = server_err(503, None);
    for attempt in 0..3u32 {
        let here = retry_delay(&err, attempt);
        let next = retry_delay(&err, attempt + 1);
        assert!(
            next > here,
            "attempt {attempt}: {next:?} not above {here:?}"
        );
    }
}

#[test]
fn jittered_stays_within_twenty_percent_of_base() {
    for base_ms in [5u64, 10, 1_000, 4_000, 60_000] {
        let base = Duration::from_millis(base_ms);
        let span = base_ms / 5;
        let low = Duration::from_millis(base_ms - span);
        let high = Duration::from_millis(base_ms + span);
        for _ in 0..128 {
            let got = jittered(base);
            assert!(
                got >= low && got <= high,
                "{got:?} outside [{low:?}, {high:?}]"
            );
        }
    }
}

#[test]
fn jittered_actually_spreads() {
    let base = Duration::from_secs(1);
    let first = jittered(base);
    assert!(
        (0..256).any(|_| jittered(base) != first),
        "256 draws all returned {first:?}, so nothing is being spread"
    );
}

#[test]
fn jittered_returns_base_when_the_span_rounds_to_zero() {
    // Under 5ms the 20% span truncates to 0 and the base is returned as-is.
    for base_ms in 0..5u64 {
        let base = Duration::from_millis(base_ms);
        assert_eq!(jittered(base), base);
    }
    assert_eq!(jittered(Duration::ZERO), Duration::ZERO);
}

/// Fetch a response carrying the given `Retry-After` header value (or none).
/// The server is returned alongside so it outlives the response.
async fn response_with_retry_after(
    header: Option<&str>,
) -> (wiremock::MockServer, reqwest::Response) {
    ensure_crypto_provider();
    let server = wiremock::MockServer::start().await;
    let mut template = wiremock::ResponseTemplate::new(429);
    if let Some(value) = header {
        template = template.insert_header("retry-after", value);
    }
    wiremock::Mock::given(wiremock::matchers::any())
        .respond_with(template)
        .mount(&server)
        .await;
    let resp = reqwest::get(server.uri()).await.unwrap();
    (server, resp)
}

#[tokio::test]
async fn parse_retry_after_reads_a_numeric_header() {
    let (_server, resp) = response_with_retry_after(Some("5")).await;
    assert_eq!(parse_retry_after(&resp), Some(5));
}

#[tokio::test]
async fn parse_retry_after_does_not_cap() {
    // The 60s cap belongs to retry_delay, not to parsing.
    let (_server, resp) = response_with_retry_after(Some("3600")).await;
    assert_eq!(parse_retry_after(&resp), Some(3600));
}

#[tokio::test]
async fn parse_retry_after_is_none_when_absent() {
    let (_server, resp) = response_with_retry_after(None).await;
    assert_eq!(parse_retry_after(&resp), None);
}

#[tokio::test]
async fn parse_retry_after_is_none_for_a_non_numeric_header() {
    for value in [
        "",
        "soon",
        "5s",
        "-3",
        "1.5",
        "Wed, 21 Oct 2015 07:28:00 GMT",
    ] {
        let (_server, resp) = response_with_retry_after(Some(value)).await;
        assert_eq!(parse_retry_after(&resp), None, "header {value:?}");
    }
}

#[tokio::test]
async fn parse_retry_after_is_none_for_zero() {
    let (_server, resp) = response_with_retry_after(Some("0")).await;
    assert_eq!(parse_retry_after(&resp), None);
}

#[tokio::test]
async fn check_response_carries_retry_after_onto_the_error() {
    let (_server, resp) = response_with_retry_after(Some("12")).await;
    let err = check_response(resp).await.unwrap_err();
    match err {
        SyncKitError::Server {
            status,
            retry_after_secs,
            ..
        } => {
            assert_eq!(status, 429);
            assert_eq!(retry_after_secs, Some(12));
        }
        other => panic!("expected a Server error, got {other:?}"),
    }
    // And the retry loop then uses it verbatim.
    assert_eq!(
        retry_delay(&server_err(429, Some(12)), 0),
        Duration::from_secs(12)
    );
}

// ── is_transient over the Http arm ──

#[tokio::test]
async fn is_transient_over_reqwest_error_kinds() {
    ensure_crypto_provider();

    // A builder error is a programming mistake: an invalid header value.
    let builder = reqwest::Client::new()
        .get("http://example.invalid/")
        .header("x-bad", "line\nbreak")
        .build()
        .unwrap_err();
    assert!(builder.is_builder());
    assert!(!is_transient(&SyncKitError::Http(builder)));

    // A redirect loop is permanent.
    let server = wiremock::MockServer::start().await;
    let uri = server.uri();
    wiremock::Mock::given(wiremock::matchers::any())
        .respond_with(wiremock::ResponseTemplate::new(302).insert_header("location", uri.as_str()))
        .mount(&server)
        .await;
    let redirect = reqwest::get(server.uri()).await.unwrap_err();
    assert!(redirect.is_redirect());
    assert!(!is_transient(&SyncKitError::Http(redirect)));

    // A body that will not decode is permanent. The uncapped reader is
    // banned in production code; a decode error is what this test needs and
    // the body is eight bytes long.
    let json_server = wiremock::MockServer::start().await;
    wiremock::Mock::given(wiremock::matchers::any())
        .respond_with(wiremock::ResponseTemplate::new(200).set_body_string("not json"))
        .mount(&json_server)
        .await;
    #[allow(clippy::disallowed_methods)]
    let decode = {
        let resp = reqwest::get(json_server.uri()).await.unwrap();
        resp.json::<u32>().await.unwrap_err()
    };
    assert!(decode.is_decode());
    assert!(!is_transient(&SyncKitError::Http(decode)));

    // A connect failure is the transient case the arm exists for.
    let connect = reqwest::get("http://127.0.0.1:1/").await.unwrap_err();
    assert!(connect.is_connect());
    assert!(!connect.is_builder() && !connect.is_redirect() && !connect.is_decode());
    assert!(is_transient(&SyncKitError::Http(connect)));
}

// ── read_body_capped ──

async fn body_of_len(len: usize) -> (wiremock::MockServer, reqwest::Response) {
    ensure_crypto_provider();
    let server = wiremock::MockServer::start().await;
    wiremock::Mock::given(wiremock::matchers::any())
        .respond_with(wiremock::ResponseTemplate::new(200).set_body_bytes(vec![b'x'; len]))
        .mount(&server)
        .await;
    let resp = reqwest::get(server.uri()).await.unwrap();
    (server, resp)
}

#[tokio::test]
async fn read_body_capped_returns_a_body_under_the_cap() {
    let (_server, resp) = body_of_len(100).await;
    let bytes = read_body_capped(resp, 1024).await.unwrap();
    assert_eq!(bytes.len(), 100);
    assert!(bytes.iter().all(|&b| b == b'x'));
}

#[tokio::test]
async fn read_body_capped_allows_a_body_exactly_at_the_cap() {
    // The check is `>`, so a body the size of the cap is legal.
    let (_server, resp) = body_of_len(256).await;
    let bytes = read_body_capped(resp, 256).await.unwrap();
    assert_eq!(bytes.len(), 256);
}

#[tokio::test]
async fn read_body_capped_rejects_an_oversized_content_length() {
    let (_server, resp) = body_of_len(2048).await;
    assert_eq!(resp.content_length(), Some(2048));
    let err = read_body_capped(resp, 1024).await.unwrap_err();
    match err {
        SyncKitError::Internal(msg) => {
            assert!(
                msg.contains("2048"),
                "message did not name the length: {msg}"
            );
            assert!(msg.contains("1024"), "message did not name the cap: {msg}");
        }
        other => panic!("expected Internal, got {other:?}"),
    }
}

#[tokio::test]
async fn read_text_capped_falls_back_to_empty_on_an_oversized_body() {
    let (_server, resp) = body_of_len(2048).await;
    assert_eq!(read_text_capped(resp, 16).await, String::new());
}

#[tokio::test]
async fn read_json_capped_propagates_the_cap_error() {
    let (_server, resp) = body_of_len(2048).await;
    let err = read_json_capped::<serde_json::Value>(resp, 16)
        .await
        .unwrap_err();
    assert!(matches!(err, SyncKitError::Internal(_)));
}

// ── the retry loop's sleep, observed on a frozen clock ──
//
// Every other retry test counts requests, and a request count cannot see
// `if attempt < max_attempts`: that guard gates only `tokio::time::sleep`,
// so flipping it to `>` (never sleep) or `<=` (sleep once more, after the
// final attempt) leaves the number of attempts untouched. These tests stamp
// the paused clock inside the operation itself, which makes each gap
// between attempts exactly the delay that was slept.

/// The gaps between consecutive clock stamps.
fn gaps(stamps: &[tokio::time::Instant]) -> Vec<Duration> {
    stamps.windows(2).map(|w| w[1] - w[0]).collect()
}

/// Assert a slept gap lands inside `jittered`'s +/-20% window for `base_ms`.
/// The windows for 1s, 2s and 4s do not overlap, so a gap identifies which
/// backoff step produced it.
#[track_caller]
fn assert_slept(got: Duration, base_ms: u64) {
    let low = Duration::from_millis(base_ms - base_ms / 5);
    let high = Duration::from_millis(base_ms + base_ms / 5);
    assert!(
        got >= low && got <= high,
        "slept {got:?}, outside the +/-20% window [{low:?}, {high:?}] around {base_ms}ms"
    );
}

#[tokio::test(start_paused = true)]
async fn retry_request_sleeps_before_each_replay_and_not_after_the_last() {
    let client = SyncKitClient::new(test_config());
    let stamps = std::sync::Mutex::new(Vec::new());
    let start = tokio::time::Instant::now();

    let err = client
        .retry_request(Idempotency::ReadOnly, || {
            stamps.lock().unwrap().push(tokio::time::Instant::now());
            async { Result::<reqwest::Response>::Err(server_err(503, None)) }
        })
        .await
        .unwrap_err();
    let end = tokio::time::Instant::now();

    assert!(matches!(err, SyncKitError::Server { status: 503, .. }));
    let stamps = stamps.into_inner().unwrap();
    assert_eq!(stamps.len(), 4, "MAX_RETRIES is 3, so 1 try plus 3 replays");

    // Nothing is slept before the first try.
    assert_eq!(stamps[0] - start, Duration::ZERO);
    let gaps = gaps(&stamps);
    assert_slept(gaps[0], 1_000);
    assert_slept(gaps[1], 2_000);
    assert_slept(gaps[2], 4_000);
    // And nothing after the last: the loop gives up the instant the final
    // attempt fails. A guard that fired on `attempt == max_attempts` would
    // burn an 8s backoff here for no replay.
    assert_eq!(
        end - stamps[3],
        Duration::ZERO,
        "slept after the final attempt, which buys nothing"
    );
}

#[tokio::test(start_paused = true)]
async fn retry_request_neither_replays_nor_sleeps_an_unsafe_operation() {
    let client = SyncKitClient::new(test_config());
    let calls = std::sync::Mutex::new(0u32);
    let start = tokio::time::Instant::now();

    let err = client
        .retry_request(Idempotency::Unsafe, || {
            *calls.lock().unwrap() += 1;
            async { Result::<reqwest::Response>::Err(server_err(503, None)) }
        })
        .await
        .unwrap_err();

    assert!(matches!(err, SyncKitError::Server { status: 503, .. }));
    assert_eq!(*calls.lock().unwrap(), 1, "Unsafe gets exactly one attempt");
    // max_attempts is 0 here, so the sleep guard must not fire even once:
    // the caller is told the request failed with no delay bought for a
    // replay that is never going to happen.
    assert_eq!(tokio::time::Instant::now() - start, Duration::ZERO);
}

#[tokio::test(start_paused = true)]
async fn retry_request_json_sleeps_before_each_replay_and_not_after_the_last() {
    let client = SyncKitClient::new(test_config());
    let stamps = std::sync::Mutex::new(Vec::new());
    let start = tokio::time::Instant::now();

    let err = client
        .retry_request_json::<_, _, serde_json::Value>(Idempotency::ReadOnly, || {
            stamps.lock().unwrap().push(tokio::time::Instant::now());
            async { Result::<reqwest::Response>::Err(server_err(503, None)) }
        })
        .await
        .unwrap_err();
    let end = tokio::time::Instant::now();

    assert!(matches!(err, SyncKitError::Server { status: 503, .. }));
    let stamps = stamps.into_inner().unwrap();
    assert_eq!(stamps.len(), 4, "MAX_RETRIES is 3, so 1 try plus 3 replays");
    assert_eq!(stamps[0] - start, Duration::ZERO);
    let gaps = gaps(&stamps);
    assert_slept(gaps[0], 1_000);
    assert_slept(gaps[1], 2_000);
    assert_slept(gaps[2], 4_000);
    assert_eq!(
        end - stamps[3],
        Duration::ZERO,
        "slept after the final attempt, which buys nothing"
    );
}

#[tokio::test]
async fn retry_request_json_sleeps_between_body_read_failures() {
    // The body-read arm has its own copy of the guard, reached only when
    // the request succeeds and the parse does not.
    ensure_crypto_provider();
    let server = wiremock::MockServer::start().await;
    wiremock::Mock::given(wiremock::matchers::any())
        .respond_with(wiremock::ResponseTemplate::new(200))
        .mount(&server)
        .await;

    // Fetch every response BEFORE freezing the clock. An empty body is
    // Content-Length 0, so hyper ends the stream without another socket
    // read: inside the loop there is no I/O left, and the only thing that
    // can advance a paused clock is the loop's own sleep.
    let mut responses = std::collections::VecDeque::new();
    for _ in 0..4 {
        responses.push_back(reqwest::get(server.uri()).await.unwrap());
    }
    let responses = std::sync::Mutex::new(responses);
    let stamps = std::sync::Mutex::new(Vec::new());

    let client = SyncKitClient::new(test_config());
    tokio::time::pause();
    let start = tokio::time::Instant::now();
    let err = client
        .retry_request_json::<_, _, serde_json::Value>(Idempotency::ReadOnly, || {
            stamps.lock().unwrap().push(tokio::time::Instant::now());
            let resp = responses
                .lock()
                .unwrap()
                .pop_front()
                .expect("the loop asked for a fifth attempt");
            async move { Result::<reqwest::Response>::Ok(resp) }
        })
        .await
        .unwrap_err();
    let end = tokio::time::Instant::now();

    // An empty body is not JSON, so every attempt fails in the parse.
    assert!(matches!(err, SyncKitError::Json(_)), "got {err:?}");
    let stamps = stamps.into_inner().unwrap();
    assert_eq!(stamps.len(), 4, "MAX_RETRIES is 3, so 1 try plus 3 replays");
    assert_eq!(stamps[0] - start, Duration::ZERO);
    let gaps = gaps(&stamps);
    assert_slept(gaps[0], 1_000);
    assert_slept(gaps[1], 2_000);
    assert_slept(gaps[2], 4_000);
    assert_eq!(
        end - stamps[3],
        Duration::ZERO,
        "slept after the final attempt, which buys nothing"
    );
}

/// Serve one chunked HTTP/1.1 response carrying `len` bytes and no
/// `Content-Length`. The whole response goes out in a single write so the
/// body reaches `bytes_stream` as one item.
async fn chunked_body_of_len(len: usize) -> reqwest::Response {
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    ensure_crypto_provider();
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        let (mut sock, _) = listener.accept().await.unwrap();
        // Drain the request head before replying. A single `read` is not
        // enough: TCP may split the request across segments, and answering
        // a half-read request races the client's own write.
        let mut request = Vec::new();
        let mut scratch = [0u8; 256];
        while !request.windows(4).any(|w| w == b"\r\n\r\n") {
            let n = sock.read(&mut scratch).await.unwrap();
            assert_ne!(n, 0, "the client closed before sending a request head");
            request.extend_from_slice(&scratch[..n]);
        }
        let mut out = b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n".to_vec();
        out.extend_from_slice(format!("{len:x}\r\n").as_bytes());
        out.extend_from_slice(&vec![b'x'; len]);
        out.extend_from_slice(b"\r\n0\r\n\r\n");
        sock.write_all(&out).await.unwrap();
        sock.flush().await.unwrap();
    });
    reqwest::get(format!("http://{addr}/")).await.unwrap()
}

#[tokio::test]
async fn read_body_capped_rejects_an_oversized_body_that_declares_no_length() {
    // Every other oversized case is caught by the Content-Length fast path,
    // which leaves the running `buf.len() + chunk.len()` total unobserved.
    // A chunked body has no declared length, so that accumulation is the
    // only thing standing between the cap and an unbounded read: with the
    // buffer still empty it is the incoming chunk's own size that has to
    // trip the limit.
    let resp = chunked_body_of_len(100).await;
    assert_eq!(
        resp.content_length(),
        None,
        "the fast path must not be what rejects this"
    );
    let err = read_body_capped(resp, 50).await.unwrap_err();
    match err {
        SyncKitError::Internal(msg) => {
            assert!(msg.contains("50"), "message did not name the cap: {msg}");
        }
        other => panic!("expected Internal, got {other:?}"),
    }
}