matrix-rust-sdk

// Copyright 2020 Damir Jelić

// Copyright 2020 The Matrix.org Foundation C.I.C.

//

// Licensed under the Apache License, Version 2.0 (the "License");

// you may not use this file except in compliance with the License.

// You may obtain a copy of the License at

//

//     http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing, software

// distributed under the License is distributed on an "AS IS" BASIS,

// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

// See the License for the specific language governing permissions and

// limitations under the License.

#[cfg(feature = "e2e-encryption")]

use std::sync::Arc;

use std::{

    collections::{BTreeMap, BTreeSet, HashMap},

    fmt,

    ops::Deref,

};

use eyeball::{SharedObservable, Subscriber};

use eyeball_im::{Vector, VectorDiff};

use futures_util::Stream;

use matrix_sdk_common::{cross_process_lock::CrossProcessLockConfig, timer};

#[cfg(feature = "e2e-encryption")]

use matrix_sdk_crypto::{

    CollectStrategy, DecryptionSettings, EncryptionSettings, OlmError, OlmMachine,

    TrustRequirement, store::DynCryptoStore, store::types::RoomPendingKeyBundleDetails,

    types::requests::ToDeviceRequest,

};

#[cfg(doc)]

use ruma::DeviceId;

#[cfg(feature = "e2e-encryption")]

use ruma::events::room::{history_visibility::HistoryVisibility, member::MembershipState};

use ruma::{

    OwnedRoomId, OwnedUserId, RoomId, UserId,

    api::client::{self as api, sync::sync_events::v5},

    events::{

        StateEvent, StateEventType,

        ignored_user_list::IgnoredUserListEventContent,

        push_rules::{PushRulesEvent, PushRulesEventContent},

        room::member::SyncRoomMemberEvent,

    },

    push::Ruleset,

    time::Instant,

};

use tokio::sync::{Mutex, broadcast};

#[cfg(feature = "e2e-encryption")]

use tokio::sync::{RwLock, RwLockReadGuard};

use tracing::{Level, debug, enabled, info, instrument, warn};

#[cfg(feature = "e2e-encryption")]

use crate::RoomMemberships;

use crate::{

    RoomStateFilter, SessionMeta, StateStore,

    deserialized_responses::DisplayName,

    error::{Error, Result},

    event_cache::store::{EventCacheStoreLock, EventCacheStoreLockState},

    media::store::MediaStoreLock,

    response_processors::{self as processors, Context},

    room::{

        Room, RoomInfoNotableUpdate, RoomInfoNotableUpdateReasons, RoomMembersUpdate, RoomState,

    },

    store::{

        AvatarCache, BaseStateStore, DynStateStore, MemoryStore, Result as StoreResult,

        RoomLoadSettings, StateChanges, StateStoreDataKey, StateStoreDataValue, StateStoreExt,

        StoreConfig, ambiguity_map::AmbiguityCache,

    },

    sync::{RoomUpdates, SyncResponse},

};

/// A no (network) IO client implementation.

///

/// This client is a state machine that receives responses and events and

/// accordingly updates its state. It is not designed to be used directly, but

/// rather through `matrix_sdk::Client`.

///

/// ```rust

/// use matrix_sdk_base::{

///     BaseClient, DmRoomDefinition, ThreadingSupport, store::StoreConfig,

/// };

/// use matrix_sdk_common::cross_process_lock::CrossProcessLockConfig;

///

/// let client = BaseClient::new(

///     StoreConfig::new(CrossProcessLockConfig::multi_process(

///         "cross-process-holder-name".to_owned(),

///     )),

///     ThreadingSupport::Disabled,

///     DmRoomDefinition::default(),

/// );

/// ```

#[derive(Clone)]

pub struct BaseClient {

    /// The state store.

    pub(crate) state_store: BaseStateStore,

    /// The store used by the event cache.

    event_cache_store: EventCacheStoreLock,

    /// The store used by the media cache.

    media_store: MediaStoreLock,

