io_jmap/

client.rs

//! Standard, blocking JMAP client.
//!
//! Wraps a single boxed stream plus the bearer token and discovered
//! [`JmapSession`], with one method per common coroutine. [`JmapClientStd::new`]
//! takes a pre-connected stream; with one of the TLS features enabled,
//! [`JmapClientStd::connect`] handles `https://` URLs end-to-end.
//!
//! Run [`JmapClientStd::session_get`] once after construction to populate the
//! session; subsequent calls resolve `accountId` and `apiUrl` from it.
//!
//! # Example
//!
//! ```rust,no_run
//! use io_jmap::client::JmapClientStd;
//! use pimalaya_stream::tls::Tls;
//! use secrecy::SecretString;
//! use url::Url;
//!
//! let url: Url = "https://api.example.com/jmap/session/".parse().unwrap();
//! let auth = SecretString::from("Bearer xyz");
//!
//! let mut client = JmapClientStd::connect(&url, &Tls::default(), auth).unwrap();
//! let session = client.session_get(&url).unwrap();
//!
//! println!("logged in as {}", session.username);
//! ```

#[cfg(any(
    feature = "rustls-aws",
    feature = "rustls-ring",
    feature = "native-tls"
))]
use core::time::Duration;
use core::{any::Any, fmt};

#[cfg(any(
    feature = "rustls-aws",
    feature = "rustls-ring",
    feature = "native-tls"
))]
use alloc::string::ToString;
use alloc::{boxed::Box, collections::BTreeMap, string::String, vec, vec::Vec};

use std::io::{self, Read, Write};

#[cfg(any(
    feature = "rustls-aws",
    feature = "rustls-ring",
    feature = "native-tls"
))]
use pimalaya_stream::{std::stream::StreamStd, tls::Tls};
use secrecy::SecretString;
use thiserror::Error;
use url::Url;

use crate::{
    coroutine::*,
    rfc8620::{
        blob_download::*,
        blob_upload::*,
        changes::JmapChangesOutput,
        coroutine::JmapRedirectYield,
        push_subscription::{get::*, set::*},
        request::{JmapRequest, JmapResponse},
        send::*,
        session::JmapSession,
        session_get::*,
    },
    rfc8621::{
        email::{changes::*, copy::*, get::*, import::*, parse::*, query::*, set::*},
        email_submission::{cancel::*, get::*, query::*, set::*},
        identity::{get::*, set::*},
        mailbox::{changes::*, get::*, query::*, set::*},
        thread::{changes::*, get::*},
        vacation_response::{get::*, set::*, *},
    },
    rfc9610::{
        address_book::{changes::*, get::*, set::*},
        contact_card::{changes::*, copy::*, get::*, query::*, set::*},
    },
};

