jmap_base_client/

client.rs

//! Auth-agnostic base JMAP HTTP client (RFC 8620).
//!
//! Provides [`JmapClient`] for session fetch, API calls, blob transfer,
//! SSE event streaming, and [`extract_response`] for parsing method results.

use std::sync::Arc;
use std::time::Duration;

use futures::StreamExt;

use crate::auth::{AuthProvider, DefaultTransport, TransportConfig};
use crate::error::ClientError;
use crate::request::Session;
use crate::sse::{SseFrame, parse_sse_block};

/// Internal state threaded through the `subscribe_events` unfold loop.
struct SseStreamState<S> {
    stream: S,
    /// Accumulates raw bytes from the HTTP stream before UTF-8 decoding.
    /// Incomplete multi-byte sequences remain here until the next chunk
    /// completes them, preventing stream termination when a codepoint is
    /// split across adjacent chunks.
    raw_buf: Vec<u8>,
    buf: String,
    /// Byte offset from which the next delimiter scan begins.
    /// Must always be a valid UTF-8 char boundary of `buf`.
    scan_from: usize,
}

/// Per-client configuration for timeouts and body size limits.
///
/// Use [`ClientConfig::default()`] for production defaults (30s timeout, RFC-safe caps).
///
/// This type is `#[non_exhaustive]`: callers outside this crate must use
/// `..ClientConfig::default()` when constructing it, allowing new fields to
/// be added in minor versions without breaking callers.
///
/// # Field-type rationale (bd:JMAP-6r7c.21)
///
/// The fields mix `u64` and `usize` integer types. The split is deliberate
/// and tracks the underlying transport's expectations:
///
/// - **HTTP body caps are `u64`** (`max_session_body`, `max_call_body`,
///   `max_download_body`, `max_upload_response_body`). `reqwest` reports
///   `Content-Length` as `u64`, and an HTTP body can in principle exceed
///   `usize::MAX` on a 32-bit target (4 GiB) without exceeding 64-bit
///   limits; the cap comparison happens before accumulation.
/// - **In-memory frame/message caps are `usize`** (`max_sse_frame`,
///   `max_ws_message`). These bound a `Vec<u8>` or `String` that must
///   fit in process memory; `usize` is the address-space-native type
///   and matches what `tokio_tungstenite::tungstenite::protocol::WebSocketConfig`
///   expects.
///
/// A future minor release MAY consolidate on `u64` with internal
/// `usize::try_from` casts at the call sites that need it. The mixed-int
/// API is a small ergonomic cost (callers cannot pass the same untyped
/// integer literal to both kinds of field without a `_u64` or `_usize`
/// suffix or `as usize` cast) traded for transparency to the underlying
/// transport's contract.
///
/// # No cross-field invariants are enforced
///
/// `validate()` checks each field independently. There is intentionally
/// no constraint that `max_sse_frame <= max_call_body` or similar — they
/// govern different threats (per-frame buffer cap vs whole-body cap)
/// and a deployment may rationally configure them in either order.
/// Document the chosen values in your own deployment notes.
#[non_exhaustive]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ClientConfig {
    /// Timeout for HTTP request/response cycles (fetch_session, call, upload_blob, download_blob).
    /// Does NOT apply to SSE or WebSocket streams (which are indefinite by nature).
    /// Must be > 0. Use `Duration::from_secs(30)` for a 30-second timeout.
    /// Default: 30 seconds.
    ///
    /// `Duration::ZERO` is forbidden because `reqwest::RequestBuilder::timeout`
    /// treats `Duration::ZERO` as "no per-request timeout" (only the
    /// client-level `connect_timeout` applies), not "instant fail". A future
    /// maintainer who thinks "zero means disable timeout, why not allow it?"
    /// is reading reqwest's semantics correctly but missing that **this
    /// crate intentionally forbids the no-timeout configuration**: a
    /// caller-visible 30-second-by-default timeout protects against
    /// indefinite hangs on stalled servers and against a slowloris-style
    /// resource leak in JMAP clients that hold one outstanding request per
    /// account. `validate()` enforces the positive-timeout invariant
    /// (bd:JMAP-6r7c.29). Do not relax this without re-deriving the
    /// no-timeout DoS argument.
    pub request_timeout: std::time::Duration,
    /// Maximum response body for fetch_session. Default: 1 MiB.
    pub max_session_body: u64,
    /// Maximum response body for call(). Default: 8 MiB.
    pub max_call_body: u64,
    /// Maximum response body for download_blob(). Default: 64 MiB.
    pub max_download_body: u64,
    /// Maximum size in bytes of the JSON response body returned by the
    /// server in reply to `upload_blob()`. Does NOT cap the size of the
    /// blob being uploaded — the JMAP server enforces that via its
    /// `maxSizeUpload` capability. This field caps only the small JSON
    /// envelope the server returns describing the stored blob.
    /// Default: 1 MiB.
    pub max_upload_response_body: u64,
    /// Maximum byte length of a single SSE frame; applied independently to
    /// the raw incoming bytes (pre-UTF-8 decode) and the decoded text.
    /// Protects against memory exhaustion from a hostile or misbehaving
    /// server that sends a single very large frame. Must be > 0.
    /// Default: 1 MiB.
    ///
    /// **Memory residency note (bd:JMAP-6lsm.7):** because the raw byte
    /// buffer and the decoded text buffer are tracked separately, the
    /// in-flight footprint while parsing a single frame can momentarily
    /// reach ~2 × `max_sse_frame` (raw bytes accumulated up to the limit,
    /// plus decoded text not yet drained). If you tune this value for a
    /// tight memory budget, plan for that 2× peak. The independent
    /// tracking is correct for the streaming UTF-8 decoder (which needs
    /// the raw buffer to be at least one full frame to handle split
    /// multi-byte sequences across HTTP chunks); see `decode_utf8_chunk`
    /// for the rationale.
    pub max_sse_frame: usize,
    /// Maximum byte length of a single WebSocket message (and frame). Mirrors
    /// `max_sse_frame` for the WebSocket transport. Used by
    /// [`JmapClient::connect_ws_session`] and threaded through to
    /// `tokio_tungstenite::tungstenite::protocol::WebSocketConfig`'s
    /// `max_message_size` and `max_frame_size`. Must be > 0.
    /// Default: 1 MiB. (bd:JMAP-6lsm.5)
    pub max_ws_message: usize,
}

impl Default for ClientConfig {
    fn default() -> Self {
        ClientConfig {
            request_timeout: std::time::Duration::from_secs(30),
            max_session_body: 1024 * 1024,
            max_call_body: 8 * 1024 * 1024,
            max_download_body: 64 * 1024 * 1024,
            max_upload_response_body: 1024 * 1024,
            max_sse_frame: 1024 * 1024,
            max_ws_message: 1024 * 1024,
        }
    }
}

impl ClientConfig {
    /// Validate that all config fields satisfy their constraints.
    ///
    /// Called automatically by [`JmapClient::new`].  Callers may also call
    /// this directly to pre-validate a config before passing it to the
    /// constructor.
    ///
    /// # Errors
    ///
    /// Returns [`ClientError::InvalidArgument`] when any field is zero or
    /// out-of-range.
    pub fn validate(&self) -> Result<(), ClientError> {
        if self.max_session_body == 0 {
            return Err(ClientError::InvalidArgument(
                "ClientConfig.max_session_body must be > 0".into(),
            ));
        }
        if self.max_call_body == 0 {
            return Err(ClientError::InvalidArgument(
                "ClientConfig.max_call_body must be > 0".into(),
            ));
        }
        if self.max_download_body == 0 {
            return Err(ClientError::InvalidArgument(
                "ClientConfig.max_download_body must be > 0".into(),
            ));
        }
        if self.max_upload_response_body == 0 {
            return Err(ClientError::InvalidArgument(
                "ClientConfig.max_upload_response_body must be > 0".into(),
            ));
        }
        if self.request_timeout == std::time::Duration::ZERO {
            // Duration::ZERO would let reqwest run requests without a
            // per-request timeout (reqwest treats ZERO as "no timeout").
            // Reject explicitly — see ClientConfig::request_timeout
            // rustdoc for the DoS-resistance rationale (bd:JMAP-6r7c.29).
            return Err(ClientError::InvalidArgument(
                "ClientConfig.request_timeout must be > 0; Duration::ZERO would disable the per-request timeout in reqwest, not 'fail immediately'. Use Duration::from_secs(30) or similar."
                    .into(),
            ));
        }
        if self.max_sse_frame == 0 {
            return Err(ClientError::InvalidArgument(
                "ClientConfig.max_sse_frame must be > 0".into(),
            ));
        }
        if self.max_ws_message == 0 {
            return Err(ClientError::InvalidArgument(
                "ClientConfig.max_ws_message must be > 0".into(),
            ));
        }
        Ok(())
    }
}

