ruma_client/

client.rs

use std::{
    borrow::Cow,
    sync::{Arc, Mutex},
    time::Duration,
};

use assign::assign;
use async_stream::try_stream;
use futures_core::stream::Stream;
use ruma::{
    DeviceId,
    api::{
        AppserviceUserIdentity, OutgoingRequest, SupportedVersions,
        auth_scheme::{AuthScheme, SendAccessToken},
        client::{
            account::register::{self, RegistrationKind},
            session::login::{self, v3::LoginInfo},
            sync::sync_events,
            uiaa::{MatrixUserIdentifier, UserIdentifier},
        },
        path_builder::{PathBuilder, SinglePath, VersionHistory},
    },
    presence::PresenceState,
};

use crate::{Error, HttpClient, ResponseError, ResponseResult, send_customized_request};

mod builder;

pub use self::builder::ClientBuilder;

/// How the client should handle access tokens when making requests.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[non_exhaustive]
pub enum TokenMode {
    #[default]
    /// Send access token only when the endpoint requires it.
    SendIfRequired,

    /// Always send access token to all endpoints.
    SendAlways,

    /// Use application service token handling.
    AppService,
}

/// A client for the Matrix client-server API.
#[derive(Clone, Debug)]
pub struct Client<C>(Arc<ClientData<C>>);

/// Data contained in Client's Rc
#[derive(Debug)]
struct ClientData<C> {
    /// The URL of the homeserver to connect to.
    homeserver_url: String,

    /// The underlying HTTP client.
    http_client: C,

    /// The access token, if logged in.
    access_token: Mutex<Option<String>>,

    /// Mode for handling tokens when making requests.
    token_mode: TokenMode,

    /// The (known) Matrix versions the homeserver supports.
    supported_matrix_versions: SupportedVersions,
}

impl Client<()> {
    /// Creates a new client builder.
    pub fn builder() -> ClientBuilder {
        ClientBuilder::new()
    }
}

impl<C> Client<C> {
    /// Get a copy of the current `access_token`, if any.
    ///
    /// Useful for serializing and persisting the session to be restored later.
    pub fn access_token(&self) -> Option<String> {
        self.0.access_token.lock().expect("session mutex was poisoned").clone()
    }
}

impl<C: HttpClient> Client<C> {
    /// Makes a request to a Matrix API endpoint.
    pub async fn send_request<R>(&self, request: R) -> ResponseResult<C, R>
    where
        R: OutgoingRequest,
        for<'a> R::Authentication: AuthScheme<Input<'a> = SendAccessToken<'a>>,
        R::PathBuilder: SupportedPathBuilder,
    {
        self.send_customized_request(request, |_| Ok(())).await
    }

    /// Makes a request to a Matrix API endpoint including additional URL parameters.
    pub async fn send_customized_request<R, F>(
        &self,
        request: R,
        customize: F,
    ) -> ResponseResult<C, R>
    where
        R: OutgoingRequest,
        for<'a> R::Authentication: AuthScheme<Input<'a> = SendAccessToken<'a>>,
        R::PathBuilder: SupportedPathBuilder,
        F: FnOnce(&mut http::Request<C::RequestBody>) -> Result<(), ResponseError<C, R>>,
    {
        let token_mode = self.0.token_mode;
        let access_token = self.access_token();

        let send_access_token = match (token_mode, access_token.as_deref()) {
            (TokenMode::AppService, Some(at)) => SendAccessToken::Appservice(at),
            (TokenMode::SendIfRequired, Some(at)) => SendAccessToken::IfRequired(at),
            (TokenMode::SendAlways, Some(at)) => SendAccessToken::Always(at),
            (_, None) => SendAccessToken::None,
        };

        send_customized_request(
            &self.0.http_client,
            &self.0.homeserver_url,
            send_access_token,
            R::PathBuilder::get_path_builder_input(self),
            request,
            customize,
        )
        .await
    }

    /// Makes a request to a Matrix API endpoint as a virtual user.
    ///
    /// This method is meant to be used by application services when interacting with the
    /// client-server API.
    pub async fn send_request_as<R>(
        &self,
        identity: AppserviceUserIdentity<'_>,
        request: R,
    ) -> ResponseResult<C, R>
    where
        R: OutgoingRequest,
        for<'a> R::Authentication: AuthScheme<Input<'a> = SendAccessToken<'a>>,
        R::PathBuilder: SupportedPathBuilder,
    {
        self.send_customized_request(request, |request| {
            Ok(identity.maybe_add_to_uri(request.uri_mut())?)
        })
        .await
    }

