refactor(sdk): Move the RoomLatestEvents* types in their own module.

This patch moves the `RoomLatestEvents*` types in their own new
`room_latest_events` module.

Author: Hywan

crates/matrix-sdk/src/latest_events/mod.rs

@@ -48,6 +48,7 @@
 
 mod error;
 mod latest_event;
+mod room_latest_events;
 
 use std::{
     collections::{HashMap, HashSet},
@@ -61,19 +62,17 @@ use futures_util::FutureExt;
 use latest_event::LatestEvent;
 pub use latest_event::{LatestEventValue, LocalLatestEventValue, RemoteLatestEventValue};
 use matrix_sdk_common::executor::{AbortOnDrop, JoinHandleExt as _, spawn};
-use ruma::{EventId, OwnedEventId, OwnedRoomId, RoomId};
+use room_latest_events::RoomLatestEvents;
+use ruma::{EventId, OwnedRoomId, RoomId};
 use tokio::{
     select,
-    sync::{
-        OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock, RwLockReadGuard, RwLockWriteGuard,
-        broadcast, mpsc,
-    },
+    sync::{RwLock, RwLockReadGuard, RwLockWriteGuard, broadcast, mpsc},
 };
 use tracing::{error, warn};
 
 use crate::{
     client::WeakClient,
-    event_cache::{EventCache, EventCacheError, RoomEventCache, RoomEventCacheGenericUpdate},
+    event_cache::{EventCache, RoomEventCacheGenericUpdate},
     room::WeakRoom,
     send_queue::{RoomSendQueueUpdate, SendQueue, SendQueueUpdate},
 };
@@ -447,200 +446,6 @@ enum LatestEventQueueUpdate {
     },
 }
 
