feat: implement msc4308 + automatic background catchup of thread subscriptions

Cargo.toml

@@ -59,7 +59,7 @@ proptest = { version = "1.6.0", default-features = false, features = ["std"] }
 rand = "0.8.5"
 reqwest = { version = "0.12.12", default-features = false }
 rmp-serde = "1.3.0"
-ruma = { git = "https://github.com/ruma/ruma", rev = "57049282e3a74f67f86e4eb2382a3e649b57cc2b", features = [
+ruma = { git = "https://github.com/ruma/ruma", rev = "51fb51a560027fd330b43398a278922cacbc825c", features = [
     "client-api-c",
     "compat-upload-signatures",
     "compat-arbitrary-length-ids",

crates/matrix-sdk-base/src/lib.rs

@@ -62,7 +62,7 @@ pub use room::{
 };
 pub use store::{
     ComposerDraft, ComposerDraftType, QueueWedgeError, StateChanges, StateStore, StateStoreDataKey,
-    StateStoreDataValue, StoreError,
+    StateStoreDataValue, StoreError, ThreadSubscriptionCatchupToken,
 };
 pub use utils::{
     MinimalRoomMemberEvent, MinimalStateEvent, OriginalMinimalStateEvent, RedactedMinimalStateEvent,

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

@@ -46,7 +46,8 @@ use crate::{
     MinimalRoomMemberEvent, RoomMemberships, StateStoreDataKey, StateStoreDataValue,
     deserialized_responses::{DisplayName, RawAnySyncOrStrippedState},
     store::{
-        QueueWedgeError, StoredThreadSubscription, traits::compare_thread_subscription_bump_stamps,
+        QueueWedgeError, StoredThreadSubscription,
+        traits::{ThreadSubscriptionCatchupToken, compare_thread_subscription_bump_stamps},
     },
 };
 
@@ -87,6 +88,7 @@ struct MemoryStoreInner {
     dependent_send_queue_events: BTreeMap<OwnedRoomId, Vec<DependentQueuedRequest>>,
     seen_knock_requests: BTreeMap<OwnedRoomId, BTreeMap<OwnedEventId, OwnedUserId>>,
     thread_subscriptions: BTreeMap<OwnedRoomId, BTreeMap<OwnedEventId, StoredThreadSubscription>>,
+    thread_subscriptions_catchup_tokens: Option<Vec<ThreadSubscriptionCatchupToken>>,
 }
 
 /// In-memory, non-persistent implementation of the `StateStore`.
@@ -183,6 +185,10 @@ impl StateStore for MemoryStore {
                 .get(room_id)
                 .cloned()
                 .map(StateStoreDataValue::SeenKnockRequests),
+            StateStoreDataKey::ThreadSubscriptionsCatchupTokens => inner
+                .thread_subscriptions_catchup_tokens
+                .clone()
+                .map(StateStoreDataValue::ThreadSubscriptionsCatchupTokens),
         })
     }
 
@@ -246,6 +252,12 @@ impl StateStore for MemoryStore {
                         .expect("Session data is not a set of seen join request ids"),
                 );
             }
+            StateStoreDataKey::ThreadSubscriptionsCatchupTokens => {
+                inner.thread_subscriptions_catchup_tokens =
+                    Some(value.into_thread_subscriptions_catchup_tokens().expect(
+                        "Session data is not a list of thread subscription catchup tokens",
+                    ));
+            }
         }
 
         Ok(())
@@ -276,6 +288,9 @@ impl StateStore for MemoryStore {
             StateStoreDataKey::SeenKnockRequests(room_id) => {
                 inner.seen_knock_requests.remove(room_id);
             }
+            StateStoreDataKey::ThreadSubscriptionsCatchupTokens => {
+                inner.thread_subscriptions_catchup_tokens = None;
+            }
         }
         Ok(())
     }

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

@@ -93,7 +93,8 @@ pub use self::{
     },
     traits::{
         ComposerDraft, ComposerDraftType, DynStateStore, IntoStateStore, ServerInfo, StateStore,
-        StateStoreDataKey, StateStoreDataValue, StateStoreExt, WellKnownResponse,
+        StateStoreDataKey, StateStoreDataValue, StateStoreExt, ThreadSubscriptionCatchupToken,
+        WellKnownResponse,
     },
 };
 

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