/// Auth-agnostic JMAP base HTTP client.
///
/// Construct with [`JmapClient::new`] or [`JmapClient::new_plain`].
/// Extension-specific clients (`jmap-chat-client`, `jmap-mail-client`) depend
/// on this crate and add their method implementations via `impl JmapClient`.
///
/// # Thread-safety (bd:JMAP-6r7c.25)
///
/// `JmapClient` is `Send + Sync + Clone`. Share by clone across threads or
/// `tokio::spawn` tasks; the underlying `reqwest::Client` is reference-counted
/// (Arc-backed) and the [`AuthProvider`] trait requires `Send + Sync` on every
/// implementation. A compile-time assertion in this crate's test suite pins
/// the `Send + Sync` contract: a future refactor that adds a non-`Sync` field
/// (e.g. `Rc<T>`, `RefCell<T>`, `Cell<T>`) would break the assertion in CI
/// before any downstream consumer is exposed.
#[derive(Clone)]
pub struct JmapClient {
    pub(crate) base_url: url::Url,
    pub(crate) auth: Arc<dyn AuthProvider>,
    pub(crate) http: reqwest::Client,
    pub(crate) config: ClientConfig,
}

impl std::fmt::Debug for JmapClient {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("JmapClient")
            .field("base_url", &self.base_url)
            .field("config", &self.config)
            .finish_non_exhaustive()
    }
}

impl JmapClient {
    /// Create a new client.
    ///
    /// `transport` configures the underlying HTTP client (TLS trust roots,
    /// client certificates, timeouts). `auth` injects per-request credentials
    /// (Bearer token, Basic credentials, or none). The two are independent so
    /// any transport can be paired with any credential scheme — for example,
    /// `CustomCaTransport` with `BearerAuth`. `base_url` must be the server
    /// origin (scheme, host, optional port) with no path, query, or fragment
    /// — e.g. `"https://100.64.1.1:8008"`. Trailing slashes are normalized
    /// away by the URL parser and are therefore accepted.
    pub fn new(
        transport: impl TransportConfig,
        auth: impl AuthProvider + 'static,
        base_url: &str,
        config: ClientConfig,
    ) -> Result<Self, ClientError> {
        // Parse-don't-validate: parse_base_url returns a fully-validated
        // url::Url so the constructor body can read as four facts (parse,
        // validate config, build transport, construct Self) rather than
        // a procedure (bd:JMAP-6lsm.26).
        let parsed = parse_base_url(base_url)?;
        config.validate()?;
        // Unwrap the opaque HttpClient at construction time (bd:JMAP-6r7c.36).
        // The wrapper exists to keep reqwest::Client out of the
        // TransportConfig trait signature; once the client is owned by
        // JmapClient, the inner reqwest::Client is the working type for
        // internal request building. into_inner is pub(crate) so external
        // code cannot reach inside the opaque wrapper.
        let http = transport.build_client()?.into_inner();
        Ok(Self {
            base_url: parsed,
            auth: Arc::new(auth),
            http,
            config,
        })
    }

    /// Convenience constructor for servers with publicly-trusted TLS.
    ///
    /// Equivalent to `JmapClient::new(DefaultTransport, auth, base_url, config)`.
    /// Use [`JmapClient::new`] when you need a custom transport (e.g.
    /// `CustomCaTransport` for a private-CA server).
    pub fn new_plain(
        auth: impl AuthProvider + 'static,
        base_url: &str,
        config: ClientConfig,
    ) -> Result<Self, ClientError> {
        Self::new(DefaultTransport, auth, base_url, config)
    }

    /// Create a new client sharing an existing `Arc<dyn AuthProvider>`
    /// (bd:JMAP-6r7c.27).
    ///
    /// `JmapClient::new` and `new_plain` take `auth` by value and wrap it
    /// in a fresh `Arc` internally. That is the ergonomic case for a
    /// caller constructing one client with one auth provider. It is the
    /// wrong shape for a caller who:
    ///
    /// - Operates multiple `JmapClient` instances against different
    ///   shards or accounts but uses the **same** credential holder
    ///   (e.g. a shared OAuth token-refresh state machine, a shared
    ///   service-account principal).
    /// - Wants a credential refresh in one client to be visible to all
    ///   sibling clients without rebuilding each one.
    ///
    /// This constructor takes a pre-built `Arc<dyn AuthProvider>` so
    /// callers can clone the Arc and pass clones to multiple
    /// `JmapClient::new_with_shared_auth` calls. The auth provider is
    /// then shared by reference-count, and any interior-mutable state
    /// (e.g. an `RwLock<TokenState>` inside a custom `OAuthAuth`
    /// implementation that holds a refreshable bearer) is genuinely
    /// shared across all sibling clients.
    ///
    /// Arguments mirror [`JmapClient::new`] otherwise.
    ///
    /// ```rust,ignore
    /// use std::sync::Arc;
    /// use jmap_base_client::{AuthProvider, BearerAuth, DefaultTransport, JmapClient, ClientConfig};
    ///
    /// let auth: Arc<dyn AuthProvider> = Arc::new(BearerAuth::new("token")?);
    /// let shard_a = JmapClient::new_with_shared_auth(
    ///     DefaultTransport,
    ///     auth.clone(),
    ///     "https://a.example.com",
    ///     ClientConfig::default(),
    /// )?;
    /// let shard_b = JmapClient::new_with_shared_auth(
    ///     DefaultTransport,
    ///     auth,
    ///     "https://b.example.com",
    ///     ClientConfig::default(),
    /// )?;
    /// ```
    pub fn new_with_shared_auth(
        transport: impl TransportConfig,
        auth: Arc<dyn AuthProvider>,
        base_url: &str,
        config: ClientConfig,
    ) -> Result<Self, ClientError> {
        let parsed = parse_base_url(base_url)?;
        config.validate()?;
        // Unwrap the opaque HttpClient at construction time (bd:JMAP-6r7c.36).
        let http = transport.build_client()?.into_inner();
        Ok(Self {
            base_url: parsed,
            auth,
            http,
            config,
        })
    }

    /// Apply the auth header (if any) to a request builder.
    ///
    /// Centralises the repeated `if let Some(...) = self.auth.auth_header()` pattern
    /// that every HTTP method uses. Callers: `fetch_session`, `call`,
    /// `subscribe_events`, `upload_blob`, `download_blob`.
    pub(crate) fn inject_auth(&self, builder: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
        if let Some(header) = self.auth.auth_header() {
            builder.header(header.name(), header.expose_value())
        } else {
            builder
        }
    }

    /// Returns `Err(ClientError::AuthFailed)` when the HTTP status indicates an
    /// authentication or authorization failure.
    ///
    /// Specifically handles:
    /// - 401 Unauthorized (RFC 7235 §3.1) — missing or invalid credentials
    /// - 403 Forbidden (RFC 7235 §3.2) — credentials present but insufficient
    ///
    /// Called before reading the response body so callers can distinguish
    /// permanent auth failures from transient errors without consuming the body.
    pub(crate) fn check_auth_status(status: reqwest::StatusCode) -> Result<(), ClientError> {
        if status == reqwest::StatusCode::UNAUTHORIZED || status == reqwest::StatusCode::FORBIDDEN {
            Err(ClientError::AuthFailed(status.as_u16()))
        } else {
            Ok(())
        }
    }

    /// Returns `Err(ClientError::RateLimited)` for HTTP 429 Too Many Requests
    /// (RFC 6585 §4), with `retry_after` taken from the `Retry-After` header
    /// (RFC 9110 §10.2.3).
    ///
    /// Called after [`Self::check_auth_status`] and before
    /// `error_for_status`, which would otherwise turn the 429 into an
    /// opaque [`ClientError::Http`].
    pub(crate) fn check_rate_limit(resp: &reqwest::Response) -> Result<(), ClientError> {
        if resp.status() == reqwest::StatusCode::TOO_MANY_REQUESTS {
            let retry_after = resp
                .headers()
                .get(reqwest::header::RETRY_AFTER)
                .and_then(|v| v.to_str().ok());
            Err(ClientError::rate_limited(
                retry_after,
                std::time::SystemTime::now(),
            ))
        } else {
            Ok(())
        }
    }
}

