Skip to content

Commit 67b1de6

Browse files
razvpHywan
razvp
authored and
Hywan
committed
feat(state-stores): Add StateStore::upsert_thread_subscriptions() method for bulk upsert
1 parent 1af22a7 commit 67b1de6

8 files changed

Lines changed: 332 additions & 1 deletion

File tree

‎crates/matrix-sdk-base/CHANGELOG.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,8 @@ All notable changes to this project will be documented in this file.
5050
([#5817](https://github.com/matrix-org/matrix-rust-sdk/pull/5817))
5151
- `ComposerDraft` can now store attachments alongside text messages.
5252
([#5794](https://github.com/matrix-org/matrix-rust-sdk/pull/5794))
53+
- Add `StateStore::upsert_thread_subscriptions()` method for bulk upserts.
54+
([#5848](https://github.com/matrix-org/matrix-rust-sdk/pull/5848))
5355

5456
## [0.14.1] - 2025-09-10
5557

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

Lines changed: 185 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -111,6 +111,8 @@ pub trait StateStoreIntegrationTests {
111111
async fn test_thread_subscriptions(&self) -> TestResult;
112112
/// Test thread subscription bumpstamp semantics.
113113
async fn test_thread_subscriptions_bumpstamps(&self) -> TestResult;
114+
/// Test thread subscriptions bulk upsert, including bumpstamp semantics.
115+
async fn test_thread_subscriptions_bulk_upsert(&self) -> TestResult;
114116
}
115117

116118
impl StateStoreIntegrationTests for DynStateStore {
@@ -2004,6 +2006,183 @@ impl StateStoreIntegrationTests for DynStateStore {
20042006

20052007
Ok(())
20062008
}
2009+
2010+
async fn test_thread_subscriptions_bulk_upsert(&self) -> TestResult {
2011+
let threads = [
2012+
event_id!("$t1"),
2013+
event_id!("$t2"),
2014+
event_id!("$t3"),
2015+
event_id!("$t4"),
2016+
event_id!("$t5"),
2017+
event_id!("$t6"),
2018+
];
2019+
// Helper for building the input for `upsert_thread_subscriptions()`,
2020+
// which is of the type: Vec<(&RoomId, &EventId, StoredThreadSubscription)>
2021+
let build_subscription_updates = |subs: &[StoredThreadSubscription]| {
2022+
threads
2023+
.iter()
2024+
.zip(subs)
2025+
.map(|(&event_id, &sub)| (room_id(), event_id, sub))
2026+
.collect::<Vec<_>>()
2027+
};
2028+
2029+
// Test bump_stamp logic
2030+
let initial_subscriptions = build_subscription_updates(&[
2031+
StoredThreadSubscription {
2032+
status: ThreadSubscriptionStatus::Unsubscribed,
2033+
bump_stamp: None,
2034+
},
2035+
StoredThreadSubscription {
2036+
status: ThreadSubscriptionStatus::Unsubscribed,
2037+
bump_stamp: Some(14),
2038+
},
2039+
StoredThreadSubscription {
2040+
status: ThreadSubscriptionStatus::Unsubscribed,
2041+
bump_stamp: None,
2042+
},
2043+
StoredThreadSubscription {
2044+
status: ThreadSubscriptionStatus::Unsubscribed,
2045+
bump_stamp: Some(210),
2046+
},
2047+
StoredThreadSubscription {
2048+
status: ThreadSubscriptionStatus::Unsubscribed,
2049+
bump_stamp: Some(5),
2050+
},
2051+
StoredThreadSubscription {
2052+
status: ThreadSubscriptionStatus::Unsubscribed,
2053+
bump_stamp: Some(100),
2054+
},
2055+
]);
2056+
2057+
let update_subscriptions = build_subscription_updates(&[
2058+
StoredThreadSubscription {
2059+
status: ThreadSubscriptionStatus::Subscribed { automatic: true },
2060+
bump_stamp: None,
2061+
},
2062+
StoredThreadSubscription {
2063+
status: ThreadSubscriptionStatus::Subscribed { automatic: true },
2064+
bump_stamp: None,
2065+
},
2066+
StoredThreadSubscription {
2067+
status: ThreadSubscriptionStatus::Subscribed { automatic: true },
2068+
bump_stamp: Some(1101),
2069+
},
2070+
StoredThreadSubscription {
2071+
status: ThreadSubscriptionStatus::Subscribed { automatic: true },
2072+
bump_stamp: Some(222),
2073+
},
2074+
StoredThreadSubscription {
2075+
status: ThreadSubscriptionStatus::Subscribed { automatic: true },
2076+
bump_stamp: Some(1),
2077+
},
2078+
StoredThreadSubscription {
2079+
status: ThreadSubscriptionStatus::Subscribed { automatic: true },
2080+
bump_stamp: Some(100),
2081+
},
2082+
]);
2083+
2084+
let expected_subscriptions = build_subscription_updates(&[
2085+
// Status should be updated, because prev and new bump_stamp are both None
2086+
StoredThreadSubscription {
2087+
status: ThreadSubscriptionStatus::Subscribed { automatic: true },
2088+
bump_stamp: None,
2089+
},
2090+
// Status should be updated, but keep initial bump_stamp (new is None)
2091+
StoredThreadSubscription {
2092+
status: ThreadSubscriptionStatus::Subscribed { automatic: true },
2093+
bump_stamp: Some(14),
2094+
},
2095+
// Status should be updated and also bump_stamp should be updated (initial was None)
2096+
StoredThreadSubscription {
2097+
status: ThreadSubscriptionStatus::Subscribed { automatic: true },
2098+
bump_stamp: Some(1101),
2099+
},
2100+
// Status should be updated and also bump_stamp should be updated (initial was lower)
2101+
StoredThreadSubscription {
2102+
status: ThreadSubscriptionStatus::Subscribed { automatic: true },
2103+
bump_stamp: Some(222),
2104+
},
2105+
// Status shouldn't change, as new bump_stamp is lower
2106+
StoredThreadSubscription {
2107+
status: ThreadSubscriptionStatus::Unsubscribed,
2108+
bump_stamp: Some(5),
2109+
},
2110+
// Status shouldn't change, as bump_stamp is equal to the previous one
2111+
StoredThreadSubscription {
2112+
status: ThreadSubscriptionStatus::Unsubscribed,
2113+
bump_stamp: Some(100),
2114+
},
2115+
]);
2116+
2117+
// Set the initial subscriptions
2118+
self.upsert_thread_subscriptions(initial_subscriptions.clone()).await?;
2119+
2120+
// Assert the subscriptions have been added
2121+
for (room_id, thread_id, expected_sub) in &initial_subscriptions {
2122+
let stored_subscription = self.load_thread_subscription(room_id, thread_id).await?;
2123+
assert_eq!(stored_subscription, Some(*expected_sub));
2124+
}
2125+
2126+
// Update subscriptions
2127+
self.upsert_thread_subscriptions(update_subscriptions).await?;
2128+
2129+
// Assert the expected subscriptions and bump_stamps
2130+
for (room_id, thread_id, expected_sub) in &expected_subscriptions {
2131+
let stored_subscription = self.load_thread_subscription(room_id, thread_id).await?;
2132+
assert_eq!(stored_subscription, Some(*expected_sub));
2133+
}
2134+
2135+
// Test just state changes, but first remove previous subscriptions
2136+
for (room_id, thread_id, _) in &expected_subscriptions {
2137+
self.remove_thread_subscription(room_id, thread_id).await?;
2138+
}
2139+
2140+
let initial_subscriptions = build_subscription_updates(&[
2141+
StoredThreadSubscription {
2142+
status: ThreadSubscriptionStatus::Unsubscribed,
2143+
bump_stamp: Some(1),
2144+
},
2145+
StoredThreadSubscription {
2146+
status: ThreadSubscriptionStatus::Subscribed { automatic: false },
2147+
bump_stamp: Some(1),
2148+
},
2149+
StoredThreadSubscription {
2150+
status: ThreadSubscriptionStatus::Subscribed { automatic: true },
2151+
bump_stamp: Some(1),
2152+
},
2153+
]);
2154+
2155+
self.upsert_thread_subscriptions(initial_subscriptions.clone()).await?;
2156+
2157+
for (room_id, thread_id, expected_sub) in &initial_subscriptions {
2158+
let stored_subscription = self.load_thread_subscription(room_id, thread_id).await?;
2159+
assert_eq!(stored_subscription, Some(*expected_sub));
2160+
}
2161+
2162+
let update_subscriptions = build_subscription_updates(&[
2163+
StoredThreadSubscription {
2164+
status: ThreadSubscriptionStatus::Subscribed { automatic: true },
2165+
bump_stamp: Some(2),
2166+
},
2167+
StoredThreadSubscription {
2168+
status: ThreadSubscriptionStatus::Unsubscribed,
2169+
bump_stamp: Some(2),
2170+
},
2171+
StoredThreadSubscription {
2172+
status: ThreadSubscriptionStatus::Subscribed { automatic: false },
2173+
bump_stamp: Some(2),
2174+
},
2175+
]);
2176+
2177+
self.upsert_thread_subscriptions(update_subscriptions.clone()).await?;
2178+
2179+
for (room_id, thread_id, expected_sub) in &update_subscriptions {
2180+
let stored_subscription = self.load_thread_subscription(room_id, thread_id).await?;
2181+
assert_eq!(stored_subscription, Some(*expected_sub));
2182+
}
2183+
2184+
Ok(())
2185+
}
20072186
}
20082187

20092188
/// Macro building to allow your StateStore implementation to run the entire
@@ -2196,6 +2375,12 @@ macro_rules! statestore_integration_tests {
21962375
let store = get_store().await?.into_state_store();
21972376
store.test_thread_subscriptions_bumpstamps().await
21982377
}
2378+
2379+
#[async_test]
2380+
async fn test_thread_subscriptions_bulk_upsert() -> TestResult {
2381+
let store = get_store().await?.into_state_store();
2382+
store.test_thread_subscriptions_bulk_upsert().await
2383+
}
21992384
}
22002385
};
22012386
}

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

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1017,6 +1017,33 @@ impl StateStore for MemoryStore {
10171017
Ok(())
10181018
}
10191019

1020+
async fn upsert_thread_subscriptions(
1021+
&self,
1022+
updates: Vec<(&RoomId, &EventId, StoredThreadSubscription)>,
1023+
) -> Result<(), Self::Error> {
1024+
let mut inner = self.inner.write().unwrap();
1025+
1026+
for (room_id, thread_id, mut new) in updates {
1027+
let room_subs = inner.thread_subscriptions.entry(room_id.to_owned()).or_default();
1028+
1029+
if let Some(previous) = room_subs.get(thread_id) {
1030+
if *previous == new {
1031+
continue;
1032+
}
1033+
if !compare_thread_subscription_bump_stamps(
1034+
previous.bump_stamp,
1035+
&mut new.bump_stamp,
1036+
) {
1037+
continue;
1038+
}
1039+
}
1040+
1041+
room_subs.insert(thread_id.to_owned(), new);
1042+
}
1043+
1044+
Ok(())
1045+
}
1046+
10201047
async fn load_thread_subscription(
10211048
&self,
10221049
room: &RoomId,

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

Lines changed: 22 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -487,7 +487,7 @@ pub trait StateStore: AsyncTraitDeps {
487487
/// bumpstamp is kept.
488488
///
489489
/// If the new thread subscription has a bumpstamp that's lower than or
490-
/// equal to a previously one, the existing subscription is kept, i.e.
490+
/// equal to a previous one, the existing subscription is kept, i.e.
491491
/// this method must have no effect.
492492
async fn upsert_thread_subscription(
493493
&self,
@@ -496,6 +496,20 @@ pub trait StateStore: AsyncTraitDeps {
496496
subscription: StoredThreadSubscription,
497497
) -> Result<(), Self::Error>;
498498

499+
/// Inserts or updates multiple thread subscriptions.
500+
///
501+
/// If the new thread subscription hasn't set a bumpstamp, and there was a
502+
/// previous subscription in the database with a bumpstamp, the existing
503+
/// bumpstamp is kept.
504+
///
505+
/// If the new thread subscription has a bumpstamp that's lower than or
506+
/// equal to a previous one, the existing subscription is kept, i.e.
507+
/// this method must have no effect.
508+
async fn upsert_thread_subscriptions(
509+
&self,
510+
updates: Vec<(&RoomId, &EventId, StoredThreadSubscription)>,
511+
) -> Result<(), Self::Error>;
512+
499513
/// Remove a previous thread subscription for a given room and thread.
500514
///
501515
/// Note: removing an unknown thread subscription is a no-op.
@@ -828,6 +842,13 @@ impl<T: StateStore> StateStore for EraseStateStoreError<T> {
828842
self.0.upsert_thread_subscription(room, thread_id, subscription).await.map_err(Into::into)
829843
}
830844

845+
async fn upsert_thread_subscriptions(
846+
&self,
847+
updates: Vec<(&RoomId, &EventId, StoredThreadSubscription)>,
848+
) -> Result<(), Self::Error> {
849+
self.0.upsert_thread_subscriptions(updates).await.map_err(Into::into)
850+
}
851+
831852
async fn load_thread_subscription(
832853
&self,
833854
room: &RoomId,

‎crates/matrix-sdk-indexeddb/CHANGELOG.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@ All notable changes to this project will be documented in this file.
1919
([#5819](https://github.com/matrix-org/matrix-rust-sdk/pull/5819))
2020
- [**breaking**] `IndexeddbCryptoStore::get_withheld_info` now returns `Result<Option<RoomKeyWithheldEntry>, ...>`.
2121
([#5737](https://github.com/matrix-org/matrix-rust-sdk/pull/5737))
22+
- Implement `StateStore::upsert_thread_subscriptions()` method for bulk upserts.
23+
([#5848](https://github.com/matrix-org/matrix-rust-sdk/pull/5848))
2224

2325
### Performance
2426

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

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1978,6 +1978,47 @@ impl_state_store!({
19781978
Ok(())
19791979
}
19801980

1981+
async fn upsert_thread_subscriptions(
1982+
&self,
1983+
updates: Vec<(&RoomId, &EventId, StoredThreadSubscription)>,
1984+
) -> Result<()> {
1985+
let tx = self
1986+
.inner
1987+
.transaction(keys::THREAD_SUBSCRIPTIONS)
1988+
.with_mode(TransactionMode::Readwrite)
1989+
.build()?;
1990+
let obj = tx.object_store(keys::THREAD_SUBSCRIPTIONS)?;
1991+
1992+
for (room_id, thread_id, subscription) in updates {
1993+
let encoded_key = self.encode_key(keys::THREAD_SUBSCRIPTIONS, (room_id, thread_id));
1994+
let mut new = PersistedThreadSubscription::from(subscription);
1995+
1996+
// See if there's a previous subscription.
1997+
if let Some(previous_value) = obj.get(&encoded_key).await? {
1998+
let previous: PersistedThreadSubscription =
1999+
self.deserialize_value(&previous_value)?;
2000+
2001+
// If the previous status is the same as the new one, don't do anything.
2002+
if new == previous {
2003+
continue;
2004+
}
2005+
if !compare_thread_subscription_bump_stamps(
2006+
previous.bump_stamp,
2007+
&mut new.bump_stamp,
2008+
) {
2009+
continue;
2010+
}
2011+
}
2012+
2013+
let serialized_value = self.serialize_value(&new);
2014+
obj.put(&serialized_value?).with_key(encoded_key).build()?;
2015+
}
2016+
2017+
tx.commit().await?;
2018+
2019+
Ok(())
2020+
}
2021+
19812022
async fn load_thread_subscription(
19822023
&self,
19832024
room: &RoomId,

‎crates/matrix-sdk-sqlite/CHANGELOG.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@ All notable changes to this project will be documented in this file.
1717

1818
- Implement a new constructor that allows to open `SqliteCryptoStore` with a cryptographic key
1919
([#5472](https://github.com/matrix-org/matrix-rust-sdk/pull/5472))
20+
- Implement `StateStore::upsert_thread_subscriptions()` method for bulk upserts.
21+
([#5848](https://github.com/matrix-org/matrix-rust-sdk/pull/5848))
2022

2123
### Refactor
2224
- [breaking] Change the logic for opening a store so as to use a `Secret` enum in the function `open_with_pool` instead of a `passphrase`

0 commit comments

Comments
 (0)