//! Internal HTTP plumbing shared across the client modules.
//!
//! Response-status to [`SyncKitError`] mapping, idempotency-key generation,
//! size-capped body reads (so a hostile server cannot exhaust memory on a
//! control response), wire-version negotiation, and JWT-expiry parsing. This
//! module is `pub(crate)`: none of it is part of the SDK's public API.

use base64::Engine;

use crate::{
    crypto,
    error::{Result, SyncKitError},
    types::{ChangeEntry, Hlc, PullChangeEntry, WireChangeEntry},
};

use super::{BASE_DELAY, MAX_RETRIES, SyncKitClient};

/// Proof, supplied at the call site, that retrying an operation is sound.
///
/// Retry replays the request, so it is only safe when the server treats the
/// operation idempotently. Requiring this argument makes that judgement an
/// explicit, reviewable decision, and the judgement has runtime teeth:
/// [`may_retry`](Self::may_retry) gates the retry loop, so an
/// [`Unsafe`](Self::Unsafe) operation physically cannot be auto-replayed.
///
/// The variants split "safe to replay" into a **read** and a **write that the
/// server dedups**, and the write variant must *name* what makes replay
/// idempotent. That is the forcing function: a mutating create tagged
/// auto-retryable without naming its dedup key does not compile, so the
/// `ota_create_release`/`ota_register_artifact` class, a `CREATE` that silently
/// relied on a server unique constraint nobody had written down, cannot recur
/// unnoticed.
#[derive(Clone, Copy, Debug)]
pub(super) enum Idempotency {
    /// A read with no server-side effect (a `GET`, or a `POST` that only reads).
    /// Replay is trivially safe.
    ReadOnly,
    /// A write whose replay collapses to a single effect because the server
    /// dedups on `on`, a content-addressed key, an optimistic version, or a
    /// `UNIQUE` constraint. `on` documents *why* replay is harmless, at the call
    /// site, and naming it is mandatory.
    IdempotentWrite {
        /// The key/constraint the server dedups on (e.g. `"(app_id, version)
        /// unique"`). Documentation, surfaced in review and logs.
        on: &'static str,
    },
    /// Carries a client-generated idempotency key (e.g. push's `batch_id`) so
    /// the server collapses duplicate deliveries into one effect.
    Keyed,
    /// Not idempotent and not idempotency-keyed: replaying it would mint
    /// duplicate external state (e.g. a second Stripe Checkout session). The
    /// retry helpers attempt it exactly once.
    Unsafe,
}

impl Idempotency {
    /// Whether the retry helpers may replay this operation on a transient error.
    fn may_retry(self) -> bool {
        matches!(
            self,
            Idempotency::ReadOnly | Idempotency::IdempotentWrite { .. } | Idempotency::Keyed
        )
    }

    /// The documented dedup basis for an idempotent write (`None` for the other
    /// variants), surfaced in the retry log so a replayed request records *why*
    /// replay was safe.
    fn dedup_basis(self) -> Option<&'static str> {
        match self {
            Idempotency::IdempotentWrite { on } => Some(on),
            _ => None,
        }
    }
}

/// Upper bound on a control/JSON/error response body read into memory. Blob
/// bodies stream through [`super::SyncKitClient::blob_download`]'s own 4 GiB cap;
/// every *other* response, JSON control replies, OTA manifests, SSE-open errors,
/// 4xx/5xx error bodies, is small, so a hostile or buggy server streaming a
/// multi-gigabyte body into one of them is a pure OOM lever. 8 MiB is generous
/// for any legitimate control body.
pub(super) const MAX_CONTROL_BODY_BYTES: usize = 8 * 1024 * 1024;

/// Read a response body into memory with a hard byte cap, the ONE sanctioned
/// way to buffer a non-blob body.
///
/// `reqwest`'s `Response::json`/`text`/`bytes` read the whole body with no size
/// limit, so a hostile server can stream an arbitrarily large body and OOM the
/// client (worst case: the public, unauthenticated OTA updater check). This
/// drains `bytes_stream()` and aborts the instant the running total exceeds
/// `limit`, and fast-rejects on an honest oversized `Content-Length` before
/// reading a byte. Those raw reader methods are banned by `clippy.toml`
/// (`disallowed-methods`) so a new call site cannot reintroduce an uncapped read.
pub(super) async fn read_body_capped(
    resp: reqwest::Response,
    limit: usize,
) -> Result<bytes::Bytes> {
    use tokio_stream::StreamExt;
    if let Some(len) = resp.content_length()
        && len > limit as u64
    {
        return Err(SyncKitError::Internal(format!(
            "response Content-Length {len} exceeds {limit}-byte cap"
        )));
    }
    let mut stream = resp.bytes_stream();
    let mut buf = bytes::BytesMut::new();
    while let Some(chunk) = stream.next().await {
        let chunk = chunk.map_err(SyncKitError::Http)?;
        if buf.len() + chunk.len() > limit {
            return Err(SyncKitError::Internal(format!(
                "response body exceeds {limit}-byte cap"
            )));
        }
        buf.extend_from_slice(&chunk);
    }
    Ok(buf.freeze())
}

