feat(send_queue): send redactions via the send queue

Signed-off-by: Johannes Marbach <n0-0ne+github@mailbox.org>

Author: Johennes

crates/matrix-sdk-base/src/store/send_queue.rs

@@ -117,6 +117,14 @@ pub enum QueuedRequestKind {
         #[serde(default)]
         accumulated: Vec<AccumulatedSentMediaInfo>,
     },
+
+    /// A redaction of another event to send.
+    Redaction {
+        /// The ID of the event to redact.
+        redacts: OwnedEventId,
+        /// The reason for the event being redacted.
+        reason: Option<String>,
+    },
 }
 
 impl From<SerializableEventContent> for QueuedRequestKind {
@@ -421,12 +429,28 @@ pub enum SentRequestKey {
 
     /// The parent transaction returned an uploaded resource URL.
     Media(SentMediaInfo),
+
+    /// The parent transaction returned a redaction event when it succeeded.
+    Redaction {
+        /// The event ID returned by the server.
+        event_id: OwnedEventId,
+
+        /// The ID of the redacted event.
+        redacts: OwnedEventId,
+
+        /// The reason for the event being redacted.
+        reason: Option<String>,
+    },
 }
 
 impl SentRequestKey {
     /// Converts the current parent key into an event id, if possible.
     pub fn into_event_id(self) -> Option<OwnedEventId> {
-        as_variant!(self, Self::Event { event_id, .. } => event_id)
+        match self {
+            Self::Event { event_id, .. } => Some(event_id),
+            Self::Redaction { event_id, .. } => Some(event_id),
+            _ => None,
+        }
     }
 
     /// Converts the current parent key into information about a sent media, if

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

@@ -126,7 +126,12 @@ pub(crate) enum AggregationKind {
     },
 
     /// An event has been redacted.
-    Redaction,
+    Redaction {
+        /// Whether this aggregation results from the local echo of a redaction.
+        /// Local echoes of redactions are applied reversibly whereas remote
+        /// echoes of redactions are applied irreversibly.
+        is_local: bool,
+    },
 
     /// An event has been edited.
     ///
@@ -237,8 +242,15 @@ impl Aggregation {
                 }
             }
 
-            AggregationKind::Redaction => {
-                if event.content().is_redacted() {
+            AggregationKind::Redaction { is_local } => {
+                if *is_local {
+                    if event.is_being_redacted {
+                        ApplyAggregationResult::LeftItemIntact
+                    } else {
+                        *event = Cow::Owned(event.with_is_being_redacted(true));
+                        ApplyAggregationResult::UpdatedItem
+                    }
+                } else if event.content().is_redacted() {
                     ApplyAggregationResult::LeftItemIntact
                 } else {
                     let new_item = event.redact(&rules.redaction);
@@ -352,9 +364,18 @@ impl Aggregation {
                 ApplyAggregationResult::Error(AggregationError::CantUndoPollEnd)
             }
 
-            AggregationKind::Redaction => {
-                // Redactions are not reversible.
-                ApplyAggregationResult::Error(AggregationError::CantUndoRedaction)
+            AggregationKind::Redaction { is_local } => {
+                if *is_local {
+                    if event.is_being_redacted {
+                        *event = Cow::Owned(event.with_is_being_redacted(false));
+                        ApplyAggregationResult::UpdatedItem
+                    } else {
+                        ApplyAggregationResult::LeftItemIntact
+                    }
+                } else {
+                    // Remote redactions are not reversible.
+                    ApplyAggregationResult::Error(AggregationError::CantUndoRedaction)
+                }
             }
 
             AggregationKind::Reaction { key, sender, .. } => {
@@ -477,7 +498,7 @@ impl Aggregations {
     pub fn add(&mut self, related_to: TimelineEventItemId, aggregation: Aggregation) {
         // If the aggregation is a redaction, it invalidates all the other aggregations;
         // remove them.
-        if matches!(aggregation.kind, AggregationKind::Redaction) {
+        if matches!(aggregation.kind, AggregationKind::Redaction { .. }) {
             for agg in self.related_events.remove(&related_to).unwrap_or_default() {
                 self.inverted_map.remove(&agg.own_id);
             }
@@ -488,7 +509,7 @@ impl Aggregations {
         if let Some(previous_aggregations) = self.related_events.get(&related_to)
             && previous_aggregations
                 .iter()
-                .any(|agg| matches!(agg.kind, AggregationKind::Redaction))
+                .any(|agg| matches!(agg.kind, AggregationKind::Redaction { .. }))
         {
             return;
         }
@@ -698,7 +719,7 @@ impl Aggregations {
                 AggregationKind::PollResponse { .. }
                 | AggregationKind::PollEnd { .. }
                 | AggregationKind::Edit(..)
-                | AggregationKind::Redaction
+                | AggregationKind::Redaction { .. }
                 | AggregationKind::BeaconUpdate { .. }
                 | AggregationKind::BeaconStop { .. } => {
                     // Nothing particular to do.

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

@@ -241,6 +241,7 @@ mod tests {
                 TimelineItemContent::MsgLike(MsgLikeContent::redacted()),
                 event_kind,
                 true,
+                false,
             )),
             TimelineUniqueId("local".to_owned()),
         )
@@ -284,6 +285,7 @@ mod tests {
                 )),
                 event_kind,
                 true,
+                false,
             )),
             TimelineUniqueId("local".to_owned()),
         )
@@ -331,6 +333,7 @@ mod tests {
                 ),
                 event_kind,
                 true,
+                false,
             )),
             TimelineUniqueId("local".to_owned()),
         )

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

@@ -1268,6 +1268,23 @@ impl<P: RoomDataProvider> TimelineController<P> {
             LocalEchoContent::React { key, send_handle, applies_to } => {
                 self.handle_local_reaction(key, send_handle, applies_to).await;
             }
+
+            LocalEchoContent::Redaction { redacts, send_error, .. } => {
+                self.handle_local_redaction(echo.transaction_id.clone(), redacts).await;
+
+                if let Some(send_error) = send_error {
+                    self.update_event_send_state(
+                        &echo.transaction_id,
+                        EventSendState::SendingFailed {
+                            error: Arc::new(matrix_sdk::Error::SendQueueWedgeError(Box::new(
+                                send_error,
+                            ))),
+                            is_recoverable: false,
+                        },
+                    )
+                    .await;
+                }
+            }
         }
     }
 
@@ -1308,6 +1325,34 @@ impl<P: RoomDataProvider> TimelineController<P> {
         tr.commit();
     }
 
+    /// Applies a local echo of a redaction.
+    pub(super) async fn handle_local_redaction(
+        &self,
+        txn_id: OwnedTransactionId,
+        redacts: OwnedEventId,
+    ) {
+        let mut state = self.state.write().await;
+        let mut tr = state.transaction();
+
+        let target = TimelineEventItemId::EventId(redacts);
+
+        let aggregation = Aggregation::new(
+            TimelineEventItemId::TransactionId(txn_id),
+            AggregationKind::Redaction { is_local: true },
+        );
+
+        tr.meta.aggregations.add(target.clone(), aggregation.clone());
+        find_item_and_apply_aggregation(
+            &tr.meta.aggregations,
+            &mut tr.items,
+            &target,
+            aggregation,
+            &tr.meta.room_version_rules,
+        );
+
+        tr.commit();
+    }
+
     /// Handle a single room send queue update.
     pub(crate) async fn handle_room_send_queue_update(&self, update: RoomSendQueueUpdate) {
         match update {

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

@@ -760,6 +760,7 @@ mod observable_items_tests {
                     origin: RemoteEventOrigin::Sync,
                 }),
                 false,
+                false,
             ),
             TimelineUniqueId(format!("__eid_{event_id}")),
         )
@@ -790,6 +791,7 @@ mod observable_items_tests {
                     send_handle: None,
                 }),
                 false,
+                false,
             ),
             TimelineUniqueId(format!("__tid_{transaction_id}")),
         )

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