-/// Type holding the [`LatestEvent`] for a room and for all its threads.
-#[derive(Debug)]
-struct RoomLatestEvents {
-    /// The state of this type.
-    state: Arc<RwLock<RoomLatestEventsState>>,
-}
-
-impl RoomLatestEvents {
-    /// Create a new [`RoomLatestEvents`].
-    async fn new(
-        weak_room: WeakRoom,
-        event_cache: &EventCache,
-    ) -> Result<Option<Self>, LatestEventsError> {
-        let room_id = weak_room.room_id();
-        let room_event_cache = match event_cache.for_room(room_id).await {
-            // It's fine to drop the `EventCacheDropHandles` here as the caller
-            // (`LatestEventState`) owns a clone of the `EventCache`.
-            Ok((room_event_cache, _drop_handles)) => room_event_cache,
-            Err(EventCacheError::RoomNotFound { .. }) => return Ok(None),
-            Err(err) => return Err(LatestEventsError::EventCache(err)),
-        };
-
-        Ok(Some(Self {
-            state: Arc::new(RwLock::new(RoomLatestEventsState {
-                for_the_room: Self::create_latest_event_for_inner(
-                    &weak_room,
-                    None,
-                    &room_event_cache,
-                )
-                .await,
-                per_thread: HashMap::new(),
-                weak_room,
-                room_event_cache,
-            })),
-        }))
-    }
-
-    async fn create_latest_event_for_inner(
-        weak_room: &WeakRoom,
-        thread_id: Option<&EventId>,
-        room_event_cache: &RoomEventCache,
-    ) -> LatestEvent {
-        LatestEvent::new(weak_room, thread_id, room_event_cache).await
-    }
-
-    /// Lock this type with shared read access, and return an owned lock guard.
-    async fn read(&self) -> RoomLatestEventsReadGuard {
-        RoomLatestEventsReadGuard { inner: self.state.clone().read_owned().await }
-    }
-
-    /// Lock this type with exclusive write access, and return an owned lock
-    /// guard.
-    async fn write(&self) -> RoomLatestEventsWriteGuard {
-        RoomLatestEventsWriteGuard { inner: self.state.clone().write_owned().await }
-    }
-}
-
-/// The state of [`RoomLatestEvents`].
-#[derive(Debug)]
-struct RoomLatestEventsState {
-    /// The latest event of the room.
-    for_the_room: LatestEvent,
-
-    /// The latest events for each thread.
-    per_thread: HashMap<OwnedEventId, LatestEvent>,
-
-    /// The room event cache associated to this room.
-    room_event_cache: RoomEventCache,
-
-    /// The (weak) room.
-    ///
-    /// It used to to get the power-levels of the user for this room when
-    /// computing the latest events.
-    weak_room: WeakRoom,
-}
-
-/// The owned lock guard returned by [`RoomLatestEvents::read`].
-struct RoomLatestEventsReadGuard {
-    inner: OwnedRwLockReadGuard<RoomLatestEventsState>,
-}
-
-impl RoomLatestEventsReadGuard {
-    /// Get the [`LatestEvent`] for the room.
-    fn for_room(&self) -> &LatestEvent {
-        &self.inner.for_the_room
-    }
-
-    /// Get the [`LatestEvent`] for the thread if it exists.
-    fn for_thread(&self, thread_id: &EventId) -> Option<&LatestEvent> {
-        self.inner.per_thread.get(thread_id)
-    }
-}
-
-/// The owned lock guard returned by [`RoomLatestEvents::write`].
-struct RoomLatestEventsWriteGuard {
-    inner: OwnedRwLockWriteGuard<RoomLatestEventsState>,
-}
-
-impl RoomLatestEventsWriteGuard {
-    async fn create_latest_event_for(&self, thread_id: Option<&EventId>) -> LatestEvent {
-        RoomLatestEvents::create_latest_event_for_inner(
-            &self.inner.weak_room,
-            thread_id,
-            &self.inner.room_event_cache,
-        )
-        .await
-    }
-
-    /// Check whether this [`RoomLatestEvents`] has a latest event for a
-    /// particular thread.
-    fn has_thread(&self, thread_id: &EventId) -> bool {
-        self.inner.per_thread.contains_key(thread_id)
-    }
-
-    /// Create the [`LatestEvent`] for thread `thread_id` and insert it in this
-    /// [`RoomLatestEvents`].
-    async fn create_and_insert_latest_event_for_thread(&mut self, thread_id: &EventId) {
-        let latest_event = self.create_latest_event_for(Some(thread_id)).await;
-
-        self.inner.per_thread.insert(thread_id.to_owned(), latest_event);
-    }
-
-    /// Forget the thread `thread_id`.
-    fn forget_thread(&mut self, thread_id: &EventId) {
-        self.inner.per_thread.remove(thread_id);
-    }
-
-    /// Update the latest events for the room and its threads, based on the
-    /// event cache data.
-    async fn update_with_event_cache(&mut self) {
-        // Get the power levels of the user for the current room if the `WeakRoom` is
-        // still valid.
-        //
-        // Get it once for all the updates of all the latest events for this room (be
-        // the room and its threads).
-        let room = self.inner.weak_room.get();
-        let power_levels = match &room {
-            Some(room) => {
-                let power_levels = room.power_levels().await.ok();
-
-                Some(room.own_user_id()).zip(power_levels)
-            }
-
-            None => None,
-        };
-
-        let inner = &mut *self.inner;
-        let for_the_room = &mut inner.for_the_room;
-        let per_thread = &mut inner.per_thread;
-        let room_event_cache = &inner.room_event_cache;
-
-        for_the_room.update_with_event_cache(room_event_cache, &power_levels).await;
-
-        for latest_event in per_thread.values_mut() {
-            latest_event.update_with_event_cache(room_event_cache, &power_levels).await;
-        }
-    }
-
-    /// Update the latest events for the room and its threads, based on the
-    /// send queue update.
-    async fn update_with_send_queue(&mut self, send_queue_update: &RoomSendQueueUpdate) {
-        // Get the power levels of the user for the current room if the `WeakRoom` is
-        // still valid.
-        //
-        // Get it once for all the updates of all the latest events for this room (be
-        // the room and its threads).
-        let room = self.inner.weak_room.get();
-        let power_levels = match &room {
-            Some(room) => {
-                let power_levels = room.power_levels().await.ok();
-
-                Some(room.own_user_id()).zip(power_levels)
-            }
-
-            None => None,
-        };
-
-        let inner = &mut *self.inner;
-        let for_the_room = &mut inner.for_the_room;
-        let per_thread = &mut inner.per_thread;
-        let room_event_cache = &inner.room_event_cache;
-
-        for_the_room
-            .update_with_send_queue(send_queue_update, room_event_cache, &power_levels)
-            .await;
-
-        for latest_event in per_thread.values_mut() {
-            latest_event
-                .update_with_send_queue(send_queue_update, room_event_cache, &power_levels)
-                .await;
-        }
-    }
-}
-
 /// The task responsible to listen to the [`EventCache`] and the [`SendQueue`].
 /// When an update is received and is considered relevant, a message is sent to
 /// the [`compute_latest_events_task`] to compute a new [`LatestEvent`].