    /// Log in with a username and password.
    ///
    /// In contrast to [`send_request`][Self::send_request], this method stores the access token
    /// returned by the endpoint in this client, in addition to returning it.
    pub async fn log_in(
        &self,
        user: &str,
        password: &str,
        device_id: Option<&DeviceId>,
        initial_device_display_name: Option<&str>,
    ) -> Result<login::v3::Response, Error<C::Error, ruma::api::error::Error>> {
        let login_info = LoginInfo::Password(login::v3::Password::new(
            UserIdentifier::Matrix(MatrixUserIdentifier::new(user.to_owned())),
            password.to_owned(),
        ));
        let response = self
            .send_request(assign!(login::v3::Request::new(login_info), {
                device_id: device_id.map(ToOwned::to_owned),
                initial_device_display_name: initial_device_display_name.map(ToOwned::to_owned),
            }))
            .await?;

        *self.0.access_token.lock().unwrap() = Some(response.access_token.clone());

        Ok(response)
    }

    /// Register as a guest.
    ///
    /// In contrast to [`send_request`][Self::send_request], this method stores the access token
    /// returned by the endpoint in this client, in addition to returning it.
    pub async fn register_guest(
        &self,
    ) -> Result<register::v3::Response, Error<C::Error, ruma::api::client::uiaa::UiaaResponse>>
    {
        let response = self
            .send_request(assign!(register::v3::Request::new(), { kind: RegistrationKind::Guest }))
            .await?;

        self.0.access_token.lock().unwrap().clone_from(&response.access_token);

        Ok(response)
    }

    /// Register as a new user on this server.
    ///
    /// In contrast to [`send_request`][Self::send_request], this method stores the access token
    /// returned by the endpoint in this client, in addition to returning it.
    ///
    /// The username is the local part of the returned user_id. If it is omitted from this request,
    /// the server will generate one.
    pub async fn register_user(
        &self,
        username: Option<&str>,
        password: &str,
    ) -> Result<register::v3::Response, Error<C::Error, ruma::api::client::uiaa::UiaaResponse>>
    {
        let response = self
            .send_request(assign!(register::v3::Request::new(), {
                username: username.map(ToOwned::to_owned),
                password: Some(password.to_owned())
            }))
            .await?;

        self.0.access_token.lock().unwrap().clone_from(&response.access_token);

        Ok(response)
    }

    /// Convenience method that represents repeated calls to the sync_events endpoint as a stream.
    ///
    /// # Example:
    ///
    /// ```no_run
    /// use std::time::Duration;
    ///
    /// # use ruma::presence::PresenceState;
    /// # use tokio_stream::{StreamExt as _};
    /// # let homeserver_url = "https://example.com".to_owned();
    /// # async {
    /// # let client = ruma_client::Client::builder()
    /// #     .homeserver_url(homeserver_url)
    /// #     .build::<ruma_client::http_client::Dummy>()
    /// #     .await?;
    /// # let next_batch_token = String::new();
    /// let mut sync_stream = Box::pin(client.sync(
    ///     None,
    ///     next_batch_token,
    ///     PresenceState::Online,
    ///     Some(Duration::from_secs(30)),
    /// ));
    /// while let Some(response) = sync_stream.try_next().await? {
    ///     // Do something with the data in the response...
    /// }
    /// # Result::<(), ruma_client::Error<_, _>>::Ok(())
    /// # };
    /// ```
    pub fn sync(
        &self,
        filter: Option<sync_events::v3::Filter>,
        mut since: String,
        set_presence: PresenceState,
        timeout: Option<Duration>,
    ) -> impl Stream<
        Item = Result<sync_events::v3::Response, Error<C::Error, ruma::api::error::Error>>,
    > + '_ {
        try_stream! {
            loop {
                let response = self
                    .send_request(assign!(sync_events::v3::Request::new(), {
                        filter: filter.clone(),
                        since: Some(since.clone()),
                        set_presence: set_presence.clone(),
                        timeout,
                    }))
                    .await?;

                since.clone_from(&response.next_batch);
                yield response;
            }
        }
    }
}

/// Marker trait to identify [`PathBuilder`] implementors that the [`Client`] supports.
///
/// This trait can be implemented for custom `PathBuilder`s if necessary.
pub trait SupportedPathBuilder: PathBuilder {
    /// Get the [`PathBuilder::Input`] from the [`Client`].
    fn get_path_builder_input<C>(client: &Client<C>) -> Self::Input<'_>;
}

impl SupportedPathBuilder for VersionHistory {
    fn get_path_builder_input<C>(client: &Client<C>) -> Self::Input<'_> {
        Cow::Borrowed(&client.0.supported_matrix_versions)
    }
}

impl SupportedPathBuilder for SinglePath {
    fn get_path_builder_input<C>(_client: &Client<C>) -> Self::Input<'_> {}
}