/// Read a response body into memory, capped at `limit` bytes.
///
/// Performs the Content-Length pre-check as a fast-path rejection for honest
/// oversized responses, then streams the body chunk-by-chunk and enforces the
/// cap before each accumulation. This prevents buffering a response that
/// exceeds the limit when Content-Length is absent (chunked Transfer-Encoding)
/// or under-reported by a hostile server — without per-chunk streaming, a
/// `.bytes().await` call would buffer the entire response before the
/// post-buffer length check could fire (bd:JMAP-6r7c.1).
///
/// Threat model: matters when connecting to untrusted servers (federation
/// peers, third-party JMAP services, DNS-hijacked endpoints, MITM scenarios).
/// On low-memory devices the peak allocation from naive buffering can OOM-kill
/// the client even if the post-buffer check would otherwise reject.
///
/// Returns `ClientError::ResponseTooLarge` when either the advertised
/// Content-Length or the accumulated chunk total exceeds `limit`.
pub(crate) async fn read_capped_body(
    resp: reqwest::Response,
    limit: u64,
) -> Result<Vec<u8>, ClientError> {
    if let Some(len) = resp.content_length()
        && len > limit
    {
        return Err(ClientError::ResponseTooLarge { actual: len, limit });
    }

    let mut stream = resp.bytes_stream();
    let mut body: Vec<u8> = Vec::new();
    while let Some(chunk) = stream.next().await {
        let chunk = chunk.map_err(ClientError::from_reqwest)?;
        let new_len = body.len() as u64 + chunk.len() as u64;
        if new_len > limit {
            return Err(ClientError::ResponseTooLarge {
                actual: new_len,
                limit,
            });
        }
        body.extend_from_slice(&chunk);
    }
    Ok(body)
}

impl JmapClient {
    /// Fetch the JMAP Session object from `{base_url}/.well-known/jmap` (RFC 8620 §2).
    ///
    /// The response body is capped at 1 MiB. Returns `ClientError::ResponseTooLarge`
    /// if the server sends more. Session URL fields (`apiUrl`, `uploadUrl`,
    /// `downloadUrl`, `eventSourceUrl`) are validated to have http/https scheme;
    /// a non-http scheme returns `ClientError::InvalidSession`.
    ///
    /// Returns `ClientError::AuthFailed` on HTTP 401 or 403, and
    /// `ClientError::RateLimited` on HTTP 429.
    ///
    /// # Charset
    ///
    /// The response body MUST be UTF-8-encoded JSON (RFC 8259 §8.1). A server
    /// that sends UTF-16 or UTF-32 JSON — even with a matching
    /// `charset=utf-16` `Content-Type` parameter — will fail to parse as
    /// `ClientError::Parse`; the error does not specifically call out the
    /// charset mismatch. Every shipped JMAP server uses UTF-8, but a
    /// non-conformant server can produce a confusing parse error
    /// (bd:JMAP-6r7c.28).
    pub async fn fetch_session(&self) -> Result<Session, ClientError> {
        let limit = self.config.max_session_body;
        let url = self.base_url.join(".well-known/jmap").map_err(|e| {
            // base_url was pre-validated in JmapClient::new; this path is
            // unreachable in practice but must be handled for completeness.
            ClientError::InvalidSession(format!("cannot construct session URL: {e}"))
        })?;

        let req = self.inject_auth(self.http.get(url).timeout(self.config.request_timeout));

        let resp = {
            let raw_resp = req.send().await.map_err(ClientError::from_reqwest)?;
            Self::check_auth_status(raw_resp.status())?;
            Self::check_rate_limit(&raw_resp)?;
            raw_resp
                .error_for_status()
                .map_err(ClientError::from_reqwest)?
        };

        // Stream the body chunk-by-chunk with the cap enforced before each
        // accumulation. The Content-Length pre-check inside read_capped_body
        // is a fast-path rejection for honest oversized responses; the
        // streaming cap is the actual DOS guard against a hostile server
        // that under-reports Content-Length or omits it entirely under
        // chunked Transfer-Encoding (bd:JMAP-6r7c.1).
        let body = read_capped_body(resp, limit).await?;

        let session: Session = serde_json::from_slice(&body).map_err(ClientError::from_parse)?;

        validate_session_url_schemes(&session)?;

        Ok(session)
    }

    /// POST a [`jmap_types::JmapRequest`] to `api_url` and return the parsed [`jmap_types::JmapResponse`]
    /// (RFC 8620 §3.3).
    ///
    /// `api_url` is taken as an explicit parameter (not from `self`) because the
    /// caller holds a [`Session`] and selects the correct URL from it.
    ///
    /// The response body is capped at 8 MiB. Returns `ClientError::ResponseTooLarge`
    /// if the server sends more.
    ///
    /// Returns `ClientError::AuthFailed` on HTTP 401 or 403, and
    /// `ClientError::RateLimited` on HTTP 429.
    ///
    /// # Charset
    ///
    /// The response body MUST be UTF-8-encoded JSON (RFC 8259 §8.1). A server
    /// that sends UTF-16 or UTF-32 JSON — even with a matching
    /// `charset=utf-16` `Content-Type` parameter — will fail to parse as
    /// `ClientError::Parse`; the error does not specifically call out the
    /// charset mismatch. Every shipped JMAP server uses UTF-8, but a
    /// non-conformant server can produce a confusing parse error
    /// (bd:JMAP-6r7c.28).
    ///
    /// # See also
    ///
    /// Prefer [`JmapClient::call_session`] when you have a [`Session`] —
    /// it picks the correct URL field automatically and prevents the
    /// "I passed `session.upload_url` instead of `session.api_url`"
    /// confusion (bd:JMAP-6r7c.39).
    pub async fn call(
        &self,
        api_url: &str,
        req: &jmap_types::JmapRequest,
    ) -> Result<jmap_types::JmapResponse, ClientError> {
        require_http_url(api_url)?;
        let limit = self.config.max_call_body;

        let builder = self.inject_auth(
            self.http
                .post(api_url)
                .json(req)
                .timeout(self.config.request_timeout),
        );

        let resp = {
            let raw_resp = builder.send().await.map_err(ClientError::from_reqwest)?;
            Self::check_auth_status(raw_resp.status())?;
            Self::check_rate_limit(&raw_resp)?;
            raw_resp
                .error_for_status()
                .map_err(ClientError::from_reqwest)?
        };

        // Stream the body chunk-by-chunk with the cap enforced before each
        // accumulation (bd:JMAP-6r7c.1). See `read_capped_body` for the DOS
        // rationale; without per-chunk streaming, a server that under-reports
        // or omits Content-Length can force unbounded allocation here.
        let body = read_capped_body(resp, limit).await?;

        let jmap_resp: jmap_types::JmapResponse =
            serde_json::from_slice(&body).map_err(ClientError::from_parse)?;

        Ok(jmap_resp)
    }

    /// POST a [`jmap_types::JmapRequest`] to the `api_url` field of `session`
    /// and return the parsed response (bd:JMAP-6r7c.39).
    ///
    /// Type-safe alternative to [`JmapClient::call`]: takes a [`Session`]
    /// reference and reads `session.api_url` internally. The
    /// "I passed `session.upload_url` instead of `session.api_url`"
    /// confusion is impossible at the call site because the caller does
    /// not select a URL — only `Session::api_url` is used.
    ///
    /// Same body cap, auth, and error semantics as [`JmapClient::call`].
    pub async fn call_session(
        &self,
        session: &Session,
        req: &jmap_types::JmapRequest,
    ) -> Result<jmap_types::JmapResponse, ClientError> {
        self.call(session.api_url.as_str(), req).await
    }