    /// The store used for encryption.

    ///

    /// This field is only meant to be used for `OlmMachine` initialization.

    /// All operations on it happen inside the `OlmMachine`.

    #[cfg(feature = "e2e-encryption")]

    crypto_store: Arc<DynCryptoStore>,

    /// The olm-machine that is created once the

    /// [`SessionMeta`][crate::session::SessionMeta] is set via

    /// [`BaseClient::activate`]

    #[cfg(feature = "e2e-encryption")]

    olm_machine: Arc<RwLock<Option<OlmMachine>>>,

    /// Observable of when a user is ignored/unignored.

    pub(crate) ignore_user_list_changes: SharedObservable<Vec<String>>,

    /// The strategy to use for picking recipient devices, when sending an

    /// encrypted message.

    #[cfg(feature = "e2e-encryption")]

    pub room_key_recipient_strategy: CollectStrategy,

    /// The settings to use for decrypting events.

    #[cfg(feature = "e2e-encryption")]

    pub decryption_settings: DecryptionSettings,

    /// If the client should handle verification events received when syncing.

    #[cfg(feature = "e2e-encryption")]

    pub handle_verification_events: bool,

    /// Whether the client supports threads or not.

    pub threading_support: ThreadingSupport,

    /// The definition of what is considered a DM room.

    pub dm_room_definition: DmRoomDefinition,

}

#[cfg(not(tarpaulin_include))]

impl fmt::Debug for BaseClient {

    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

        f.debug_struct("BaseClient")

            .field("session_meta", &self.state_store.session_meta())

            .field("sync_token", &self.state_store.sync_token)

            .finish_non_exhaustive()

    }

}

/// Whether this client instance supports threading or not. Currently used to

/// determine how the client handles read receipts and unread count computations

/// on the base SDK level.

///

/// Timelines on the other hand have a separate `TimelineFocus`

/// `hide_threaded_events` associated value that can be used to hide threaded

/// events but also to enable threaded read receipt sending. This is because

/// certain timeline instances should ignore threading no matter what's defined

/// at the client level. One such example are media filtered timelines which

/// should contain all the room's media no matter what thread its in (unless

/// explicitly opted into).

#[derive(Clone, Copy, Debug)]

pub enum ThreadingSupport {

    /// Threading enabled.

    Enabled {

        /// Enable client-wide thread subscriptions support (MSC4306 / MSC4308).

        ///

        /// This may cause filtering out of thread subscriptions, and loading

        /// the thread subscriptions via the sliding sync extension,

        /// when the room list service is being used.

        with_subscriptions: bool,

    },

    /// Threading disabled.

    Disabled,

}

impl BaseClient {

    /// Create a new client.

    ///

    /// # Arguments

    ///

    /// * `config` - the configuration for the stores (state store, event cache

    ///   store and crypto store).

    pub fn new(

        config: StoreConfig,

        threading_support: ThreadingSupport,

        dm_room_definition: DmRoomDefinition,

    ) -> Self {

        let store = BaseStateStore::new(config.state_store);

        BaseClient {

            state_store: store,

            event_cache_store: config.event_cache_store,

            media_store: config.media_store,

            #[cfg(feature = "e2e-encryption")]

            crypto_store: config.crypto_store,

            #[cfg(feature = "e2e-encryption")]

            olm_machine: Default::default(),

            ignore_user_list_changes: Default::default(),

            #[cfg(feature = "e2e-encryption")]

            room_key_recipient_strategy: Default::default(),

            #[cfg(feature = "e2e-encryption")]

            decryption_settings: DecryptionSettings {

                sender_device_trust_requirement: TrustRequirement::Untrusted,

            },

            #[cfg(feature = "e2e-encryption")]

            handle_verification_events: true,

            threading_support,

            dm_room_definition,

        }

    }

    /// Clones the current base client to use the same crypto store but a

    /// different, in-memory store config, and resets transient state.

    #[cfg(feature = "e2e-encryption")]