/// [`read_body_capped`] + JSON deserialize, the capped replacement for
/// `resp.json::<T>()`.
pub(super) async fn read_json_capped<T: serde::de::DeserializeOwned>(
    resp: reqwest::Response,
    limit: usize,
) -> Result<T> {
    let bytes = read_body_capped(resp, limit).await?;
    Ok(serde_json::from_slice(&bytes)?)
}

/// [`read_body_capped`] + lossy UTF-8, the capped replacement for `resp.text()`
/// on error bodies (which are only logged, so lossy decoding and a `""` fallback
/// on an oversized/failed read are fine).
pub(super) async fn read_text_capped(resp: reqwest::Response, limit: usize) -> String {
    match read_body_capped(resp, limit).await {
        Ok(bytes) => String::from_utf8_lossy(&bytes).into_owned(),
        Err(_) => String::new(),
    }
}

/// Current on-wire HLC-envelope version, written into the `__skver` tag.
const ENVELOPE_VERSION: u64 = 2;

/// The on-wire HLC-envelope version, parsed from the `__skver` tag.
///
/// Parsing is the explicit dispatch point: an unknown (future) version is a
/// hard error, never a silent fallback. That closes the X2 hazard, a protocol
/// bump that an older client cannot understand fails loudly instead of being
/// mis-decoded as a bare row and corrupting the clock.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
enum WireVersion {
    /// The current envelope: `{ __skver: 2, __skhlc, data }`.
    V2,
}

impl WireVersion {
    fn parse(tag: u64) -> Result<Self> {
        match tag {
            2 => Ok(WireVersion::V2),
            other => Err(SyncKitError::Crypto(format!(
                "unknown sync envelope version {other}; this client is too old to read it"
            ))),
        }
    }
}

impl SyncKitClient {
    /// Retry an async HTTP operation with exponential backoff.
    ///
    /// Retries on transient errors (network failures, 5xx, 429) up to [`MAX_RETRIES`]
    /// times with delays of 1s, 2s, 4s. Returns the last error if all attempts fail.
    /// Client errors (4xx except 429) are considered permanent and returned immediately.
    ///
    /// `idempotency` is the caller's proof that replay is safe, see [`Idempotency`].
    pub(super) async fn retry_request<F, Fut>(
        &self,
        idempotency: Idempotency,
        mut operation: F,
    ) -> Result<reqwest::Response>
    where
        F: FnMut() -> Fut,
        Fut: std::future::Future<Output = Result<reqwest::Response>>,
    {
        // A non-idempotent operation gets exactly one attempt: zero retries.
        let max_attempts = if idempotency.may_retry() {
            MAX_RETRIES
        } else {
            0
        };
        let mut last_err = None;

        for attempt in 0..=max_attempts {
            match operation().await {
                Ok(resp) => return Ok(resp),
                Err(err) => {
                    if !is_transient(&err) {
                        return Err(err);
                    }

                    if attempt < max_attempts {
                        let delay = retry_delay(&err, attempt);
                        tracing::debug!(
                            attempt = attempt + 1,
                            max_retries = MAX_RETRIES,
                            delay_ms = delay.as_millis() as u64,
                            error = %err,
                            idempotency_basis = idempotency.dedup_basis(),
                            "Transient error, retrying after backoff",
                        );
                        tokio::time::sleep(delay).await;
                    }

                    last_err = Some(err);
                }
            }
        }

        Err(last_err.expect("loop ran at least once"))
    }

