refactor(threads): do not store the unsubscribed state in the DB

Author: bnjbvr

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

@@ -1093,8 +1093,8 @@ impl Room {
         Ok(Arc::new(RoomPreview::new(AsyncRuntimeDropped::new(client), room_preview)))
     }
 
-    /// Toggle a MSC4306 subscription to a thread in this room, based on the
-    /// thread root event id.
+    /// Set a MSC4306 subscription to a thread in this room, based on the thread
+    /// root event id.
     ///
     /// If `subscribed` is `true`, it will subscribe to the thread, with a
     /// precision that the subscription was manually requested by the user
@@ -1135,7 +1135,6 @@ impl Room {
             matrix_sdk::room::ThreadStatus::Subscribed { automatic } => {
                 ThreadStatus::Subscribed { automatic }
             }
-            matrix_sdk::room::ThreadStatus::Unsubscribed => ThreadStatus::Unsubscribed,
         }))
     }
 }
@@ -1149,9 +1148,6 @@ pub enum ThreadStatus {
         /// mention) or if it was manually requested by the user.
         automatic: bool,
     },
-
-    /// The thread is not subscribed to.
-    Unsubscribed,
 }
 
 /// A listener for receiving new live location shares in a room.

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

@@ -1810,16 +1810,33 @@ impl StateStoreIntegrationTests for DynStateStore {
         assert_eq!(maybe_status, Some(ThreadStatus::Subscribed { automatic: false }));
 
         // We can override the thread subscription status.
-        self.upsert_thread_subscription(room_id(), first_thread, ThreadStatus::Unsubscribed)
-            .await
-            .unwrap();
+        self.upsert_thread_subscription(
+            room_id(),
+            first_thread,
+            ThreadStatus::Subscribed { automatic: false },
+        )
+        .await
+        .unwrap();
 
         // And it's correctly reflected.
         let maybe_status = self.load_thread_subscription(room_id(), first_thread).await.unwrap();
-        assert_eq!(maybe_status, Some(ThreadStatus::Unsubscribed));
+        assert_eq!(maybe_status, Some(ThreadStatus::Subscribed { automatic: false }));
         // And the second thread is still subscribed.
         let maybe_status = self.load_thread_subscription(room_id(), second_thread).await.unwrap();
         assert_eq!(maybe_status, Some(ThreadStatus::Subscribed { automatic: false }));
+
+        // We can remove a thread subscription.
+        self.remove_thread_subscription(room_id(), second_thread).await.unwrap();
+
+        // And it's correctly reflected.
+        let maybe_status = self.load_thread_subscription(room_id(), second_thread).await.unwrap();
+        assert_eq!(maybe_status, None);
+        // And the first thread is still subscribed.
+        let maybe_status = self.load_thread_subscription(room_id(), first_thread).await.unwrap();
+        assert_eq!(maybe_status, Some(ThreadStatus::Subscribed { automatic: false }));
+
+        // Removing a thread subscription for an unknown thread is a no-op.
+        self.remove_thread_subscription(room_id(), second_thread).await.unwrap();
     }
 }
 

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

@@ -982,6 +982,27 @@ impl StateStore for MemoryStore {
             .and_then(|subscriptions| subscriptions.get(thread_id))
             .copied())
     }
+
+    async fn remove_thread_subscription(
+        &self,
+        room: &RoomId,
+        thread_id: &EventId,
+    ) -> Result<(), Self::Error> {
+        let mut inner = self.inner.write().unwrap();
+
+        let Some(room_subs) = inner.thread_subscriptions.get_mut(room) else {
+            return Ok(());
+        };
+
+        room_subs.remove(thread_id);
+
+        if room_subs.is_empty() {
+            // If there are no more subscriptions for this room, remove the room entry.
+            inner.thread_subscriptions.remove(room);
+        }
+
+        Ok(())
+    }
 }
 
 #[cfg(test)]

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

@@ -462,9 +462,6 @@ pub enum ThreadStatus {
         /// manual user choice.
         automatic: bool,
     },
-    /// The thread is unsubscribed to (it won't cause any notifications or
-    /// automatic subscription anymore).
-    Unsubscribed,
 }
 
 impl ThreadStatus {
@@ -478,7 +475,6 @@ impl ThreadStatus {
                     "manual"
                 }
             }
-            ThreadStatus::Unsubscribed => "unsubscribed",
         }
     }
 
