timeline/sendq: add a Timeline::redact() method and tweak local echoes behavior

bindings/matrix-sdk-ffi/src/client.rs

@@ -147,6 +147,16 @@ pub trait ProgressWatcher: Send + Sync {
     fn transmission_progress(&self, progress: TransmissionProgress);
 }
 
+/// A listener to the global (client-wide) status of the sending queue.
+#[uniffi::export(callback_interface)]
+pub trait SendingQueueStatusListener: Sync + Send {
+    /// Called every time the sending queue has received a new status.
+    ///
+    /// This can be set automatically (in case of sending failure), or manually
+    /// via an API call.
+    fn on_value(&self, new_value: bool);
+}
+
 #[derive(Clone, Copy, uniffi::Record)]
 pub struct TransmissionProgress {
     pub current: u64,
@@ -307,6 +317,42 @@ impl Client {
 
         Ok(())
     }
+
+    /// Enables or disables the sending queue, according to the given parameter.
+    ///
+    /// The sending queue automatically disables itself whenever sending an
+    /// event with it failed (e.g., sending an event via the high-level Timeline
+    /// object), so it's required to manually re-enable it as soon as
+    /// connectivity is back on the device.
+    pub fn enable_sending_queue(&self, val: bool) {
+        if val {
+            self.inner.sending_queue().enable();
+        } else {
+            self.inner.sending_queue().disable();
+        }
+    }
+
+    /// Subscribe to the global enablement status of the sending queue, at the
+    /// client-wide level.
+    ///
+    /// The given listener will be immediately called with the initial value of
+    /// the enablement status.
+    pub fn subscribe_to_sending_queue_status(
+        &self,
+        listener: Box<dyn SendingQueueStatusListener>,
+    ) -> Arc<TaskHandle> {
+        let mut subscriber = self.inner.sending_queue().subscribe_status();
+
+        Arc::new(TaskHandle::new(RUNTIME.spawn(async move {
+            // Call with the initial value.
+            listener.on_value(subscriber.next_now());
+
+            // Call every time the value changes.
+            while let Some(next_val) = subscriber.next().await {
+                listener.on_value(next_val);
+            }
+        })))
+    }
 }
 
 impl Client {

bindings/matrix-sdk-ffi/src/timeline/mod.rs

@@ -45,13 +45,13 @@ use ruma::{
         },
         AnyMessageLikeEventContent,
     },
-    EventId,
+    EventId, OwnedTransactionId,
 };
 use tokio::{
     sync::Mutex,
     task::{AbortHandle, JoinHandle},
 };
-use tracing::{error, info, warn};
+use tracing::{error, warn};
 use uuid::Uuid;
 
 use self::content::{Reaction, ReactionSenderData, TimelineItemContent};
@@ -219,10 +219,22 @@ impl Timeline {
         Ok(())
     }
 