    /// Retry an HTTP operation and deserialize the JSON response body inside
    /// the retry loop. This ensures a transient body-read failure (truncated
    /// response, connection reset mid-body) is retried rather than surfacing
    /// as a permanent error after the server already committed the operation.
    ///
    /// `idempotency` is the caller's proof that replay is safe, see [`Idempotency`].
    pub(super) async fn retry_request_json<F, Fut, T>(
        &self,
        idempotency: Idempotency,
        mut operation: F,
    ) -> Result<T>
    where
        F: FnMut() -> Fut,
        Fut: std::future::Future<Output = Result<reqwest::Response>>,
        T: serde::de::DeserializeOwned,
    {
        // A non-idempotent operation gets exactly one attempt: zero retries.
        let max_attempts = if idempotency.may_retry() {
            MAX_RETRIES
        } else {
            0
        };
        let mut last_err = None;

        for attempt in 0..=max_attempts {
            match operation().await {
                Ok(resp) => match read_json_capped::<T>(resp, MAX_CONTROL_BODY_BYTES).await {
                    Ok(parsed) => return Ok(parsed),
                    Err(e) => {
                        let err = e;
                        if attempt < max_attempts {
                            let delay = retry_delay(&err, attempt);
                            tracing::debug!(
                                attempt = attempt + 1,
                                max_retries = MAX_RETRIES,
                                delay_ms = delay.as_millis() as u64,
                                error = %err,
                                "Response body read failed, retrying",
                            );
                            tokio::time::sleep(delay).await;
                        }
                        last_err = Some(err);
                    }
                },
                Err(err) => {
                    if !is_transient(&err) {
                        return Err(err);
                    }

                    if attempt < max_attempts {
                        let delay = retry_delay(&err, attempt);
                        tracing::debug!(
                            attempt = attempt + 1,
                            max_retries = max_attempts,
                            delay_ms = delay.as_millis() as u64,
                            error = %err,
                            "Transient error, retrying after backoff",
                        );
                        tokio::time::sleep(delay).await;
                    }

                    last_err = Some(err);
                }
            }
        }

        Err(last_err.expect("loop ran at least once"))
    }

    /// Encrypt a change entry for the wire. Every change now seals an HLC
    /// envelope (Deletes included), so the master key is always required.
    #[cfg(test)]
    pub(super) fn encrypt_change(&self, entry: ChangeEntry) -> Result<WireChangeEntry> {
        let master_key = self.require_master_key()?;
        Self::encrypt_change_with_key(entry, &master_key)
    }

    /// Decrypt a pulled legacy entry that has no encrypted payload (a pre-HLC
    /// Delete). The HLC is synthesized from the entry's `client_timestamp`. No key
    /// needed.
    #[cfg(test)]
    pub(super) fn decrypt_change_no_data(entry: PullChangeEntry) -> Result<ChangeEntry> {
        debug_assert!(entry.data.is_none());
        Ok(ChangeEntry {
            table: entry.table,
            op: entry.op,
            row_id: entry.row_id,
            hlc: Hlc::from_legacy(entry.timestamp.timestamp_millis(), entry.device_id),
            timestamp: entry.timestamp,
            data: None,
            extra: serde_json::Map::default(),
        })
    }

    /// Wrap a change's HLC and payload into one JSON envelope. Encrypting the
    /// envelope (rather than the bare row payload) is what carries the HLC inside
    /// the E2E ciphertext, including for Deletes, which have no row payload. The
    /// server stores the ciphertext opaquely and never sees the clock.
    ///
    /// The `__skver` tag is a positive, explicit version marker. A reader
    /// dispatches on it (see [`WireVersion`]) rather than structurally guessing,
    /// so a future format bump is rejected loudly instead of silently misread.
    fn hlc_envelope(hlc: &Hlc, data: Option<&serde_json::Value>) -> serde_json::Value {
        serde_json::json!({ "__skver": ENVELOPE_VERSION, "__skhlc": hlc, "data": data })
    }

