refactor(timeline): apply aggregations in a single place

Author: bnjbvr

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

@@ -40,10 +40,11 @@
 use std::collections::HashMap;
 
 use ruma::{MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedTransactionId, OwnedUserId};
-use tracing::warn;
+use tracing::{trace, warn};
 
+use super::{rfind_event_by_item_id, ObservableItemsTransaction};
 use crate::timeline::{
-    PollState, ReactionInfo, ReactionStatus, TimelineEventItemId, TimelineItemContent,
+    PollState, ReactionInfo, ReactionStatus, TimelineEventItemId, TimelineItem, TimelineItemContent,
 };
 
 /// Which kind of aggregation (related event) is this?
@@ -369,6 +370,41 @@ impl Aggregations {
     }
 }
 
+/// Find an item identified by the target identifier, and apply the aggregation
+/// onto it.
+///
+/// Returns whether the aggregation has been applied or not (so as to increment
+/// a number of updated result, for instance).
+pub(crate) fn find_item_and_apply_aggregation(
+    items: &mut ObservableItemsTransaction<'_>,
+    target: &TimelineEventItemId,
+    aggregation: Aggregation,
+) -> bool {
+    let Some((idx, event_item)) = rfind_event_by_item_id(items, &target) else {
+        warn!("couldn't find aggregation's target {target:?}");
+        return false;
+    };
+
+    let mut new_content = event_item.content().clone();
+
+    match aggregation.apply(&mut new_content) {
+        Ok(true) => {
+            trace!("applied aggregation");
+            let new_item = event_item.with_content(new_content);
+            items.replace(idx, TimelineItem::new(new_item, event_item.internal_id.to_owned()));
+            true
+        }
+        Ok(false) => {
+            trace!("applying the aggregation had no effect");
+            false
+        }
+        Err(err) => {
+            warn!("error when applying aggregation: {err}");
+            false
+        }
+    }
+}
+
 /// The result of marking an aggregation as sent.
 pub(crate) enum MarkAggregationSentResult {
     /// The aggregation has been found, and marked as sent.

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

@@ -1357,6 +1357,7 @@ impl<P: RoomDataProvider> TimelineController<P> {
         applies_to: OwnedTransactionId,
     ) {
         let mut state = self.state.write().await;
+        let mut tr = state.transaction();
 
         let target = TimelineEventItemId::TransactionId(applies_to);
 
@@ -1372,26 +1373,10 @@ impl<P: RoomDataProvider> TimelineController<P> {
             },
         );
 
-        state.meta.aggregations.add(target.clone(), aggregation.clone());
+        tr.meta.aggregations.add(target.clone(), aggregation.clone());
+        find_item_and_apply_aggregation(&mut tr.items, &target, aggregation);
 
-        let Some((item_pos, item)) = rfind_event_by_item_id(&state.items, &target) else {
-            warn!("Local item not found anymore.");
-            return;
-        };
-
-        let mut content = item.content().clone();
-        match aggregation.apply(&mut content) {
-            Ok(true) => {
-                trace!("added local reaction to local echo");
-                let internal_id = item.internal_id.clone();
-                let new_item = item.with_content(content);
-                state.items.replace(item_pos, TimelineItem::new(new_item, internal_id));
-            }
-            Ok(false) => {}
-            Err(err) => {
-                warn!("when applying local reaction to local echo: {err}");
-            }
-        }
+        tr.commit();
     }
 
     /// Handle a single room send queue update.

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