-    pub fn send(self: Arc<Self>, msg: Arc<RoomMessageEventContentWithoutRelation>) {
-        RUNTIME.spawn(async move {
-            self.inner.send((*msg).to_owned().with_relation(None).into()).await;
-        });
+    /// Queues an event in the room's sending queue so it's processed for
+    /// sending later.
+    ///
+    /// Returns an abort handle that allows to abort sending, if it hasn't
+    /// happened yet.
+    pub async fn send(
+        self: Arc<Self>,
+        msg: Arc<RoomMessageEventContentWithoutRelation>,
+    ) -> Result<Arc<AbortSendHandle>, ClientError> {
+        match self.inner.send((*msg).to_owned().with_relation(None).into()).await {
+            Ok(handle) => Ok(Arc::new(AbortSendHandle { inner: Mutex::new(Some(handle)) })),
+            Err(err) => {
+                error!("error when sending a message: {err}");
+                Err(anyhow::anyhow!(err).into())
+            }
+        }
     }
 
     pub fn send_image(
@@ -390,7 +402,9 @@ impl Timeline {
             AnyMessageLikeEventContent::UnstablePollStart(poll_start_event_content.into());
 
         RUNTIME.spawn(async move {
-            self.inner.send(event_content).await;
+            if let Err(err) = self.inner.send(event_content).await {
+                error!("unable to start poll: {err}");
+            }
         });
 
         Ok(())
@@ -409,7 +423,9 @@ impl Timeline {
             AnyMessageLikeEventContent::UnstablePollResponse(poll_response_event_content);
 
         RUNTIME.spawn(async move {
-            self.inner.send(event_content).await;
+            if let Err(err) = self.inner.send(event_content).await {
+                error!("unable to send poll response: {err}");
+            }
         });
 
         Ok(())
@@ -426,7 +442,9 @@ impl Timeline {
         let event_content = AnyMessageLikeEventContent::UnstablePollEnd(poll_end_event_content);
 
         RUNTIME.spawn(async move {
-            self.inner.send(event_content).await;
+            if let Err(err) = self.inner.send(event_content).await {
+                error!("unable to end poll: {err}");
+            }
         });
 
         Ok(())
@@ -472,7 +490,7 @@ impl Timeline {
         Ok(())
     }
 
-    pub fn send_location(
+    pub async fn send_location(
         self: Arc<Self>,
         body: String,
         geo_uri: String,
@@ -496,7 +514,8 @@ impl Timeline {
         let room_message_event_content = RoomMessageEventContentWithoutRelation::new(
             MessageType::Location(location_event_message_content),
         );
-        self.send(Arc::new(room_message_event_content))
+        // Errors are logged in `Self::send` already.
+        let _ = self.send(Arc::new(room_message_event_content)).await;
     }
 
     pub async fn toggle_reaction(&self, event_id: String, key: String) -> Result<(), ClientError> {
@@ -511,22 +530,14 @@ impl Timeline {
         Ok(())
     }
 
-    pub fn retry_send(self: Arc<Self>, txn_id: String) {
-        RUNTIME.spawn(async move {
-            if let Err(e) = self.inner.retry_send(txn_id.as_str().into()).await {
-                error!(txn_id, "Failed to retry sending: {e}");
-            }
-        });
-    }
-
-    pub fn cancel_send(self: Arc<Self>, txn_id: String) {
-        RUNTIME.spawn(async move {
-            if !self.inner.cancel_send(txn_id.as_str().into()).await {
-                info!(txn_id, "Failed to discard local echo: Not found");
-            }
-        });
-    }
-
+    /// Get the current timeline item for the given event ID, if any.
+    ///
+    /// Will return a remote event, *or* a local echo that has been sent but not
+    /// yet replaced by a remote echo.
+    ///
+    /// It's preferable to store the timeline items in the model for your UI, if
+    /// possible, instead of just storing IDs and coming back to the timeline
+    /// object to look up items.
     pub async fn get_event_timeline_item_by_event_id(
         &self,
         event_id: String,
@@ -540,6 +551,26 @@ impl Timeline {
         Ok(Arc::new(EventTimelineItem(item)))
     }
 
+    /// Get the current timeline item for the given transaction ID, if any.
+    ///
+    /// This will always return a local echo, if found.
+    ///
+    /// It's preferable to store the timeline items in the model for your UI, if
+    /// possible, instead of just storing IDs and coming back to the timeline
+    /// object to look up items.
+    pub async fn get_event_timeline_item_by_transaction_id(
+        &self,
+        transaction_id: String,
+    ) -> Result<Arc<EventTimelineItem>, ClientError> {
+        let transaction_id: OwnedTransactionId = transaction_id.into();
+        let item = self
+            .inner
+            .item_by_transaction_id(&transaction_id)
+            .await
+            .context("Item with given transaction ID not found")?;
+        Ok(Arc::new(EventTimelineItem(item)))
+    }
+
     pub async fn get_timeline_event_content_by_event_id(
         &self,
         event_id: String,
@@ -567,6 +598,52 @@ impl Timeline {
 
         latest_event.map(|item| Arc::new(EventTimelineItem(item)))
     }
+
+    /// Redacts an event from the timeline.
+    ///
+    /// Only works for events that exist as timeline items.
+    ///
+    /// If it was a local event, this will *try* to cancel it, if it was not
+    /// being sent already. If the event was a remote event, then it will be
+    /// redacted by sending a redaction request to the server.
+    ///
+    /// Returns whether the redaction did happen. It can only return false for
+    /// local events that are being processed.
+    pub async fn redact_event(
+        &self,
+        item: Arc<EventTimelineItem>,
+        reason: Option<String>,
+    ) -> Result<bool, ClientError> {
+        let removed = self
+            .inner
+            .redact(&item.0, reason.as_deref())
+            .await
+            .map_err(|err| anyhow::anyhow!(err))?;
+
+        Ok(removed)
+    }
+}
+
+#[derive(uniffi::Object)]
+pub struct AbortSendHandle {
+    inner: Mutex<Option<matrix_sdk::send_queue::AbortSendHandle>>,
+}
+
+#[uniffi::export(async_runtime = "tokio")]
+impl AbortSendHandle {
+    /// Try to abort the sending of the current event.
+    ///
+    /// If this returns `true`, then the sending could be aborted, because the
+    /// event hasn't been sent yet. Otherwise, if this returns `false`, the
+    /// event had already been sent and could not be aborted.
+    async fn abort(self: Arc<Self>) -> bool {
+        if let Some(inner) = self.inner.lock().await.take() {
+            inner.abort().await
+        } else {
+            warn!("trying to abort an send handle that's already been actioned");
+            false
+        }
+    }
 }
 
 #[derive(Debug, thiserror::Error, uniffi::Error)]
@@ -771,9 +848,6 @@ pub enum EventSendState {
     /// The local event has been sent to the server, but unsuccessfully: The
     /// sending has failed.
     SendingFailed { error: String },
-    /// Sending has been cancelled because an earlier event in the
-    /// message-sending queue failed.
-    Cancelled,
     /// The local event has been sent successfully to the server.
     Sent { event_id: String },
 }
@@ -785,7 +859,6 @@ impl From<&matrix_sdk_ui::timeline::EventSendState> for EventSendState {
         match value {
             NotSentYet => Self::NotSentYet,
             SendingFailed { error } => Self::SendingFailed { error: error.to_string() },
-            Cancelled => Self::Cancelled,
             Sent { event_id } => Self::Sent { event_id: event_id.to_string() },
         }
     }