base: store timeline to the StateStore

Author: jsparber

matrix_sdk_base/src/client.rs

@@ -25,7 +25,7 @@ use std::{
 use matrix_sdk_common::{
     deserialized_responses::{
         AmbiguityChanges, JoinedRoom, LeftRoom, MemberEvent, MembersResponse, Rooms,
-        StrippedMemberEvent, SyncResponse, SyncRoomEvent, Timeline,
+        StrippedMemberEvent, SyncResponse, SyncRoomEvent, Timeline, TimelineSlice,
     },
     instant::Instant,
     locks::RwLock,
@@ -48,7 +48,11 @@ use ruma::{
     DeviceId,
 };
 use ruma::{
-    api::client::r0::{self as api, push::get_notifications::Notification},
+    api::client::r0::{
+        self as api,
+        message::get_message_events::{Direction, Response as GetMessageEventsResponse},
+        push::get_notifications::Notification,
+    },
     events::{
         room::member::{MemberEventContent, MembershipState},
         AnyGlobalAccountDataEvent, AnyRoomAccountDataEvent, AnyStrippedStateEvent,
@@ -836,6 +840,15 @@ impl BaseClient {
             let notification_count = new_info.unread_notifications.into();
             room_info.update_notification_count(notification_count);
 
+            changes.add_timeline(
+                &room_id,
+                TimelineSlice::new(
+                    timeline.events.iter().cloned().rev().collect(),
+                    next_batch.clone(),
+                    timeline.prev_batch.clone(),
+                ),
+            );
+
             new_rooms.join.insert(
                 room_id,
                 JoinedRoom::new(
@@ -879,6 +892,14 @@ impl BaseClient {
             self.handle_room_account_data(&room_id, &new_info.account_data.events, &mut changes)
                 .await;
 
+            changes.add_timeline(
+                &room_id,
+                TimelineSlice::new(
+                    timeline.events.iter().cloned().rev().collect(),
+                    next_batch.clone(),
+                    timeline.prev_batch.clone(),
+                ),
+            );
             changes.add_room(room_info);
             new_rooms
                 .leave
@@ -949,6 +970,64 @@ impl BaseClient {
         }
     }
 
+    /// Receive a successful /messages response.
+    ///
+    /// * `response` - The successful response from /messages.
+    pub async fn receive_messages(
+        &self,
+        room_id: &RoomId,
+        direction: &Direction,
+        response: &GetMessageEventsResponse,
+    ) -> Result<Vec<SyncRoomEvent>> {
+        let mut changes = StateChanges::default();
+
+        let mut events: Vec<SyncRoomEvent> = vec![];
+        for event in &response.chunk {
+            #[allow(unused_mut)]
+            let mut event: SyncRoomEvent = event.clone().into();
+
+            #[cfg(feature = "encryption")]
+            match hoist_room_event_prev_content(&event.event) {
+                Ok(e) => {
+                    if let AnySyncRoomEvent::Message(AnySyncMessageEvent::RoomEncrypted(
+                        encrypted,
+                    )) = e
+                    {
+                        if let Some(olm) = self.olm_machine().await {
+                            if let Ok(decrypted) = olm.decrypt_room_event(&encrypted, room_id).await
+                            {
+                                event = decrypted;
+                            }
+                        }
+                    }
+                }
+                Err(error) => {
+                    warn!("Error deserializing event {:?}", error);
+                }
+            }
+
+            events.push(event);
+        }
+
+        let (chunk, start, end) = match direction {
+            Direction::Backward => {
+                (events.clone(), response.start.clone().unwrap(), response.end.clone())
+            }
+            Direction::Forward => (
+                events.iter().rev().cloned().collect(),
+                response.end.clone().unwrap(),
+                response.start.clone(),
+            ),
+        };
+
+        let timeline = TimelineSlice::new(chunk, start, end);
+        changes.add_timeline(room_id, timeline);
+
+        self.store().save_changes(&changes).await?;
+
+        Ok(events)
+    }
+
     /// Receive a get member events response and convert it to a deserialized
     /// `MembersResponse`
     ///

matrix_sdk_base/src/store/memory_store.rs

@@ -21,6 +21,7 @@ use dashmap::{DashMap, DashSet};
 use lru::LruCache;
 use matrix_sdk_common::{async_trait, instant::Instant, locks::Mutex};
 use ruma::{
+    api::client::r0::message::get_message_events::Direction,
     events::{
         presence::PresenceEvent,
         receipt::Receipt,
@@ -34,9 +35,9 @@ use ruma::{
 };
 use tracing::info;
 
-use super::{Result, RoomInfo, StateChanges, StateStore};
+use super::{Result, RoomInfo, StateChanges, StateStore, StoredTimelineSlice};
 use crate::{
-    deserialized_responses::{MemberEvent, StrippedMemberEvent},
+    deserialized_responses::{MemberEvent, StrippedMemberEvent, TimelineSlice},
     media::{MediaRequest, UniqueKey},
 };
 
@@ -66,6 +67,8 @@ pub struct MemoryStore {
     room_event_receipts:
         Arc<DashMap<RoomId, DashMap<String, DashMap<EventId, DashMap<UserId, Receipt>>>>>,
     media: Arc<Mutex<LruCache<String, Vec<u8>>>>,
+    timeline_slices: Arc<DashMap<RoomId, DashMap<String, TimelineSlice>>>,
+    event_id_to_timeline_slice: Arc<DashMap<RoomId, DashMap<EventId, String>>>,
 }
 
 impl MemoryStore {
@@ -90,6 +93,8 @@ impl MemoryStore {
             room_user_receipts: DashMap::new().into(),
             room_event_receipts: DashMap::new().into(),
             media: Arc::new(Mutex::new(LruCache::new(100))),
+            timeline_slices: DashMap::new().into(),
+            event_id_to_timeline_slice: DashMap::new().into(),
         }
     }
 
@@ -271,6 +276,23 @@ impl MemoryStore {
             }
         }
 
+        for (room, timeline) in &changes.timeline {
+            // FIXME: make sure that we don't add events already known
+            for event in &timeline.events {
+                let event_id = event.event_id();
+
+                self.event_id_to_timeline_slice
+                    .entry(room.clone())
+                    .or_insert_with(DashMap::new)
+                    .insert(event_id, timeline.start.clone());
+            }
+
+            self.timeline_slices
+                .entry(room.clone())
+                .or_insert_with(DashMap::new)
+                .insert(timeline.start.clone(), timeline.clone());
+        }
+
         info!("Saved changes in {:?}", now.elapsed());
 
         Ok(())
@@ -424,6 +446,93 @@ impl MemoryStore {
 
         Ok(())
     }
+
+    async fn get_timeline(
+        &self,
+        room_id: &RoomId,
+        start: Option<&EventId>,
+        end: Option<&EventId>,
+        limit: Option<usize>,
+        direction: Direction,
+    ) -> Result<Option<StoredTimelineSlice>> {
+        let event_id_to_timeline_slice = self.event_id_to_timeline_slice.get(room_id).unwrap();
+        let timeline_slices = self.timeline_slices.get(room_id).unwrap();
+
+        let slice = if let Some(start) = start {
+            event_id_to_timeline_slice.get(start).and_then(|s| timeline_slices.get(&*s))
+        } else {
+            self.get_sync_token().await?.and_then(|s| timeline_slices.get(&*s))
+        };
+
+        let mut slice = if let Some(slice) = slice {
+            slice
+        } else {
+            return Ok(None);
+        };
+
+        let mut timeline = StoredTimelineSlice::new(vec![], None);
+
+        match direction {
+            Direction::Forward => {
+                loop {
+                    timeline.events.append(&mut slice.events.iter().rev().cloned().collect());
+
+                    timeline.token = Some(slice.start.clone());
+
+                    let found_end = end.map_or(false, |end| {
+                        slice.events.iter().any(|event| &event.event_id() == end)
+                    });
+
+                    if found_end {
+                        break;
+                    }
+
+                    if let Some(next_slice) = timeline_slices.get(&slice.start) {
+                        slice = next_slice;
+                    } else {
+                        // No more timeline slices or a gap in the known timeline
+                        break;
+                    }
+
+                    if let Some(limit) = limit {
+                        if timeline.events.len() >= limit {
+                            break;
+                        }
+                    }
+                }
+            }
+            Direction::Backward => {
+                loop {
+                    timeline.events.append(&mut slice.events.clone());
+                    timeline.token = slice.end.clone();
+
+                    let found_end = end.map_or(false, |end| {
+                        slice.events.iter().any(|event| &event.event_id() == end)
+                    });
+
+                    if found_end {
+                        break;
+                    }
+
+                    if let Some(prev_slice) =
+                        slice.end.as_deref().and_then(|end| timeline_slices.get(end))
+                    {
+                        slice = prev_slice;
+                    } else {
+                        // No more timeline slices or we have a gap in the known timeline
+                        break;
+                    }
+
+                    if let Some(limit) = limit {
+                        if timeline.events.len() >= limit {
+                            break;
+                        }
+                    }
+                }
+            }
+        }
+        Ok(Some(timeline))
+    }
 }
 
 #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
@@ -555,6 +664,16 @@ impl StateStore for MemoryStore {
     async fn remove_media_content_for_uri(&self, uri: &MxcUri) -> Result<()> {
         self.remove_media_content_for_uri(uri).await
     }
+    async fn get_timeline(
+        &self,
+        room_id: &RoomId,
+        start: Option<&EventId>,
+        end: Option<&EventId>,
+        limit: Option<usize>,
+        direction: Direction,
+    ) -> Result<Option<StoredTimelineSlice>> {
+        self.get_timeline(room_id, start, end, limit, direction).await
+    }
 }
 
 #[cfg(test)]

matrix_sdk_base/src/store/mod.rs

@@ -23,7 +23,9 @@ use std::{
 use dashmap::DashMap;
 use matrix_sdk_common::{async_trait, locks::RwLock, AsyncTraitDeps};
 use ruma::{
-    api::client::r0::push::get_notifications::Notification,
+    api::client::r0::{
+        message::get_message_events::Direction, push::get_notifications::Notification,
+    },
     events::{
         presence::PresenceEvent,
         receipt::{Receipt, ReceiptEventContent},
@@ -39,7 +41,7 @@ use ruma::{
 use sled::Db;
 
 use crate::{
-    deserialized_responses::{MemberEvent, StrippedMemberEvent},
+    deserialized_responses::{MemberEvent, StrippedMemberEvent, SyncRoomEvent, TimelineSlice},
     media::MediaRequest,
     rooms::{RoomInfo, RoomType},
     Room, Session,
@@ -280,6 +282,29 @@ pub trait StateStore: AsyncTraitDeps {
     ///
     /// * `uri` - The `MxcUri` of the media files.
     async fn remove_media_content_for_uri(&self, uri: &MxcUri) -> Result<()>;
+
+    /// Get a slice of the timeline of a room.
+    ///
+    /// # Arguments
+    ///
+    /// * `room_id` - The id of the room for which the timeline should be
+    ///   fetched.
+    ///
+    /// * `start` - The start point from which events should be returned.
+    ///
+    /// * `end` - The end point to which events should be returned.
+    ///
+    /// * `limit` - The maximum number of events to return.
+    ///
+    /// * `direction` - The direction events should be returned.
+    async fn get_timeline(
+        &self,
+        room_id: &RoomId,
+        start: Option<&EventId>,
+        end: Option<&EventId>,
+        limit: Option<usize>,
+        direction: Direction,
+    ) -> Result<Option<StoredTimelineSlice>>;
 }
 
 /// A state store wrapper for the SDK.
@@ -453,6 +478,9 @@ pub struct StateChanges {
     pub ambiguity_maps: BTreeMap<RoomId, BTreeMap<String, BTreeSet<UserId>>>,
     /// A map of `RoomId` to a vector of `Notification`s
     pub notifications: BTreeMap<RoomId, Vec<Notification>>,
+
+    /// A mapping of `RoomId` to a `TimelineSlice`
+    pub timeline: BTreeMap<RoomId, TimelineSlice>,
 }
 
 impl StateChanges {
@@ -537,4 +565,27 @@ impl StateChanges {
     pub fn add_receipts(&mut self, room_id: &RoomId, event: ReceiptEventContent) {
         self.receipts.insert(room_id.to_owned(), event);
     }
+
+    /// Update the `StateChanges` struct with the given room with a new
+    /// `TimelineSlice`.
+    pub fn add_timeline(&mut self, room_id: &RoomId, timeline: TimelineSlice) {
+        self.timeline.insert(room_id.to_owned(), timeline);
+    }
+}
+
+/// Store state changes and pass them to the StateStore.
+#[derive(Debug, Default)]
+pub struct StoredTimelineSlice {
+    /// A start token to fetch more events if the requested slice isn't fully
+    /// known.
+    pub token: Option<String>,
+
+    /// The requested events
+    pub events: Vec<SyncRoomEvent>,
+}
+
+impl StoredTimelineSlice {
+    pub fn new(events: Vec<SyncRoomEvent>, token: Option<String>) -> Self {
+        Self { events, token }
+    }
 }