From 4adbb4aa88e23794da2d3d9ee144b488a1c9aa32 Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Mon, 1 Sep 2025 14:53:37 +0200 Subject: [PATCH] feat(sdk): add support for persisting the thread subscription catchup tokens --- crates/matrix-sdk-base/src/lib.rs | 2 +- .../matrix-sdk-base/src/store/memory_store.rs | 17 ++++++- crates/matrix-sdk-base/src/store/mod.rs | 3 +- crates/matrix-sdk-base/src/store/traits.rs | 46 +++++++++++++++++++ .../src/state_store/mod.rs | 14 +++++- crates/matrix-sdk-sqlite/src/state_store.rs | 17 ++++++- 6 files changed, 94 insertions(+), 5 deletions(-) diff --git a/crates/matrix-sdk-base/src/lib.rs b/crates/matrix-sdk-base/src/lib.rs index 15cb0af92..6d29168c2 100644 --- a/crates/matrix-sdk-base/src/lib.rs +++ b/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, diff --git a/crates/matrix-sdk-base/src/store/memory_store.rs b/crates/matrix-sdk-base/src/store/memory_store.rs index f166475fc..6ffedcf68 100644 --- a/crates/matrix-sdk-base/src/store/memory_store.rs +++ b/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>, seen_knock_requests: BTreeMap>, thread_subscriptions: BTreeMap>, + thread_subscriptions_catchup_tokens: Option>, } /// 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(()) } diff --git a/crates/matrix-sdk-base/src/store/mod.rs b/crates/matrix-sdk-base/src/store/mod.rs index a28625c67..1e5da059d 100644 --- a/crates/matrix-sdk-base/src/store/mod.rs +++ b/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, }, }; diff --git a/crates/matrix-sdk-base/src/store/traits.rs b/crates/matrix-sdk-base/src/store/traits.rs index 783dc2dab..ed37c430f 100644 --- a/crates/matrix-sdk-base/src/store/traits.rs +++ b/crates/matrix-sdk-base/src/store/traits.rs @@ -1151,6 +1151,36 @@ pub enum StateStoreDataValue { /// A list of knock request ids marked as seen in a room. SeenKnockRequests(BTreeMap), + + /// A list of tokens to continue thread subscriptions catchup. + /// + /// See documentation of [`ThreadSubscriptionCatchupToken`] for more + /// details. + ThreadSubscriptionsCatchupTokens(Vec), +} + +/// 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. +#[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, } /// Current draft of the composer for the room. @@ -1222,6 +1252,14 @@ impl StateStoreDataValue { pub fn into_seen_knock_requests(self) -> Option> { 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> { + as_variant!(self, Self::ThreadSubscriptionsCatchupTokens) + } } /// A key for key-value data. @@ -1258,6 +1296,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 +1335,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 diff --git a/crates/matrix-sdk-indexeddb/src/state_store/mod.rs b/crates/matrix-sdk-indexeddb/src/state_store/mod.rs index 4d43d0ece..82728bf17 100644 --- a/crates/matrix-sdk-indexeddb/src/state_store/mod.rs +++ b/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::>(&f)) .transpose()? .map(StateStoreDataValue::SeenKnockRequests), + StateStoreDataKey::ThreadSubscriptionsCatchupTokens => value + .map(|f| self.deserialize_value::>(&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 = diff --git a/crates/matrix-sdk-sqlite/src/state_store.rs b/crates/matrix-sdk-sqlite/src/state_store.rs index 3c5111b2c..f9f6188b5 100644 --- a/crates/matrix-sdk-sqlite/src/state_store.rs +++ b/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(()); } }