@@ -1151,6 +1151,38 @@ pub enum StateStoreDataValue {
 
     /// A list of knock request ids marked as seen in a room.
     SeenKnockRequests(BTreeMap<OwnedEventId, OwnedUserId>),
+
+    /// A list of tokens to continue thread subscriptions catchup.
+    ///
+    /// See documentation of [`ThreadSubscriptionCatchupToken`] for more
+    /// details.
+    ThreadSubscriptionsCatchupTokens(Vec<ThreadSubscriptionCatchupToken>),
+}
+
+/// Tokens to use when catching up on thread subscriptions.
+///
+/// These tokens are created when the client receives some thread subscriptions
+/// from sync, but the sync indicates that there are more thread subscriptions
+/// available on the server. In this case, it's expected that the client will
+/// call the [MSC4308] companion endpoint to catch up (back-paginate) on
+/// previous thread subscriptions.
+///
+/// [MSC4308]: https://github.com/matrix-org/matrix-spec-proposals/pull/4308
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct ThreadSubscriptionCatchupToken {
+    /// The token to use as the lower bound when fetching new threads
+    /// subscriptions.
+    ///
+    /// In sliding sync, this is the `prev_batch` value of a sliding sync
+    /// response.
+    pub from: String,
+
+    /// The token to use as the upper bound when fetching new threads
+    /// subscriptions.
+    ///
+    /// In sliding sync, it must be set to the `pos` value of the sliding sync
+    /// *request*, which response received a `prev_batch` token.
+    pub to: Option<String>,
 }
 
 /// Current draft of the composer for the room.
@@ -1222,6 +1254,14 @@ impl StateStoreDataValue {
     pub fn into_seen_knock_requests(self) -> Option<BTreeMap<OwnedEventId, OwnedUserId>> {
         as_variant!(self, Self::SeenKnockRequests)
     }
+
+    /// Get this value if it is the data for the thread subscriptions catchup
+    /// tokens.
+    pub fn into_thread_subscriptions_catchup_tokens(
+        self,
+    ) -> Option<Vec<ThreadSubscriptionCatchupToken>> {
+        as_variant!(self, Self::ThreadSubscriptionsCatchupTokens)
+    }
 }
 
 /// A key for key-value data.
@@ -1258,6 +1298,9 @@ pub enum StateStoreDataKey<'a> {
 
     /// A list of knock request ids marked as seen in a room.
     SeenKnockRequests(&'a RoomId),
+
+    /// A list of thread subscriptions catchup tokens.
+    ThreadSubscriptionsCatchupTokens,
 }
 
 impl StateStoreDataKey<'_> {
@@ -1294,6 +1337,11 @@ impl StateStoreDataKey<'_> {
     /// Key prefix to use for the
     /// [`SeenKnockRequests`][Self::SeenKnockRequests] variant.
     pub const SEEN_KNOCK_REQUESTS: &'static str = "seen_knock_requests";
+
+    /// Key prefix to use for the
+    /// [`ThreadSubscriptionsCatchupTokens`][Self::ThreadSubscriptionsCatchupTokens] variant.
+    pub const THREAD_SUBSCRIPTIONS_CATCHUP_TOKENS: &'static str =
+        "thread_subscriptions_catchup_tokens";
 }
 
 /// Compare two thread subscription changes bump stamps, given a fixed room and

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

@@ -32,7 +32,7 @@ use matrix_sdk_base::{
         StateStore, StoreError, StoredThreadSubscription, ThreadSubscriptionStatus,
     },
     MinimalRoomMemberEvent, RoomInfo, RoomMemberships, StateStoreDataKey, StateStoreDataValue,
-    ROOM_VERSION_FALLBACK, ROOM_VERSION_RULES_FALLBACK,
+    ThreadSubscriptionCatchupToken, ROOM_VERSION_FALLBACK, ROOM_VERSION_RULES_FALLBACK,
 };
 use matrix_sdk_store_encryption::{Error as EncryptionError, StoreCipher};
 use ruma::{
@@ -439,6 +439,9 @@ impl IndexeddbStateStore {
             StateStoreDataKey::SeenKnockRequests(room_id) => {
                 self.encode_key(keys::KV, (StateStoreDataKey::SEEN_KNOCK_REQUESTS, room_id))
             }
+            StateStoreDataKey::ThreadSubscriptionsCatchupTokens => {
+                self.encode_key(keys::KV, StateStoreDataKey::THREAD_SUBSCRIPTIONS_CATCHUP_TOKENS)
+            }
         }
     }
 }