    /// Open an SSE connection to `event_source_url` and return an async stream
    /// of parsed [`SseFrame`]s (RFC 8620 §7.3).
    ///
    /// # URI template expansion
    ///
    /// `Session.event_source_url` is a URI template (RFC 6570 Level-1) with
    /// variables `types`, `closeafter`, and `ping`. You **must** expand it
    /// before passing it to this function, or the server will receive the
    /// literal text `{types}` in the URL and return an error. Use
    /// [`expand_url_template`](crate::expand_url_template):
    ///
    /// ```rust,ignore
    /// let url = jmap_base_client::expand_url_template(
    ///     &session.event_source_url,
    ///     &[("types", "*"), ("closeafter", "no"), ("ping", "0")],
    /// )?;
    /// let stream = client.subscribe_events(&url, None).await?;
    /// ```
    ///
    /// If `last_event_id` is `Some`, sends a `Last-Event-ID` header so the
    /// server can resume from where the previous stream left off.
    ///
    /// Buffer growth is capped at [`ClientConfig::max_sse_frame`] bytes per
    /// frame (default: 1 MiB). If a single SSE frame exceeds this limit the
    /// stream yields `ClientError::SseFrameTooLarge` and terminates.
    ///
    /// No timeout is applied to this call or to the resulting stream.  The
    /// connect timeout (10 s, TCP only) is the only deadline enforced.  If the
    /// server stalls before sending HTTP response headers, or later goes silent
    /// on the open connection, this call or the stream will hang indefinitely.
    /// Wrap the entire call and/or stream iteration in [`tokio::time::timeout`]
    /// if you need to bound either phase.
    ///
    /// Returns `ClientError::AuthFailed` on HTTP 401 or 403 before the stream
    /// starts.
    ///
    /// # Stream drop and cancellation (bd:JMAP-6r7c.24)
    ///
    /// The returned `BoxStream` may be dropped at any point — mid-frame,
    /// while awaiting `StreamExt::next`, or from inside a `tokio::select!`
    /// losing-branch cancellation. Dropping is always safe and always
    /// synchronous:
    ///
    /// - **Partial frame bytes are discarded.** Any bytes accumulated in
    ///   the internal `raw_buf` or `buf` that have not yet been parsed
    ///   into an [`SseFrame`] are lost. There is no buffering or replay
    ///   inside the client — the server is the source of truth for what
    ///   was emitted vs what was acknowledged.
    /// - **The underlying HTTP connection is released.** The
    ///   `reqwest::Response::bytes_stream` held inside the stream is
    ///   dropped along with the stream; reqwest returns the connection
    ///   to its pool (or closes it) per its own pool policy.
    /// - **Resumption is the caller's job.** If you want to resume from
    ///   the last successfully-parsed frame, capture the most recent
    ///   `SseFrame::id` (if the server sets one) and pass it as
    ///   `last_event_id` on the next `subscribe_events` call. The
    ///   server will replay events from that point per RFC 8895 §9.
    ///
    /// `tokio::select!` cancellation is the canonical use case: a caller
    /// racing the SSE stream against a shutdown signal can drop the
    /// stream by selecting the shutdown branch without leaking the HTTP
    /// connection or memory.
    pub async fn subscribe_events(
        &self,
        event_source_url: &str,
        last_event_id: Option<&str>,
    ) -> Result<futures::stream::BoxStream<'static, Result<SseFrame, ClientError>>, ClientError>
    {
        require_http_url(event_source_url)?;
        let mut req = self
            .http
            .get(event_source_url)
            .header("Accept", "text/event-stream");
        if let Some(id) = last_event_id {
            req = req.header("Last-Event-ID", id);
        }
        let req = self.inject_auth(req);

        let resp = req.send().await.map_err(ClientError::from_reqwest)?;
        Self::check_auth_status(resp.status())?;
        let resp = resp.error_for_status().map_err(ClientError::from_reqwest)?;

        // Verify Content-Type before streaming. A misconfigured server returning
        // application/json would silently produce no events (no SSE delimiter found).
        {
            // to_ascii_lowercase(): media types are case-insensitive per RFC 7231 §3.1.1.1.
            let ct = resp
                .headers()
                .get(reqwest::header::CONTENT_TYPE)
                .and_then(|v| v.to_str().ok())
                .unwrap_or("")
                .to_ascii_lowercase();
            // RFC 7231 §3.1.1.1 / RFC 9110 §8.3: the media-type "essence" (type
            // + "/" + subtype) is bounded by ';', SP, HTAB, or end-of-string.
            // A naive starts_with("text/event-stream") would accept
            // "text/event-streamish" or "text/event-stream2" — exactly the bug
            // JMAP-6lsm.2 flagged. Split off the parameter list / trailing
            // whitespace and compare the essence exactly.
            let essence = ct
                .split(|c: char| c == ';' || c.is_whitespace())
                .next()
                .unwrap_or("");
            if essence != "text/event-stream" {
                return Err(ClientError::UnexpectedResponse(format!(
                    "subscribe_events: expected Content-Type text/event-stream, got: {ct:?}"
                )));
            }
        }

        let byte_stream = resp.bytes_stream();
        let sse_frame_limit = self.config.max_sse_frame;

        Ok(futures::stream::unfold(
            Some(SseStreamState {
                stream: byte_stream,
                raw_buf: Vec::new(),
                buf: String::new(),
                scan_from: 0, // invariant: valid UTF-8 char boundary of buf; 0 always satisfies this
            }),
            move |state| async move {
                let SseStreamState {
                    mut stream,
                    mut raw_buf,
                    mut buf,
                    mut scan_from,
                } = state?;
                loop {
                    // Search for any double-newline delimiter (LF/CRLF/CR variants).
                    // scan_from is set to old_len.saturating_sub(3) after each append
                    // so we only re-scan the overlap region.  3 bytes back is the
                    // minimum that covers all delimiter prefixes that can straddle a
                    // chunk boundary:
                    //   - `\r\n\r\n` (4 bytes): longest prefix that fits in one chunk
                    //     but is incomplete is `\r\n\r` (3 bytes) — exactly covered.
                    //   - `\n\r\n` (3 bytes): longest incomplete prefix is `\n\r` (2
                    //     bytes) — covered by the 3-byte overlap.
                    //   - `\n\n` and `\r\r` (2 bytes each): longest incomplete prefix
                    //     is 1 byte — covered by the 3-byte overlap.
                    // Since \r and \n are single-byte UTF-8 codepoints, 3 bytes back
                    // is always a valid char boundary — no adjustment needed.
                    //
                    // Do not simplify to "rescan the whole buffer" (bd:JMAP-6r7c.19).
                    // The four-delimiter min_by_key + 3-byte overlap pattern is
                    // load-bearing for four independent reasons:
                    //
                    //   1. Overlap-rescan is O(chunk_size) per chunk; whole-buffer
                    //      rescan is O(n) per chunk, i.e. O(n²) over the lifetime
                    //      of a long-lived SSE stream. JMAP push streams can run
                    //      for hours with thousands of small events; the
                    //      difference is measurable.
                    //   2. All four delimiter forms are necessary. The WHATWG
                    //      EventSource spec + RFC 8895 line-terminator rules let
                    //      a single server emit mixed line endings within one
                    //      frame. min_by_key over all four catches the mixed
                    //      `\n\r\n` case that neither `\n\n` nor `\r\n\r\n`
                    //      alone catches. Removing forms produces silent
                    //      frame-misalignment on those servers.
                    //   3. The walk-back to a UTF-8 char boundary on the
                    //      `scan_from = old_len.saturating_sub(3)` assignment
                    //      below is needed because the byte preceding scan_from
                    //      might be in the middle of a multi-byte codepoint;
                    //      `buf[scan_from..]` would panic on the next `.find()`
                    //      without it.
                    //   4. Tests in tests/client_tests.rs pin all three line-
                    //      ending variants against hand-crafted server bodies:
                    //      `test_subscribe_events_crlf_line_endings`,
                    //      `test_subscribe_events_lf_crlf_frame_delimiter`,
                    //      and `test_subscribe_events_cr_line_endings`. The
                    //      mixed `\n\r\n` case (the second test) is the
                    //      trickiest and is the canonical example of why all
                    //      four delimiter searches are present.
                    //
                    // Resist the "just rescan the whole buffer" refactor.
                    let frame_end = [
                        buf[scan_from..]
                            .find("\r\n\r\n")
                            .map(|p| (scan_from + p, 4usize)),
                        buf[scan_from..]
                            .find("\n\n")
                            .map(|p| (scan_from + p, 2usize)),
                        buf[scan_from..]
                            .find("\r\r")
                            .map(|p| (scan_from + p, 2usize)),
                        // Mixed: LF-terminated last field line followed by a
                        // CRLF-terminated blank line.  Not detected by \n\n (the
                        // LFs are separated by a CR) or \r\n\r\n (no leading \r\n).
                        buf[scan_from..]
                            .find("\n\r\n")
                            .map(|p| (scan_from + p, 3usize)),
                    ]
                    .into_iter()
                    .flatten()
                    .min_by_key(|&(pos, _)| pos);

                    if let Some((pos, delim_len)) = frame_end {
                        let frame = {
                            let slice = &buf[..pos];
                            if slice.contains('\r') {
                                slice.replace("\r\n", "\n").replace('\r', "\n")
                            } else {
                                slice.to_owned()
                            }
                        };
                        buf.drain(..pos + delim_len);
                        scan_from = 0; // 0 satisfies the UTF-8 char boundary invariant
                        let sse_frame = parse_sse_block(&frame);
                        return Some((
                            Ok(sse_frame),
                            Some(SseStreamState {
                                stream,
                                raw_buf,
                                buf,
                                scan_from,
                            }),
                        ));
                    }

                    match stream.next().await {
                        None => return None,
                        Some(Err(e)) => {
                            return Some((Err(ClientError::from_reqwest(e)), None));
                        }
                        Some(Ok(bytes)) => {
                            // Accumulate raw bytes first. A multi-byte UTF-8 codepoint
                            // may be split across adjacent HTTP chunks; decode only the
                            // valid prefix and leave the remainder in raw_buf until the
                            // next chunk completes the sequence.
                            raw_buf.extend_from_slice(&bytes);
                            // Cap raw_buf to prevent OOM on persistent invalid UTF-8 input.
                            // Use the same limit as the decoded buf cap.
                            if raw_buf.len() > sse_frame_limit {
                                return Some((
                                    Err(ClientError::SseFrameTooLarge {
                                        limit: sse_frame_limit,
                                    }),
                                    None,
                                ));
                            }
                            let old_len = buf.len();
                            decode_utf8_chunk(&mut raw_buf, &mut buf);
                            scan_from = old_len.saturating_sub(3);
                            // Walk backward to a valid UTF-8 char boundary so that
                            // buf[scan_from..] never panics on multibyte characters.
                            while scan_from > 0 && !buf.is_char_boundary(scan_from) {
                                scan_from -= 1;
                            }
                            // Guard against unbounded buffer growth from a hostile server.
                            // Yield the error and terminate (state = None).
                            if buf.len() > sse_frame_limit {
                                return Some((
                                    Err(ClientError::SseFrameTooLarge {
                                        limit: sse_frame_limit,
                                    }),
                                    None,
                                ));
                            }
                        }
                    }
                }
            },
        )
        .boxed())
    }

