feat(timeline): handle live aggregations on non-live timelines (#5060)

This makes it possible to handle reactions/redactions/edits/etc. on
non-live timelines. As a result, the pinned and focused timelines will
now get live reactions/redactions and so on. This makes it possible to
also have the thread timelines handle those live events, although it's
unclear how it will pane out in the end, when the event cache is also
involved.

Author: bnjbvr

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

@@ -392,26 +392,22 @@ async fn room_event_cache_updates_task(
 
             RoomEventCacheUpdate::UpdateTimelineEvents { diffs, origin } => {
                 trace!("Received new timeline events diffs");
-
-                // We shouldn't use the general way of adding events to timelines to
-                // non-live timelines, such as pinned events or focused timeline.
-                // These timelines should handle any live updates by themselves.
-                if !is_live {
-                    continue;
+                let origin = match origin {
+                    EventsOrigin::Sync => RemoteEventOrigin::Sync,
+                    EventsOrigin::Pagination => RemoteEventOrigin::Pagination,
+                    EventsOrigin::Cache => RemoteEventOrigin::Cache,
+                };
+
+                let has_diffs = !diffs.is_empty();
+
+                if is_live {
+                    timeline_controller.handle_remote_events_with_diffs(diffs, origin).await;
+                } else {
+                    // Only handle the remote aggregation for a non-live timeline.
+                    timeline_controller.handle_remote_aggregations(diffs, origin).await;
                 }
 
-                timeline_controller
-                    .handle_remote_events_with_diffs(
-                        diffs,
-                        match origin {
-                            EventsOrigin::Sync => RemoteEventOrigin::Sync,
-                            EventsOrigin::Pagination => RemoteEventOrigin::Pagination,
-                            EventsOrigin::Cache => RemoteEventOrigin::Cache,
-                        },
-                    )
-                    .await;
-
-                if matches!(origin, EventsOrigin::Cache) {
+                if has_diffs && matches!(origin, RemoteEventOrigin::Cache) {
                     timeline_controller.retry_event_decryption(None).await;
                 }
             }

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

@@ -753,6 +753,22 @@ impl<P: RoomDataProvider, D: Decryptor> TimelineController<P, D> {
             .await
     }
 
+    /// Only handle aggregations received as [`VectorDiff`]s.
+    pub(super) async fn handle_remote_aggregations(
+        &self,
+        diffs: Vec<VectorDiff<TimelineEvent>>,
+        origin: RemoteEventOrigin,
+    ) {
+        if diffs.is_empty() {
+            return;
+        }
+
+        let mut state = self.state.write().await;
+        state
+            .handle_remote_aggregations(diffs, origin, &self.room_data_provider, &self.settings)
+            .await
+    }
+
     pub(super) async fn clear(&self) {
         self.state.write().await.clear();
     }

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

@@ -91,6 +91,25 @@ impl TimelineState {
         transaction.commit();
     }
 
+    /// Handle remote aggregations on events as [`VectorDiff`]s.
+    pub(super) async fn handle_remote_aggregations<RoomData>(
+        &mut self,
+        diffs: Vec<VectorDiff<TimelineEvent>>,
+        origin: RemoteEventOrigin,
+        room_data: &RoomData,
+        settings: &TimelineSettings,
+    ) where
+        RoomData: RoomDataProvider,
+    {
+        if diffs.is_empty() {
+            return;
+        }
+
+        let mut transaction = self.transaction();
+        transaction.handle_remote_aggregations(diffs, origin, room_data, settings).await;
+        transaction.commit();
+    }
+
     /// 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) {

crates/matrix-sdk-ui/src/timeline/controller/state_transaction.rs

@@ -177,6 +177,159 @@ impl<'a> TimelineStateTransaction<'a> {
         self.check_invariants();
     }
 
+    async fn handle_remote_aggregation<RoomData>(
+        &mut self,
+        event: TimelineEvent,
+        position: TimelineItemPosition,
+        room_data_provider: &RoomData,
+        date_divider_adjuster: &mut DateDividerAdjuster,
+    ) where
+        RoomData: RoomDataProvider,
+    {
+        let deserialized = match event.raw().deserialize() {
+            Ok(deserialized) => deserialized,
+            Err(err) => {
+                warn!("Failed to deserialize timeline event: {err}");
+                return;
+            }
+        };
+
+        let sender = deserialized.sender().to_owned();
+        let timestamp = deserialized.origin_server_ts();
+        let event_id = deserialized.event_id().to_owned();
+        let txn_id = deserialized.transaction_id().map(ToOwned::to_owned);
+
+        if let Some(action @ TimelineAction::HandleAggregation { .. }) = TimelineAction::from_event(
+            deserialized,
+            event.raw(),
+            room_data_provider,
+            None,
+            &self.items,
+            &mut self.meta,
+        )
+        .await
+        {
+            let encryption_info = event.kind.encryption_info().cloned();
+
+            let sender_profile = room_data_provider.profile_from_user_id(&sender).await;
+
+            let ctx = TimelineEventContext {
+                sender,
+                sender_profile,
+                timestamp,
+                // These are not used when handling an aggregation.
+                read_receipts: Default::default(),
+                is_highlighted: false,
+                flow: Flow::Remote {
+                    event_id: event_id.clone(),
+                    raw_event: event.raw().clone(),
+                    encryption_info,
+                    txn_id,
+                    position,
+                },
+                // This field is not used when handling an aggregation.
+                should_add_new_items: false,
+            };
+
+            TimelineEventHandler::new(self, ctx).handle_event(date_divider_adjuster, action).await;
+        }
+    }
+
+    /// Handle a set of live remote aggregations on events as [`VectorDiff`]s.
+    ///
+    /// This is like `handle_remote_events`, with two key differences:
+    /// - it only applies to aggregated events, not all the sync events.
+    /// - it will also not add the events to the `all_remote_events` array
+    ///   itself.
+    pub(super) async fn handle_remote_aggregations<RoomData>(
+        &mut self,
+        diffs: Vec<VectorDiff<TimelineEvent>>,
+        origin: RemoteEventOrigin,
+        room_data_provider: &RoomData,
+        settings: &TimelineSettings,
+    ) where
+        RoomData: RoomDataProvider,
+    {
+        let mut date_divider_adjuster =
+            DateDividerAdjuster::new(settings.date_divider_mode.clone());
+
+        for diff in diffs {
+            match diff {
+                VectorDiff::Append { values: events } => {
+                    for event in events {
+                        self.handle_remote_aggregation(
+                            event,
+                            TimelineItemPosition::End { origin },
+                            room_data_provider,
+                            &mut date_divider_adjuster,
+                        )
+                        .await;
+                    }
+                }
+
+                VectorDiff::PushFront { value: event } => {
+                    self.handle_remote_aggregation(
+                        event,
+                        TimelineItemPosition::Start { origin },
+                        room_data_provider,
+                        &mut date_divider_adjuster,
+                    )
+                    .await;
+                }
+
+                VectorDiff::PushBack { value: event } => {
+                    self.handle_remote_aggregation(
+                        event,
+                        TimelineItemPosition::End { origin },
+                        room_data_provider,
+                        &mut date_divider_adjuster,
+                    )
+                    .await;
+                }
+
+                VectorDiff::Insert { index: event_index, value: event } => {
+                    self.handle_remote_aggregation(
+                        event,
+                        TimelineItemPosition::At { event_index, origin },
+                        room_data_provider,
+                        &mut date_divider_adjuster,
+                    )
+                    .await;
+                }
+
+                VectorDiff::Set { index: event_index, value: event } => {
+                    if let Some(timeline_item_index) = self
+                        .items
+                        .all_remote_events()
+                        .get(event_index)
+                        .and_then(|meta| meta.timeline_item_index)
+                    {
+                        self.handle_remote_aggregation(
+                            event,
+                            TimelineItemPosition::UpdateAt { timeline_item_index },
+                            room_data_provider,
+                            &mut date_divider_adjuster,
+                        )
+                        .await;
+                    } else {
+                        warn!(event_index, "Set update dropped because there wasn't any attached timeline item index.");
+                    }
+                }
+
+                VectorDiff::Remove { .. } | VectorDiff::Clear => {
+                    // Do nothing. An aggregated redaction comes with a
+                    // redaction event, or as a redacted event in the first
+                    // place.
+                }
+
+                v => unimplemented!("{v:?}"),
+            }
+        }
+
+        self.adjust_date_dividers(date_divider_adjuster);
+        self.check_invariants();
+    }
+
     fn check_invariants(&self) {
         self.check_no_duplicate_read_receipts();
         self.check_no_unused_unique_ids();

crates/matrix-sdk-ui/tests/integration/timeline/focus_event.rs

@@ -185,7 +185,7 @@ async fn test_new_focused() {
 }
 
 #[async_test]
-async fn test_focused_timeline_does_not_react() {
+async fn test_live_aggregations_are_reflected_on_focused_timelines() {
     let room_id = room_id!("!a98sd12bjh:example.org");
     let (client, server) = logged_in_client_with_server().await;
     let sync_settings = SyncSettings::new().timeout(Duration::from_millis(3000));
@@ -254,8 +254,16 @@ async fn test_focused_timeline_does_not_react() {
     let _response = client.sync_once(sync_settings.clone()).await.unwrap();
     server.reset().await;
 
-    // Nothing was received by the focused event timeline
-    assert_pending!(timeline_stream);
+    // We only receive one updated for the reaction, from the timeline stream.
+    assert_let!(Some(timeline_updates) = timeline_stream.next().await);
+    assert_eq!(timeline_updates.len(), 1);
+    assert_let!(VectorDiff::Set { index: 1, value: item } = &timeline_updates[0]);
+
+    let event_item = item.as_event().unwrap();
+    assert_eq!(event_item.content().as_message().unwrap().body(), "yolo");
+    let reactions = event_item.content().reactions().cloned().unwrap_or_default();
+    assert_eq!(reactions.len(), 1);
+    let _ = reactions["👍"][*BOB];
 }
 
 #[async_test]

crates/matrix-sdk-ui/tests/integration/timeline/pinned_event.rs

@@ -440,7 +440,7 @@ async fn test_pinned_timeline_with_pinned_utd_on_sync_contains_it() {
 }
 
 #[async_test]
-async fn test_edited_events_are_not_reflected_in_sync() {
+async fn test_edited_events_are_reflected_in_sync() {
     let server = MatrixMockServer::new().await;
     let client = server.client_builder().build().await;
     let room_id = room_id!("!test:localhost");
@@ -452,11 +452,11 @@ async fn test_edited_events_are_not_reflected_in_sync() {
         .server_ts(MilliSecondsSinceUnixEpoch::now())
         .into_raw_sync();
 
-    // Mock /event for some timeline events
+    // Mock /event for some timeline events.
     mock_events_endpoint(&server, room_id, vec![pinned_event]).await;
 
     // Load initial timeline items: a text message and a `m.room.pinned_events` with
-    // event $1
+    // event $1.
     let room = PinnedEventsSync::new(room_id)
         .with_pinned_event_ids(vec!["$1"])
         .mock_and_sync(&client, &server)
@@ -470,7 +470,7 @@ async fn test_edited_events_are_not_reflected_in_sync() {
         "there should be no live back-pagination status for a focused timeline"
     );
 
-    // Load timeline items
+    // Load timeline items.
     let (items, mut timeline_stream) = timeline.subscribe().await;
 
     assert_eq!(items.len(), 1 + 1); // event item + a date divider
@@ -479,31 +479,40 @@ async fn test_edited_events_are_not_reflected_in_sync() {
     assert_pending!(timeline_stream);
 
     let edited_event = f
-        .text_msg("edited message!")
+        .text_msg("* edited message!")
         .edit(
             event_id!("$1"),
-            RoomMessageEventContentWithoutRelation::text_plain("* edited message!"),
+            RoomMessageEventContentWithoutRelation::text_plain("edited message!"),
         )
         .event_id(event_id!("$2"))
         .server_ts(MilliSecondsSinceUnixEpoch::now())
         .into_raw_sync();
 
-    // Mock /event for some timeline events
+    // Mock /event for some timeline events.
     mock_events_endpoint(&server, room_id, vec![edited_event.clone()]).await;
 
-    // Load new pinned event contents from sync, where $2 is and edit on $1
+    // Load new pinned event contents from sync, where $2 is and edit on $1.
     let _ = PinnedEventsSync::new(room_id)
         .with_timeline_events(vec![edited_event])
         .mock_and_sync(&client, &server)
         .await
         .expect("Sync failed");
 
-    // The edit does not replace the original event
+    assert_let!(Some(timeline_updates) = timeline_stream.next().await);
+    assert_eq!(timeline_updates.len(), 1);
+
+    // The edit does replace the original event.
+    assert_let!(VectorDiff::Set { index: 1, value } = &timeline_updates[0]);
+    let event = value.as_event().unwrap();
+    assert_eq!(event.event_id().unwrap(), event_id!("$1"));
+    assert_eq!(event.content().as_message().unwrap().body(), "edited message!");
+
+    // That's all, folks!
     assert_pending!(timeline_stream);
 }
 
 #[async_test]
-async fn test_redacted_events_are_not_reflected_in_sync() {
+async fn test_redacted_events_are_reflected_in_sync() {
     let server = MatrixMockServer::new().await;
     let client = server.client_builder().build().await;
     let room_id = room_id!("!test:localhost");
@@ -557,7 +566,15 @@ async fn test_redacted_events_are_not_reflected_in_sync() {
         .await
         .expect("Sync failed");
 
-    // The redaction does not replace the original event
+    assert_let!(Some(timeline_updates) = timeline_stream.next().await);
+    assert_eq!(timeline_updates.len(), 1);
+
+    // The redaction takes place.
+    assert_let!(VectorDiff::Set { index: 1, value } = &timeline_updates[0]);
+    let event = value.as_event().unwrap();
+    assert!(event.content().is_redacted());
+
+    // That's all, folks!
     assert_pending!(timeline_stream);
 }