@@ -488,7 +484,6 @@ impl ThreadStatus {
         match s {
             "automatic" => Some(ThreadStatus::Subscribed { automatic: true }),
             "manual" => Some(ThreadStatus::Subscribed { automatic: false }),
-            "unsubscribed" => Some(ThreadStatus::Unsubscribed),
             _ => None,
         }
     }

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

@@ -481,17 +481,22 @@ pub trait StateStore: AsyncTraitDeps {
     ) -> Result<Vec<DependentQueuedRequest>, Self::Error>;
 
     /// Insert or update a thread subscription for a given room and thread.
-    ///
-    /// Note: there's no way to remove a thread subscription, because it's
-    /// either subscribed to, or unsubscribed to, after it's been saved for
-    /// the first time.
     async fn upsert_thread_subscription(
         &self,
         room: &RoomId,
         thread_id: &EventId,
         status: ThreadStatus,
     ) -> Result<(), Self::Error>;
 
+    /// Remove a previous thread subscription for a given room and thread.
+    ///
+    /// Note: removing an unknown thread subscription is a no-op.
+    async fn remove_thread_subscription(
+        &self,
+        room: &RoomId,
+        thread_id: &EventId,
+    ) -> Result<(), Self::Error>;
+
     /// Loads the current thread subscription for a given room and thread.
     ///
     /// Returns `None` if there was no entry for the given room/thread pair.
@@ -811,6 +816,14 @@ impl<T: StateStore> StateStore for EraseStateStoreError<T> {
     ) -> Result<Option<ThreadStatus>, Self::Error> {
         self.0.load_thread_subscription(room, thread_id).await.map_err(Into::into)
     }
+
+    async fn remove_thread_subscription(
+        &self,
+        room: &RoomId,
+        thread_id: &EventId,
+    ) -> Result<(), Self::Error> {
+        self.0.remove_thread_subscription(room, thread_id).await.map_err(Into::into)
+    }
 }
 
 /// Convenience functionality for state stores.

crates/matrix-sdk-indexeddb/src/state_store/mod.rs

@@ -1837,6 +1837,21 @@ impl_state_store!({
 
         Ok(Some(status))
     }
+
+    async fn remove_thread_subscription(&self, room: &RoomId, thread_id: &EventId) -> Result<()> {
+        let encoded_key = self.encode_key(keys::THREAD_SUBSCRIPTIONS, (room, thread_id));
+
+        self.inner
+            .transaction_on_one_with_mode(
+                keys::THREAD_SUBSCRIPTIONS,
+                IdbTransactionMode::Readwrite,
+            )?
+            .object_store(keys::THREAD_SUBSCRIPTIONS)?
+            .delete(&encoded_key)?
+            .await?;
+
+        Ok(())
+    }
 });
 
 /// A room member.

crates/matrix-sdk-sqlite/src/state_store.rs

@@ -2142,6 +2142,25 @@ impl StateStore for SqliteStateStore {
             })
             .transpose()?)
     }
+
+    async fn remove_thread_subscription(
+        &self,
+        room_id: &RoomId,
+        thread_id: &EventId,
+    ) -> Result<(), Self::Error> {
+        let room_id = self.encode_key(keys::THREAD_SUBSCRIPTIONS, room_id);
+        let thread_id = self.encode_key(keys::THREAD_SUBSCRIPTIONS, thread_id);
+
+        self.acquire()
+            .await?
+            .execute(
+                "DELETE FROM thread_subscriptions WHERE room_id = ? AND event_id = ?",
+                (room_id, thread_id),
+            )
+            .await?;
+
+        Ok(())
+    }
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]

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