@@ -880,9 +685,9 @@ mod tests {
             assert!(rooms.contains_key(room_id_1));
 
             // Room 0 contains zero thread latest events.
-            assert!(rooms.get(room_id_0).unwrap().read().await.inner.per_thread.is_empty());
+            assert!(rooms.get(room_id_0).unwrap().read().await.per_thread().is_empty());
             // Room 1 contains zero thread latest events.
-            assert!(rooms.get(room_id_1).unwrap().read().await.inner.per_thread.is_empty());
+            assert!(rooms.get(room_id_1).unwrap().read().await.per_thread().is_empty());
         }
 
         // Now let's listen to one thread respectively for two rooms.
@@ -899,17 +704,17 @@ mod tests {
             assert!(rooms.contains_key(room_id_2));
 
             // Room 0 contains zero thread latest events.
-            assert!(rooms.get(room_id_0).unwrap().read().await.inner.per_thread.is_empty());
+            assert!(rooms.get(room_id_0).unwrap().read().await.per_thread().is_empty());
             // Room 1 contains one thread latest event…
             let room_1 = rooms.get(room_id_1).unwrap().read().await;
-            assert_eq!(room_1.inner.per_thread.len(), 1);
+            assert_eq!(room_1.per_thread().len(), 1);
             // … which is thread 1.0.
-            assert!(room_1.inner.per_thread.contains_key(thread_id_1_0));
+            assert!(room_1.per_thread().contains_key(thread_id_1_0));
             // Room 2 contains one thread latest event…
             let room_2 = rooms.get(room_id_2).unwrap().read().await;
-            assert_eq!(room_2.inner.per_thread.len(), 1);
+            assert_eq!(room_2.per_thread().len(), 1);
             // … which is thread 2.0.
-            assert!(room_2.inner.per_thread.contains_key(thread_id_2_0));
+            assert!(room_2.per_thread().contains_key(thread_id_2_0));
         }
     }
 
@@ -939,7 +744,7 @@ mod tests {
             assert!(rooms.contains_key(room_id_0));
 
             // Room 0 contains zero thread latest events.
-            assert!(rooms.get(room_id_0).unwrap().read().await.inner.per_thread.is_empty());
+            assert!(rooms.get(room_id_0).unwrap().read().await.per_thread().is_empty());
         }
 
         // Now let's forget about room 0.
@@ -980,9 +785,9 @@ mod tests {
 
             // Room 0 contains one thread latest event…
             let room_0 = rooms.get(room_id_0).unwrap().read().await;
-            assert_eq!(room_0.inner.per_thread.len(), 1);
+            assert_eq!(room_0.per_thread().len(), 1);
             // … which is thread 0.0.
-            assert!(room_0.inner.per_thread.contains_key(thread_id_0_0));
+            assert!(room_0.per_thread().contains_key(thread_id_0_0));
         }
 
         // Now let's forget about the thread.
@@ -996,7 +801,7 @@ mod tests {
             assert!(rooms.contains_key(room_id_0));
 
             // But the thread has been removed.
-            assert!(rooms.get(room_id_0).unwrap().read().await.inner.per_thread.is_empty());
+            assert!(rooms.get(room_id_0).unwrap().read().await.per_thread().is_empty());
         }
     }