@@ -591,6 +594,10 @@ impl_state_store!({
                 .map(|f| self.deserialize_value::<BTreeMap<OwnedEventId, OwnedUserId>>(&f))
                 .transpose()?
                 .map(StateStoreDataValue::SeenKnockRequests),
+            StateStoreDataKey::ThreadSubscriptionsCatchupTokens => value
+                .map(|f| self.deserialize_value::<Vec<ThreadSubscriptionCatchupToken>>(&f))
+                .transpose()?
+                .map(StateStoreDataValue::ThreadSubscriptionsCatchupTokens),
         };
 
         Ok(value)
@@ -632,6 +639,11 @@ impl_state_store!({
                     .into_seen_knock_requests()
                     .expect("Session data is not a set of seen knock request ids"),
             ),
+            StateStoreDataKey::ThreadSubscriptionsCatchupTokens => self.serialize_value(
+                &value
+                    .into_thread_subscriptions_catchup_tokens()
+                    .expect("Session data is not a list of thread subscription catchup tokens"),
+            ),
         };
 
         let tx =

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

@@ -40,7 +40,7 @@ use ruma::{
 use rusqlite::{OptionalExtension, Transaction};
 use serde::{Deserialize, Serialize};
 use tokio::fs;
-use tracing::{debug, warn};
+use tracing::{debug, trace, warn};
 
 use crate::{
     error::{Error, Result},
@@ -416,6 +416,9 @@ impl SqliteStateStore {
             StateStoreDataKey::SeenKnockRequests(room_id) => {
                 Cow::Owned(format!("{}:{room_id}", StateStoreDataKey::SEEN_KNOCK_REQUESTS))
             }
+            StateStoreDataKey::ThreadSubscriptionsCatchupTokens => {
+                Cow::Borrowed(StateStoreDataKey::THREAD_SUBSCRIPTIONS_CATCHUP_TOKENS)
+            }
         };
 
         self.encode_key(keys::KV_BLOB, &*key_s)
@@ -1037,6 +1040,11 @@ impl StateStore for SqliteStateStore {
                     StateStoreDataKey::SeenKnockRequests(_) => {
                         StateStoreDataValue::SeenKnockRequests(self.deserialize_value(&data)?)
                     }
+                    StateStoreDataKey::ThreadSubscriptionsCatchupTokens => {
+                        StateStoreDataValue::ThreadSubscriptionsCatchupTokens(
+                            self.deserialize_value(&data)?,
+                        )
+                    }
                 })
             })
             .transpose()
@@ -1077,6 +1085,11 @@ impl StateStore for SqliteStateStore {
                     .into_seen_knock_requests()
                     .expect("Session data is not a set of seen knock request ids"),
             )?,
+            StateStoreDataKey::ThreadSubscriptionsCatchupTokens => self.serialize_value(
+                &value
+                    .into_thread_subscriptions_catchup_tokens()
+                    .expect("Session data is not a list of thread subscription catchup tokens"),
+            )?,
         };
 
         self.acquire()
@@ -2125,9 +2138,11 @@ impl StateStore for SqliteStateStore {
         if let Some(previous) = self.load_thread_subscription(room_id, thread_id).await? {
             if previous == new {
                 // No need to update anything.
+                trace!("not saving thread subscription because the subscription is the same");
                 return Ok(());
             }
             if !compare_thread_subscription_bump_stamps(previous.bump_stamp, &mut new.bump_stamp) {
+                trace!("not saving thread subscription because we have a newer bump stamp");
                 return Ok(());
             }
         }

crates/matrix-sdk-ui/src/room_list_service/mod.rs

@@ -155,6 +155,15 @@ impl RoomListService {
                 enabled: Some(true),
             }));
 
+        if client.enabled_thread_subscriptions() {
+            builder = builder.with_thread_subscriptions_extension(
+                assign!(http::request::ThreadSubscriptions::default(), {
+                    enabled: Some(true),
+                    limit: Some(ruma::uint!(10))
+                }),
+            );
+        }
+
         if share_pos {
             // We don't deal with encryption device messages here so this is safe
             builder = builder.share_pos();

crates/matrix-sdk/src/client/builder/mod.rs

@@ -608,6 +608,7 @@ impl ClientBuilder {
 
         let event_cache = OnceCell::new();
         let latest_events = OnceCell::new();
+        let thread_subscriptions_catchup = OnceCell::new();
 
         #[cfg(feature = "experimental-search")]
         let search_index =
@@ -632,6 +633,7 @@ impl ClientBuilder {
             self.cross_process_store_locks_holder_name,
             #[cfg(feature = "experimental-search")]
             search_index,
+            thread_subscriptions_catchup,
         )
         .await;