@@ -53,8 +53,9 @@ use tracing::{debug, error, field::debug, info, instrument, trace, warn};
 use super::{
     algorithms::{rfind_event_by_id, rfind_event_by_item_id},
     controller::{
-        Aggregation, AggregationKind, ObservableItemsTransaction, ObservableItemsTransactionEntry,
-        PendingEdit, PendingEditKind, TimelineMetadata, TimelineStateTransaction,
+        find_item_and_apply_aggregation, Aggregation, AggregationKind, ObservableItemsTransaction,
+        ObservableItemsTransactionEntry, PendingEdit, PendingEditKind, TimelineMetadata,
+        TimelineStateTransaction,
     },
     date_dividers::DateDividerAdjuster,
     event_item::{
@@ -411,23 +412,13 @@ impl<'a, 'o> TimelineEventHandler<'a, 'o> {
 
                 AnyMessageLikeEventContent::Sticker(content) => {
                     if should_add {
-                        let mut content = TimelineItemContent::Sticker(Sticker {
-                            content,
-                            reactions: Default::default(),
-                        });
-
-                        if let Some(event_id) = self.ctx.flow.event_id() {
-                            // Applying the cache to remote events only because local echoes
-                            // don't have an event ID that could be referenced by responses yet.
-                            if let Err(err) = self.meta.aggregations.apply(
-                                &TimelineEventItemId::EventId(event_id.to_owned()),
-                                &mut content,
-                            ) {
-                                warn!("discarding sticker aggregations: {err}");
-                            }
-                        }
-
-                        self.add_item(content, None);
+                        self.add_item(
+                            TimelineItemContent::Sticker(Sticker {
+                                content,
+                                reactions: Default::default(),
+                            }),
+                            None,
+                        );
                     }
                 }
 
@@ -587,15 +578,10 @@ impl<'a, 'o> TimelineEventHandler<'a, 'o> {
 
         let edit_json = edit_json.flatten();
 
-        let mut content =
-            TimelineItemContent::message(msg, edit_content, self.items, Default::default());
-        if let Err(err) =
-            self.meta.aggregations.apply(&self.ctx.flow.timeline_item_id(), &mut content)
-        {
-            warn!("discarding message aggregations: {err}");
-        }
-
-        self.add_item(content, edit_json);
+        self.add_item(
+            TimelineItemContent::message(msg, edit_content, self.items, Default::default()),
+            edit_json,
+        );
     }
 
     #[instrument(skip_all, fields(replacement_event_id = ?replacement.event_id))]
@@ -732,7 +718,7 @@ impl<'a, 'o> TimelineEventHandler<'a, 'o> {
     /// [`crate::timeline::TimelineController::handle_local_echo`].
     #[instrument(skip_all, fields(relates_to_event_id = ?c.relates_to.event_id))]
     fn handle_reaction(&mut self, c: ReactionEventContent) {
-        let reacted_to_event_id = &c.relates_to.event_id;
+        let target = TimelineEventItemId::EventId(c.relates_to.event_id);
 
         // Add the aggregation to the manager.
         let reaction_status = match &self.ctx.flow {
@@ -754,28 +740,10 @@ impl<'a, 'o> TimelineEventHandler<'a, 'o> {
                 reaction_status,
             },
         );
-        self.meta
-            .aggregations
-            .add(TimelineEventItemId::EventId(reacted_to_event_id.clone()), aggregation.clone());
 
-        let Some((idx, event_item)) = rfind_event_by_id(self.items, reacted_to_event_id) else {
-            warn!("couldn't find reaction's target {reacted_to_event_id:?}");
-            return;
-        };
-
-        let mut new_content = event_item.content().clone();
-        match aggregation.apply(&mut new_content) {
-            Ok(true) => {
-                trace!("added reaction");
-                let new_item = event_item.with_content(new_content);
-                self.items
-                    .replace(idx, TimelineItem::new(new_item, event_item.internal_id.to_owned()));
-                self.result.items_updated += 1;
-            }
-            Ok(false) => {}
-            Err(err) => {
-                warn!("error when applying reaction aggregation: {err}");
-            }
+        self.meta.aggregations.add(target.clone(), aggregation.clone());
+        if find_item_and_apply_aggregation(self.items, &target, aggregation) {
+            self.result.items_updated += 1;
         }
     }
 
@@ -872,21 +840,14 @@ impl<'a, 'o> TimelineEventHandler<'a, 'o> {
             .unzip();
 
         let poll_state = PollState::new(c, edit_content, Default::default());
-        let mut content = TimelineItemContent::Poll(poll_state);
-        if let Err(err) =
-            self.meta.aggregations.apply(&self.ctx.flow.timeline_item_id(), &mut content)
-        {
-            warn!("discarding poll aggregations: {err}");
-        }
 
         let edit_json = edit_json.flatten();
 
-        self.add_item(content, edit_json);
+        self.add_item(TimelineItemContent::Poll(poll_state), edit_json);
     }
 
     fn handle_poll_response(&mut self, c: UnstablePollResponseEventContent) {
-        let start_event_id = c.relates_to.event_id;
-
+        let target = TimelineEventItemId::EventId(c.relates_to.event_id);
         let aggregation = Aggregation::new(
             self.ctx.flow.timeline_item_id(),
             AggregationKind::PollResponse {
@@ -895,60 +856,21 @@ impl<'a, 'o> TimelineEventHandler<'a, 'o> {
                 answers: c.poll_response.answers,
             },
         );
-
-        self.meta
-            .aggregations
-            .add(TimelineEventItemId::EventId(start_event_id.clone()), aggregation.clone());
-
-        let Some((item_pos, item)) = rfind_event_by_id(self.items, &start_event_id) else {
-            return;
-        };
-
-        let mut new_content = item.content().clone();
-        match aggregation.apply(&mut new_content) {
-            Ok(true) => {
-                trace!("adding poll response.");
-                self.items.replace(
-                    item_pos,
-                    TimelineItem::new(item.with_content(new_content), item.internal_id.clone()),
-                );
-                self.result.items_updated += 1;
-            }
-            Ok(false) => {}
-            Err(err) => {
-                warn!("discarding poll response: {err}");
-            }
+        self.meta.aggregations.add(target.clone(), aggregation.clone());
+        if find_item_and_apply_aggregation(self.items, &target, aggregation) {
+            self.result.items_updated += 1;
         }
     }
 
     fn handle_poll_end(&mut self, c: UnstablePollEndEventContent) {
-        let start_event_id = c.relates_to.event_id;
-
+        let target = TimelineEventItemId::EventId(c.relates_to.event_id);
         let aggregation = Aggregation::new(
             self.ctx.flow.timeline_item_id(),
             AggregationKind::PollEnd { end_date: self.ctx.timestamp },
         );
-        self.meta
-            .aggregations
-            .add(TimelineEventItemId::EventId(start_event_id.clone()), aggregation.clone());
-
-        let Some((item_pos, item)) = rfind_event_by_id(self.items, &start_event_id) else {
-            return;
-        };
-
-        let mut new_content = item.content().clone();
-        match aggregation.apply(&mut new_content) {
-            Ok(true) => {
-                trace!("Ending poll.");
-                let new_item = item.with_content(new_content);
-                self.items
-                    .replace(item_pos, TimelineItem::new(new_item, item.internal_id.to_owned()));
-                self.result.items_updated += 1;
-            }
-            Ok(false) => {}
-            Err(err) => {
-                warn!("discarding poll end: {err}");
-            }
+        self.meta.aggregations.add(target.clone(), aggregation.clone());
+        if find_item_and_apply_aggregation(self.items, &target, aggregation) {
+            self.result.items_updated += 1;
         }
     }
 
@@ -1044,11 +966,18 @@ impl<'a, 'o> TimelineEventHandler<'a, 'o> {
     ///    timeline item being added here.
     fn add_item(
         &mut self,
-        content: TimelineItemContent,
+        mut content: TimelineItemContent,
         edit_json: Option<Raw<AnySyncTimelineEvent>>,
     ) {
         self.result.item_added = true;
 
+        // Apply any pending or stashed aggregations.
+        if let Err(err) =
+            self.meta.aggregations.apply(&self.ctx.flow.timeline_item_id(), &mut content)
+        {
+            warn!("discarding aggregations: {err}");
+        }
+
         let sender = self.ctx.sender.to_owned();
         let sender_profile = TimelineDetails::from_initial_value(self.ctx.sender_profile.clone());
         let timestamp = self.ctx.timestamp;