    pub async fn clone_with_in_memory_state_store(

        &self,

        cross_process_mode: CrossProcessLockConfig,

        handle_verification_events: bool,

    ) -> Result<Self> {

        let config = StoreConfig::new(cross_process_mode).state_store(MemoryStore::new());

        let config = config.crypto_store(self.crypto_store.clone());

        let copy = Self {

            state_store: BaseStateStore::new(config.state_store),

            event_cache_store: config.event_cache_store,

            media_store: config.media_store,

            // We copy the crypto store as well as the `OlmMachine` for two reasons:

            // 1. The `self.crypto_store` is the same as the one used inside the `OlmMachine`.

            // 2. We need to ensure that the parent and child use the same data and caches inside

            //    the `OlmMachine` so the various ratchets and places where new randomness gets

            //    introduced don't diverge, i.e. one-time keys that get generated by the Olm Account

            //    or Olm sessions when they encrypt or decrypt messages.

            crypto_store: self.crypto_store.clone(),

            olm_machine: self.olm_machine.clone(),

            ignore_user_list_changes: Default::default(),

            room_key_recipient_strategy: self.room_key_recipient_strategy.clone(),

            decryption_settings: self.decryption_settings.clone(),

            handle_verification_events,

            threading_support: self.threading_support,

            dm_room_definition: self.dm_room_definition.clone(),

        };

        copy.state_store.derive_from_other(&self.state_store).await?;

        Ok(copy)

    }

    /// Clones the current base client to use the same crypto store but a

    /// different, in-memory store config, and resets transient state.

    #[cfg(not(feature = "e2e-encryption"))]

    #[allow(clippy::unused_async)]

    pub async fn clone_with_in_memory_state_store(

        &self,

        cross_process_store_config: CrossProcessLockConfig,

        _handle_verification_events: bool,

    ) -> Result<Self> {

        let config = StoreConfig::new(cross_process_store_config).state_store(MemoryStore::new());

        Ok(Self::new(config, ThreadingSupport::Disabled, DmRoomDefinition::default()))

    }

    /// Get the session meta information.

    ///

    /// If the client is currently logged in, this will return a

    /// [`SessionMeta`] object which contains the user ID and device ID.

    /// Otherwise it returns `None`.

    pub fn session_meta(&self) -> Option<&SessionMeta> {

        self.state_store.session_meta()

    }

    /// Get all the rooms this client knows about.

    pub fn rooms(&self) -> Vec<Room> {

        self.state_store.rooms()

    }

    /// Get all the rooms this client knows about, filtered by room state.

    pub fn rooms_filtered(&self, filter: RoomStateFilter) -> Vec<Room> {

        self.state_store.rooms_filtered(filter)

    }

    /// Get a stream of all the rooms changes, in addition to the existing

    /// rooms.

    pub fn rooms_stream(

        &self,

    ) -> (Vector<Room>, impl Stream<Item = Vec<VectorDiff<Room>>> + use<>) {

        self.state_store.rooms_stream()

    }

    /// Lookup the Room for the given RoomId, or create one, if it didn't exist

    /// yet in the store

    pub fn get_or_create_room(&self, room_id: &RoomId, room_state: RoomState) -> Room {

        self.state_store.get_or_create_room(room_id, room_state)

    }

    /// Get a reference to the state store.

    pub fn state_store(&self) -> &DynStateStore {

        self.state_store.deref()

    }

    /// Get a reference to the event cache store.

    pub fn event_cache_store(&self) -> &EventCacheStoreLock {

        &self.event_cache_store

    }

    /// Get a reference to the media store.

    pub fn media_store(&self) -> &MediaStoreLock {

        &self.media_store

    }

    /// Check whether the client has been activated.

    ///

    /// See [`BaseClient::activate`] to know what it means.

    pub fn is_active(&self) -> bool {

        self.state_store.session_meta().is_some()

    }

    /// Activate the client.

    ///

    /// A client is considered active when:

    ///