    /// Open a WebSocket connection to `ws_url` using this client's
    /// configured `max_ws_message` byte cap.
    ///
    /// Convenience wrapper around [`connect_ws_with_limit`](crate::connect_ws_with_limit) that
    /// passes [`ClientConfig::max_ws_message`] as the per-message /
    /// per-frame byte cap. Mirrors the [`Self::subscribe_events`]
    /// pattern of "the JmapClient method uses ClientConfig; the free
    /// function takes an explicit value".
    ///
    /// `ws_url` must come from the session document's WebSocket capability
    /// URL. `auth_header` is an optional `(name, value)` pair for the
    /// upgrade request; the auth provider on this client is NOT used here
    /// because some servers attach WebSocket auth via cookie or session
    /// header rather than the same scheme as HTTP requests.
    ///
    /// # Security
    ///
    /// The `auth_header` value is a credential and must not be logged or
    /// echoed back to other systems. Treat it with the same care as a
    /// [`BearerAuth`](crate::BearerAuth) token. Transport errors raised by this
    /// method are constructed without the original credential bytes, but
    /// downstream code that inspects [`ClientError`] should still avoid
    /// printing or storing the `auth_header` itself.
    ///
    /// Returns [`ClientError::InvalidArgument`] for any URL that is not
    /// `wss://` (plaintext `ws://` is rejected).
    /// See [`connect_ws_with_limit`](crate::connect_ws_with_limit) for full error semantics.
    pub async fn connect_ws_session(
        &self,
        ws_url: &str,
        auth_header: Option<crate::auth::AuthHeader<'_>>,
    ) -> Result<crate::ws::WsSession, ClientError> {
        crate::ws::connect_ws_with_limit(ws_url, auth_header, self.config.max_ws_message).await
    }

    /// Open an SSE connection via a [`Session`]-supplied
    /// URL template (bd:JMAP-6r7c.64).
    ///
    /// Type-safe convenience wrapper over [`Self::subscribe_events`] —
    /// expands `session.event_source_url` internally using the
    /// caller-supplied [`SubscribeEventsSessionParams`] template
    /// variables. The caller cannot accidentally pass `session.api_url`
    /// or any other URL field because no URL is exposed at the call
    /// site.
    ///
    /// Template variables that are `None` expand to an empty string,
    /// per the RFC 8620 §7.3 default-omission semantics. Most JMAP
    /// servers accept `types=` (subscribe to all types),
    /// `closeafter=` (stay-open), and `ping=` (no server pings) as
    /// defaults; if your server requires explicit values, supply them
    /// via the params struct.
    pub async fn subscribe_events_session(
        &self,
        session: &Session,
        params: SubscribeEventsSessionParams<'_>,
    ) -> Result<futures::stream::BoxStream<'static, Result<SseFrame, ClientError>>, ClientError>
    {
        let SubscribeEventsSessionParams {
            types,
            close_after,
            ping,
            last_event_id,
        } = params;

        // RFC 8620 §7.3 event_source_url variables: types, closeafter, ping.
        // ping is u32; format to a String only when needed so the empty-default
        // path stays allocation-free.
        let ping_owned = ping.map(|p| p.to_string());
        let vars = [
            ("types", types.unwrap_or("")),
            ("closeafter", close_after.map_or("", CloseAfter::as_str)),
            ("ping", ping_owned.as_deref().unwrap_or("")),
        ];
        let expanded = crate::blob::expand_url_template(session.event_source_url.as_str(), &vars)?;
        self.subscribe_events(&expanded, last_event_id).await
    }

    /// Like [`Self::subscribe_events`], but reconnects automatically when
    /// the connection drops (RFC 8620 §7.3).
    ///
    /// Each reconnect sends `Last-Event-ID` with the most recent
    /// [`SseFrame::id`] seen (or `last_event_id` if no frame carried one
    /// yet), so the server can replay changes the client missed. Frames are
    /// yielded exactly as `subscribe_events` yields them; reconnects are
    /// invisible to the caller.
    ///
    /// # Retry rules
    ///
    /// - **Retried:** the server closing the stream, transport failures
    ///   (connect, mid-stream body errors), and HTTP 408, 429 and 5xx.
    /// - **Fatal** (yielded as the last item, then the stream ends):
    ///   `AuthFailed`, other HTTP 4xx, a wrong `Content-Type`,
    ///   `SseFrameTooLarge`, and invalid URLs or header values (including
    ///   a `Last-Event-ID` that is not a valid HTTP field value). Retrying
    ///   these cannot succeed.
    ///
    /// The first connection is made without delay. A reconnect after a
    /// connection that delivered at least one frame is also immediate
    /// (`closeafter=state` servers close after every event). Otherwise each
    /// consecutive failure waits per [`SseReconnectPolicy`]: the delay
    /// doubles from `initial_delay` up to `max_delay`, then equal jitter
    /// picks a uniform delay in `[d/2, d]` so many clients do not reconnect
    /// in lockstep. After `max_retries` consecutive failures the stream
    /// yields the last error (or just ends, if the last failure was a clean
    /// close) and stops.
    ///
    /// The returned stream owns a clone of this client; drop it to stop.
    pub fn subscribe_events_reconnecting(
        &self,
        event_source_url: &str,
        last_event_id: Option<&str>,
        policy: SseReconnectPolicy,
    ) -> futures::stream::BoxStream<'static, Result<SseFrame, ClientError>> {
        let state = SseReconnectState {
            client: self.clone(),
            url: event_source_url.to_owned(),
            last_event_id: last_event_id.map(str::to_owned),
            policy,
            inner: None,
            delivered: false,
            failures: 0,
        };
        futures::stream::unfold(Some(state), |state| async move {
            let mut st = state?;
            loop {
                let last_err;
                if let Some(inner) = st.inner.as_mut() {
                    match inner.next().await {
                        Some(Ok(frame)) => {
                            if let Some(id) = &frame.id {
                                st.last_event_id = Some(id.clone());
                            }
                            st.delivered = true;
                            st.failures = 0;
                            return Some((Ok(frame), Some(st)));
                        }
                        Some(Err(e)) if !sse_error_is_retryable(&e) => {
                            return Some((Err(e), None));
                        }
                        Some(Err(e)) => last_err = Some(e),
                        None => last_err = None,
                    }
                    st.inner = None;
                    if st.delivered {
                        continue;
                    }
                } else {
                    if st.failures > 0 {
                        tokio::time::sleep(st.policy.delay_for(st.failures)).await;
                    }
                    match st
                        .client
                        .subscribe_events(&st.url, st.last_event_id.as_deref())
                        .await
                    {
                        Ok(stream) => {
                            st.inner = Some(stream);
                            st.delivered = false;
                            continue;
                        }
                        Err(e) if !sse_error_is_retryable(&e) => return Some((Err(e), None)),
                        Err(e) => last_err = Some(e),
                    }
                }
                // Reached only on a failed attempt: a connect error, or a
                // connection that ended before delivering any frame.
                st.failures = st.failures.saturating_add(1);
                if st.policy.max_retries.is_some_and(|max| st.failures > max) {
                    return last_err.map(|e| (Err(e), None));
                }
            }
        })
        .boxed()
    }
}

