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

@@ -263,14 +263,7 @@ impl EventCache {
 
         room_cache
             .inner
-            .append_events(
-                &**store,
-                events,
-                prev_batch,
-                Default::default(),
-                Default::default(),
-                Default::default(),
-            )
+            .append_events(&**store, events, prev_batch, Default::default(), Default::default())
             .await?;
 
         Ok(())
@@ -477,6 +470,30 @@ 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 fact 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,
         store: &dyn EventCacheStore,
@@ -486,10 +503,12 @@ impl RoomEventCacheInner {
             store,
             updates.timeline,
             updates.ephemeral.clone(),
-            updates.account_data,
             updates.ambiguity_changes,
         )
         .await?;
+
+        self.handle_account_data(updates.account_data);
+
         Ok(())
     }
 
@@ -498,7 +517,6 @@ impl RoomEventCacheInner {
         store: &dyn EventCacheStore,
         timeline: Timeline,
         ephemeral: Vec<Raw<AnySyncEphemeralRoomEvent>>,
-        account_data: Vec<Raw<AnyRoomAccountDataEvent>>,
         ambiguity_changes: BTreeMap<OwnedEventId, AmbiguityChange>,
     ) -> Result<()> {
         if timeline.limited {
@@ -510,7 +528,7 @@ impl RoomEventCacheInner {
             // Clear internal state (events, pagination tokens, etc.).
             store.clear_room(self.room.room_id()).await?;
 
-            // Propagate to observers.
+            // Propagate to observers. We ignore the fact there aren't any.
             let _ = self.sender.send(RoomEventCacheUpdate::Clear);
         }
 
@@ -520,7 +538,6 @@ impl RoomEventCacheInner {
             store,
             timeline.events,
             timeline.prev_batch,
-            account_data,
             ephemeral,
             ambiguity_changes,
         )
@@ -534,14 +551,8 @@ impl RoomEventCacheInner {
         store: &dyn EventCacheStore,
         updates: LeftRoomUpdate,
     ) -> Result<()> {
-        self.handle_timeline(
-            store,
-            updates.timeline,
-            Vec::new(),
-            Vec::new(),
-            updates.ambiguity_changes,
-        )
-        .await?;
+        self.handle_timeline(store, updates.timeline, Vec::new(), updates.ambiguity_changes)
+            .await?;
         Ok(())
     }
 
@@ -552,14 +563,12 @@ impl RoomEventCacheInner {
         store: &dyn EventCacheStore,
         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(());
@@ -585,12 +594,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(())
     }
@@ -738,13 +743,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 has been moved.
+        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>>,