    /// Split a decrypted payload back into `(hlc, data)`.
    ///
    /// Dispatch is explicit, not structural:
    /// - A `__skver` tag means a versioned envelope; the version is parsed via
    ///   [`WireVersion::parse`], which **errors loudly** on an unknown future
    ///   version rather than silently falling back to a bare-row read (which
    ///   would corrupt the clock, the X2 hazard).
    /// - No `__skver` but an embedded `__skhlc` that parses is a gen-1 envelope
    ///   (predates the version tag).
    /// - Anything else is a legacy bare-row payload whose HLC is synthesized
    ///   from `node` + `timestamp_ms`.
    fn split_hlc_envelope(
        decrypted: serde_json::Value,
        node: crate::ids::DeviceId,
        timestamp_ms: i64,
    ) -> Result<(Hlc, Option<serde_json::Value>)> {
        if let Some(obj) = decrypted.as_object() {
            if let Some(tag) = obj.get("__skver") {
                // Explicit version present: dispatch, rejecting unknown loudly.
                let tag = tag.as_u64().ok_or_else(|| {
                    SyncKitError::Crypto("envelope __skver tag is not an integer".into())
                })?;
                return match WireVersion::parse(tag)? {
                    WireVersion::V2 => {
                        let hlc = obj
                            .get("__skhlc")
                            .and_then(|v| serde_json::from_value::<Hlc>(v.clone()).ok())
                            .ok_or_else(|| {
                                SyncKitError::Crypto("v2 envelope missing __skhlc".into())
                            })?;
                        let data = obj.get("data").cloned().filter(|v| !v.is_null());
                        Ok((hlc, data))
                    }
                };
            }
            // gen-1 envelope: embedded HLC, no version tag.
            if let Some(hlc) = obj
                .get("__skhlc")
                .and_then(|v| serde_json::from_value::<Hlc>(v.clone()).ok())
            {
                let data = obj.get("data").cloned().filter(|v| !v.is_null());
                return Ok((hlc, data));
            }
        }
        // Legacy bare-row payload: the decrypted value is the row data itself.
        Ok((Hlc::from_legacy(timestamp_ms, node), Some(decrypted)))
    }

    /// Encrypt with a pre-loaded key. Used by `push()` to avoid per-entry lock
    /// acquisition. Every change (including Deletes) is encrypted, because the HLC
    /// envelope always needs sealing. The ciphertext is bound to its
    /// `(table, row_id)` address via AEAD associated data, so a server cannot
    /// relocate it to another row/table without the open failing closed.
    pub(super) fn encrypt_change_with_key(
        entry: ChangeEntry,
        master_key: &[u8; 32],
    ) -> Result<WireChangeEntry> {
        let ctx = crypto::AeadContext::entry(&entry.table, &entry.row_id);
        let envelope = Self::hlc_envelope(&entry.hlc, entry.data.as_ref());
        let encrypted_data = Some(crypto::encrypt_json_aad(&envelope, master_key, &ctx)?);

        Ok(WireChangeEntry {
            table: entry.table,
            op: entry.op,
            row_id: entry.row_id,
            timestamp: entry.timestamp,
            data: encrypted_data,
        })
    }

    /// Decrypt the data field of a pulled change entry.
    #[cfg(test)]
    pub(super) fn decrypt_change(&self, entry: PullChangeEntry) -> Result<ChangeEntry> {
        if entry.data.is_some() {
            let master_key = self.require_master_key()?;
            Self::decrypt_change_with_key(entry, &master_key)
        } else {
            Self::decrypt_change_no_data(entry)
        }
    }

    /// Decrypt with a pre-loaded key, preserving `device_id` and `seq` in a [`PulledChange`].
    ///
    /// Used by `pull_rich()` to produce conflict-detection-ready results.
    pub(super) fn decrypt_change_to_pulled(
        entry: PullChangeEntry,
        master_key: &[u8; 32],
    ) -> Result<crate::types::PulledChange> {
        let device_id = entry.device_id;
        let seq = entry.seq;
        let decrypted = Self::decrypt_change_with_key(entry, master_key)?;
        Ok(crate::types::PulledChange {
            entry: decrypted,
            device_id,
            seq,
        })
    }

    /// Decrypt a pulled entry during a rotation window, selecting the key by the
    /// entry's `key_id`. Generic over the decrypt step so both the plain-pull
    /// (`ChangeEntry`) and rich-pull (`PulledChange`) paths share exactly this
    /// selection logic, including the unknown-`key_id` fallback that tries the
    /// primary key then the pending key. Previously the live pull path
    /// reimplemented a fallback-less variant while the tested one sat unused.
    pub(super) fn decrypt_with_rotation_keys<T, F>(
        entry: PullChangeEntry,
        primary_key: &[u8; 32],
        primary_key_id: i32,
        pending_key: &[u8; 32],
        pending_key_id: i32,
        decrypt_fn: &F,
    ) -> Result<T>
    where
        F: Fn(PullChangeEntry, &[u8; 32]) -> Result<T>,
    {
        let effective_key_id = entry.key_id.unwrap_or(1);
        if effective_key_id == pending_key_id {
            decrypt_fn(entry, pending_key)
        } else if effective_key_id == primary_key_id || effective_key_id <= 1 {
            decrypt_fn(entry, primary_key)
        } else {
            // Unknown key_id, try primary, then fall back to pending.
            match decrypt_fn(entry.clone(), primary_key) {
                Ok(result) => Ok(result),
                Err(_) => decrypt_fn(entry, pending_key),
            }
        }
    }