/// Backoff policy for [`JmapClient::subscribe_events_reconnecting`].
///
/// Start from [`Default`] and adjust the public fields:
///
/// ```rust
/// # use std::time::Duration;
/// # use jmap_base_client::SseReconnectPolicy;
/// let mut policy = SseReconnectPolicy::default();
/// policy.max_delay = Duration::from_secs(300);
/// policy.max_retries = Some(10);
/// ```
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SseReconnectPolicy {
    /// Delay before the first retry; doubled after each further consecutive
    /// failure. Default: 1 s.
    pub initial_delay: Duration,
    /// Upper bound on the delay before jitter is applied. Default: 60 s.
    pub max_delay: Duration,
    /// Consecutive failed attempts to retry before giving up; `None` retries
    /// forever. A frame arriving resets the count. Default: `None`.
    pub max_retries: Option<u32>,
}

impl Default for SseReconnectPolicy {
    fn default() -> Self {
        Self {
            initial_delay: Duration::from_secs(1),
            max_delay: Duration::from_secs(60),
            max_retries: None,
        }
    }
}

impl SseReconnectPolicy {
    /// Delay before retry number `failures` (1-based): exponential backoff
    /// `d = min(initial_delay * 2^(failures-1), max_delay)`, then equal
    /// jitter, uniform in `[d/2, d]`.
    fn delay_for(&self, failures: u32) -> Duration {
        use std::hash::BuildHasher as _;
        let factor = 2u32.saturating_pow(failures.saturating_sub(1));
        let base = self
            .initial_delay
            .saturating_mul(factor)
            .min(self.max_delay);
        let half = base / 2;
        let span = u64::try_from(half.as_nanos()).unwrap_or(u64::MAX);
        // ponytail: jitter needs spread, not unpredictability, so a freshly
        // keyed std SipHasher stands in for an RNG dependency.
        let r = std::collections::hash_map::RandomState::new().hash_one(());
        half + Duration::from_nanos(r % span.saturating_add(1))
    }
}

/// Unfold state for [`JmapClient::subscribe_events_reconnecting`].
struct SseReconnectState {
    client: JmapClient,
    url: String,
    last_event_id: Option<String>,
    policy: SseReconnectPolicy,
    inner: Option<futures::stream::BoxStream<'static, Result<SseFrame, ClientError>>>,
    /// The current connection has yielded at least one frame.
    delivered: bool,
    /// Consecutive failed attempts since the last delivered frame.
    failures: u32,
}

/// `true` for errors a later reconnect can plausibly fix: transport
/// failures and HTTP 408 / 429 / 5xx. Builder errors (an invalid URL or
/// header value, such as a `Last-Event-ID` carrying control characters)
/// have no status either, but resending the same request cannot fix them.
fn sse_error_is_retryable(e: &ClientError) -> bool {
    match e {
        ClientError::Http(h) => {
            !h.is_builder() && h.status().is_none_or(|s| s == 408 || s == 429 || s >= 500)
        }
        _ => false,
    }
}

/// The RFC 8620 §7.3 `closeafter` event-source variable, which "MUST be
/// one of" these two values.
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum CloseAfter {
    /// `"state"`: "The server MUST end the HTTP response after pushing a
    /// state event."
    State,
    /// `"no"`: "The connection is persisted by the server as a standard
    /// event-source resource."
    No,
}

impl CloseAfter {
    /// The wire value substituted into the URL template.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::State => "state",
            Self::No => "no",
        }
    }
}

/// Parameters for [`JmapClient::subscribe_events_session`]
/// (bd:JMAP-6r7c.64).
///
/// Carries the RFC 8620 §7.3 event_source_url template variables
/// (`types`, `closeafter`, `ping`) plus the optional Last-Event-ID
/// header (RFC 8895 §9). `None` template values expand to an empty
/// string per the RFC's default-omission semantics.
///
/// Construct with a struct literal:
///
/// ```rust,ignore
/// client.subscribe_events_session(&session, SubscribeEventsSessionParams {
///     types: Some("Email,Mailbox"),
///     close_after: Some(CloseAfter::State),
///     ping: Some(60),
///     last_event_id: Some("evt-1234"),
/// }).await?;
/// ```
#[derive(Debug, Clone, Copy, Default)]
pub struct SubscribeEventsSessionParams<'a> {
    /// `types` template variable: comma-separated list of JMAP data-type
    /// names to subscribe to (`"Email,Mailbox"`), or `"*"` for all types
    /// the server supports. `None` expands to an empty string.
    pub types: Option<&'a str>,
    /// `closeafter` template variable (see [`CloseAfter`]). `None`
    /// expands to an empty string.
    pub close_after: Option<CloseAfter>,
    /// `ping` template variable: server-ping interval in seconds, or
    /// `0` to disable. `None` expands to an empty string.
    pub ping: Option<u32>,
    /// Optional `Last-Event-ID` header for resumption (RFC 8895 §9).
    pub last_event_id: Option<&'a str>,
}

/// Find the method response matching `call_id` in `resp` and deserialize its
/// arguments into `T`.
///
/// Returns [`ClientError::MethodNotFound`] if no invocation with the given
/// call_id exists. Returns [`ClientError::MethodError`] if any invocation
/// with the matching call_id is a JMAP `"error"` response (RFC 8620 §3.6.2),
/// or [`ClientError::Parse`] if that error object has no string `type` or a
/// malformed `description` or `limit`.
///
/// # Multiple invocations sharing a call_id
///
/// Per RFC 8620 §3.2, a single method call may produce multiple invocations
/// in the response — for example, `Foo/copy` with `onSuccessDestroyOriginal:
/// true` produces both a `Foo/copy` and an implicit `Foo/set` invocation,
/// both stamped with the same call_id (RFC 8620 §5.8 example, lines 3158–
/// 3180). This function handles that case by:
///
/// 1. **Errors take precedence.** If any invocation matching `call_id`
///    has method name `"error"`, this function returns that error. A
///    success response cannot mask a sibling error response with the same
///    call_id — silently returning the success while the server reported
///    failure would be data loss for the caller.
/// 2. Otherwise, the **first** non-error invocation matching `call_id`
///    is deserialized into `T`. In the §5.8 implicit-method case both
///    invocations are successes; the first is the primary response and
///    is what the caller wants.
///
/// # Contract: `call_id` is the only matcher (bd:JMAP-6r7c.23)
///
/// The method-name field of the matching invocation is **not** checked
/// against `T`. The function trusts the caller's choice of `call_id` to
/// identify the invocation and trusts the server's choice of method name
/// to be consistent with the request it was answering. Two consequences
/// callers should be aware of:
///
/// - **Wrong `T` may still parse.** If a caller asks for
///   `extract_response::<EmailGetResponse>(&resp, "r1")` but the matching
///   invocation is actually a `Mailbox/get` response, `serde_json::from_value`
///   will succeed on any structural overlap between the two shapes (both
///   carry `accountId`, an `ids`/`list` field, etc.) and return a
///   default-shaped `EmailGetResponse`. The function cannot detect this
///   because it has no view of what method name the caller expected.
/// - **Server ordering is trusted.** When multiple non-error invocations
///   share a `call_id`, the function returns the first one in
///   `resp.method_responses` order. RFC 8620 §5.8 implies the primary
///   response is first, but the spec does not normatively require it; a
///   non-conformant server that returns the implicit method first would
///   silently get the wrong invocation deserialized as `T`.
///
/// Callers that need method-name verification or want to disambiguate
/// among multiple non-error matches should iterate
/// `resp.method_responses` directly. The field is public and each
/// [`jmap_types::Invocation`] has public `name`, `arguments` and `call_id`
/// fields.
///
/// # Panics
///
/// This function does not catch panics from `T`'s [`serde::Deserialize`]
/// implementation. If a custom `T` type's `deserialize` impl panics — e.g.
/// because of an `.unwrap()` on a sub-field — the panic propagates through
/// `extract_response` to the caller's await point. The standard derived
/// `Deserialize` impls in the workspace type crates (`jmap-types`,
/// `jmap-mail-types`, etc.) do not panic; this caveat only affects
/// hand-rolled `Deserialize` impls outside the workspace
/// (bd:JMAP-6r7c.44).
///
/// This function is `pub` so extension crates (`jmap-chat-client`,
/// `jmap-mail-client`) can use it to extract typed results from a
/// [`jmap_types::JmapResponse`] without depending on internal details.
pub fn extract_response<T: serde::de::DeserializeOwned>(
    resp: &jmap_types::JmapResponse,
    call_id: &str,
) -> Result<T, ClientError> {
    // Two-pass scan: errors take precedence per §3.6.1, so look for an error
    // invocation first; otherwise return the first non-error invocation.
    if let Some(args) = find_error_args(resp, call_id) {
        let err = <jmap_types::JmapError as serde::Deserialize>::deserialize(args)
            .map_err(ClientError::from_parse)?;
        return Err(ClientError::MethodError(err));
    }

    // The early return above already handled the error case; every remaining
    // candidate is a success invocation, so we just take the first one.
    let inv = resp
        .method_responses
        .iter()
        .find(|inv| inv.call_id == call_id)
        .ok_or_else(|| ClientError::MethodNotFound(call_id.to_owned()))?;
    <T as serde::Deserialize>::deserialize(&inv.arguments).map_err(ClientError::from_parse)
}