/// Errors returned by [`JmapClientStd`].
#[derive(Debug, Error)]
pub enum JmapClientStdError {
    /// The raw send coroutine failed.
    #[error(transparent)]
    Send(#[from] JmapSendError),
    /// The session fetch coroutine failed.
    #[error(transparent)]
    SessionGet(#[from] JmapSessionGetError),
    /// The blob upload coroutine failed.
    #[error(transparent)]
    BlobUpload(#[from] JmapBlobUploadError),
    /// The blob download coroutine failed.
    #[error(transparent)]
    BlobDownload(#[from] JmapBlobDownloadError),
    /// The `PushSubscription/get` coroutine failed.
    #[error(transparent)]
    PushSubscriptionGet(#[from] JmapPushSubscriptionGetError),
    /// The `PushSubscription/set` coroutine failed.
    #[error(transparent)]
    PushSubscriptionSet(#[from] JmapPushSubscriptionSetError),
    /// The `Mailbox/get` coroutine failed.
    #[error(transparent)]
    MailboxGet(#[from] JmapMailboxGetError),
    /// The `Mailbox/query` coroutine failed.
    #[error(transparent)]
    MailboxQuery(#[from] JmapMailboxQueryError),
    /// The `Mailbox/set` coroutine failed.
    #[error(transparent)]
    MailboxSet(#[from] JmapMailboxSetError),
    /// The `Mailbox/changes` coroutine failed.
    #[error(transparent)]
    MailboxChanges(#[from] JmapMailboxChangesError),
    /// The `Email/get` coroutine failed.
    #[error(transparent)]
    EmailGet(#[from] JmapEmailGetError),
    /// The `Email/query` coroutine failed.
    #[error(transparent)]
    EmailQuery(#[from] JmapEmailQueryError),
    /// The `Email/set` coroutine failed.
    #[error(transparent)]
    EmailSet(#[from] JmapEmailSetError),
    /// The `Email/changes` coroutine failed.
    #[error(transparent)]
    EmailChanges(#[from] JmapEmailChangesError),
    /// The `Email/copy` coroutine failed.
    #[error(transparent)]
    EmailCopy(#[from] JmapEmailCopyError),
    /// The `Email/import` coroutine failed.
    #[error(transparent)]
    EmailImport(#[from] JmapEmailImportError),
    /// The `Email/parse` coroutine failed.
    #[error(transparent)]
    EmailParse(#[from] JmapEmailParseError),
    /// The `Thread/get` coroutine failed.
    #[error(transparent)]
    ThreadGet(#[from] JmapThreadGetError),
    /// The `Thread/changes` coroutine failed.
    #[error(transparent)]
    ThreadChanges(#[from] JmapThreadChangesError),
    /// The `Identity/get` coroutine failed.
    #[error(transparent)]
    IdentityGet(#[from] JmapIdentityGetError),
    /// The `Identity/set` coroutine failed.
    #[error(transparent)]
    IdentitySet(#[from] JmapIdentitySetError),
    /// The `EmailSubmission/get` coroutine failed.
    #[error(transparent)]
    EmailSubmissionGet(#[from] JmapEmailSubmissionGetError),
    /// The `EmailSubmission/query` coroutine failed.
    #[error(transparent)]
    EmailSubmissionQuery(#[from] JmapEmailSubmissionQueryError),
    /// The `EmailSubmission/set` coroutine failed.
    #[error(transparent)]
    EmailSubmissionSet(#[from] JmapEmailSubmissionSetError),
    /// The `EmailSubmission/set` cancel coroutine failed.
    #[error(transparent)]
    EmailSubmissionCancel(#[from] JmapEmailSubmissionCancelError),
    /// The `VacationResponse/get` coroutine failed.
    #[error(transparent)]
    VacationResponseGet(#[from] JmapVacationResponseGetError),
    /// The `VacationResponse/set` coroutine failed.
    #[error(transparent)]
    VacationResponseSet(#[from] JmapVacationResponseSetError),
    /// The `AddressBook/get` coroutine failed.
    #[error(transparent)]
    AddressBookGet(#[from] JmapAddressBookGetError),
    /// The `AddressBook/set` coroutine failed.
    #[error(transparent)]
    AddressBookSet(#[from] JmapAddressBookSetError),
    /// The `AddressBook/changes` coroutine failed.
    #[error(transparent)]
    AddressBookChanges(#[from] JmapAddressBookChangesError),
    /// The `ContactCard/get` coroutine failed.
    #[error(transparent)]
    ContactCardGet(#[from] JmapContactCardGetError),
    /// The `ContactCard/query` coroutine failed.
    #[error(transparent)]
    ContactCardQuery(#[from] JmapContactCardQueryError),
    /// The `ContactCard/set` coroutine failed.
    #[error(transparent)]
    ContactCardSet(#[from] JmapContactCardSetError),
    /// The `ContactCard/changes` coroutine failed.
    #[error(transparent)]
    ContactCardChanges(#[from] JmapContactCardChangesError),
    /// The `ContactCard/copy` coroutine failed.
    #[error(transparent)]
    ContactCardCopy(#[from] JmapContactCardCopyError),
    /// The underlying stream failed to read or write.
    #[error(transparent)]
    Io(#[from] io::Error),
    /// The TCP connection or the TLS negotiation failed.
    #[cfg(any(
        feature = "rustls-aws",
        feature = "rustls-ring",
        feature = "native-tls"
    ))]
    #[error(transparent)]
    Tls(#[from] anyhow::Error),
    /// The URL to connect to carries no host.
    #[cfg(any(
        feature = "rustls-aws",
        feature = "rustls-ring",
        feature = "native-tls"
    ))]
    #[error("JMAP URL `{0}` has no host")]
    UrlMissingHost(String),
    /// The URL to connect to carries a scheme the client cannot open.
    #[cfg(any(
        feature = "rustls-aws",
        feature = "rustls-ring",
        feature = "native-tls"
    ))]
    #[error(
        "JMAP URL `{url}` has unsupported scheme `{scheme}` (expected `http`, `https`, `jmap` or `jmaps`)"
    )]
    UrlUnsupportedScheme {
        /// The URL the client was asked to open.
        url: String,
        /// The unsupported scheme of that URL.
        scheme: String,
    },
    /// The server answered with a redirect during a non-redirectable
    /// operation.
    #[error("JMAP server redirected to `{0}` during a non-redirectable operation")]
    UnexpectedRedirect(Url),
    /// A method requiring the session ran before [`JmapClientStd::session_get`].
    #[error("JMAP client missing session; call `session_get` first")]
    MissingSession,
}

const READ_BUFFER_SIZE: usize = 16 * 1024;

/// Std-blocking JMAP client wrapping a single boxed stream.
pub struct JmapClientStd {
    /// The wrapped stream the coroutines read from and write to.
    pub stream: Box<dyn JmapStream>,
    /// The pre-formatted HTTP `Authorization` header value.
    pub http_auth: SecretString,
    /// The session discovered by [`Self::session_get`], if any.
    pub session: Option<JmapSession>,
}

impl JmapClientStd {
    /// Builds a client around `stream`. The caller is responsible for opening
    /// the connection (TCP, TLS handshake if needed) and for the bearer token /
    /// authorization header value.
    pub fn new<S: Read + Write + Send + 'static>(stream: S, http_auth: SecretString) -> Self {
        Self {
            stream: Box::new(stream),
            http_auth,
            session: None,
        }
    }

    /// Default ALPN list for JMAP TLS handshakes: `["http/1.1"]` (JMAP rides
    /// on HTTP/1.1). Exposed so config-based callers can share one source of
    /// truth.
    pub fn default_alpn() -> Vec<String> {
        vec![String::from("http/1.1")]
    }

    /// Resumes any standard-shape coroutine (`Yield = JmapYield`) against the
    /// wrapped stream until it terminates.
    ///
    /// Redirect-aware coroutines ([`JmapSessionGet`], [`JmapBlobUpload`],
    /// [`JmapBlobDownload`]) and the streaming
    /// [`JmapEventSource`](crate::rfc8620::event_source::subscribe::JmapEventSource)
    /// have their own per-method loops.
    pub fn run<C, T, E>(&mut self, mut coroutine: C) -> Result<T, JmapClientStdError>
    where
        C: JmapCoroutine<Yield = JmapYield, Return = Result<T, E>>,
        JmapClientStdError: From<E>,
    {
        let mut buf = [0u8; READ_BUFFER_SIZE];
        let mut arg: Option<&[u8]> = None;

        loop {
            match coroutine.resume(arg.take()) {
                JmapCoroutineState::Complete(Ok(out)) => return Ok(out),
                JmapCoroutineState::Complete(Err(err)) => return Err(err.into()),
                JmapCoroutineState::Yielded(JmapYield::WantsRead) => {
                    let n = self.stream.read(&mut buf)?;
                    arg = Some(&buf[..n]);
                }
                JmapCoroutineState::Yielded(JmapYield::WantsWrite(bytes)) => {
                    self.stream.write_all(&bytes)?;
                    arg = None;
                }
            }
        }
    }

    /// Builds a client from a pre-connected stream and an already-discovered
    /// [`JmapSession`]. Skips [`Self::session_get`].
    pub fn from_parts<S: Read + Write + Send + 'static>(
        stream: S,
        http_auth: SecretString,
        session: JmapSession,
    ) -> Self {
        Self {
            stream: Box::new(stream),
            http_auth,
            session: Some(session),
        }
    }

    /// Connects to `url`, doing a TLS handshake for `https` / `jmaps` (plain
    /// TCP for `http` / `jmap`). ALPN comes from `tls.rustls.alpn` (see
    /// [`Self::default_alpn`]); empty vec skips ALPN.
    #[cfg(any(
        feature = "rustls-aws",
        feature = "rustls-ring",
        feature = "native-tls"
    ))]
    pub fn connect(
        url: &Url,
        tls: &Tls,
        http_auth: SecretString,
    ) -> Result<Self, JmapClientStdError> {
        let host = url
            .host_str()
            .ok_or_else(|| JmapClientStdError::UrlMissingHost(url.to_string()))?;

        let stream = match url.scheme() {
            "http" | "jmap" => StreamStd::connect_tcp(host, url.port().unwrap_or(80))?,
            "https" | "jmaps" => StreamStd::connect_tls(host, url.port().unwrap_or(443), tls)?,
            scheme => {
                return Err(JmapClientStdError::UrlUnsupportedScheme {
                    url: url.to_string(),
                    scheme: scheme.to_string(),
                });
            }
        };

        // NOTE: 5s per-read (not per-operation) timeout so the watch loop
        // polls its shutdown atomic between SSE push frames; large JMAP
        // responses keep working as long as TCP packets keep arriving.
        stream.set_read_timeout(Some(Duration::from_secs(5)))?;

        Ok(Self {
            stream: Box::new(stream),
            http_auth,
            session: None,
        })
    }

    /// Replaces the underlying stream; useful when `apiUrl`, `uploadUrl` or
    /// `downloadUrl` resolves to a different authority than the first
    /// connection target, or after a redirect.
    pub fn set_stream<S: Read + Write + Send + 'static>(&mut self, stream: S) {
        self.stream = Box::new(stream);
    }

    /// Returns the cached session, if [`Self::session_get`] has run.
    pub fn session(&self) -> Option<&JmapSession> {
        self.session.as_ref()
    }

    /// Returns the pre-formatted HTTP `Authorization` header value.
    pub fn http_auth(&self) -> &SecretString {
        &self.http_auth
    }

    fn session_or_err(&self) -> Result<&JmapSession, JmapClientStdError> {
        self.session
            .as_ref()
            .ok_or(JmapClientStdError::MissingSession)
    }

    /// Runs [`JmapSessionGet`] and caches the discovered session.
    ///
    /// `url` is either a base URL for `/.well-known/jmap` discovery or a
    /// direct session endpoint. A 3xx response terminates with
    /// [`JmapClientStdError::UnexpectedRedirect`].
    pub fn session_get(&mut self, url: &Url) -> Result<&JmapSession, JmapClientStdError> {
        let mut coroutine = JmapSessionGet::new(&self.http_auth, url);
        let mut buf = [0u8; READ_BUFFER_SIZE];
        let mut arg: Option<&[u8]> = None;

        loop {
            match coroutine.resume(arg.take()) {
                JmapCoroutineState::Complete(Ok(JmapSessionGetOutput { session, .. })) => {
                    self.session = Some(session);
                    return Ok(self.session.as_ref().unwrap());
                }
                JmapCoroutineState::Complete(Err(err)) => return Err(err.into()),
                JmapCoroutineState::Yielded(JmapRedirectYield::WantsRead) => {
                    let n = self.stream.read(&mut buf)?;
                    arg = Some(&buf[..n]);
                }
                JmapCoroutineState::Yielded(JmapRedirectYield::WantsWrite(bytes)) => {
                    self.stream.write_all(&bytes)?;
                    arg = None;
                }
                JmapCoroutineState::Yielded(JmapRedirectYield::WantsRedirect { url, .. }) => {
                    return Err(JmapClientStdError::UnexpectedRedirect(url));
                }
            }
        }
    }

    /// Sends a raw JMAP request and returns the raw [`JmapResponse`]. Useful
    /// for passthrough CLIs and ad-hoc requests with custom `using`
    /// capabilities.
    pub fn send_raw(&mut self, request: JmapRequest) -> Result<JmapResponse, JmapClientStdError> {
        let session = self.session_or_err()?;
        let coroutine = JmapSend::new(&self.http_auth, &session.api_url, request)?;
        let out = self.run(coroutine)?;
        Ok(out.response)
    }

    /// Uploads a blob to `upload_url` (RFC 8620 §6.1). The caller must resolve
    /// the session's `uploadUrl` template (e.g. substitute `{accountId}`).
    /// A 3xx response terminates with [`JmapClientStdError::UnexpectedRedirect`].
    pub fn blob_upload(
        &mut self,
        upload_url: &Url,
        content_type: &str,
        data: Vec<u8>,
    ) -> Result<JmapBlobUploadOutput, JmapClientStdError> {
        let mut coroutine = JmapBlobUpload::new(&self.http_auth, upload_url, content_type, data);
        let mut buf = [0u8; READ_BUFFER_SIZE];
        let mut arg: Option<&[u8]> = None;

        loop {
            match coroutine.resume(arg.take()) {
                JmapCoroutineState::Complete(Ok(out)) => return Ok(out),
                JmapCoroutineState::Complete(Err(err)) => return Err(err.into()),
                JmapCoroutineState::Yielded(JmapRedirectYield::WantsRead) => {
                    let n = self.stream.read(&mut buf)?;
                    arg = Some(&buf[..n]);
                }
                JmapCoroutineState::Yielded(JmapRedirectYield::WantsWrite(bytes)) => {
                    self.stream.write_all(&bytes)?;
                    arg = None;
                }
                JmapCoroutineState::Yielded(JmapRedirectYield::WantsRedirect { url, .. }) => {
                    return Err(JmapClientStdError::UnexpectedRedirect(url));
                }
            }
        }
    }

    /// Downloads a blob from `download_url` (RFC 8620 §6.2). The caller must
    /// resolve the session's `downloadUrl` template. A 3xx response terminates
    /// with [`JmapClientStdError::UnexpectedRedirect`].
    pub fn blob_download(&mut self, download_url: &Url) -> Result<Vec<u8>, JmapClientStdError> {
        let mut coroutine = JmapBlobDownload::new(&self.http_auth, download_url);
        let mut buf = [0u8; READ_BUFFER_SIZE];
        let mut arg: Option<&[u8]> = None;

        loop {
            match coroutine.resume(arg.take()) {
                JmapCoroutineState::Complete(Ok(out)) => return Ok(out.data),
                JmapCoroutineState::Complete(Err(err)) => return Err(err.into()),
                JmapCoroutineState::Yielded(JmapRedirectYield::WantsRead) => {
                    let n = self.stream.read(&mut buf)?;
                    arg = Some(&buf[..n]);
                }
                JmapCoroutineState::Yielded(JmapRedirectYield::WantsWrite(bytes)) => {
                    self.stream.write_all(&bytes)?;
                    arg = None;
                }
                JmapCoroutineState::Yielded(JmapRedirectYield::WantsRedirect { url, .. }) => {
                    return Err(JmapClientStdError::UnexpectedRedirect(url));
                }
            }
        }
    }

    /// Runs [`JmapPushSubscriptionGet`] (`PushSubscription/get`).
    pub fn push_subscription_get(
        &mut self,
        opts: JmapPushSubscriptionGetOptions,
    ) -> Result<JmapPushSubscriptionGetOutput, JmapClientStdError> {
        let coroutine =
            JmapPushSubscriptionGet::new(self.session_or_err()?, &self.http_auth, opts)?;
        self.run(coroutine)
    }

    /// Runs [`JmapPushSubscriptionSet`] (`PushSubscription/set`).
    pub fn push_subscription_set(
        &mut self,
        args: JmapPushSubscriptionSetArgs,
    ) -> Result<JmapPushSubscriptionSetOutput, JmapClientStdError> {
        let coroutine =
            JmapPushSubscriptionSet::new(self.session_or_err()?, &self.http_auth, args)?;
        self.run(coroutine)
    }

    /// Runs [`JmapMailboxGet`] (`Mailbox/get`).
    pub fn mailbox_get(
        &mut self,
        opts: JmapMailboxGetOptions,
    ) -> Result<JmapMailboxGetOutput, JmapClientStdError> {
        let coroutine = JmapMailboxGet::new(self.session_or_err()?, &self.http_auth, opts)?;
        self.run(coroutine)
    }

    /// Runs [`JmapMailboxQuery`] (batched `Mailbox/query` +
    /// `Mailbox/get`).
    pub fn mailbox_query(
        &mut self,
        opts: JmapMailboxQueryOptions,
    ) -> Result<JmapMailboxQueryOutput, JmapClientStdError> {
        let coroutine = JmapMailboxQuery::new(self.session_or_err()?, &self.http_auth, opts)?;
        self.run(coroutine)
    }

    /// Runs [`JmapMailboxSet`] (`Mailbox/set`).
    pub fn mailbox_set(
        &mut self,
        args: JmapMailboxSetArgs,
    ) -> Result<JmapMailboxSetOutput, JmapClientStdError> {
        let coroutine = JmapMailboxSet::new(self.session_or_err()?, &self.http_auth, args)?;
        self.run(coroutine)
    }

    /// Runs [`JmapMailboxChanges`] (`Mailbox/changes`).
    pub fn mailbox_changes(
        &mut self,
        since_state: impl Into<String>,
        opts: JmapMailboxChangesOptions,
    ) -> Result<JmapChangesOutput, JmapClientStdError> {
        let coroutine =
            JmapMailboxChanges::new(self.session_or_err()?, &self.http_auth, since_state, opts)?;
        self.run(coroutine)
    }

    /// Runs [`JmapEmailGet`] (`Email/get`).
    pub fn email_get(
        &mut self,
        ids: Vec<String>,
        opts: JmapEmailGetOptions,
    ) -> Result<JmapEmailGetOutput, JmapClientStdError> {
        let coroutine = JmapEmailGet::new(self.session_or_err()?, &self.http_auth, ids, opts)?;
        self.run(coroutine)
    }

    /// Runs [`JmapEmailQuery`] (batched `Email/query` + `Email/get`).
    pub fn email_query(
        &mut self,
        opts: JmapEmailQueryOptions,
    ) -> Result<JmapEmailQueryOutput, JmapClientStdError> {
        let coroutine = JmapEmailQuery::new(self.session_or_err()?, &self.http_auth, opts)?;
        self.run(coroutine)
    }

    /// Runs [`JmapEmailSet`] (`Email/set`).
    pub fn email_set(
        &mut self,
        args: JmapEmailSetArgs,
    ) -> Result<JmapEmailSetOutput, JmapClientStdError> {
        let coroutine = JmapEmailSet::new(self.session_or_err()?, &self.http_auth, args)?;
        self.run(coroutine)
    }

    /// Runs [`JmapEmailChanges`] (`Email/changes`).
    pub fn email_changes(
        &mut self,
        since_state: impl Into<String>,
        opts: JmapEmailChangesOptions,
    ) -> Result<JmapChangesOutput, JmapClientStdError> {
        let coroutine =
            JmapEmailChanges::new(self.session_or_err()?, &self.http_auth, since_state, opts)?;
        self.run(coroutine)
    }

    /// Runs [`JmapEmailCopy`] (`Email/copy`).
    pub fn email_copy(
        &mut self,
        from_account_id: impl Into<String>,
        emails: BTreeMap<String, JmapEmailCopyArgs>,
    ) -> Result<JmapEmailCopyOutput, JmapClientStdError> {
        let coroutine = JmapEmailCopy::new(
            self.session_or_err()?,
            &self.http_auth,
            from_account_id,
            emails,
        )?;
        self.run(coroutine)
    }

    /// Runs [`JmapEmailImport`] (`Email/import`).
    pub fn email_import(
        &mut self,
        emails: BTreeMap<String, JmapEmailImportArgs>,
    ) -> Result<JmapEmailImportOutput, JmapClientStdError> {
        let coroutine = JmapEmailImport::new(self.session_or_err()?, &self.http_auth, emails)?;
        self.run(coroutine)
    }

    /// Runs [`JmapEmailParse`] (`Email/parse`).
    pub fn email_parse(
        &mut self,
        blob_ids: Vec<String>,
        opts: JmapEmailParseOptions,
    ) -> Result<JmapEmailParseOutput, JmapClientStdError> {
        let coroutine =
            JmapEmailParse::new(self.session_or_err()?, &self.http_auth, blob_ids, opts)?;
        self.run(coroutine)
    }

    /// Runs [`JmapThreadGet`] (`Thread/get`).
    pub fn thread_get(
        &mut self,
        ids: Vec<String>,
    ) -> Result<JmapThreadGetOutput, JmapClientStdError> {
        let coroutine = JmapThreadGet::new(self.session_or_err()?, &self.http_auth, ids)?;
        self.run(coroutine)
    }

    /// Runs [`JmapThreadChanges`] (`Thread/changes`).
    pub fn thread_changes(
        &mut self,
        since_state: impl Into<String>,
        opts: JmapThreadChangesOptions,
    ) -> Result<JmapChangesOutput, JmapClientStdError> {
        let coroutine =
            JmapThreadChanges::new(self.session_or_err()?, &self.http_auth, since_state, opts)?;
        self.run(coroutine)
    }

    /// Runs [`JmapIdentityGet`] (`Identity/get`).
    pub fn identity_get(
        &mut self,
        opts: JmapIdentityGetOptions,
    ) -> Result<JmapIdentityGetOutput, JmapClientStdError> {
        let coroutine = JmapIdentityGet::new(self.session_or_err()?, &self.http_auth, opts)?;
        self.run(coroutine)
    }

    /// Runs [`JmapIdentitySet`] (`Identity/set`).
    pub fn identity_set(
        &mut self,
        args: JmapIdentitySetArgs,
    ) -> Result<JmapIdentitySetOutput, JmapClientStdError> {
        let coroutine = JmapIdentitySet::new(self.session_or_err()?, &self.http_auth, args)?;
        self.run(coroutine)
    }

    /// Runs [`JmapEmailSubmissionGet`] (`EmailSubmission/get`).
    pub fn email_submission_get(
        &mut self,
        opts: JmapEmailSubmissionGetOptions,
    ) -> Result<JmapEmailSubmissionGetOutput, JmapClientStdError> {
        let coroutine = JmapEmailSubmissionGet::new(self.session_or_err()?, &self.http_auth, opts)?;
        self.run(coroutine)
    }

    /// Runs [`JmapEmailSubmissionQuery`] (batched
    /// `EmailSubmission/query` + `EmailSubmission/get`).
    pub fn email_submission_query(
        &mut self,
        opts: JmapEmailSubmissionQueryOptions,
    ) -> Result<JmapEmailSubmissionQueryOutput, JmapClientStdError> {
        let coroutine =
            JmapEmailSubmissionQuery::new(self.session_or_err()?, &self.http_auth, opts)?;
        self.run(coroutine)
    }

    /// Runs [`JmapEmailSubmissionSet`] (`EmailSubmission/set`).
    pub fn email_submission_set(
        &mut self,
        submissions: BTreeMap<String, JmapEmailSubmissionCreate>,
    ) -> Result<JmapEmailSubmissionSetOutput, JmapClientStdError> {
        let coroutine =
            JmapEmailSubmissionSet::new(self.session_or_err()?, &self.http_auth, submissions)?;
        self.run(coroutine)
    }

    /// Runs [`JmapEmailSubmissionCancel`] (`EmailSubmission/set` with
    /// `undoStatus: "canceled"`).
    pub fn email_submission_cancel(
        &mut self,
        ids: Vec<String>,
    ) -> Result<JmapEmailSubmissionCancelOutput, JmapClientStdError> {
        let coroutine =
            JmapEmailSubmissionCancel::new(self.session_or_err()?, &self.http_auth, ids)?;
        self.run(coroutine)
    }

    /// Runs [`JmapVacationResponseGet`]; returns the singleton, if any.
    pub fn vacation_response_get(
        &mut self,
    ) -> Result<Option<JmapVacationResponse>, JmapClientStdError> {
        let coroutine = JmapVacationResponseGet::new(self.session_or_err()?, &self.http_auth)?;
        Ok(self.run(coroutine)?.vacation_response)
    }

    /// Runs [`JmapVacationResponseSet`]; returns the updated singleton if the
    /// server echoed it back.
    pub fn vacation_response_set(
        &mut self,
        patch: JmapVacationResponseUpdate,
    ) -> Result<Option<JmapVacationResponse>, JmapClientStdError> {
        let coroutine =
            JmapVacationResponseSet::new(self.session_or_err()?, &self.http_auth, patch)?;
        Ok(self.run(coroutine)?.updated)
    }

    /// Runs [`JmapAddressBookGet`] (`AddressBook/get`).
    pub fn address_book_get(
        &mut self,
        opts: JmapAddressBookGetOptions,
    ) -> Result<JmapAddressBookGetOutput, JmapClientStdError> {
        let coroutine = JmapAddressBookGet::new(self.session_or_err()?, &self.http_auth, opts)?;
        self.run(coroutine)
    }

    /// Runs [`JmapAddressBookSet`] (`AddressBook/set`).
    pub fn address_book_set(
        &mut self,
        args: JmapAddressBookSetArgs,
    ) -> Result<JmapAddressBookSetOutput, JmapClientStdError> {
        let coroutine = JmapAddressBookSet::new(self.session_or_err()?, &self.http_auth, args)?;
        self.run(coroutine)
    }

    /// Runs [`JmapAddressBookChanges`] (`AddressBook/changes`).
    pub fn address_book_changes(
        &mut self,
        since_state: impl Into<String>,
        opts: JmapAddressBookChangesOptions,
    ) -> Result<JmapChangesOutput, JmapClientStdError> {
        let coroutine = JmapAddressBookChanges::new(
            self.session_or_err()?,
            &self.http_auth,
            since_state,
            opts,
        )?;
        self.run(coroutine)
    }

    /// Runs [`JmapContactCardGet`] (`ContactCard/get`).
    pub fn contact_card_get(
        &mut self,
        opts: JmapContactCardGetOptions,
    ) -> Result<JmapContactCardGetOutput, JmapClientStdError> {
        let coroutine = JmapContactCardGet::new(self.session_or_err()?, &self.http_auth, opts)?;
        self.run(coroutine)
    }

    /// Runs [`JmapContactCardQuery`] (batched `ContactCard/query` +
    /// `ContactCard/get`).
    pub fn contact_card_query(
        &mut self,
        opts: JmapContactCardQueryOptions,
    ) -> Result<JmapContactCardQueryOutput, JmapClientStdError> {
        let coroutine = JmapContactCardQuery::new(self.session_or_err()?, &self.http_auth, opts)?;
        self.run(coroutine)
    }

    /// Runs [`JmapContactCardSet`] (`ContactCard/set`).
    pub fn contact_card_set(
        &mut self,
        args: JmapContactCardSetArgs,
    ) -> Result<JmapContactCardSetOutput, JmapClientStdError> {
        let coroutine = JmapContactCardSet::new(self.session_or_err()?, &self.http_auth, args)?;
        self.run(coroutine)
    }

    /// Runs [`JmapContactCardChanges`] (`ContactCard/changes`).
    pub fn contact_card_changes(
        &mut self,
        since_state: impl Into<String>,
        opts: JmapContactCardChangesOptions,
    ) -> Result<JmapChangesOutput, JmapClientStdError> {
        let coroutine = JmapContactCardChanges::new(
            self.session_or_err()?,
            &self.http_auth,
            since_state,
            opts,
        )?;
        self.run(coroutine)
    }

    /// Runs [`JmapContactCardCopy`] (`ContactCard/copy`).
    pub fn contact_card_copy(
        &mut self,
        from_account_id: impl Into<String>,
        cards: BTreeMap<String, JmapContactCardCopyArgs>,
    ) -> Result<JmapContactCardCopyOutput, JmapClientStdError> {
        let coroutine = JmapContactCardCopy::new(
            self.session_or_err()?,
            &self.http_auth,
            from_account_id,
            cards,
        )?;
        self.run(coroutine)
    }
}

impl fmt::Debug for JmapClientStd {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("JmapClientStd")
            .field("http_auth", &self.http_auth)
            .field("session", &self.session)
            .finish_non_exhaustive()
    }
}

/// Erased stream the client resumes coroutines against: auto-implemented for
/// any blocking `Read + Write + Send + 'static`. `Send` flows through the
/// `Box<dyn …>` so [`JmapClientStd`] can move between worker threads;
/// [`Self::as_any_mut`] lets specialized callers downcast back to the
/// concrete stream.
pub trait JmapStream: Read + Write + Send + Any {
    /// Upcasts the stream to [`Any`] for downcasting to the concrete type.
    fn as_any_mut(&mut self) -> &mut dyn Any;
}

impl<T: Read + Write + Send + Any> JmapStream for T {
    fn as_any_mut(&mut self) -> &mut dyn Any {
        self
    }
}