@@ -689,6 +689,7 @@ mod tests {
             TimelineItemContent::MsgLike(MsgLikeContent::redacted()),
             event_kind,
             false,
+            false,
         )
     }
 

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

@@ -830,8 +830,13 @@ impl<'a, 'o> TimelineEventHandler<'a, 'o> {
         }
 
         let target = TimelineEventItemId::EventId(redacted.clone());
-        let aggregation =
-            Aggregation::new(self.ctx.flow.timeline_item_id(), AggregationKind::Redaction);
+        let aggregation = Aggregation::new(
+            self.ctx.flow.timeline_item_id(),
+            AggregationKind::Redaction {
+                // We can only get here for remote echoes of redactions.
+                is_local: false,
+            },
+        );
         self.meta.aggregations.add(target.clone(), aggregation.clone());
 
         find_item_and_apply_aggregation(
@@ -953,6 +958,7 @@ impl<'a, 'o> TimelineEventHandler<'a, 'o> {
             content,
             kind,
             is_room_encrypted,
+            false,
         );
 
         // Apply any pending or stashed aggregations.

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

@@ -86,6 +86,8 @@ pub struct EventTimelineItem {
     ///
     /// May be false when we don't know about the room encryption status yet.
     pub(super) is_room_encrypted: bool,
+    /// Whether a redaction for this event is currently being sent.
+    pub(super) is_being_redacted: bool,
 }
 
 #[derive(Clone, Debug)]
@@ -127,6 +129,7 @@ impl EventTimelineItem {
         content: TimelineItemContent,
         kind: EventTimelineItemKind,
         is_room_encrypted: bool,
+        is_being_redacted: bool,
     ) -> Self {
         Self {
             sender,
@@ -137,6 +140,7 @@ impl EventTimelineItem {
             content,
             kind,
             is_room_encrypted,
+            is_being_redacted,
         }
     }
 
@@ -493,9 +497,15 @@ impl EventTimelineItem {
             content,
             kind,
             is_room_encrypted: self.is_room_encrypted,
+            is_being_redacted: false,
         }
     }
 
+    /// Clone the current event item, and update its `is_being_redacted`.
+    pub(super) fn with_is_being_redacted(&self, is_being_redacted: bool) -> Self {
+        Self { is_being_redacted, ..self.clone() }
+    }
+
     pub(super) fn handle(&self) -> TimelineItemHandle<'_> {
         match &self.kind {
             EventTimelineItemKind::Local(local) => {

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

@@ -173,6 +173,12 @@ impl TestTimeline {
         txn_id
     }
 
+    async fn handle_local_redaction(&self, redacts: OwnedEventId) -> OwnedTransactionId {
+        let txn_id = TransactionId::new();
+        self.controller.handle_local_redaction(txn_id.clone(), redacts).await;
+        txn_id
+    }
+
     async fn handle_back_paginated_event(&self, event: Raw<AnyTimelineEvent>) {
         let timeline_event = TimelineEvent::from_plaintext(event.cast());
         self.controller

crates/matrix-sdk-ui/src/timeline/tests/redaction.rs

@@ -23,6 +23,7 @@ use ruma::{
         StateEventContentChange, reaction::RedactedReactionEventContent,
         room::message::OriginalSyncRoomMessageEvent,
     },
+    owned_event_id,
 };
 use stream_assert::{assert_next_matches, assert_pending};
 
@@ -203,3 +204,34 @@ async fn test_reaction_redaction_timeline_filter() {
     assert_eq!(item.content().reactions().cloned().unwrap_or_default().len(), 0);
     assert_eq!(timeline.controller.items().await.len(), 2);
 }
+
+#[async_test]
+async fn test_local_and_remote_echo_of_redaction() {
+    let timeline = TestTimeline::new();
+    let mut stream = timeline.subscribe_events().await;
+
+    let f = &timeline.factory;
+
+    // Send a message.
+    let event_id = owned_event_id!("$1");
+    timeline
+        .handle_live_event(f.text_msg("Hello, world!").sender(&ALICE).event_id(&event_id))
+        .await;
+    let item = assert_next_matches!(stream, VectorDiff::PushBack { value } => value);
+    assert!(!item.content().is_redacted());
+    assert!(!item.is_being_redacted);
+
+    // Now redact the message. We first emit the local echo of the redaction event.
+    // The timeline event should be marked as being under redaction.
+    timeline.handle_local_redaction(event_id.clone()).await;
+    let item = assert_next_matches!(stream, VectorDiff::Set { index: 0, value } => value);
+    assert!(!item.content().is_redacted());
+    assert!(item.is_being_redacted);
+
+    // Then comes the remote echo of the redaction event. The timeline event should
+    // now be redacted.
+    timeline.handle_live_event(f.redaction(&event_id).sender(&ALICE)).await;
+    let item = assert_next_matches!(stream, VectorDiff::Set { index: 0, value } => value);
+    assert!(item.content().is_redacted());
+    assert!(!item.is_being_redacted);
+}