    /// Encrypt a change for a **group** changelog, sealed under the group's GCK
    /// and bound to `(group_id, table, row_id)` as AEAD associated data. Same HLC
    /// envelope as the personal path; the extra `group_id` binding makes a
    /// ciphertext non-relocatable across groups (the p1 crypto guarantee).
    pub(super) fn encrypt_group_change_with_key(
        group_id: &str,
        entry: ChangeEntry,
        gck: &[u8; 32],
    ) -> Result<WireChangeEntry> {
        let ctx = crypto::AeadContext::group_entry(group_id, &entry.table, &entry.row_id);
        let envelope = Self::hlc_envelope(&entry.hlc, entry.data.as_ref());
        let encrypted_data = Some(crypto::encrypt_json_aad(&envelope, gck, &ctx)?);

        Ok(WireChangeEntry {
            table: entry.table,
            op: entry.op,
            row_id: entry.row_id,
            timestamp: entry.timestamp,
            data: encrypted_data,
        })
    }

    /// Decrypt a pulled group entry under the group's GCK, preserving `device_id`
    /// and `seq` in a [`PulledChange`]. The counterpart to
    /// [`encrypt_group_change_with_key`](Self::encrypt_group_change_with_key).
    /// Group pull has no master-key-rotation window (a group's key rotates
    /// server-side, re-encrypted in place), so this is the whole decrypt path.
    pub(super) fn decrypt_group_change_to_pulled(
        group_id: &str,
        entry: PullChangeEntry,
        gck: &[u8; 32],
    ) -> Result<crate::types::PulledChange> {
        let device_id = entry.device_id;
        let seq = entry.seq;
        let ctx = crypto::AeadContext::group_entry(group_id, &entry.table, &entry.row_id);
        let (hlc, data) = match entry.data {
            Some(ref value) => {
                let decrypted = crypto::decrypt_json_aad(value, gck, &ctx)?;
                Self::split_hlc_envelope(decrypted, device_id, entry.timestamp.timestamp_millis())?
            }
            None => (
                Hlc::from_legacy(entry.timestamp.timestamp_millis(), device_id),
                None,
            ),
        };
        Ok(crate::types::PulledChange {
            entry: ChangeEntry {
                table: entry.table,
                op: entry.op,
                row_id: entry.row_id,
                hlc,
                timestamp: entry.timestamp,
                data,
                extra: serde_json::Map::default(),
            },
            device_id,
            seq,
        })
    }

    /// Decrypt with a pre-loaded key. Used by `pull()` to avoid per-entry lock acquisition.
    pub(super) fn decrypt_change_with_key(
        entry: PullChangeEntry,
        master_key: &[u8; 32],
    ) -> Result<ChangeEntry> {
        let ctx = crypto::AeadContext::entry(&entry.table, &entry.row_id);
        let (hlc, data) = match entry.data {
            Some(ref value) => {
                let decrypted = crypto::decrypt_json_aad(value, master_key, &ctx)?;
                Self::split_hlc_envelope(
                    decrypted,
                    entry.device_id,
                    entry.timestamp.timestamp_millis(),
                )?
            }
            None => (
                Hlc::from_legacy(entry.timestamp.timestamp_millis(), entry.device_id),
                None,
            ),
        };

        Ok(ChangeEntry {
            table: entry.table,
            op: entry.op,
            row_id: entry.row_id,
            hlc,
            timestamp: entry.timestamp,
            data,
            extra: serde_json::Map::default(),
        })
    }
}

/// Extract the `exp` claim from a JWT without verifying the signature.
///
/// JWTs are `header.payload.signature` where the payload is base64url-encoded JSON.
/// We decode the payload segment and read the `exp` field. Returns `None` if
/// the token is malformed or `exp` is missing.
pub(super) fn jwt_exp(token: &str) -> Option<i64> {
    let payload = token.split('.').nth(1)?;
    let bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD
        .decode(payload)
        .ok()?;
    let claims: serde_json::Value = serde_json::from_slice(&bytes).ok()?;
    claims["exp"].as_i64()
}