    /// 1. It has a `SessionMeta` (user ID, device ID and access token),

    /// 2. Has loaded cached data from storage,

    /// 3. If encryption is enabled, it also initialized or restored its

    ///    `OlmMachine`.

    ///

    /// # Arguments

    ///

    /// * `session_meta` - The meta of a session that the user already has from

    ///   a previous login call.

    ///

    /// * `custom_account` - A custom

    ///   [`matrix_sdk_crypto::vodozemac::olm::Account`] to be used for the

    ///   identity and one-time keys of this [`BaseClient`]. If no account is

    ///   provided, a new default one or one from the store will be used. If an

    ///   account is provided and one already exists in the store for this

    ///   [`UserId`]/[`DeviceId`] combination, an error will be raised. This is

    ///   useful if one wishes to create identity keys before knowing the

    ///   user/device IDs, e.g., to use the identity key as the device ID.

    ///

    /// * `room_load_settings` — Specify how many rooms must be restored; use

    ///   `::default()` if you don't know which value to pick.

    ///

    /// # Panics

    ///

    /// This method panics if it is called twice.

    ///

    /// [`UserId`]: ruma::UserId

    pub async fn activate(

        &self,

        session_meta: SessionMeta,

        room_load_settings: RoomLoadSettings,

        #[cfg(feature = "e2e-encryption")] custom_account: Option<

            crate::crypto::vodozemac::olm::Account,

        >,

    ) -> Result<()> {

        debug!(user_id = ?session_meta.user_id, device_id = ?session_meta.device_id, "Activating the client");

        self.state_store.load_rooms(&session_meta.user_id, room_load_settings).await?;

        self.state_store.load_sync_token().await?;

        self.state_store.set_session_meta(session_meta);

        #[cfg(feature = "e2e-encryption")]

        self.regenerate_olm(custom_account).await?;

        Ok(())

    }

    /// Recreate an `OlmMachine` from scratch.

    ///

    /// In particular, this will clear all its caches.

    #[cfg(feature = "e2e-encryption")]

    pub async fn regenerate_olm(

        &self,

        custom_account: Option<crate::crypto::vodozemac::olm::Account>,

    ) -> Result<()> {

        tracing::debug!("regenerating OlmMachine");

        let session_meta = self.session_meta().ok_or(Error::OlmError(OlmError::MissingSession))?;

        // Recreate the `OlmMachine` and wipe the in-memory cache in the store

        // because we suspect it has stale data.

        let olm_machine = OlmMachine::with_store(

            &session_meta.user_id,

            &session_meta.device_id,

            self.crypto_store.clone(),

            custom_account,

        )

        .await

        .map_err(OlmError::from)?;

        *self.olm_machine.write().await = Some(olm_machine);

        Ok(())

    }

    /// Get the current, if any, sync token of the client.

    /// This will be None if the client didn't sync at least once.

    pub async fn sync_token(&self) -> Option<String> {

        self.state_store.sync_token.read().await.clone()

    }

    /// User has knocked on a room.

    ///

    /// Update the internal and cached state accordingly. Return the final Room.

    pub async fn room_knocked(&self, room_id: &RoomId) -> Result<Room> {

        let room = self.state_store.get_or_create_room(room_id, RoomState::Knocked);

        if room.state() != RoomState::Knocked {

            let store_guard = self.state_store.lock().lock().await;

            // We are no longer joined to the room, so the invite acceptance details are no

            // longer relevant.

            #[cfg(feature = "e2e-encryption")]

            if let Some(olm_machine) = self.olm_machine().await.as_ref() {

                olm_machine.store().clear_room_pending_key_bundle(room_id).await?

            }

            room.update_and_save_room_info_with_store_guard(&store_guard, |mut info| {

                info.mark_as_knocked();

                info.mark_state_partially_synced();

                info.mark_members_missing(); // the own member event changed

                (info, RoomInfoNotableUpdateReasons::MEMBERSHIP)

            })

            .await?;

        }

        Ok(room)

    }

