event cache: internalize handling of the account data

(By moving handling of the fully read marker into the event cache
itself.)

Author: bnjbvr

crates/matrix-sdk-ui/src/timeline/builder.rs

@@ -185,18 +185,17 @@ impl TimelineBuilder {
                             inner.clear().await;
                         }
 
-                        RoomEventCacheUpdate::Append {
-                            events,
-                            account_data,
-                            ephemeral,
-                            ambiguity_changes,
-                        } => {
+                        RoomEventCacheUpdate::UpdateReadMarker { event_id } => {
+                            trace!(target = %event_id, "Handling fully read marker.");
+                            inner.handle_fully_read_marker(event_id).await;
+                        }
+
+                        RoomEventCacheUpdate::Append { events, ephemeral, ambiguity_changes } => {
                             trace!("Received new events");
 
-                            // TODO: (bnjbvr) account_data and ephemeral should be handled by the
-                            // event cache, and we should replace this with a simple
-                            // `handle_add_events`.
-                            inner.handle_sync_events(events, account_data, ephemeral).await;
+                            // TODO: (bnjbvr) ephemeral should be handled by the event cache, and
+                            // we should replace this with a simple `add_events_at`.
+                            inner.handle_sync_events(events, ephemeral).await;
 
                             let member_ambiguity_changes = ambiguity_changes
                                 .values()

crates/matrix-sdk-ui/src/timeline/inner/mod.rs

@@ -41,8 +41,8 @@ use ruma::{
             message::{MessageType, Relation},
             redaction::RoomRedactionEventContent,
         },
-        AnyMessageLikeEventContent, AnyRoomAccountDataEvent, AnySyncEphemeralRoomEvent,
-        AnySyncMessageLikeEvent, AnySyncTimelineEvent, MessageLikeEventType,
+        AnyMessageLikeEventContent, AnySyncEphemeralRoomEvent, AnySyncMessageLikeEvent,
+        AnySyncTimelineEvent, MessageLikeEventType,
     },
     serde::Raw,
     EventId, OwnedEventId, OwnedTransactionId, RoomVersionId, TransactionId, UserId,
@@ -426,22 +426,17 @@ impl<P: RoomDataProvider> TimelineInner<P> {
         self.state.write().await.clear();
     }
 
+    pub(super) async fn handle_fully_read_marker(&self, fully_read_event_id: OwnedEventId) {
+        self.state.write().await.handle_fully_read_marker(fully_read_event_id);
+    }
+
     pub(super) async fn handle_sync_events(
         &self,
         events: Vec<SyncTimelineEvent>,
-        account_data: Vec<Raw<AnyRoomAccountDataEvent>>,
         ephemeral: Vec<Raw<AnySyncEphemeralRoomEvent>>,
     ) {
         let mut state = self.state.write().await;
-        state
-            .handle_sync_events(
-                events,
-                account_data,
-                ephemeral,
-                &self.room_data_provider,
-                &self.settings,
-            )
-            .await;
+        state.handle_sync_events(events, ephemeral, &self.room_data_provider, &self.settings).await;
     }
 
     #[cfg(test)]

crates/matrix-sdk-ui/src/timeline/inner/state.rs

@@ -23,7 +23,7 @@ use ruma::events::receipt::ReceiptEventContent;
 use ruma::{
     events::{
         relation::Annotation, room::redaction::RoomRedactionEventContent,
-        AnyMessageLikeEventContent, AnyRoomAccountDataEvent, AnySyncEphemeralRoomEvent,
+        AnyMessageLikeEventContent, AnySyncEphemeralRoomEvent,
     },
     push::Action,
     serde::Raw,
@@ -109,11 +109,18 @@ impl TimelineInnerState {
         handle_many_res
     }
 
+    /// Marks the given event as fully read, using the read marker received from
+    /// sync.
+    pub(super) fn handle_fully_read_marker(&mut self, fully_read_event_id: OwnedEventId) {
+        let mut txn = self.transaction();
+        txn.set_fully_read_event(fully_read_event_id);
+        txn.commit();
+    }
+
     #[instrument(skip_all)]
     pub(super) async fn handle_sync_events<P: RoomDataProvider>(
         &mut self,
         events: Vec<SyncTimelineEvent>,
-        account_data: Vec<Raw<AnyRoomAccountDataEvent>>,
         ephemeral: Vec<Raw<AnySyncEphemeralRoomEvent>>,
         room_data_provider: &P,
         settings: &TimelineInnerSettings,
@@ -128,20 +135,6 @@ impl TimelineInnerState {
         )
         .await;
 
-        trace!("Handling account data");
-        for raw_event in account_data {
-            match raw_event.deserialize() {
-                Ok(AnyRoomAccountDataEvent::FullyRead(ev)) => {
-                    txn.set_fully_read_event(ev.content.event_id);
-                }
-                Ok(_) => {}
-                Err(e) => {
-                    let event_type = raw_event.get_field::<String>("type").ok().flatten();
-                    warn!(event_type, "Failed to deserialize account data: {e}");
-                }
-            }
-        }
-
         if !ephemeral.is_empty() {
             trace!("Handling ephemeral room events");
             let own_user_id = room_data_provider.own_user_id();

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

@@ -252,13 +252,7 @@ impl EventCache {
 
         room_cache
             .inner
-            .replace_all_events_by(
-                events,
-                prev_batch,
-                Default::default(),
-                Default::default(),
-                Default::default(),
-            )
+            .replace_all_events_by(events, prev_batch, Default::default(), Default::default())
             .await?;
 
         Ok(())
@@ -462,22 +456,47 @@ impl RoomEventCacheInner {
         }
     }
 
+    fn handle_account_data(&self, account_data: Vec<Raw<AnyRoomAccountDataEvent>>) {
+        trace!("Handling account data");
+        for raw_event in account_data {
+            match raw_event.deserialize() {
+                Ok(AnyRoomAccountDataEvent::FullyRead(ev)) => {
+                    // Propagate to observers. (We ignore the error if there aren't any.)
+                    let _ = self.sender.send(RoomEventCacheUpdate::UpdateReadMarker {
+                        event_id: ev.content.event_id,
+                    });
+                }
+
+                Ok(_) => {
+                    // We're not interested in other room account data updates,
+                    // at this point.
+                }
+
+                Err(e) => {
+                    let event_type = raw_event.get_field::<String>("type").ok().flatten();
+                    warn!(event_type, "Failed to deserialize account data: {e}");
+                }
+            }
+        }
+    }
+
     async fn handle_joined_room_update(&self, updates: JoinedRoomUpdate) -> Result<()> {
         self.handle_timeline(
             updates.timeline,
             updates.ephemeral.clone(),
-            updates.account_data,
             updates.ambiguity_changes,
         )
         .await?;
+
+        self.handle_account_data(updates.account_data);
+
         Ok(())
     }
 
     async fn handle_timeline(
         &self,
         timeline: Timeline,
         ephemeral: Vec<Raw<AnySyncEphemeralRoomEvent>>,
-        account_data: Vec<Raw<AnyRoomAccountDataEvent>>,
         ambiguity_changes: BTreeMap<OwnedEventId, AmbiguityChange>,
     ) -> Result<()> {
         if timeline.limited {
@@ -489,7 +508,6 @@ impl RoomEventCacheInner {
             self.replace_all_events_by(
                 timeline.events,
                 timeline.prev_batch,
-                account_data,
                 ephemeral,
                 ambiguity_changes,
             )
@@ -501,7 +519,6 @@ impl RoomEventCacheInner {
             self.append_new_events(
                 timeline.events,
                 timeline.prev_batch,
-                account_data,
                 ephemeral,
                 ambiguity_changes,
             )
@@ -512,8 +529,7 @@ impl RoomEventCacheInner {
     }
 
     async fn handle_left_room_update(&self, updates: LeftRoomUpdate) -> Result<()> {
-        self.handle_timeline(updates.timeline, Vec::new(), Vec::new(), updates.ambiguity_changes)
-            .await?;
+        self.handle_timeline(updates.timeline, Vec::new(), updates.ambiguity_changes).await?;
         Ok(())
     }
 
@@ -523,7 +539,6 @@ impl RoomEventCacheInner {
         &self,
         events: Vec<SyncTimelineEvent>,
         prev_batch: Option<String>,
-        account_data: Vec<Raw<AnyRoomAccountDataEvent>>,
         ephemeral: Vec<Raw<AnySyncEphemeralRoomEvent>>,
         ambiguity_changes: BTreeMap<OwnedEventId, AmbiguityChange>,
     ) -> Result<()> {
@@ -541,7 +556,6 @@ impl RoomEventCacheInner {
             room_events,
             events,
             prev_batch,
-            account_data,
             ephemeral,
             ambiguity_changes,
         )
@@ -554,15 +568,13 @@ impl RoomEventCacheInner {
         &self,
         events: Vec<SyncTimelineEvent>,
         prev_batch: Option<String>,
-        account_data: Vec<Raw<AnyRoomAccountDataEvent>>,
         ephemeral: Vec<Raw<AnySyncEphemeralRoomEvent>>,
         ambiguity_changes: BTreeMap<OwnedEventId, AmbiguityChange>,
     ) -> Result<()> {
         self.append_events_locked_impl(
             self.events.write().await,
             events,
             prev_batch,
-            account_data,
             ephemeral,
             ambiguity_changes,
         )
@@ -579,14 +591,12 @@ impl RoomEventCacheInner {
         mut room_events: RwLockWriteGuard<'_, RoomEvents>,
         events: Vec<SyncTimelineEvent>,
         prev_batch: Option<String>,
-        account_data: Vec<Raw<AnyRoomAccountDataEvent>>,
         ephemeral: Vec<Raw<AnySyncEphemeralRoomEvent>>,
         ambiguity_changes: BTreeMap<OwnedEventId, AmbiguityChange>,
     ) -> Result<()> {
         if events.is_empty()
             && prev_batch.is_none()
             && ephemeral.is_empty()
-            && account_data.is_empty()
             && ambiguity_changes.is_empty()
         {
             return Ok(());
@@ -608,12 +618,8 @@ impl RoomEventCacheInner {
             self.pagination_token_notifier.notify_one();
         }
 
-        let _ = self.sender.send(RoomEventCacheUpdate::Append {
-            events,
-            account_data,
-            ephemeral,
-            ambiguity_changes,
-        });
+        let _ =
+            self.sender.send(RoomEventCacheUpdate::Append { events, ephemeral, ambiguity_changes });
 
         Ok(())
     }
@@ -819,13 +825,16 @@ pub enum RoomEventCacheUpdate {
     /// The room has been cleared from events.
     Clear,
 
+    /// The fully read marker has moved to a different event.
+    UpdateReadMarker {
+        /// Event at which the read marker is now pointing.
+        event_id: OwnedEventId,
+    },
+
     /// The room has new events.
     Append {
         /// All the new events that have been added to the room's timeline.
         events: Vec<SyncTimelineEvent>,
-        /// XXX: this is temporary, until account data lives in the event cache
-        /// — or will it live there?
-        account_data: Vec<Raw<AnyRoomAccountDataEvent>>,
         /// XXX: this is temporary, until read receipts are handled in the event
         /// cache
         ephemeral: Vec<Raw<AnySyncEphemeralRoomEvent>>,