/// Returns `true` if the token's `exp` claim is within [`TOKEN_EXPIRY_BUFFER_SECS`]
/// of the current time (or already past). Returns `false` if the token cannot
/// be decoded, in that case, let the server decide.
#[cfg(test)]
pub(super) fn token_is_expired(token: &str) -> bool {
    let Some(exp) = jwt_exp(token) else {
        return false;
    };
    let now = chrono::Utc::now().timestamp();
    now >= exp - super::TOKEN_EXPIRY_BUFFER_SECS
}

/// Check an HTTP response for errors, returning the response on success.
///
/// For 429 responses, parses the `Retry-After` header so the retry loop
/// can use it instead of the default exponential backoff delay.
pub(super) async fn check_response(resp: reqwest::Response) -> Result<reqwest::Response> {
    let status = resp.status().as_u16();
    if status >= 400 {
        let retry_after_secs = parse_retry_after(&resp);
        // Capped read: this error path sits in front of every request (including
        // the otherwise-capped blob/SSE streams), so an uncapped `resp.text()`
        // here let a hostile server bypass those caps with a giant error body.
        let message = read_text_capped(resp, MAX_CONTROL_BODY_BYTES).await;
        return Err(SyncKitError::Server {
            status,
            message,
            retry_after_secs,
        });
    }
    Ok(resp)
}

/// Extract the `Retry-After` header from an HTTP response as seconds.
/// Returns `None` if the header is absent, non-numeric, or zero.
fn parse_retry_after(resp: &reqwest::Response) -> Option<u64> {
    resp.headers()
        .get("retry-after")
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.parse::<u64>().ok())
        .filter(|&secs| secs > 0)
}

/// Compute the backoff delay for a retry attempt.
///
/// Uses the server's `Retry-After` header (if present on a 429) capped at
/// 60 seconds. Otherwise exponential backoff (1s, 2s, 4s) with ±20% jitter so
/// many clients that hit a transient blip simultaneously don't retry in lockstep
/// and re-amplify the load.
fn retry_delay(err: &SyncKitError, attempt: u32) -> std::time::Duration {
    if let SyncKitError::Server {
        retry_after_secs: Some(secs),
        ..
    } = err
    {
        let capped = (*secs).min(60);
        return std::time::Duration::from_secs(capped);
    }
    jittered(BASE_DELAY * 2u32.pow(attempt))
}

/// Spread a backoff delay by ±20% to avoid synchronized retries (thundering herd).
fn jittered(base: std::time::Duration) -> std::time::Duration {
    use rand::RngExt;
    let millis = base.as_millis() as u64;
    let span = millis / 5; // 20%
    if span == 0 {
        return base;
    }
    // Uniform in [millis - span, millis + span].
    let delta = rand::rng().random_range(0..=2 * span);
    std::time::Duration::from_millis(millis - span + delta)
}

/// Returns true if the error is transient and worth retrying.
///
/// Transient errors:
/// - Network-level failures (connection refused, timeout, DNS, etc.)
/// - Server errors (5xx)
/// - Rate limiting (429)
///
/// Permanent errors (not retried):
/// - Client errors (4xx except 429), bad request, auth failure, not found, etc.
/// - Serialization errors, encryption errors, missing session, etc.
pub(super) fn is_transient(err: &SyncKitError) -> bool {
    match err {
        SyncKitError::Http(e) => {
            // Transport errors (timeout, connect, DNS) are transient.
            // Builder errors are programming mistakes, redirect loops and
            // body decode failures are permanent, retrying won't help.
            !e.is_builder() && !e.is_redirect() && !e.is_decode()
        }
        SyncKitError::Server { status, .. } => {
            // 5xx = server error (transient), 429 = rate limited (transient)
            *status >= 500 || *status == 429
        }
        // Everything else (auth, crypto, serialization, keychain) is permanent.
        // Note: some keychain errors (e.g. locked keychain, unavailable
        // secret-service) are conceptually transient, but keychain operations
        // are synchronous and not wrapped by retry_request.
        _ => false,
    }
}

#[cfg(test)]
mod tests {
    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 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,
        };
        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,
        };

        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,
        };

        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,
        };

        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,
            };
            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,
        };
        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,
        };
        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,
        }
    }

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