@@ -3675,6 +3675,7 @@ impl Room {
     /// - An `Ok` result if the subscription was successful, or if the server
     ///   skipped an automatic subscription (as the user unsubscribed from the
     ///   thread after the event causing the automatic subscription).
+    #[instrument(skip(self), fields(room_id = %self.room_id()))]
     pub async fn subscribe_thread(
         &self,
         thread_root: OwnedEventId,
@@ -3733,6 +3734,7 @@ impl Room {
     /// - An `Ok` result if the unsubscription was successful, or the thread was
     ///   already unsubscribed.
     /// - A 404 error if the event isn't known, or isn't a thread root.
+    #[instrument(skip(self), fields(room_id = %self.room_id()))]
     pub async fn unsubscribe_thread(&self, thread_root: OwnedEventId) -> Result<()> {
         self.client
             .send(unsubscribe_thread::unstable::Request::new(
@@ -3741,11 +3743,10 @@ impl Room {
             ))
             .await?;
 
+        trace!("Server acknowledged the thread subscription removal; removed it from db too");
+
         // Immediately save the result into the database.
-        self.client
-            .state_store()
-            .upsert_thread_subscription(self.room_id(), &thread_root, ThreadStatus::Unsubscribed)
-            .await?;
+        self.client.state_store().remove_thread_subscription(self.room_id(), &thread_root).await?;
 
         Ok(())
     }
@@ -3766,6 +3767,7 @@ impl Room {
     ///   event couldn't be found, or the event isn't a thread.
     /// - An error if the request fails for any other reason, such as a network
     ///   error.
+    #[instrument(skip(self), fields(room_id = %self.room_id()))]
     pub async fn fetch_thread_subscription(
         &self,
         thread_root: OwnedEventId,
@@ -3778,36 +3780,29 @@ impl Room {
             ))
             .await;
 
-        match result {
-            Ok(response) => Ok(Some(ThreadStatus::Subscribed { automatic: response.automatic })),
+        let sub = match result {
+            Ok(response) => Some(ThreadStatus::Subscribed { automatic: response.automatic }),
             Err(http_error) => match http_error.as_client_api_error() {
-                Some(error) if error.status_code == StatusCode::NOT_FOUND => {
-                    // At this point the server returned no subscriptions, which can mean that the
-                    // endpoint doesn't exist (not enabled/implemented yet on the server), or that
-                    // the thread doesn't exist, or that the user has unsubscribed from it
-                    // previously.
-                    //
-                    // If we had any information about prior unsubscription, we can use it here to
-                    // return something slightly more precise than what the server returned.
-                    let stored_status = self
-                        .client
-                        .state_store()
-                        .load_thread_subscription(self.room_id(), &thread_root)
-                        .await?;
-
-                    if let Some(ThreadStatus::Unsubscribed) = stored_status {
-                        // The thread was unsubscribed from before, so maintain this information.
-                        Ok(Some(ThreadStatus::Unsubscribed))
-                    } else {
-                        // We either have stale information (the thread was marked as subscribed
-                        // to, but the server said it wasn't), or we didn't have any information.
-                        // Return unknown.
-                        Ok(None)
-                    }
-                }
-                _ => Err(http_error.into()),
+                Some(error) if error.status_code == StatusCode::NOT_FOUND => None,
+                _ => return Err(http_error.into()),
             },
+        };
+
+        // Keep the database in sync.
+        if let Some(sub) = &sub {
+            self.client
+                .state_store()
+                .upsert_thread_subscription(self.room_id(), &thread_root, *sub)
+                .await?;
+        } else {
+            // If the subscription was not found, remove it from the database.
+            self.client
+                .state_store()
+                .remove_thread_subscription(self.room_id(), &thread_root)
+                .await?;
         }
+
+        Ok(sub)
     }
 }
 

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

@@ -56,10 +56,10 @@ async fn test_subscribe_thread() {
 
     room.unsubscribe_thread(root_id.clone()).await.unwrap();
 
-    // Now, if I retry to get the subscription status for this thread, it's
-    // unsubscribed.
+    // Now, if I retry to get the subscription status for this thread, it doesn't
+    // exist anymore.
     let subscription = room.fetch_thread_subscription(root_id.clone()).await.unwrap();
-    assert_matches!(subscription, Some(ThreadStatus::Unsubscribed));
+    assert_matches!(subscription, None);
 
     // Subscribing automatically to the thread may also return a `M_SKIPPED` error
     // that should be non-fatal.
@@ -76,5 +76,5 @@ async fn test_subscribe_thread() {
 
     // And in this case, the thread is still unsubscribed.
     let subscription = room.fetch_thread_subscription(root_id).await.unwrap();
-    assert_matches!(subscription, Some(ThreadStatus::Unsubscribed));
+    assert_matches!(subscription, None);
 }

labs/multiverse/src/widgets/room_view/mod.rs

@@ -534,7 +534,6 @@ impl RoomView {
                                         "subscribed (manual)"
                                     }
                                 }
-                                ThreadStatus::Unsubscribed => "unsubscribed",
                             }
                         ));
                     }