feat(sdk): Bulk process thread subscription updates from sync and companion enpoint

Author: razvp

crates/matrix-sdk/CHANGELOG.md

@@ -41,6 +41,9 @@ All notable changes to this project will be documented in this file.
   ([#5678](https://github.com/matrix-org/matrix-rust-sdk/pull/5678).
 - Add new API to decline calls ([MSC4310](https://github.com/matrix-org/matrix-spec-proposals/pull/4310)): `Room::make_decline_call_event` and `Room::subscribe_to_call_decline_events`
   ([#5614](https://github.com/matrix-org/matrix-rust-sdk/pull/5614))
+- Use `StateStore::upsert_thread_subscriptions()` to bulk process thread subscription updates received
+  via the sync response or from the MSC4308 companion endpoint.
+  ([#5848](https://github.com/matrix-org/matrix-rust-sdk/pull/5848))
 
 ### Refactor
 

crates/matrix-sdk/src/client/thread_subscriptions.rs

@@ -28,7 +28,7 @@ use matrix_sdk_base::{
 use matrix_sdk_common::executor::spawn;
 use once_cell::sync::OnceCell;
 use ruma::{
-    OwnedEventId, OwnedRoomId,
+    EventId, OwnedEventId, OwnedRoomId, RoomId,
     api::client::threads::get_thread_subscriptions_changes::unstable::{
         ThreadSubscription, ThreadUnsubscription,
     },
@@ -179,71 +179,69 @@ impl ThreadSubscriptionCatchup {
         unsubscribed: BTreeMap<OwnedRoomId, BTreeMap<OwnedEventId, ThreadUnsubscription>>,
         token: Option<ThreadSubscriptionCatchupToken>,
     ) -> Result<()> {
+        // Precompute the updates so we don't hold the guard for too long.
+        let maybe_updates = self.build_subscription_updates(&subscribed, &unsubscribed);
         let Some(guard) = self.lock().await else {
             // Client is shutting down.
             return Ok(());
         };
+        // Save the token even if there aren't any updates.
         self.save_catchup_token(&guard, token).await?;
-        self.store_subscriptions(&guard, subscribed, unsubscribed).await?;
+        if let Some(updates) = maybe_updates {
+            trace!(
+                "saving {} new subscriptions and {} unsubscriptions",
+                subscribed.values().map(|by_room| by_room.len()).sum::<usize>(),
+                unsubscribed.values().map(|by_room| by_room.len()).sum::<usize>(),
+            );
+            guard.client.state_store().upsert_thread_subscriptions(updates).await?;
+        }
         Ok(())
     }
 
-    async fn store_subscriptions(
+    /// Internal helper for building the thread subscription updates Vec.
+    fn build_subscription_updates<'a>(
         &self,
-        guard: &GuardedStoreAccess,
-        subscribed: BTreeMap<OwnedRoomId, BTreeMap<OwnedEventId, ThreadSubscription>>,
-        unsubscribed: BTreeMap<OwnedRoomId, BTreeMap<OwnedEventId, ThreadUnsubscription>>,
-    ) -> Result<()> {
+        subscribed: &'a BTreeMap<OwnedRoomId, BTreeMap<OwnedEventId, ThreadSubscription>>,
+        unsubscribed: &'a BTreeMap<OwnedRoomId, BTreeMap<OwnedEventId, ThreadUnsubscription>>,
+    ) -> Option<Vec<(&'a RoomId, &'a EventId, StoredThreadSubscription)>> {
         if subscribed.is_empty() && unsubscribed.is_empty() {
-            // Nothing to do.
-            return Ok(());
+            return None;
         }
 
-        trace!(
-            "saving {} new subscriptions and {} unsubscriptions",
-            subscribed.values().map(|by_room| by_room.len()).sum::<usize>(),
-            unsubscribed.values().map(|by_room| by_room.len()).sum::<usize>(),
-        );
+        let mut updates: Vec<(&RoomId, &EventId, StoredThreadSubscription)> =
+            Vec::with_capacity(unsubscribed.len() + subscribed.len());
 
         // Take into account the new unsubscriptions.
         for (room_id, room_map) in unsubscribed {
             for (event_id, thread_sub) in room_map {
-                guard
-                    .client
-                    .state_store()
-                    .upsert_thread_subscription(
-                        &room_id,
-                        &event_id,
-                        StoredThreadSubscription {
-                            status: ThreadSubscriptionStatus::Unsubscribed,
-                            bump_stamp: Some(thread_sub.bump_stamp.into()),
-                        },
-                    )
-                    .await?;
+                updates.push((
+                    room_id,
+                    event_id,
+                    StoredThreadSubscription {
+                        status: ThreadSubscriptionStatus::Unsubscribed,
+                        bump_stamp: Some(thread_sub.bump_stamp.into()),
+                    },
+                ));
             }
         }
 
         // Take into account the new subscriptions.
         for (room_id, room_map) in subscribed {
             for (event_id, thread_sub) in room_map {
-                guard
-                    .client
-                    .state_store()
-                    .upsert_thread_subscription(
-                        &room_id,
-                        &event_id,
-                        StoredThreadSubscription {
-                            status: ThreadSubscriptionStatus::Subscribed {
-                                automatic: thread_sub.automatic,
-                            },
-                            bump_stamp: Some(thread_sub.bump_stamp.into()),
+                updates.push((
+                    room_id,
+                    event_id,
+                    StoredThreadSubscription {
+                        status: ThreadSubscriptionStatus::Subscribed {
+                            automatic: thread_sub.automatic,
                         },
-                    )
-                    .await?;
+                        bump_stamp: Some(thread_sub.bump_stamp.into()),
+                    },
+                ));
             }
         }
 
-        Ok(())
+        Some(updates)
     }
 
     /// Internal helper to lock writes to the thread subscriptions catchup
@@ -348,16 +346,28 @@ impl ThreadSubscriptionCatchup {
 
             match client.send(req).await {
                 Ok(resp) => {
+                    // Precompute the updates so we don't hold the guard for too long.
+                    let maybe_updates =
+                        this.build_subscription_updates(&resp.subscribed, &resp.unsubscribed);
+
                     let guard = this
                         .lock()
                         .await
                         .expect("a client instance is alive, so the locking should not fail");
 
-                    if let Err(err) =
-                        this.store_subscriptions(&guard, resp.subscribed, resp.unsubscribed).await
-                    {
-                        warn!("Failed to store caught up thread subscriptions: {err}");
-                        continue;
+                    if let Some(updates) = maybe_updates {
+                        trace!(
+                            "saving {} new subscriptions and {} unsubscriptions",
+                            resp.subscribed.values().map(|by_room| by_room.len()).sum::<usize>(),
+                            resp.unsubscribed.values().map(|by_room| by_room.len()).sum::<usize>(),
+                        );
+
+                        if let Err(err) =
+                            guard.client.state_store().upsert_thread_subscriptions(updates).await
+                        {
+                            warn!("Failed to store caught up thread subscriptions: {err}");
+                            continue;
+                        }
                     }
 
                     // Refresh the tokens, as the list might have changed while we sent the