/// Return the full JMAP `"error"` object for `call_id`, including any
/// properties beyond `type` and `description`.
///
/// RFC 8620 §3.6.2: "Other properties may be present with further
/// information". [`ClientError::MethodError`] carries the properties
/// [`jmap_types::JmapError`] models; call this with the same `resp` and
/// `call_id` after [`extract_response`] returns `MethodError` to read the
/// rest.
///
/// Uses the same lookup as [`extract_response`]: the first invocation
/// matching `call_id` whose method name is `"error"`. Returns `None` if
/// there is no such invocation or its arguments are not a JSON object.
pub fn method_error_details<'a>(
    resp: &'a jmap_types::JmapResponse,
    call_id: &str,
) -> Option<&'a serde_json::Map<String, serde_json::Value>> {
    find_error_args(resp, call_id)?.as_object()
}

/// Arguments of the first `"error"` invocation matching `call_id`.
fn find_error_args<'a>(
    resp: &'a jmap_types::JmapResponse,
    call_id: &str,
) -> Option<&'a serde_json::Value> {
    resp.method_responses
        .iter()
        .find(|inv| inv.call_id == call_id && inv.name == "error")
        .map(|inv| &inv.arguments)
}

/// Decode as much valid UTF-8 as possible from `raw` into `buf`, draining
/// consumed (or definitively-invalid) bytes from `raw`.
///
/// Three cases:
/// - `raw` is fully valid UTF-8: pushed entirely to `buf`, `raw` cleared.
/// - `raw` ends with an **incomplete** multi-byte sequence (`error_len == None`):
///   the valid prefix is pushed to `buf` and drained from `raw`; the incomplete
///   head bytes stay in `raw` for the next chunk to complete them.
/// - `raw` contains a **definitively invalid** byte sequence (`error_len == Some(n)`):
///   the valid prefix is pushed to `buf`; the valid prefix AND the `n` invalid
///   bytes are drained from `raw` so they do not accumulate.
///
/// The caller is responsible for capping `raw.len()` before calling this
/// function; unbounded growth from a hostile server that never completes a
/// sequence is prevented by the 1 MiB `SSE_BUF_SIZE_LIMIT` check in
/// `subscribe_events`.
fn decode_utf8_chunk(raw: &mut Vec<u8>, buf: &mut String) {
    match std::str::from_utf8(raw) {
        Ok(s) => {
            buf.push_str(s);
            raw.clear();
        }
        Err(e) => {
            let valid_up_to = e.valid_up_to();
            // valid_up_to is the documented prefix length that IS valid UTF-8
            // by construction (std::str::Utf8Error contract). The crate-root
            // #![forbid(unsafe_code)] rules out std::str::from_utf8_unchecked,
            // which would be the canonical zero-cost path for "I have a slice
            // I know is valid UTF-8 but the type system doesn't". With unsafe
            // forbidden, .expect() is the cheapest legal alternative; the
            // expect message tracks the soundness argument so a future
            // reviewer does not "fix" this with unwrap_or_default()
            // (which would silently drop valid UTF-8 prefix bytes on a
            // hypothetical impossible failure path) (bd:JMAP-6r7c.54).
            buf.push_str(
                std::str::from_utf8(&raw[..valid_up_to])
                    .expect("valid_up_to is a valid UTF-8 boundary"),
            );
            match e.error_len() {
                Some(n) => {
                    // Definitively invalid: drain the valid prefix AND the
                    // invalid bytes so they do not accumulate in raw.
                    let drain_end = (valid_up_to + n).min(raw.len());
                    raw.drain(..drain_end);
                }
                None => {
                    // Incomplete multi-byte sequence: drain only the valid
                    // prefix.  The incomplete head stays in raw until the next
                    // chunk arrives with the missing continuation bytes.
                    raw.drain(..valid_up_to);
                }
            }
        }
    }
}

/// Extract the URL scheme as a borrowed slice of `url`.
///
/// Returns the prefix before `"://"`. The returned slice is in the
/// *original* case of the input (callers must use [`str::eq_ignore_ascii_case`]
/// for the comparison per RFC 3986 §3.1, which says only schemes are
/// case-insensitive). Returns `None` if `"://"` is not present in `url`.
/// URL templates containing `{variable}` syntax are handled correctly
/// because the extraction is a prefix scan, not a full URL parse.
///
/// Returning a borrowed slice rather than a `to_ascii_lowercase()` String
/// avoids allocating on every scheme check; previously each call to
/// `url_scheme` produced a fresh String the size of the URL prefix
/// (bd:JMAP-6lsm.10).
fn url_scheme(url: &str) -> Option<&str> {
    url.split_once("://").map(|(scheme, _)| scheme)
}

/// `true` if `url`'s scheme prefix is `http` or `https` (case-insensitive,
/// RFC 3986 §3.1).
fn is_http_or_https(url: &str) -> bool {
    url_scheme(url)
        .is_some_and(|s| s.eq_ignore_ascii_case("http") || s.eq_ignore_ascii_case("https"))
}

/// Parse and validate a JMAP base URL.
///
/// The input must be:
/// - non-empty
/// - syntactically valid (parses with [`url::Url::parse`])
/// - http or https scheme (case-insensitive per RFC 3986 §3.1)
/// - origin-only: no explicit path component, no query string, no fragment.
///   "Origin" here means scheme + host + optional port — the form a JMAP
///   server is identified by. Trailing slashes are accepted (the URL
///   parser normalizes them away).
///
/// Examples that PASS: `https://jmap.example.com`,
/// `https://jmap.example.com/`, `https://10.0.0.1:8008`,
/// `http://localhost:8080`.
///
/// Examples that FAIL: `""`, `https://example.com/api`,
/// `https://example.com?query=1`, `https://example.com#fragment`,
/// `ftp://example.com`.
///
/// Extracted from [`JmapClient::new`] so the constructor reads as
/// 'parse-don't-validate' rather than a six-step inline procedure
/// (bd:JMAP-6lsm.26).
fn parse_base_url(base_url: &str) -> Result<url::Url, ClientError> {
    if base_url.is_empty() {
        return Err(ClientError::InvalidArgument(
            "base_url may not be empty".into(),
        ));
    }
    let parsed = url::Url::parse(base_url)
        .map_err(|e| ClientError::InvalidArgument(format!("base_url is not a valid URL: {e}")))?;
    // Reject RFC 3986 user-info (`user:password@host`) before any further
    // validation. The url crate's Display impl echoes user-info verbatim
    // (lossless-round-trip, NOT safe-display — RFC 3986 §7.5 warns
    // against the latter), so an unrejected user-info value would surface
    // through every subsequent error message, every reqwest::Error's
    // url() field, and any tracing instrumentation that captures the
    // base URL or a derived ClientError. JMAP authenticates via the
    // Authorization header (AuthProvider trait), not URL user-info; the
    // user-info component has no legitimate use here (bd:JMAP-6r7c.58).
    //
    // The error message DOES NOT echo `base_url` back — doing so would
    // route the password into the error chain we are trying to keep it
    // out of.
    if !parsed.username().is_empty() || parsed.password().is_some() {
        return Err(ClientError::InvalidArgument(
            "base_url must not contain user-info (`user:password@host`); pass credentials via \
             AuthProvider instead"
                .into(),
        ));
    }
    let scheme = parsed.scheme();
    if scheme != "http" && scheme != "https" {
        return Err(ClientError::InvalidArgument(format!(
            "base_url scheme must be http or https, got: {scheme:?}"
        )));
    }
    let path = parsed.path();
    // url::Url::path() returns "/" for root-only URLs (no path segments);
    // any value other than "/" means the URL contains an explicit path component.
    if path != "/" {
        return Err(ClientError::InvalidArgument(format!(
            "base_url must not have a path component, got: {path:?}"
        )));
    }
    if parsed.query().is_some() {
        return Err(ClientError::InvalidArgument(
            "base_url must not have a query string".into(),
        ));
    }
    if parsed.fragment().is_some() {
        return Err(ClientError::InvalidArgument(
            "base_url must not have a fragment".into(),
        ));
    }
    Ok(parsed)
}