    /// The user has joined a room using this specific client.

    ///

    /// This method should be called if the user accepts an invite or if they

    /// join a public room.

    ///

    /// The method will create a [`Room`] object if one does not exist yet and

    /// set the state of the [`Room`] to [`RoomState::Joined`]. The [`Room`]

    /// object will be persisted in the cache. Please note that the [`Room`]

    /// will be a stub until a sync has been received with the full room

    /// state using [`BaseClient::receive_sync_response`].

    ///

    /// Update the internal and cached state accordingly. Return the final Room.

    ///

    /// # Arguments

    ///

    /// * `room_id` - The unique ID identifying the joined room.

    /// * `inviter` - When joining this room in response to an invitation, the

    ///   inviter should be recorded before sending the join request to the

    ///   server. Providing the inviter here ensures that the

    ///   [`RoomPendingKeyBundleDetails`] are stored for this room.

    ///

    /// # Examples

    ///

    /// ```rust

    /// # use matrix_sdk_base::{BaseClient, store::StoreConfig, RoomState, ThreadingSupport, DmRoomDefinition};

    /// # use ruma::{OwnedRoomId, OwnedUserId, RoomId};

    /// use matrix_sdk_common::cross_process_lock::CrossProcessLockConfig;

    /// # async {

    /// # let client = BaseClient::new(

    ///     StoreConfig::new(CrossProcessLockConfig::multi_process("example")),

    ///     ThreadingSupport::Disabled,

    ///     DmRoomDefinition::default()

    /// );

    /// # async fn send_join_request() -> anyhow::Result<OwnedRoomId> { todo!() }

    /// # async fn maybe_get_inviter(room_id: &RoomId) -> anyhow::Result<Option<OwnedUserId>> { todo!() }

    /// # let room_id: &RoomId = todo!();

    /// let maybe_inviter = maybe_get_inviter(room_id).await?;

    /// let room_id = send_join_request().await?;

    /// let room = client.room_joined(&room_id, maybe_inviter).await?;

    ///

    /// assert_eq!(room.state(), RoomState::Joined);

    /// # matrix_sdk_test::TestResult::Ok(()) };

    /// ```

    pub async fn room_joined(

        &self,

        room_id: &RoomId,

        inviter: Option<OwnedUserId>,

    ) -> Result<Room> {

        let room = self.state_store.get_or_create_room(room_id, RoomState::Joined);

        // If the state isn't `RoomState::Joined` then this means that we knew about

        // this room before. Let's modify the existing state now.

        if room.state() != RoomState::Joined {

            let store_guard = self.state_store_lock().lock().await;

            #[cfg(feature = "e2e-encryption")]

            {

                // If our previous state was an invite and we're now in the joined state, this

                // means that the user has explicitly accepted an invite. Let's

                // remember some details about the invite.

                //

                // This is somewhat of a workaround for our lack of cryptographic membership.

                // Later on we will decide if historic room keys should be accepted

                // based on this info. If a user has accepted an invite and we receive a room

                // key bundle shortly after, we might accept it. If we don't do

                // this, the homeserver could trick us into accepting any historic room key

                // bundle.

                let previous_state = room.state();

                if previous_state == RoomState::Invited

                    && let Some(inviter) = inviter

                    && let Some(olm_machine) = self.olm_machine().await.as_ref()

                {

                    olm_machine.store().store_room_pending_key_bundle(room_id, &inviter).await?

                }

            }

            #[cfg(not(feature = "e2e-encryption"))]

            {

                // suppress unused argument warning

                let _ = inviter;

            }

            room.update_and_save_room_info_with_store_guard(&store_guard, |mut info| {

                info.mark_as_joined();

                info.mark_state_partially_synced();

                info.mark_members_missing(); // the own member event changed

                (info, RoomInfoNotableUpdateReasons::MEMBERSHIP)

            })

            .await?;

        }

        Ok(room)

    }

    /// User has left a room.

    ///