feat(room): add a new function that will only subscribe to a thread if needed

Author: bnjbvr

crates/matrix-sdk/src/event_cache/room/threads.rs

@@ -20,7 +20,6 @@ use eyeball_im::VectorDiff;
 use matrix_sdk_base::{
     event_cache::{Event, Gap},
     linked_chunk::{ChunkContent, Position},
-    store::ThreadSubscription,
 };
 use ruma::OwnedEventId;
 use tokio::sync::broadcast::{Receiver, Sender};
@@ -358,35 +357,17 @@ pub async fn should_subscribe_thread(
 /// Optionally subscribe to new threads, if the client enabled automatic thread
 /// subscription support.
 pub async fn subscribe_to_new_threads(room: &Room, new_thread_subs: AutomaticThreadSubscriptions) {
-    // If there's no subscriptions, or the client hasn't enabled thread
+    // If there's no new subscriptions, or the client hasn't enabled thread
     // subscriptions, we don't have anything to do.
     if new_thread_subs.is_empty() || !room.client.enabled_thread_subscriptions() {
         return;
     }
-
     for (thread_root, subscribe_up_to_event_id) in new_thread_subs {
-        let previous_status = match room.load_or_fetch_thread_subscription(&thread_root).await {
-            Ok(status) => status,
-            Err(err) => {
-                warn!(%thread_root, "couldn't fetch thread subscription: {err}");
-                continue;
-            }
-        };
-
-        match previous_status {
-            Some(ThreadSubscription { .. }) => {
-                // Already subscribed, nothing to do.
-                trace!(%thread_root, "already subscribed to thread");
-            }
-            None => {
-                // Send an automatic subscription!
-                let automatic = Some(subscribe_up_to_event_id);
-                if let Err(err) = room.subscribe_thread(thread_root.clone(), automatic).await {
-                    warn!(%thread_root, "couldn't subscribe to thread: {err}");
-                } else {
-                    trace!(%thread_root, "subscribed to thread");
-                }
-            }
+        let automatic = Some(subscribe_up_to_event_id);
+        if let Err(err) = room.subscribe_thread_if_needed(&thread_root, automatic).await {
+            warn!(%thread_root, "couldn't subscribe to thread: {err}");
+        } else {
+            trace!(%thread_root, "subscribed to thread");
         }
     }
 }

crates/matrix-sdk/src/room/mod.rs

@@ -3765,6 +3765,43 @@ impl Room {
         }
     }
 
+    /// Subscribe to a thread if needed, based on a current subscription to it.
+    ///
+    /// This is like [`Self::subscribe_thread`], but it first checks if the user
+    /// has already subscribed to a thread, so as to minimize sending
+    /// unnecessary subscriptions which would be ignored by the server.
+    pub async fn subscribe_thread_if_needed(
+        &self,
+        thread_root: &EventId,
+        automatic: Option<OwnedEventId>,
+    ) -> Result<()> {
+        let prev_sub = self.load_or_fetch_thread_subscription(thread_root).await?;
+
+        match prev_sub {
+            Some(prev_sub) => {
+                if prev_sub.automatic {
+                    // Either we have a manual subscription (which overrides an
+                    // automatic one), or we have another
+                    // automatic one (and we can't order these), so send the
+                    // request to the server.
+                } else {
+                    // The previous subscription was manual. There's no point sending the request
+                    // here:
+                    // - if the new subscription is automatic, since a manual subscription is higher
+                    // priority than an automatic one, it will be ignored by the server.
+                    // - if the new subscription is manual, this is an idempotent operation.
+                    return Ok(());
+                }
+            }
+
+            None => {
+                // There's no prior subscription, so we can safely send one.
+            }
+        }
+
+        self.subscribe_thread(thread_root.to_owned(), automatic).await
+    }
+
     /// Unsubscribe from a given thread in this room.
     ///
     /// # Arguments

crates/matrix-sdk/tests/integration/room/thread.rs

@@ -79,6 +79,82 @@ async fn test_subscribe_thread() {
     assert_matches!(subscription, None);
 }
 
+#[async_test]
+async fn test_subscribe_thread_if_needed() {
+    let server = MatrixMockServer::new().await;
+    let client = server.client_builder().build().await;
+
+    let room_id = room_id!("!test:example.org");
+    let room = server.sync_joined_room(&client, room_id).await;
+
+    // The function `subscribe_thread_if_needed` will automatically subscribe to the
+    // thread, if there's no prior manual subscription, whether the new
+    // subscription is automatic or not.
+    for (root_id, automatic) in [
+        (owned_event_id!("$root"), None),
+        (owned_event_id!("$woot"), Some(owned_event_id!("$woot"))),
+    ] {
+        server
+            .mock_put_thread_subscription()
+            .match_room_id(room_id.to_owned())
+            .match_thread_id(root_id.clone())
+            .ok()
+            .mock_once()
+            .mount()
+            .await;
+
+        room.subscribe_thread_if_needed(&root_id, automatic).await.unwrap();
+    }
+
+    // The function `subscribe_thread_if_needed` will automatically subscribe to the
+    // thread, if there's a previous automatic subscription, whether the new
+    // subscription is automatic or not.
+    for (root_id, automatic) in [
+        (owned_event_id!("$foot"), None),
+        (owned_event_id!("$toot"), Some(owned_event_id!("$toot"))),
+    ] {
+        server
+            .mock_get_thread_subscription()
+            .match_room_id(room_id.to_owned())
+            .match_thread_id(root_id.clone())
+            .ok(true)
+            .mock_once()
+            .mount()
+            .await;
+
+        server
+            .mock_put_thread_subscription()
+            .match_room_id(room_id.to_owned())
+            .match_thread_id(root_id.clone())
+            .ok()
+            .mock_once()
+            .mount()
+            .await;
+
+        room.subscribe_thread_if_needed(&root_id, automatic).await.unwrap();
+    }
+
+    // The function `subscribe_thread_if_needed` is a no-op if there's a prior
+    // manual subscription, whether the new subscription is automatic or not.
+    for (root_id, automatic) in [
+        (owned_event_id!("$root"), None),
+        (owned_event_id!("$woot"), Some(owned_event_id!("$woot"))),
+    ] {
+        server
+            .mock_get_thread_subscription()
+            .match_room_id(room_id.to_owned())
+            .match_thread_id(root_id.clone())
+            .ok(false)
+            .mock_once()
+            .mount()
+            .await;
+
+        // No-op! (The PUT endpoint hasn't been mocked, so this would result in a 404 if
+        // it were trying to hit it.)
+        room.subscribe_thread_if_needed(&root_id, automatic).await.unwrap();
+    }
+}
+
 #[async_test]
 async fn test_thread_push_rule_is_triggered_for_subscribed_threads() {
     // This test checks that the evaluation of push rules for threads will correctly