/// Validate that `url` uses an http or https scheme.
///
/// Called at the top of each public method that accepts a URL parameter
/// (`call`, `subscribe_events`, `upload_blob`, `download_blob`).  This is a
/// defense-in-depth check: the primary protection is
/// [`validate_session_url_schemes`] which rejects bad URLs in the Session
/// document at fetch time.  This check makes each individual call site
/// self-defending against accidentally passing a non-http URL (e.g. from a
/// test fixture or a misused API).
///
/// Returns [`ClientError::InvalidArgument`] if the scheme is not http/https.
pub(crate) fn require_http_url(url: &str) -> Result<(), ClientError> {
    if !is_http_or_https(url) {
        return Err(ClientError::InvalidArgument(format!(
            "URL must have http or https scheme, got: {url:?}"
        )));
    }
    Ok(())
}

/// Validate the *schemes only* of the four session URL fields.
///
/// Three of the four (`upload_url`, `download_url`, `event_source_url`) are
/// RFC 6570 URI templates carrying `{accountId}`, `{blobId}`, `{types}`,
/// etc. They cannot be fully parsed as URLs without first being expanded
/// with the relevant variables, but the scheme prefix is always concrete
/// (templates put the scheme on the left of `://`). This function does the
/// minimal check that the scheme prefix is `http://` or `https://` — it
/// does NOT verify that the templates carry the required variables, that
/// the host is well-formed, or that the path is reachable.
///
/// Renamed from `validate_session_urls` for accuracy (bd:JMAP-6lsm.23):
/// the name implied stronger validation than the function actually
/// performs.
fn validate_session_url_schemes(session: &Session) -> Result<(), ClientError> {
    // Mixed types (JmapUrl, JmapUrlTemplate) deliberately — both expose
    // .as_str() identically. Iterating &str values keeps the validation
    // loop oblivious to the typed-URL distinction (bd:JMAP-6r7c.40),
    // which is the right scope for a scheme check.
    for url in [
        session.api_url.as_str(),
        session.upload_url.as_str(),
        session.download_url.as_str(),
        session.event_source_url.as_str(),
    ] {
        if !is_http_or_https(url) {
            return Err(ClientError::InvalidSession(format!(
                "session URL has non-http/https scheme: {url:?}"
            )));
        }
    }
    Ok(())
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::SseReconnectPolicy;
    use super::decode_utf8_chunk;
    use std::time::Duration;

    /// Oracle: the `SseReconnectPolicy` / `subscribe_events_reconnecting`
    /// contract — retry `n` waits a uniform delay in `[d/2, d]` where
    /// `d = min(initial_delay * 2^(n-1), max_delay)`. With 100 ms / 1 s the
    /// expected `d` per retry is 100, 200, 400, 800, then capped at 1000
    /// (also for an absurd retry count, which must not overflow).
    #[test]
    fn reconnect_delay_doubles_caps_and_jitters() {
        let policy = SseReconnectPolicy {
            initial_delay: Duration::from_millis(100),
            max_delay: Duration::from_secs(1),
            ..SseReconnectPolicy::default()
        };
        for (failures, d_ms) in [
            (1, 100),
            (2, 200),
            (3, 400),
            (4, 800),
            (5, 1000),
            (500, 1000),
        ] {
            let d = Duration::from_millis(d_ms);
            let mut seen = std::collections::HashSet::new();
            for _ in 0..64 {
                let got = policy.delay_for(failures);
                assert!(
                    got >= d / 2 && got <= d,
                    "retry {failures}: {got:?} outside [{:?}, {d:?}]",
                    d / 2
                );
                seen.insert(got);
            }
            assert!(seen.len() > 1, "retry {failures}: delay must be jittered");
        }
    }

    /// Oracle: all-ASCII bytes pushed to buf; raw cleared.
    #[test]
    fn decode_utf8_chunk_all_ascii() {
        let mut raw = b"hello".to_vec();
        let mut buf = String::new();
        decode_utf8_chunk(&mut raw, &mut buf);
        assert_eq!(buf, "hello");
        assert!(raw.is_empty());
    }

    /// Oracle: complete multi-byte codepoint (U+00E9 café = 0xC3 0xA9) pushed fully.
    #[test]
    fn decode_utf8_chunk_complete_multibyte() {
        let mut raw = "café".as_bytes().to_vec();
        let mut buf = String::new();
        decode_utf8_chunk(&mut raw, &mut buf);
        assert_eq!(buf, "café");
        assert!(raw.is_empty());
    }

    /// Oracle: first byte of a 2-byte sequence (U+00E9 = 0xC3 0xA9) arrives alone.
    /// error_len == None (incomplete) — the byte must be RETAINED in raw, not dropped.
    /// Regression test for the bug where valid_up_to.max(1) discarded this byte.
    #[test]
    fn decode_utf8_chunk_incomplete_head_retained() {
        let mut raw = vec![0xC3u8]; // first byte of 2-byte sequence
        let mut buf = String::new();
        decode_utf8_chunk(&mut raw, &mut buf);
        assert_eq!(buf, "", "no complete codepoints to push");
        assert_eq!(raw, vec![0xC3u8], "incomplete head must stay in raw");
    }

    /// Oracle: valid ASCII prefix then first byte of a 2-byte sequence.
    /// Valid prefix goes to buf; incomplete head stays in raw.
    #[test]
    fn decode_utf8_chunk_prefix_then_incomplete_head() {
        let mut raw = vec![b'a', b'b', 0xC3u8];
        let mut buf = String::new();
        decode_utf8_chunk(&mut raw, &mut buf);
        assert_eq!(buf, "ab");
        assert_eq!(raw, vec![0xC3u8], "incomplete head must stay in raw");
    }

    /// Oracle: two-call simulation — incomplete sequence completed on second call.
    /// This is the exact HTTP chunk-split scenario the fix targets.
    #[test]
    fn decode_utf8_chunk_split_sequence_completed() {
        // Chunk 1: only the first byte of U+00E9 (é = 0xC3 0xA9)
        let mut raw = vec![0xC3u8];
        let mut buf = String::new();
        decode_utf8_chunk(&mut raw, &mut buf);
        assert_eq!(raw, vec![0xC3u8], "incomplete head retained after chunk 1");

        // Chunk 2: completion byte
        raw.push(0xA9u8);
        decode_utf8_chunk(&mut raw, &mut buf);
        assert_eq!(buf, "é", "character fully decoded after chunk 2");
        assert!(raw.is_empty());
    }

    /// Oracle: definitively invalid byte (0xFF is never valid in UTF-8) is drained.
    #[test]
    fn decode_utf8_chunk_invalid_byte_drained() {
        let mut raw = vec![0xFFu8];
        let mut buf = String::new();
        decode_utf8_chunk(&mut raw, &mut buf);
        assert_eq!(buf, "");
        assert!(raw.is_empty(), "definitively invalid byte must be drained");
    }

    /// Oracle: valid prefix then definitively invalid byte — prefix pushed and both drained.
    #[test]
    fn decode_utf8_chunk_prefix_then_invalid_drained() {
        let mut raw = vec![b'a', b'b', 0xFFu8];
        let mut buf = String::new();
        decode_utf8_chunk(&mut raw, &mut buf);
        assert_eq!(buf, "ab");
        assert!(raw.is_empty(), "prefix and invalid byte must be drained");
    }

    /// Compile-time assertion that [`JmapClient`] is `Send + Sync + Clone`
    /// (bd:JMAP-6r7c.25). A future refactor that adds a non-`Sync` field
    /// (e.g. `Rc<T>`, `RefCell<T>`, `Cell<T>`) would fail this test in CI
    /// before any downstream consumer that shares a `JmapClient` across
    /// tokio tasks is exposed. The body is fenceposts: the function bodies
    /// of `assert_send`, `assert_sync`, and `assert_clone` never run; what
    /// matters is that the trait-bound monomorphization succeeds.
    #[test]
    fn jmap_client_is_send_sync_clone() {
        fn assert_send<T: Send>() {}
        fn assert_sync<T: Sync>() {}
        fn assert_clone<T: Clone>() {}
        assert_send::<super::JmapClient>();
        assert_sync::<super::JmapClient>();
        assert_clone::<super::JmapClient>();
    }
}