1use std::{
16 cmp::Reverse,
17 collections::{BTreeMap, BTreeSet, HashMap, HashSet},
18 str::FromStr as _,
19 sync::Arc,
20};
21
22use async_trait::async_trait;
23use gloo_utils::format::JsValueSerdeExt;
24use growable_bloom_filter::GrowableBloom;
25use indexed_db_futures::{
26 KeyRange, cursor::CursorDirection, database::Database, error::OpenDbError, prelude::*,
27 transaction::TransactionMode,
28};
29use matrix_sdk_base::{
30 MinimalRoomMemberEvent, ROOM_VERSION_FALLBACK, ROOM_VERSION_RULES_FALLBACK, RoomInfo,
31 RoomMemberships, StateStoreDataKey, StateStoreDataValue, ThreadSubscriptionCatchupToken,
32 deserialized_responses::{DisplayName, RawAnySyncOrStrippedState},
33 store::{
34 ChildTransactionId, ComposerDraft, DependentQueuedRequest, DependentQueuedRequestKind,
35 QueuedRequest, QueuedRequestKind, RoomLoadSettings, SentRequestKey,
36 SerializableEventContent, StateChanges, StateStore, StoreError, StoredThreadSubscription,
37 SupportedVersionsResponse, ThreadSubscriptionStatus, WellKnownResponse,
38 compare_thread_subscription_bump_stamps,
39 },
40 ttl::TtlValue,
41};
42use matrix_sdk_store_encryption::{Error as EncryptionError, StoreCipher};
43use ruma::{
44 CanonicalJsonObject, EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedMxcUri,
45 OwnedRoomId, OwnedTransactionId, OwnedUserId, RoomId, TransactionId, UserId,
46 api::client::discovery::get_capabilities::v3::Capabilities,
47 canonical_json::{RedactedBecause, redact},
48 events::{
49 AnyGlobalAccountDataEvent, AnyRoomAccountDataEvent, AnySyncStateEvent,
50 GlobalAccountDataEventType, RoomAccountDataEventType, StateEventType, SyncStateEvent,
51 presence::PresenceEvent,
52 receipt::{Receipt, ReceiptThread, ReceiptType},
53 room::member::{
54 MembershipState, RoomMemberEventContent, StrippedRoomMemberEvent, SyncRoomMemberEvent,
55 },
56 },
57 profile::{UserProfile, UserProfileUpdate},
58 serde::Raw,
59};
60use serde::{Deserialize, Serialize, de::DeserializeOwned, ser::Error};
61use tracing::{debug, warn};
62use wasm_bindgen::JsValue;
63
64mod migrations;
65
66pub use self::migrations::MigrationConflictStrategy;
67use self::migrations::{upgrade_inner_db, upgrade_meta_db};
68use crate::{error::GenericError, serializer::safe_encode::traits::SafeEncode};
69
70#[derive(Debug, thiserror::Error)]
71pub enum IndexeddbStateStoreError {
72 #[error(transparent)]
73 Json(#[from] serde_json::Error),
74 #[error(transparent)]
75 Encryption(#[from] EncryptionError),
76 #[error("DomException {name} ({code}): {message}")]
77 DomException { name: String, message: String, code: u16 },
78 #[error(transparent)]
79 StoreError(#[from] StoreError),
80 #[error(
81 "Can't migrate {name} from {old_version} to {new_version} without deleting data. \
82 See MigrationConflictStrategy for ways to configure."
83 )]
84 MigrationConflict { name: String, old_version: u32, new_version: u32 },
85}
86
87impl From<GenericError> for IndexeddbStateStoreError {
88 fn from(value: GenericError) -> Self {
89 Self::StoreError(value.into())
90 }
91}
92
93impl From<web_sys::DomException> for IndexeddbStateStoreError {
94 fn from(frm: web_sys::DomException) -> IndexeddbStateStoreError {
95 IndexeddbStateStoreError::DomException {
96 name: frm.name(),
97 message: frm.message(),
98 code: frm.code(),
99 }
100 }
101}
102
103impl From<IndexeddbStateStoreError> for StoreError {
104 fn from(e: IndexeddbStateStoreError) -> Self {
105 match e {
106 IndexeddbStateStoreError::Json(e) => StoreError::Json(e),
107 IndexeddbStateStoreError::StoreError(e) => e,
108 IndexeddbStateStoreError::Encryption(e) => StoreError::Encryption(e),
109 _ => StoreError::backend(e),
110 }
111 }
112}
113
114impl From<indexed_db_futures::error::DomException> for IndexeddbStateStoreError {
115 fn from(value: indexed_db_futures::error::DomException) -> Self {
116 web_sys::DomException::from(value).into()
117 }
118}
119
120impl From<indexed_db_futures::error::SerialisationError> for IndexeddbStateStoreError {
121 fn from(value: indexed_db_futures::error::SerialisationError) -> Self {
122 Self::Json(serde_json::Error::custom(value.to_string()))
123 }
124}
125
126impl From<indexed_db_futures::error::UnexpectedDataError> for IndexeddbStateStoreError {
127 fn from(value: indexed_db_futures::error::UnexpectedDataError) -> Self {
128 IndexeddbStateStoreError::StoreError(StoreError::backend(value))
129 }
130}
131
132impl From<indexed_db_futures::error::JSError> for IndexeddbStateStoreError {
133 fn from(value: indexed_db_futures::error::JSError) -> Self {
134 GenericError::from(value.to_string()).into()
135 }
136}
137
138impl From<indexed_db_futures::error::Error> for IndexeddbStateStoreError {
139 fn from(value: indexed_db_futures::error::Error) -> Self {
140 use indexed_db_futures::error::Error;
141 match value {
142 Error::DomException(e) => e.into(),
143 Error::Serialisation(e) => e.into(),
144 Error::MissingData(e) => e.into(),
145 Error::Unknown(e) => e.into(),
146 }
147 }
148}
149
150impl From<OpenDbError> for IndexeddbStateStoreError {
151 fn from(value: OpenDbError) -> Self {
152 match value {
153 OpenDbError::Base(error) => error.into(),
154 _ => GenericError::from(value.to_string()).into(),
155 }
156 }
157}
158
159mod keys {
160 pub const INTERNAL_STATE: &str = "matrix-sdk-state";
161 pub const BACKUPS_META: &str = "backups";
162
163 pub const ACCOUNT_DATA: &str = "account_data";
164
165 pub const PROFILES: &str = "profiles";
167 pub const DISPLAY_NAMES: &str = "display_names";
168 pub const USER_IDS: &str = "user_ids";
169
170 pub const ROOM_STATE: &str = "room_state";
171 pub const ROOM_INFOS: &str = "room_infos";
172 pub const PRESENCE: &str = "presence";
173 pub const ROOM_ACCOUNT_DATA: &str = "room_account_data";
174 pub const ROOM_SEND_QUEUE: &str = "room_send_queue";
176 pub const DEPENDENT_SEND_QUEUE: &str = "room_dependent_send_queue";
178 pub const THREAD_SUBSCRIPTIONS: &str = "room_thread_subscriptions";
179
180 pub const STRIPPED_ROOM_STATE: &str = "stripped_room_state";
181 pub const STRIPPED_USER_IDS: &str = "stripped_user_ids";
182
183 pub const ROOM_USER_RECEIPTS: &str = "room_user_receipts";
184 pub const ROOM_EVENT_RECEIPTS: &str = "room_event_receipts";
185
186 pub const GLOBAL_PROFILES: &str = "global_profiles";
187 pub const UNAVAILABLE_GLOBAL_PROFILES: &str = "unavailable_global_profiles";
188
189 pub const CUSTOM: &str = "custom";
190 pub const KV: &str = "kv";
191
192 pub const ALL_STORES: &[&str] = &[
194 ACCOUNT_DATA,
195 PROFILES,
196 DISPLAY_NAMES,
197 USER_IDS,
198 ROOM_STATE,
199 ROOM_INFOS,
200 PRESENCE,
201 ROOM_ACCOUNT_DATA,
202 STRIPPED_ROOM_STATE,
203 STRIPPED_USER_IDS,
204 ROOM_USER_RECEIPTS,
205 ROOM_EVENT_RECEIPTS,
206 ROOM_SEND_QUEUE,
207 THREAD_SUBSCRIPTIONS,
208 DEPENDENT_SEND_QUEUE,
209 GLOBAL_PROFILES,
210 UNAVAILABLE_GLOBAL_PROFILES,
211 CUSTOM,
212 KV,
213 ];
214
215 pub const STORE_KEY: &str = "store_key";
218}
219
220pub use keys::ALL_STORES;
221use matrix_sdk_base::store::QueueWedgeError;
222
223fn serialize_value(store_cipher: Option<&StoreCipher>, event: &impl Serialize) -> Result<JsValue> {
225 Ok(match store_cipher {
226 Some(cipher) => {
227 let data = serde_json::to_vec(event)?;
228 JsValue::from_serde(&cipher.encrypt_value_data(data)?)?
229 }
230 None => JsValue::from_serde(event)?,
231 })
232}
233
234fn deserialize_value<T: DeserializeOwned>(
236 store_cipher: Option<&StoreCipher>,
237 event: &JsValue,
238) -> Result<T> {
239 match store_cipher {
240 Some(cipher) => {
241 use zeroize::Zeroize;
242 let mut plaintext = cipher.decrypt_value_data(event.into_serde()?)?;
243 let ret = serde_json::from_slice(&plaintext);
244 plaintext.zeroize();
245 Ok(ret?)
246 }
247 None => Ok(event.into_serde()?),
248 }
249}
250
251fn encode_key<T>(store_cipher: Option<&StoreCipher>, table_name: &str, key: T) -> JsValue
252where
253 T: SafeEncode,
254{
255 match store_cipher {
256 Some(cipher) => key.as_secure_string(table_name, cipher),
257 None => key.as_encoded_string(),
258 }
259 .into()
260}
261
262fn encode_to_range<T>(
263 store_cipher: Option<&StoreCipher>,
264 table_name: &str,
265 key: T,
266) -> KeyRange<JsValue>
267where
268 T: SafeEncode,
269{
270 match store_cipher {
271 Some(cipher) => key.encode_to_range_secure(table_name, cipher),
272 None => key.encode_to_range(),
273 }
274}
275
276#[derive(Debug)]
278pub struct IndexeddbStateStoreBuilder {
279 name: Option<String>,
280 passphrase: Option<String>,
281 migration_conflict_strategy: MigrationConflictStrategy,
282}
283
284impl IndexeddbStateStoreBuilder {
285 fn new() -> Self {
286 Self {
287 name: None,
288 passphrase: None,
289 migration_conflict_strategy: MigrationConflictStrategy::BackupAndDrop,
290 }
291 }
292
293 pub fn name(mut self, value: String) -> Self {
295 self.name = Some(value);
296 self
297 }
298
299 pub fn passphrase(mut self, value: String) -> Self {
303 self.passphrase = Some(value);
304 self
305 }
306
307 pub fn migration_conflict_strategy(mut self, value: MigrationConflictStrategy) -> Self {
311 self.migration_conflict_strategy = value;
312 self
313 }
314
315 pub async fn build(self) -> Result<IndexeddbStateStore> {
316 let migration_strategy = self.migration_conflict_strategy.clone();
317 let name = self.name.unwrap_or_else(|| "state".to_owned());
318
319 let meta_name = format!("{name}::{}", keys::INTERNAL_STATE);
320
321 let (meta, store_cipher) = upgrade_meta_db(&meta_name, self.passphrase.as_deref()).await?;
322 let inner =
323 upgrade_inner_db(&name, store_cipher.as_deref(), migration_strategy, &meta).await?;
324
325 Ok(IndexeddbStateStore { name, inner, meta, store_cipher })
326 }
327}
328
329pub struct IndexeddbStateStore {
330 name: String,
331 pub(crate) inner: Database,
332 pub(crate) meta: Database,
333 pub(crate) store_cipher: Option<Arc<StoreCipher>>,
334}
335
336#[cfg(not(tarpaulin_include))]
337impl std::fmt::Debug for IndexeddbStateStore {
338 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
339 f.debug_struct("IndexeddbStateStore").field("name", &self.name).finish()
340 }
341}
342
343type Result<A, E = IndexeddbStateStoreError> = std::result::Result<A, E>;
344
345impl IndexeddbStateStore {
346 pub fn builder() -> IndexeddbStateStoreBuilder {
348 IndexeddbStateStoreBuilder::new()
349 }
350
351 pub fn version(&self) -> u32 {
353 self.inner.version() as u32
354 }
355
356 pub fn meta_version(&self) -> u32 {
358 self.meta.version() as u32
359 }
360
361 pub async fn has_backups(&self) -> Result<bool> {
363 Ok(self
364 .meta
365 .transaction(keys::BACKUPS_META)
366 .with_mode(TransactionMode::Readonly)
367 .build()?
368 .object_store(keys::BACKUPS_META)?
369 .count()
370 .await?
371 > 0)
372 }
373
374 pub async fn latest_backup(&self) -> Result<Option<String>> {
376 if let Some(mut cursor) = self
377 .meta
378 .transaction(keys::BACKUPS_META)
379 .with_mode(TransactionMode::Readonly)
380 .build()?
381 .object_store(keys::BACKUPS_META)?
382 .open_cursor()
383 .with_direction(CursorDirection::Prev)
384 .await?
385 && let Some(record) = cursor.next_record::<JsValue>().await?
386 {
387 Ok(record.as_string())
388 } else {
389 Ok(None)
390 }
391 }
392
393 fn serialize_value(&self, event: &impl Serialize) -> Result<JsValue> {
395 serialize_value(self.store_cipher.as_deref(), event)
396 }
397
398 fn deserialize_value<T: DeserializeOwned>(&self, event: &JsValue) -> Result<T> {
400 deserialize_value(self.store_cipher.as_deref(), event)
401 }
402
403 fn encode_key<T>(&self, table_name: &str, key: T) -> JsValue
404 where
405 T: SafeEncode,
406 {
407 encode_key(self.store_cipher.as_deref(), table_name, key)
408 }
409
410 fn encode_to_range<T>(&self, table_name: &str, key: T) -> KeyRange<JsValue>
411 where
412 T: SafeEncode,
413 {
414 encode_to_range(self.store_cipher.as_deref(), table_name, key)
415 }
416
417 pub async fn get_user_ids_inner(
420 &self,
421 room_id: &RoomId,
422 memberships: RoomMemberships,
423 stripped: bool,
424 ) -> Result<Vec<OwnedUserId>> {
425 let store_name = if stripped { keys::STRIPPED_USER_IDS } else { keys::USER_IDS };
426
427 let tx = self.inner.transaction(store_name).with_mode(TransactionMode::Readonly).build()?;
428 let store = tx.object_store(store_name)?;
429 let range = self.encode_to_range(store_name, room_id);
430
431 let user_ids = if memberships.is_empty() {
432 store
434 .get_all()
435 .with_query(&range)
436 .await?
437 .filter_map(Result::ok)
438 .filter_map(|f| self.deserialize_value::<RoomMember>(&f).ok().map(|m| m.user_id))
439 .collect::<Vec<_>>()
440 } else {
441 let mut user_ids = Vec::new();
442 let cursor = store.open_cursor().with_query(&range).await?;
443
444 if let Some(mut cursor) = cursor {
445 while let Some(value) = cursor.next_record().await? {
446 let member = self.deserialize_value::<RoomMember>(&value)?;
447
448 if memberships.matches(&member.membership) {
449 user_ids.push(member.user_id);
450 }
451 }
452 }
453
454 user_ids
455 };
456
457 Ok(user_ids)
458 }
459
460 async fn get_custom_value_for_js(&self, jskey: &JsValue) -> Result<Option<Vec<u8>>> {
461 self.inner
462 .transaction(keys::CUSTOM)
463 .with_mode(TransactionMode::Readonly)
464 .build()?
465 .object_store(keys::CUSTOM)?
466 .get(jskey)
467 .await?
468 .map(|f| self.deserialize_value(&f))
469 .transpose()
470 }
471
472 fn encode_kv_data_key(&self, key: StateStoreDataKey<'_>) -> JsValue {
473 match key {
477 StateStoreDataKey::SyncToken => {
478 self.encode_key(StateStoreDataKey::SYNC_TOKEN, StateStoreDataKey::SYNC_TOKEN)
479 }
480 StateStoreDataKey::SupportedVersions => self.encode_key(
481 StateStoreDataKey::SUPPORTED_VERSIONS,
482 StateStoreDataKey::SUPPORTED_VERSIONS,
483 ),
484 StateStoreDataKey::WellKnown => {
485 self.encode_key(keys::KV, StateStoreDataKey::WELL_KNOWN)
486 }
487 StateStoreDataKey::Filter(filter_name) => {
488 self.encode_key(StateStoreDataKey::FILTER, (StateStoreDataKey::FILTER, filter_name))
489 }
490 StateStoreDataKey::UserAvatarUrl(user_id) => {
491 self.encode_key(keys::KV, (StateStoreDataKey::USER_AVATAR_URL, user_id))
492 }
493 StateStoreDataKey::RecentlyVisitedRooms(user_id) => {
494 self.encode_key(keys::KV, (StateStoreDataKey::RECENTLY_VISITED_ROOMS, user_id))
495 }
496 StateStoreDataKey::UtdHookManagerData => {
497 self.encode_key(keys::KV, StateStoreDataKey::UTD_HOOK_MANAGER_DATA)
498 }
499 StateStoreDataKey::OneTimeKeyAlreadyUploaded => {
500 self.encode_key(keys::KV, StateStoreDataKey::ONE_TIME_KEY_ALREADY_UPLOADED)
501 }
502 StateStoreDataKey::ComposerDraft(room_id, thread_root) => {
503 if let Some(thread_root) = thread_root {
504 self.encode_key(
505 keys::KV,
506 (StateStoreDataKey::COMPOSER_DRAFT, (room_id, thread_root)),
507 )
508 } else {
509 self.encode_key(keys::KV, (StateStoreDataKey::COMPOSER_DRAFT, room_id))
510 }
511 }
512 StateStoreDataKey::SeenKnockRequests(room_id) => {
513 self.encode_key(keys::KV, (StateStoreDataKey::SEEN_KNOCK_REQUESTS, room_id))
514 }
515 StateStoreDataKey::ThreadSubscriptionsCatchupTokens => {
516 self.encode_key(keys::KV, StateStoreDataKey::THREAD_SUBSCRIPTIONS_CATCHUP_TOKENS)
517 }
518 StateStoreDataKey::HomeserverCapabilities => {
519 self.encode_key(keys::KV, StateStoreDataKey::HOMESERVER_CAPABILITIES)
520 }
521 }
522 }
523}
524
525#[derive(Serialize, Deserialize)]
528struct PersistedQueuedRequest {
529 pub room_id: OwnedRoomId,
531
532 kind: Option<QueuedRequestKind>,
535 transaction_id: OwnedTransactionId,
536
537 pub error: Option<QueueWedgeError>,
538
539 priority: Option<usize>,
540
541 #[serde(default = "created_now")]
543 created_at: MilliSecondsSinceUnixEpoch,
544
545 is_wedged: Option<bool>,
549
550 event: Option<SerializableEventContent>,
551}
552
553fn created_now() -> MilliSecondsSinceUnixEpoch {
554 MilliSecondsSinceUnixEpoch::now()
555}
556
557impl PersistedQueuedRequest {
558 fn into_queued_request(self) -> Option<QueuedRequest> {
559 let kind = self.kind.or_else(|| self.event.map(QueuedRequestKind::from))?;
560
561 let error = match self.is_wedged {
562 Some(true) => {
563 Some(QueueWedgeError::GenericApiError {
565 msg: "local echo failed to send in a previous session".into(),
566 })
567 }
568 _ => self.error,
569 };
570
571 let priority = self.priority.unwrap_or(0);
573
574 Some(QueuedRequest {
575 kind,
576 transaction_id: self.transaction_id,
577 error,
578 priority,
579 created_at: self.created_at,
580 })
581 }
582}
583
584#[derive(Serialize, Deserialize, PartialEq)]
585struct PersistedThreadSubscription {
586 status: String,
587 bump_stamp: Option<u64>,
588}
589
590impl From<StoredThreadSubscription> for PersistedThreadSubscription {
591 fn from(value: StoredThreadSubscription) -> Self {
592 Self { status: value.status.as_str().to_owned(), bump_stamp: value.bump_stamp }
593 }
594}
595
596#[cfg(target_family = "wasm")]
605macro_rules! impl_state_store {
606 ({ $($body:tt)* }) => {
607 #[async_trait(?Send)]
608 impl StateStore for IndexeddbStateStore {
609 type Error = IndexeddbStateStoreError;
610
611 $($body)*
612 }
613 };
614}
615
616#[cfg(not(target_family = "wasm"))]
617macro_rules! impl_state_store {
618 ({ $($body:tt)* }) => {
619 impl IndexeddbStateStore {
620 $($body)*
621 }
622 };
623}
624
625impl_state_store!({
626 async fn get_kv_data(&self, key: StateStoreDataKey<'_>) -> Result<Option<StateStoreDataValue>> {
627 let encoded_key = self.encode_kv_data_key(key);
628
629 let value = self
630 .inner
631 .transaction(keys::KV)
632 .with_mode(TransactionMode::Readonly)
633 .build()?
634 .object_store(keys::KV)?
635 .get(&encoded_key)
636 .await?;
637
638 let value = match key {
639 StateStoreDataKey::SyncToken => value
640 .map(|f| self.deserialize_value::<String>(&f))
641 .transpose()?
642 .map(StateStoreDataValue::SyncToken),
643 StateStoreDataKey::SupportedVersions => value
644 .map(|f| self.deserialize_value::<TtlValue<SupportedVersionsResponse>>(&f))
645 .transpose()?
646 .map(StateStoreDataValue::SupportedVersions),
647 StateStoreDataKey::WellKnown => value
648 .map(|f| self.deserialize_value::<TtlValue<Option<WellKnownResponse>>>(&f))
649 .transpose()?
650 .map(StateStoreDataValue::WellKnown),
651 StateStoreDataKey::Filter(_) => value
652 .map(|f| self.deserialize_value::<String>(&f))
653 .transpose()?
654 .map(StateStoreDataValue::Filter),
655 StateStoreDataKey::UserAvatarUrl(_) => value
656 .map(|f| self.deserialize_value::<OwnedMxcUri>(&f))
657 .transpose()?
658 .map(StateStoreDataValue::UserAvatarUrl),
659 StateStoreDataKey::RecentlyVisitedRooms(_) => value
660 .map(|f| self.deserialize_value::<Vec<OwnedRoomId>>(&f))
661 .transpose()?
662 .map(StateStoreDataValue::RecentlyVisitedRooms),
663 StateStoreDataKey::UtdHookManagerData => value
664 .map(|f| self.deserialize_value::<GrowableBloom>(&f))
665 .transpose()?
666 .map(StateStoreDataValue::UtdHookManagerData),
667 StateStoreDataKey::OneTimeKeyAlreadyUploaded => value
668 .map(|f| self.deserialize_value::<bool>(&f))
669 .transpose()?
670 .map(|_| StateStoreDataValue::OneTimeKeyAlreadyUploaded),
671 StateStoreDataKey::ComposerDraft(_, _) => value
672 .map(|f| self.deserialize_value::<ComposerDraft>(&f))
673 .transpose()?
674 .map(StateStoreDataValue::ComposerDraft),
675 StateStoreDataKey::SeenKnockRequests(_) => value
676 .map(|f| self.deserialize_value::<BTreeMap<OwnedEventId, OwnedUserId>>(&f))
677 .transpose()?
678 .map(StateStoreDataValue::SeenKnockRequests),
679 StateStoreDataKey::ThreadSubscriptionsCatchupTokens => value
680 .map(|f| self.deserialize_value::<Vec<ThreadSubscriptionCatchupToken>>(&f))
681 .transpose()?
682 .map(StateStoreDataValue::ThreadSubscriptionsCatchupTokens),
683 StateStoreDataKey::HomeserverCapabilities => value
684 .map(|f| self.deserialize_value::<TtlValue<Capabilities>>(&f))
685 .transpose()?
686 .map(StateStoreDataValue::HomeserverCapabilities),
687 };
688
689 Ok(value)
690 }
691
692 async fn set_kv_data(
693 &self,
694 key: StateStoreDataKey<'_>,
695 value: StateStoreDataValue,
696 ) -> Result<()> {
697 let encoded_key = self.encode_kv_data_key(key);
698
699 let serialized_value = match key {
700 StateStoreDataKey::SyncToken => self
701 .serialize_value(&value.into_sync_token().expect("Session data not a sync token")),
702 StateStoreDataKey::SupportedVersions => self.serialize_value(
703 &value
704 .into_supported_versions()
705 .expect("Session data not containing supported versions"),
706 ),
707 StateStoreDataKey::WellKnown => self.serialize_value(
708 &value.into_well_known().expect("Session data not containing well-known"),
709 ),
710 StateStoreDataKey::Filter(_) => {
711 self.serialize_value(&value.into_filter().expect("Session data not a filter"))
712 }
713 StateStoreDataKey::UserAvatarUrl(_) => self.serialize_value(
714 &value.into_user_avatar_url().expect("Session data not an user avatar url"),
715 ),
716 StateStoreDataKey::RecentlyVisitedRooms(_) => self.serialize_value(
717 &value
718 .into_recently_visited_rooms()
719 .expect("Session data not a recently visited room list"),
720 ),
721 StateStoreDataKey::UtdHookManagerData => self.serialize_value(
722 &value.into_utd_hook_manager_data().expect("Session data not UtdHookManagerData"),
723 ),
724 StateStoreDataKey::OneTimeKeyAlreadyUploaded => self.serialize_value(&true),
725 StateStoreDataKey::ComposerDraft(_, _) => self.serialize_value(
726 &value.into_composer_draft().expect("Session data not a composer draft"),
727 ),
728 StateStoreDataKey::SeenKnockRequests(_) => self.serialize_value(
729 &value
730 .into_seen_knock_requests()
731 .expect("Session data is not a set of seen knock request ids"),
732 ),
733 StateStoreDataKey::ThreadSubscriptionsCatchupTokens => self.serialize_value(
734 &value
735 .into_thread_subscriptions_catchup_tokens()
736 .expect("Session data is not a list of thread subscription catchup tokens"),
737 ),
738 StateStoreDataKey::HomeserverCapabilities => self.serialize_value(
739 &value
740 .into_homeserver_capabilities()
741 .expect("Session data is not a homeserver capabilities"),
742 ),
743 };
744
745 let tx = self.inner.transaction(keys::KV).with_mode(TransactionMode::Readwrite).build()?;
746
747 let obj = tx.object_store(keys::KV)?;
748
749 obj.put(&serialized_value?).with_key(encoded_key).build()?;
750
751 tx.commit().await?;
752
753 Ok(())
754 }
755
756 async fn remove_kv_data(&self, key: StateStoreDataKey<'_>) -> Result<()> {
757 let encoded_key = self.encode_kv_data_key(key);
758
759 let tx = self.inner.transaction(keys::KV).with_mode(TransactionMode::Readwrite).build()?;
760 let obj = tx.object_store(keys::KV)?;
761
762 obj.delete(&encoded_key).build()?;
763
764 tx.commit().await?;
765
766 Ok(())
767 }
768
769 async fn save_changes(&self, changes: &StateChanges) -> Result<()> {
770 let mut stores: HashSet<&'static str> = [
771 (changes.sync_token.is_some(), keys::KV),
772 (!changes.ambiguity_maps.is_empty(), keys::DISPLAY_NAMES),
773 (!changes.account_data.is_empty(), keys::ACCOUNT_DATA),
774 (!changes.presence.is_empty(), keys::PRESENCE),
775 (
776 !changes.profiles.is_empty() || !changes.profiles_to_delete.is_empty(),
777 keys::PROFILES,
778 ),
779 (!changes.room_account_data.is_empty(), keys::ROOM_ACCOUNT_DATA),
780 (!changes.receipts.is_empty(), keys::ROOM_EVENT_RECEIPTS),
781 (!changes.global_profiles.is_empty(), keys::GLOBAL_PROFILES),
782 (
783 !changes.global_profiles.is_empty()
784 || !changes.unavailable_global_profiles.is_empty(),
785 keys::UNAVAILABLE_GLOBAL_PROFILES,
786 ),
787 ]
788 .iter()
789 .filter_map(|(id, key)| if *id { Some(*key) } else { None })
790 .collect();
791
792 if !changes.state.is_empty() {
793 stores.extend([
794 keys::ROOM_STATE,
795 keys::USER_IDS,
796 keys::STRIPPED_USER_IDS,
797 keys::STRIPPED_ROOM_STATE,
798 keys::PROFILES,
799 ]);
800 }
801
802 if !changes.redactions.is_empty() {
803 stores.extend([keys::ROOM_STATE, keys::ROOM_INFOS]);
804 }
805
806 if !changes.room_infos.is_empty() {
807 stores.insert(keys::ROOM_INFOS);
808 }
809
810 if !changes.stripped_state.is_empty() {
811 stores.extend([keys::STRIPPED_ROOM_STATE, keys::STRIPPED_USER_IDS]);
812 }
813
814 if !changes.receipts.is_empty() {
815 stores.extend([keys::ROOM_EVENT_RECEIPTS, keys::ROOM_USER_RECEIPTS])
816 }
817
818 if stores.is_empty() {
819 return Ok(());
821 }
822
823 let stores: Vec<&'static str> = stores.into_iter().collect();
824 let tx = self.inner.transaction(stores).with_mode(TransactionMode::Readwrite).build()?;
825
826 if let Some(s) = &changes.sync_token {
827 tx.object_store(keys::KV)?
828 .put(&self.serialize_value(s)?)
829 .with_key(self.encode_kv_data_key(StateStoreDataKey::SyncToken))
830 .build()?;
831 }
832
833 if !changes.ambiguity_maps.is_empty() {
834 let store = tx.object_store(keys::DISPLAY_NAMES)?;
835 for (room_id, ambiguity_maps) in &changes.ambiguity_maps {
836 for (display_name, map) in ambiguity_maps {
837 let key = self.encode_key(
838 keys::DISPLAY_NAMES,
839 (
840 room_id,
841 display_name
842 .as_normalized_str()
843 .unwrap_or_else(|| display_name.as_raw_str()),
844 ),
845 );
846
847 store.put(&self.serialize_value(&map)?).with_key(key).build()?;
848 }
849 }
850 }
851
852 if !changes.account_data.is_empty() {
853 let store = tx.object_store(keys::ACCOUNT_DATA)?;
854 for (event_type, event) in &changes.account_data {
855 store
856 .put(&self.serialize_value(&event)?)
857 .with_key(self.encode_key(keys::ACCOUNT_DATA, event_type))
858 .build()?;
859 }
860 }
861
862 if !changes.room_account_data.is_empty() {
863 let store = tx.object_store(keys::ROOM_ACCOUNT_DATA)?;
864 for (room, events) in &changes.room_account_data {
865 for (event_type, event) in events {
866 let key = self.encode_key(keys::ROOM_ACCOUNT_DATA, (room, event_type));
867 store.put(&self.serialize_value(&event)?).with_key(key).build()?;
868 }
869 }
870 }
871
872 if !changes.state.is_empty() {
873 let state = tx.object_store(keys::ROOM_STATE)?;
874 let profiles = tx.object_store(keys::PROFILES)?;
875 let user_ids = tx.object_store(keys::USER_IDS)?;
876 let stripped_state = tx.object_store(keys::STRIPPED_ROOM_STATE)?;
877 let stripped_user_ids = tx.object_store(keys::STRIPPED_USER_IDS)?;
878
879 for (room, user_ids) in &changes.profiles_to_delete {
880 for user_id in user_ids {
881 let key = self.encode_key(keys::PROFILES, (room, user_id));
882 profiles.delete(&key).build()?;
883 }
884 }
885
886 for (room, event_types) in &changes.state {
887 let profile_changes = changes.profiles.get(room);
888
889 for (event_type, events) in event_types {
890 for (state_key, raw_event) in events {
891 let key = self.encode_key(keys::ROOM_STATE, (room, event_type, state_key));
892 state
893 .put(&self.serialize_value(&raw_event)?)
894 .with_key(key.clone())
895 .build()?;
896 stripped_state.delete(&key).build()?;
897
898 if *event_type == StateEventType::RoomMember {
899 let event =
900 match raw_event.deserialize_as_unchecked::<SyncRoomMemberEvent>() {
901 Ok(ev) => ev,
902 Err(e) => {
903 let event_id: Option<String> =
904 raw_event.get_field("event_id").ok().flatten();
905 debug!(event_id, "Failed to deserialize member event: {e}");
906 continue;
907 }
908 };
909
910 let key = (room, state_key);
911
912 stripped_user_ids
913 .delete(&self.encode_key(keys::STRIPPED_USER_IDS, key))
914 .build()?;
915
916 user_ids
917 .put(&self.serialize_value(&RoomMember::from(&event))?)
918 .with_key(self.encode_key(keys::USER_IDS, key))
919 .build()?;
920
921 if let Some(profile) =
922 profile_changes.and_then(|p| p.get(event.state_key()))
923 {
924 profiles
925 .put(&self.serialize_value(&profile)?)
926 .with_key(self.encode_key(keys::PROFILES, key))
927 .build()?;
928 }
929 }
930 }
931 }
932 }
933 }
934
935 if !changes.room_infos.is_empty() {
936 let room_infos = tx.object_store(keys::ROOM_INFOS)?;
937 for (room_id, room_info) in &changes.room_infos {
938 room_infos
939 .put(&self.serialize_value(&room_info)?)
940 .with_key(self.encode_key(keys::ROOM_INFOS, room_id))
941 .build()?;
942 }
943 }
944
945 if !changes.presence.is_empty() {
946 let store = tx.object_store(keys::PRESENCE)?;
947 for (sender, event) in &changes.presence {
948 store
949 .put(&self.serialize_value(&event)?)
950 .with_key(self.encode_key(keys::PRESENCE, sender))
951 .build()?;
952 }
953 }
954
955 if !changes.stripped_state.is_empty() {
956 let store = tx.object_store(keys::STRIPPED_ROOM_STATE)?;
957 let user_ids = tx.object_store(keys::STRIPPED_USER_IDS)?;
958
959 for (room, event_types) in &changes.stripped_state {
960 for (event_type, events) in event_types {
961 for (state_key, raw_event) in events {
962 let key = self
963 .encode_key(keys::STRIPPED_ROOM_STATE, (room, event_type, state_key));
964 store.put(&self.serialize_value(&raw_event)?).with_key(key).build()?;
965
966 if *event_type == StateEventType::RoomMember {
967 let event = match raw_event
968 .deserialize_as_unchecked::<StrippedRoomMemberEvent>()
969 {
970 Ok(ev) => ev,
971 Err(e) => {
972 let event_id: Option<String> =
973 raw_event.get_field("event_id").ok().flatten();
974 debug!(
975 event_id,
976 "Failed to deserialize stripped member event: {e}"
977 );
978 continue;
979 }
980 };
981
982 let key = (room, state_key);
983
984 user_ids
985 .put(&self.serialize_value(&RoomMember::from(&event))?)
986 .with_key(self.encode_key(keys::STRIPPED_USER_IDS, key))
987 .build()?;
988 }
989 }
990 }
991 }
992 }
993
994 if !changes.receipts.is_empty() {
995 let room_user_receipts = tx.object_store(keys::ROOM_USER_RECEIPTS)?;
996 let room_event_receipts = tx.object_store(keys::ROOM_EVENT_RECEIPTS)?;
997
998 for (room, content) in &changes.receipts {
999 for (event_id, receipts) in &content.0 {
1000 for (receipt_type, receipts) in receipts {
1001 for (user_id, receipt) in receipts {
1002 let key = match receipt.thread.as_str() {
1003 Some(thread_id) => self.encode_key(
1004 keys::ROOM_USER_RECEIPTS,
1005 (room, receipt_type, thread_id, user_id),
1006 ),
1007 None => self.encode_key(
1008 keys::ROOM_USER_RECEIPTS,
1009 (room, receipt_type, user_id),
1010 ),
1011 };
1012
1013 if let Some((old_event, _)) =
1014 room_user_receipts.get(&key).await?.and_then(|f| {
1015 self.deserialize_value::<(OwnedEventId, Receipt)>(&f).ok()
1016 })
1017 {
1018 let key = match receipt.thread.as_str() {
1019 Some(thread_id) => self.encode_key(
1020 keys::ROOM_EVENT_RECEIPTS,
1021 (room, receipt_type, thread_id, old_event, user_id),
1022 ),
1023 None => self.encode_key(
1024 keys::ROOM_EVENT_RECEIPTS,
1025 (room, receipt_type, old_event, user_id),
1026 ),
1027 };
1028 room_event_receipts.delete(&key).build()?;
1029 }
1030
1031 room_user_receipts
1032 .put(&self.serialize_value(&(event_id, receipt))?)
1033 .with_key(key)
1034 .build()?;
1035
1036 let key = match receipt.thread.as_str() {
1038 Some(thread_id) => self.encode_key(
1039 keys::ROOM_EVENT_RECEIPTS,
1040 (room, receipt_type, thread_id, event_id, user_id),
1041 ),
1042 None => self.encode_key(
1043 keys::ROOM_EVENT_RECEIPTS,
1044 (room, receipt_type, event_id, user_id),
1045 ),
1046 };
1047 room_event_receipts
1048 .put(&self.serialize_value(&(user_id, receipt))?)
1049 .with_key(key)
1050 .build()?;
1051 }
1052 }
1053 }
1054 }
1055 }
1056
1057 if !changes.redactions.is_empty() {
1058 let state = tx.object_store(keys::ROOM_STATE)?;
1059 let room_info = tx.object_store(keys::ROOM_INFOS)?;
1060
1061 for (room_id, redactions) in &changes.redactions {
1062 let range = self.encode_to_range(keys::ROOM_STATE, room_id);
1063 let Some(mut cursor) = state.open_cursor().with_query(&range).await? else {
1064 continue;
1065 };
1066
1067 let mut redaction_rules = None;
1068
1069 while let Some(value) = cursor.next_record().await? {
1070 let Some(key) = cursor.key::<JsValue>()? else {
1071 break;
1072 };
1073
1074 let raw_evt = self.deserialize_value::<Raw<AnySyncStateEvent>>(&value)?;
1075 if let Ok(Some(event_id)) = raw_evt.get_field::<OwnedEventId>("event_id")
1076 && let Some(redaction) = redactions.get(&event_id)
1077 {
1078 let redaction_rules = match &redaction_rules {
1079 Some(r) => r,
1080 None => {
1081 let value = room_info
1082 .get(&self.encode_key(keys::ROOM_INFOS, room_id))
1083 .await?
1084 .and_then(|f| self.deserialize_value::<RoomInfo>(&f).ok())
1085 .map(|info| info.room_version_rules_or_default())
1086 .unwrap_or_else(|| {
1087 warn!(
1088 ?room_id,
1089 "Unable to get the room version rules, \
1090 defaulting to rules for room version \
1091 {ROOM_VERSION_FALLBACK}"
1092 );
1093 ROOM_VERSION_RULES_FALLBACK
1094 })
1095 .redaction;
1096 redaction_rules.get_or_insert(value)
1097 }
1098 };
1099
1100 let redacted = redact(
1101 raw_evt.deserialize_as::<CanonicalJsonObject>()?,
1102 redaction_rules,
1103 Some(RedactedBecause::from_raw_event(redaction)?),
1104 )
1105 .map_err(StoreError::Redaction)?;
1106 state.put(&self.serialize_value(&redacted)?).with_key(key).build()?;
1107 }
1108 }
1109 }
1110 }
1111
1112 if !changes.global_profiles.is_empty() {
1113 let store = tx.object_store(keys::GLOBAL_PROFILES)?;
1114 let unavailable_store = tx.object_store(keys::UNAVAILABLE_GLOBAL_PROFILES)?;
1115 for (user_id, profile_update) in &changes.global_profiles {
1116 let key = self.encode_key(keys::GLOBAL_PROFILES, user_id);
1117 match profile_update {
1118 UserProfileUpdate::Updated(profile_changes) => {
1119 let existing: Option<UserProfile> = store
1120 .get(&key)
1121 .await?
1122 .map(|f| self.deserialize_value(&f))
1123 .transpose()?;
1124
1125 let mut profile = existing.unwrap_or_default();
1126 profile.apply(profile_changes.clone());
1127
1128 unavailable_store.delete(&key).build()?;
1131
1132 store.put(&self.serialize_value(&profile)?).with_key(key).build()?;
1133 }
1134 UserProfileUpdate::Dropped => {
1135 store.delete(&key).build()?;
1136 }
1137 _ => {
1138 warn!(%user_id, "Unhandled UserProfileUpdate variant; ignoring");
1139 }
1140 }
1141 }
1142 }
1143
1144 if !changes.unavailable_global_profiles.is_empty() {
1145 let store = tx.object_store(keys::UNAVAILABLE_GLOBAL_PROFILES)?;
1146 for (user_id, created_at) in &changes.unavailable_global_profiles {
1147 let key = self.encode_key(keys::GLOBAL_PROFILES, user_id);
1150 store.put(&self.serialize_value(created_at)?).with_key(key).build()?;
1151 }
1152 }
1153
1154 tx.commit().await.map_err(|e| e.into())
1155 }
1156
1157 async fn get_presence_event(&self, user_id: &UserId) -> Result<Option<Raw<PresenceEvent>>> {
1158 self.inner
1159 .transaction(keys::PRESENCE)
1160 .with_mode(TransactionMode::Readonly)
1161 .build()?
1162 .object_store(keys::PRESENCE)?
1163 .get(&self.encode_key(keys::PRESENCE, user_id))
1164 .await?
1165 .map(|f| self.deserialize_value(&f))
1166 .transpose()
1167 }
1168
1169 async fn get_presence_events(
1170 &self,
1171 user_ids: &[OwnedUserId],
1172 ) -> Result<Vec<Raw<PresenceEvent>>> {
1173 if user_ids.is_empty() {
1174 return Ok(Vec::new());
1175 }
1176
1177 let txn =
1178 self.inner.transaction(keys::PRESENCE).with_mode(TransactionMode::Readonly).build()?;
1179 let store = txn.object_store(keys::PRESENCE)?;
1180
1181 let mut events = Vec::with_capacity(user_ids.len());
1182
1183 for user_id in user_ids {
1184 if let Some(event) = store
1185 .get(&self.encode_key(keys::PRESENCE, user_id))
1186 .await?
1187 .map(|f| self.deserialize_value(&f))
1188 .transpose()?
1189 {
1190 events.push(event)
1191 }
1192 }
1193
1194 Ok(events)
1195 }
1196
1197 async fn get_state_event(
1198 &self,
1199 room_id: &RoomId,
1200 event_type: StateEventType,
1201 state_key: &str,
1202 ) -> Result<Option<RawAnySyncOrStrippedState>> {
1203 Ok(self
1204 .get_state_events_for_keys(room_id, event_type, &[state_key])
1205 .await?
1206 .into_iter()
1207 .next())
1208 }
1209
1210 async fn get_state_events(
1211 &self,
1212 room_id: &RoomId,
1213 event_type: StateEventType,
1214 ) -> Result<Vec<RawAnySyncOrStrippedState>> {
1215 let stripped_range =
1216 self.encode_to_range(keys::STRIPPED_ROOM_STATE, (room_id, &event_type));
1217 let stripped_events = self
1218 .inner
1219 .transaction(keys::STRIPPED_ROOM_STATE)
1220 .with_mode(TransactionMode::Readonly)
1221 .build()?
1222 .object_store(keys::STRIPPED_ROOM_STATE)?
1223 .get_all()
1224 .with_query(&stripped_range)
1225 .await?
1226 .filter_map(Result::ok)
1227 .filter_map(|f| {
1228 self.deserialize_value(&f).ok().map(RawAnySyncOrStrippedState::Stripped)
1229 })
1230 .collect::<Vec<_>>();
1231
1232 if !stripped_events.is_empty() {
1233 return Ok(stripped_events);
1234 }
1235
1236 let range = self.encode_to_range(keys::ROOM_STATE, (room_id, event_type));
1237 Ok(self
1238 .inner
1239 .transaction(keys::ROOM_STATE)
1240 .with_mode(TransactionMode::Readonly)
1241 .build()?
1242 .object_store(keys::ROOM_STATE)?
1243 .get_all()
1244 .with_query(&range)
1245 .await?
1246 .filter_map(Result::ok)
1247 .filter_map(|f| self.deserialize_value(&f).ok().map(RawAnySyncOrStrippedState::Sync))
1248 .collect::<Vec<_>>())
1249 }
1250
1251 async fn get_state_events_for_keys(
1252 &self,
1253 room_id: &RoomId,
1254 event_type: StateEventType,
1255 state_keys: &[&str],
1256 ) -> Result<Vec<RawAnySyncOrStrippedState>> {
1257 if state_keys.is_empty() {
1258 return Ok(Vec::new());
1259 }
1260
1261 let mut events = Vec::with_capacity(state_keys.len());
1262
1263 {
1264 let txn = self
1265 .inner
1266 .transaction(keys::STRIPPED_ROOM_STATE)
1267 .with_mode(TransactionMode::Readonly)
1268 .build()?;
1269 let store = txn.object_store(keys::STRIPPED_ROOM_STATE)?;
1270
1271 for state_key in state_keys {
1272 if let Some(event) =
1273 store
1274 .get(&self.encode_key(
1275 keys::STRIPPED_ROOM_STATE,
1276 (room_id, &event_type, state_key),
1277 ))
1278 .await?
1279 .map(|f| self.deserialize_value(&f))
1280 .transpose()?
1281 {
1282 events.push(RawAnySyncOrStrippedState::Stripped(event));
1283 }
1284 }
1285
1286 if !events.is_empty() {
1287 return Ok(events);
1288 }
1289 }
1290
1291 let txn = self
1292 .inner
1293 .transaction(keys::ROOM_STATE)
1294 .with_mode(TransactionMode::Readonly)
1295 .build()?;
1296 let store = txn.object_store(keys::ROOM_STATE)?;
1297
1298 for state_key in state_keys {
1299 if let Some(event) = store
1300 .get(&self.encode_key(keys::ROOM_STATE, (room_id, &event_type, state_key)))
1301 .await?
1302 .map(|f| self.deserialize_value(&f))
1303 .transpose()?
1304 {
1305 events.push(RawAnySyncOrStrippedState::Sync(event));
1306 }
1307 }
1308
1309 Ok(events)
1310 }
1311
1312 async fn get_profile(
1313 &self,
1314 room_id: &RoomId,
1315 user_id: &UserId,
1316 ) -> Result<Option<MinimalRoomMemberEvent>> {
1317 self.inner
1318 .transaction(keys::PROFILES)
1319 .with_mode(TransactionMode::Readonly)
1320 .build()?
1321 .object_store(keys::PROFILES)?
1322 .get(&self.encode_key(keys::PROFILES, (room_id, user_id)))
1323 .await?
1324 .map(|f| self.deserialize_value(&f))
1325 .transpose()
1326 }
1327
1328 async fn get_profiles<'a>(
1329 &self,
1330 room_id: &RoomId,
1331 user_ids: &'a [OwnedUserId],
1332 ) -> Result<BTreeMap<&'a UserId, MinimalRoomMemberEvent>> {
1333 if user_ids.is_empty() {
1334 return Ok(BTreeMap::new());
1335 }
1336
1337 let txn =
1338 self.inner.transaction(keys::PROFILES).with_mode(TransactionMode::Readonly).build()?;
1339 let store = txn.object_store(keys::PROFILES)?;
1340
1341 let mut profiles = BTreeMap::new();
1342 for user_id in user_ids {
1343 if let Some(profile) = store
1344 .get(&self.encode_key(keys::PROFILES, (room_id, user_id)))
1345 .await?
1346 .map(|f| self.deserialize_value(&f))
1347 .transpose()?
1348 {
1349 profiles.insert(user_id.as_ref(), profile);
1350 }
1351 }
1352
1353 Ok(profiles)
1354 }
1355
1356 async fn get_room_infos(&self, room_load_settings: &RoomLoadSettings) -> Result<Vec<RoomInfo>> {
1357 let transaction = self
1358 .inner
1359 .transaction(keys::ROOM_INFOS)
1360 .with_mode(TransactionMode::Readonly)
1361 .build()?;
1362
1363 let object_store = transaction.object_store(keys::ROOM_INFOS)?;
1364
1365 Ok(match room_load_settings {
1366 RoomLoadSettings::All => object_store
1367 .get_all()
1368 .await?
1369 .map(|room_info| self.deserialize_value::<RoomInfo>(&room_info?))
1370 .collect::<Result<_>>()?,
1371
1372 RoomLoadSettings::One(room_id) => {
1373 match object_store.get(&self.encode_key(keys::ROOM_INFOS, room_id)).await? {
1374 Some(room_info) => vec![self.deserialize_value::<RoomInfo>(&room_info)?],
1375 None => vec![],
1376 }
1377 }
1378 })
1379 }
1380
1381 async fn get_users_with_display_name(
1382 &self,
1383 room_id: &RoomId,
1384 display_name: &DisplayName,
1385 ) -> Result<BTreeSet<OwnedUserId>> {
1386 self.inner
1387 .transaction(keys::DISPLAY_NAMES)
1388 .with_mode(TransactionMode::Readonly)
1389 .build()?
1390 .object_store(keys::DISPLAY_NAMES)?
1391 .get(&self.encode_key(
1392 keys::DISPLAY_NAMES,
1393 (
1394 room_id,
1395 display_name.as_normalized_str().unwrap_or_else(|| display_name.as_raw_str()),
1396 ),
1397 ))
1398 .await?
1399 .map(|f| self.deserialize_value::<BTreeSet<OwnedUserId>>(&f))
1400 .unwrap_or_else(|| Ok(Default::default()))
1401 }
1402
1403 async fn get_users_with_display_names<'a>(
1404 &self,
1405 room_id: &RoomId,
1406 display_names: &'a [DisplayName],
1407 ) -> Result<HashMap<&'a DisplayName, BTreeSet<OwnedUserId>>> {
1408 let mut map = HashMap::new();
1409
1410 if display_names.is_empty() {
1411 return Ok(map);
1412 }
1413
1414 let txn = self
1415 .inner
1416 .transaction(keys::DISPLAY_NAMES)
1417 .with_mode(TransactionMode::Readonly)
1418 .build()?;
1419 let store = txn.object_store(keys::DISPLAY_NAMES)?;
1420
1421 for display_name in display_names {
1422 if let Some(user_ids) = store
1423 .get(
1424 &self.encode_key(
1425 keys::DISPLAY_NAMES,
1426 (
1427 room_id,
1428 display_name
1429 .as_normalized_str()
1430 .unwrap_or_else(|| display_name.as_raw_str()),
1431 ),
1432 ),
1433 )
1434 .await?
1435 .map(|f| self.deserialize_value::<BTreeSet<OwnedUserId>>(&f))
1436 .transpose()?
1437 {
1438 map.insert(display_name, user_ids);
1439 }
1440 }
1441
1442 Ok(map)
1443 }
1444
1445 async fn get_account_data_event(
1446 &self,
1447 event_type: GlobalAccountDataEventType,
1448 ) -> Result<Option<Raw<AnyGlobalAccountDataEvent>>> {
1449 self.inner
1450 .transaction(keys::ACCOUNT_DATA)
1451 .with_mode(TransactionMode::Readonly)
1452 .build()?
1453 .object_store(keys::ACCOUNT_DATA)?
1454 .get(&self.encode_key(keys::ACCOUNT_DATA, event_type))
1455 .await?
1456 .map(|f| self.deserialize_value(&f))
1457 .transpose()
1458 }
1459
1460 async fn get_room_account_data_event(
1461 &self,
1462 room_id: &RoomId,
1463 event_type: RoomAccountDataEventType,
1464 ) -> Result<Option<Raw<AnyRoomAccountDataEvent>>> {
1465 self.inner
1466 .transaction(keys::ROOM_ACCOUNT_DATA)
1467 .with_mode(TransactionMode::Readonly)
1468 .build()?
1469 .object_store(keys::ROOM_ACCOUNT_DATA)?
1470 .get(&self.encode_key(keys::ROOM_ACCOUNT_DATA, (room_id, event_type)))
1471 .await?
1472 .map(|f| self.deserialize_value(&f))
1473 .transpose()
1474 }
1475
1476 async fn get_user_room_receipt_event(
1477 &self,
1478 room_id: &RoomId,
1479 receipt_type: ReceiptType,
1480 receipt_thread: &ReceiptThread,
1481 user_id: &UserId,
1482 ) -> Result<Option<(OwnedEventId, Receipt)>> {
1483 let key = match receipt_thread.as_str() {
1484 Some(thread_id) => self
1485 .encode_key(keys::ROOM_USER_RECEIPTS, (room_id, receipt_type, thread_id, user_id)),
1486 None => self.encode_key(keys::ROOM_USER_RECEIPTS, (room_id, receipt_type, user_id)),
1487 };
1488 self.inner
1489 .transaction(keys::ROOM_USER_RECEIPTS)
1490 .with_mode(TransactionMode::Readonly)
1491 .build()?
1492 .object_store(keys::ROOM_USER_RECEIPTS)?
1493 .get(&key)
1494 .await?
1495 .map(|f| self.deserialize_value(&f))
1496 .transpose()
1497 }
1498
1499 async fn get_event_room_receipt_events(
1500 &self,
1501 room_id: &RoomId,
1502 receipt_type: ReceiptType,
1503 receipt_thread: &ReceiptThread,
1504 event_id: &EventId,
1505 ) -> Result<Vec<(OwnedUserId, Receipt)>> {
1506 let range = match receipt_thread.as_str() {
1507 Some(thread_id) => self.encode_to_range(
1508 keys::ROOM_EVENT_RECEIPTS,
1509 (room_id, receipt_type, thread_id, event_id),
1510 ),
1511 None => {
1512 self.encode_to_range(keys::ROOM_EVENT_RECEIPTS, (room_id, receipt_type, event_id))
1513 }
1514 };
1515 let tx = self
1516 .inner
1517 .transaction(keys::ROOM_EVENT_RECEIPTS)
1518 .with_mode(TransactionMode::Readonly)
1519 .build()?;
1520 let store = tx.object_store(keys::ROOM_EVENT_RECEIPTS)?;
1521
1522 Ok(store
1523 .get_all()
1524 .with_query(&range)
1525 .await?
1526 .filter_map(Result::ok)
1527 .filter_map(|f| self.deserialize_value(&f).ok())
1528 .collect::<Vec<_>>())
1529 }
1530
1531 async fn get_custom_value(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
1532 let jskey = &JsValue::from_str(core::str::from_utf8(key).map_err(StoreError::Codec)?);
1533 self.get_custom_value_for_js(jskey).await
1534 }
1535
1536 async fn set_custom_value(&self, key: &[u8], value: Vec<u8>) -> Result<Option<Vec<u8>>> {
1537 let jskey = JsValue::from_str(core::str::from_utf8(key).map_err(StoreError::Codec)?);
1538
1539 let prev = self.get_custom_value_for_js(&jskey).await?;
1540
1541 let tx =
1542 self.inner.transaction(keys::CUSTOM).with_mode(TransactionMode::Readwrite).build()?;
1543
1544 tx.object_store(keys::CUSTOM)?
1545 .put(&self.serialize_value(&value)?)
1546 .with_key(jskey)
1547 .build()?;
1548
1549 tx.commit().await.map_err(IndexeddbStateStoreError::from)?;
1550 Ok(prev)
1551 }
1552
1553 async fn remove_custom_value(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
1554 let jskey = JsValue::from_str(core::str::from_utf8(key).map_err(StoreError::Codec)?);
1555
1556 let prev = self.get_custom_value_for_js(&jskey).await?;
1557
1558 let tx =
1559 self.inner.transaction(keys::CUSTOM).with_mode(TransactionMode::Readwrite).build()?;
1560
1561 tx.object_store(keys::CUSTOM)?.delete(&jskey).build()?;
1562
1563 tx.commit().await.map_err(IndexeddbStateStoreError::from)?;
1564 Ok(prev)
1565 }
1566
1567 async fn remove_room(&self, room_id: &RoomId) -> Result<()> {
1568 let direct_stores = [keys::ROOM_INFOS, keys::ROOM_SEND_QUEUE, keys::DEPENDENT_SEND_QUEUE];
1571
1572 let prefixed_stores = [
1575 keys::PROFILES,
1576 keys::DISPLAY_NAMES,
1577 keys::USER_IDS,
1578 keys::ROOM_STATE,
1579 keys::ROOM_ACCOUNT_DATA,
1580 keys::ROOM_EVENT_RECEIPTS,
1581 keys::ROOM_USER_RECEIPTS,
1582 keys::STRIPPED_ROOM_STATE,
1583 keys::STRIPPED_USER_IDS,
1584 keys::THREAD_SUBSCRIPTIONS,
1585 ];
1586
1587 let all_stores = {
1588 let mut v = Vec::new();
1589 v.extend(prefixed_stores);
1590 v.extend(direct_stores);
1591 v
1592 };
1593
1594 let tx =
1595 self.inner.transaction(all_stores).with_mode(TransactionMode::Readwrite).build()?;
1596
1597 for store_name in direct_stores {
1598 tx.object_store(store_name)?.delete(&self.encode_key(store_name, room_id)).build()?;
1599 }
1600
1601 for store_name in prefixed_stores {
1602 let store = tx.object_store(store_name)?;
1603 let range = self.encode_to_range(store_name, room_id);
1604 for key in store.get_all_keys::<JsValue>().with_query(&range).await? {
1605 store.delete(&key?).build()?;
1606 }
1607 }
1608
1609 tx.commit().await.map_err(|e| e.into())
1610 }
1611
1612 async fn get_user_ids(
1613 &self,
1614 room_id: &RoomId,
1615 memberships: RoomMemberships,
1616 ) -> Result<Vec<OwnedUserId>> {
1617 let ids = self.get_user_ids_inner(room_id, memberships, true).await?;
1618 if !ids.is_empty() {
1619 return Ok(ids);
1620 }
1621 self.get_user_ids_inner(room_id, memberships, false).await
1622 }
1623
1624 async fn save_send_queue_request(
1625 &self,
1626 room_id: &RoomId,
1627 transaction_id: OwnedTransactionId,
1628 created_at: MilliSecondsSinceUnixEpoch,
1629 kind: QueuedRequestKind,
1630 priority: usize,
1631 ) -> Result<()> {
1632 let encoded_key = self.encode_key(keys::ROOM_SEND_QUEUE, room_id);
1633
1634 let tx = self
1635 .inner
1636 .transaction(keys::ROOM_SEND_QUEUE)
1637 .with_mode(TransactionMode::Readwrite)
1638 .build()?;
1639
1640 let obj = tx.object_store(keys::ROOM_SEND_QUEUE)?;
1641
1642 let prev = obj.get(&encoded_key).await?;
1647
1648 let mut prev = prev.map_or_else(
1649 || Ok(Vec::new()),
1650 |val| self.deserialize_value::<Vec<PersistedQueuedRequest>>(&val),
1651 )?;
1652
1653 prev.push(PersistedQueuedRequest {
1655 room_id: room_id.to_owned(),
1656 kind: Some(kind),
1657 transaction_id,
1658 error: None,
1659 is_wedged: None,
1660 event: None,
1661 priority: Some(priority),
1662 created_at,
1663 });
1664
1665 obj.put(&self.serialize_value(&prev)?).with_key(encoded_key).build()?;
1667
1668 tx.commit().await?;
1669
1670 Ok(())
1671 }
1672
1673 async fn update_send_queue_request(
1674 &self,
1675 room_id: &RoomId,
1676 transaction_id: &TransactionId,
1677 kind: QueuedRequestKind,
1678 ) -> Result<bool> {
1679 let encoded_key = self.encode_key(keys::ROOM_SEND_QUEUE, room_id);
1680
1681 let tx = self
1682 .inner
1683 .transaction(keys::ROOM_SEND_QUEUE)
1684 .with_mode(TransactionMode::Readwrite)
1685 .build()?;
1686
1687 let obj = tx.object_store(keys::ROOM_SEND_QUEUE)?;
1688
1689 let prev = obj.get(&encoded_key).await?;
1694
1695 let mut prev = prev.map_or_else(
1696 || Ok(Vec::new()),
1697 |val| self.deserialize_value::<Vec<PersistedQueuedRequest>>(&val),
1698 )?;
1699
1700 if let Some(entry) = prev.iter_mut().find(|entry| entry.transaction_id == transaction_id) {
1702 entry.kind = Some(kind);
1703 entry.error = None;
1705 entry.is_wedged = None;
1707 entry.event = None;
1708
1709 obj.put(&self.serialize_value(&prev)?).with_key(encoded_key).build()?;
1711 tx.commit().await?;
1712
1713 Ok(true)
1714 } else {
1715 Ok(false)
1716 }
1717 }
1718
1719 async fn remove_send_queue_request(
1720 &self,
1721 room_id: &RoomId,
1722 transaction_id: &TransactionId,
1723 ) -> Result<bool> {
1724 let encoded_key = self.encode_key(keys::ROOM_SEND_QUEUE, room_id);
1725
1726 let tx = self
1727 .inner
1728 .transaction([keys::ROOM_SEND_QUEUE, keys::DEPENDENT_SEND_QUEUE])
1729 .with_mode(TransactionMode::Readwrite)
1730 .build()?;
1731
1732 let obj = tx.object_store(keys::ROOM_SEND_QUEUE)?;
1733
1734 if let Some(val) = obj.get(&encoded_key).await? {
1739 let mut prev = self.deserialize_value::<Vec<PersistedQueuedRequest>>(&val)?;
1740 if let Some(pos) = prev.iter().position(|item| item.transaction_id == transaction_id) {
1741 prev.remove(pos);
1742
1743 if prev.is_empty() {
1744 obj.delete(&encoded_key).build()?;
1745 } else {
1746 obj.put(&self.serialize_value(&prev)?).with_key(encoded_key).build()?;
1747 }
1748
1749 tx.commit().await?;
1750 return Ok(true);
1751 }
1752 }
1753
1754 Ok(false)
1755 }
1756
1757 async fn load_send_queue_requests(&self, room_id: &RoomId) -> Result<Vec<QueuedRequest>> {
1758 let encoded_key = self.encode_key(keys::ROOM_SEND_QUEUE, room_id);
1759
1760 let prev = self
1763 .inner
1764 .transaction(keys::ROOM_SEND_QUEUE)
1765 .with_mode(TransactionMode::Readwrite)
1766 .build()?
1767 .object_store(keys::ROOM_SEND_QUEUE)?
1768 .get(&encoded_key)
1769 .await?;
1770
1771 let mut prev = prev.map_or_else(
1772 || Ok(Vec::new()),
1773 |val| self.deserialize_value::<Vec<PersistedQueuedRequest>>(&val),
1774 )?;
1775
1776 prev.sort_by_key(|item| Reverse(item.priority.unwrap_or(0)));
1778
1779 Ok(prev.into_iter().filter_map(PersistedQueuedRequest::into_queued_request).collect())
1780 }
1781
1782 async fn update_send_queue_request_status(
1783 &self,
1784 room_id: &RoomId,
1785 transaction_id: &TransactionId,
1786 error: Option<QueueWedgeError>,
1787 ) -> Result<()> {
1788 let encoded_key = self.encode_key(keys::ROOM_SEND_QUEUE, room_id);
1789
1790 let tx = self
1791 .inner
1792 .transaction(keys::ROOM_SEND_QUEUE)
1793 .with_mode(TransactionMode::Readwrite)
1794 .build()?;
1795
1796 let obj = tx.object_store(keys::ROOM_SEND_QUEUE)?;
1797
1798 if let Some(val) = obj.get(&encoded_key).await? {
1799 let mut prev = self.deserialize_value::<Vec<PersistedQueuedRequest>>(&val)?;
1800 if let Some(request) =
1801 prev.iter_mut().find(|item| item.transaction_id == transaction_id)
1802 {
1803 request.is_wedged = None;
1804 request.error = error;
1805 obj.put(&self.serialize_value(&prev)?).with_key(encoded_key).build()?;
1806 }
1807 }
1808
1809 tx.commit().await?;
1810
1811 Ok(())
1812 }
1813
1814 async fn load_rooms_with_unsent_requests(&self) -> Result<Vec<OwnedRoomId>> {
1815 let tx = self
1816 .inner
1817 .transaction(keys::ROOM_SEND_QUEUE)
1818 .with_mode(TransactionMode::Readwrite)
1819 .build()?;
1820
1821 let obj = tx.object_store(keys::ROOM_SEND_QUEUE)?;
1822
1823 let all_entries = obj
1824 .get_all()
1825 .await?
1826 .map(|item| self.deserialize_value::<Vec<PersistedQueuedRequest>>(&item?))
1827 .collect::<Result<Vec<Vec<PersistedQueuedRequest>>, _>>()?
1828 .into_iter()
1829 .flat_map(|vec| vec.into_iter().map(|item| item.room_id))
1830 .collect::<BTreeSet<_>>();
1831
1832 Ok(all_entries.into_iter().collect())
1833 }
1834
1835 async fn save_dependent_queued_request(
1836 &self,
1837 room_id: &RoomId,
1838 parent_txn_id: &TransactionId,
1839 own_txn_id: ChildTransactionId,
1840 created_at: MilliSecondsSinceUnixEpoch,
1841 content: DependentQueuedRequestKind,
1842 ) -> Result<()> {
1843 let encoded_key = self.encode_key(keys::DEPENDENT_SEND_QUEUE, room_id);
1844
1845 let tx = self
1846 .inner
1847 .transaction(keys::DEPENDENT_SEND_QUEUE)
1848 .with_mode(TransactionMode::Readwrite)
1849 .build()?;
1850
1851 let obj = tx.object_store(keys::DEPENDENT_SEND_QUEUE)?;
1852
1853 let prev = obj.get(&encoded_key).await?;
1856
1857 let mut prev = prev.map_or_else(
1858 || Ok(Vec::new()),
1859 |val| self.deserialize_value::<Vec<DependentQueuedRequest>>(&val),
1860 )?;
1861
1862 prev.push(DependentQueuedRequest {
1864 kind: content,
1865 parent_transaction_id: parent_txn_id.to_owned(),
1866 own_transaction_id: own_txn_id,
1867 parent_key: None,
1868 created_at,
1869 });
1870
1871 obj.put(&self.serialize_value(&prev)?).with_key(encoded_key).build()?;
1873
1874 tx.commit().await?;
1875
1876 Ok(())
1877 }
1878
1879 async fn update_dependent_queued_request(
1880 &self,
1881 room_id: &RoomId,
1882 own_transaction_id: &ChildTransactionId,
1883 new_content: DependentQueuedRequestKind,
1884 ) -> Result<bool> {
1885 let encoded_key = self.encode_key(keys::DEPENDENT_SEND_QUEUE, room_id);
1886
1887 let tx = self
1888 .inner
1889 .transaction(keys::DEPENDENT_SEND_QUEUE)
1890 .with_mode(TransactionMode::Readwrite)
1891 .build()?;
1892
1893 let obj = tx.object_store(keys::DEPENDENT_SEND_QUEUE)?;
1894
1895 let prev = obj.get(&encoded_key).await?;
1898
1899 let mut prev = prev.map_or_else(
1900 || Ok(Vec::new()),
1901 |val| self.deserialize_value::<Vec<DependentQueuedRequest>>(&val),
1902 )?;
1903
1904 let mut found = false;
1906 for entry in prev.iter_mut() {
1907 if entry.own_transaction_id == *own_transaction_id {
1908 found = true;
1909 entry.kind = new_content;
1910 break;
1911 }
1912 }
1913
1914 if found {
1915 obj.put(&self.serialize_value(&prev)?).with_key(encoded_key).build()?;
1916 tx.commit().await?;
1917 }
1918
1919 Ok(found)
1920 }
1921
1922 async fn mark_dependent_queued_requests_as_ready(
1923 &self,
1924 room_id: &RoomId,
1925 parent_txn_id: &TransactionId,
1926 parent_key: SentRequestKey,
1927 ) -> Result<usize> {
1928 let encoded_key = self.encode_key(keys::DEPENDENT_SEND_QUEUE, room_id);
1929
1930 let tx = self
1931 .inner
1932 .transaction(keys::DEPENDENT_SEND_QUEUE)
1933 .with_mode(TransactionMode::Readwrite)
1934 .build()?;
1935
1936 let obj = tx.object_store(keys::DEPENDENT_SEND_QUEUE)?;
1937
1938 let prev = obj.get(&encoded_key).await?;
1941
1942 let mut prev = prev.map_or_else(
1943 || Ok(Vec::new()),
1944 |val| self.deserialize_value::<Vec<DependentQueuedRequest>>(&val),
1945 )?;
1946
1947 let mut num_updated = 0;
1949 for entry in prev.iter_mut().filter(|entry| entry.parent_transaction_id == parent_txn_id) {
1950 entry.parent_key = Some(parent_key.clone());
1951 num_updated += 1;
1952 }
1953
1954 if num_updated > 0 {
1955 obj.put(&self.serialize_value(&prev)?).with_key(encoded_key).build()?;
1956 tx.commit().await?;
1957 }
1958
1959 Ok(num_updated)
1960 }
1961
1962 async fn remove_dependent_queued_request(
1963 &self,
1964 room_id: &RoomId,
1965 txn_id: &ChildTransactionId,
1966 ) -> Result<bool> {
1967 let encoded_key = self.encode_key(keys::DEPENDENT_SEND_QUEUE, room_id);
1968
1969 let tx = self
1970 .inner
1971 .transaction(keys::DEPENDENT_SEND_QUEUE)
1972 .with_mode(TransactionMode::Readwrite)
1973 .build()?;
1974
1975 let obj = tx.object_store(keys::DEPENDENT_SEND_QUEUE)?;
1976
1977 if let Some(val) = obj.get(&encoded_key).await? {
1980 let mut prev = self.deserialize_value::<Vec<DependentQueuedRequest>>(&val)?;
1981 if let Some(pos) = prev.iter().position(|item| item.own_transaction_id == *txn_id) {
1982 prev.remove(pos);
1983
1984 if prev.is_empty() {
1985 obj.delete(&encoded_key).build()?;
1986 } else {
1987 obj.put(&self.serialize_value(&prev)?).with_key(encoded_key).build()?;
1988 }
1989
1990 tx.commit().await?;
1991 return Ok(true);
1992 }
1993 }
1994
1995 Ok(false)
1996 }
1997
1998 async fn load_dependent_queued_requests(
1999 &self,
2000 room_id: &RoomId,
2001 ) -> Result<Vec<DependentQueuedRequest>> {
2002 let encoded_key = self.encode_key(keys::DEPENDENT_SEND_QUEUE, room_id);
2003
2004 let prev = self
2006 .inner
2007 .transaction(keys::DEPENDENT_SEND_QUEUE)
2008 .with_mode(TransactionMode::Readwrite)
2009 .build()?
2010 .object_store(keys::DEPENDENT_SEND_QUEUE)?
2011 .get(&encoded_key)
2012 .await?;
2013
2014 prev.map_or_else(
2015 || Ok(Vec::new()),
2016 |val| self.deserialize_value::<Vec<DependentQueuedRequest>>(&val),
2017 )
2018 }
2019
2020 async fn upsert_thread_subscriptions(
2021 &self,
2022 updates: Vec<(&RoomId, &EventId, StoredThreadSubscription)>,
2023 ) -> Result<()> {
2024 let tx = self
2025 .inner
2026 .transaction(keys::THREAD_SUBSCRIPTIONS)
2027 .with_mode(TransactionMode::Readwrite)
2028 .build()?;
2029 let obj = tx.object_store(keys::THREAD_SUBSCRIPTIONS)?;
2030
2031 for (room_id, thread_id, subscription) in updates {
2032 let encoded_key = self.encode_key(keys::THREAD_SUBSCRIPTIONS, (room_id, thread_id));
2033 let mut new = PersistedThreadSubscription::from(subscription);
2034
2035 if let Some(previous_value) = obj.get(&encoded_key).await? {
2037 let previous: PersistedThreadSubscription =
2038 self.deserialize_value(&previous_value)?;
2039
2040 if new == previous {
2043 continue;
2044 }
2045 if !compare_thread_subscription_bump_stamps(
2046 previous.bump_stamp,
2047 &mut new.bump_stamp,
2048 ) {
2049 continue;
2050 }
2051 }
2052
2053 let serialized_value = self.serialize_value(&new);
2054 obj.put(&serialized_value?).with_key(encoded_key).build()?;
2055 }
2056
2057 tx.commit().await?;
2058
2059 Ok(())
2060 }
2061
2062 async fn load_thread_subscription(
2063 &self,
2064 room: &RoomId,
2065 thread_id: &EventId,
2066 ) -> Result<Option<StoredThreadSubscription>> {
2067 let encoded_key = self.encode_key(keys::THREAD_SUBSCRIPTIONS, (room, thread_id));
2068
2069 let js_value = self
2070 .inner
2071 .transaction(keys::THREAD_SUBSCRIPTIONS)
2072 .with_mode(TransactionMode::Readonly)
2073 .build()?
2074 .object_store(keys::THREAD_SUBSCRIPTIONS)?
2075 .get(&encoded_key)
2076 .await?;
2077
2078 let Some(js_value) = js_value else {
2079 return Ok(None);
2081 };
2082
2083 let sub: PersistedThreadSubscription = self.deserialize_value(&js_value)?;
2084
2085 let status = ThreadSubscriptionStatus::from_str(&sub.status).map_err(|_| {
2086 StoreError::InvalidData {
2087 details: format!(
2088 "invalid thread status for room {room} and thread {thread_id}: {}",
2089 sub.status
2090 ),
2091 }
2092 })?;
2093
2094 Ok(Some(StoredThreadSubscription { status, bump_stamp: sub.bump_stamp }))
2095 }
2096
2097 async fn remove_thread_subscription(&self, room: &RoomId, thread_id: &EventId) -> Result<()> {
2098 let encoded_key = self.encode_key(keys::THREAD_SUBSCRIPTIONS, (room, thread_id));
2099
2100 let transaction = self
2101 .inner
2102 .transaction(keys::THREAD_SUBSCRIPTIONS)
2103 .with_mode(TransactionMode::Readwrite)
2104 .build()?;
2105 transaction.object_store(keys::THREAD_SUBSCRIPTIONS)?.delete(&encoded_key).await?;
2106 transaction.commit().await?;
2107
2108 Ok(())
2109 }
2110
2111 async fn get_global_profile(&self, user_id: &UserId) -> Result<Option<UserProfile>> {
2112 let transaction = self
2113 .inner
2114 .transaction(keys::GLOBAL_PROFILES)
2115 .with_mode(TransactionMode::Readonly)
2116 .build()?;
2117 let store = transaction.object_store(keys::GLOBAL_PROFILES)?;
2118 let key = self.encode_key(keys::GLOBAL_PROFILES, user_id);
2119
2120 store.get(&key).await?.map(|f| self.deserialize_value(&f)).transpose()
2121 }
2122
2123 async fn get_global_profiles<'a>(
2124 &self,
2125 user_ids: &'a [OwnedUserId],
2126 ) -> Result<BTreeMap<&'a UserId, UserProfile>> {
2127 let transaction = self
2128 .inner
2129 .transaction(keys::GLOBAL_PROFILES)
2130 .with_mode(TransactionMode::Readonly)
2131 .build()?;
2132 let store = transaction.object_store(keys::GLOBAL_PROFILES)?;
2133
2134 let mut profiles = BTreeMap::new();
2135 for user_id in user_ids {
2136 let key = self.encode_key(keys::GLOBAL_PROFILES, user_id);
2137 if let Some(value) = store.get(&key).await? {
2138 profiles.insert(user_id.as_ref(), self.deserialize_value(&value)?);
2139 }
2140 }
2141
2142 Ok(profiles)
2143 }
2144
2145 async fn get_missing_global_profiles<'a>(
2146 &self,
2147 user_ids: &'a [OwnedUserId],
2148 ) -> Result<Vec<&'a UserId>> {
2149 let transaction = self
2150 .inner
2151 .transaction([keys::GLOBAL_PROFILES, keys::UNAVAILABLE_GLOBAL_PROFILES])
2152 .with_mode(TransactionMode::Readonly)
2153 .build()?;
2154 let profiles = transaction.object_store(keys::GLOBAL_PROFILES)?;
2155 let unavailable = transaction.object_store(keys::UNAVAILABLE_GLOBAL_PROFILES)?;
2156
2157 let mut missing = Vec::new();
2159 for user_id in user_ids {
2160 let key = self.encode_key(keys::GLOBAL_PROFILES, user_id);
2162 if profiles.count().with_query(&key).await? == 0
2163 && unavailable.count().with_query(&key).await? == 0
2164 {
2165 missing.push(user_id.as_ref());
2166 }
2167 }
2168
2169 Ok(missing)
2170 }
2171
2172 #[allow(clippy::unused_async)]
2173 async fn optimize(&self) -> Result<()> {
2174 Ok(())
2175 }
2176
2177 #[allow(clippy::unused_async)]
2178 async fn get_size(&self) -> Result<Option<usize>> {
2179 Ok(None)
2180 }
2181
2182 #[allow(clippy::unused_async)]
2183 async fn close(&self) -> Result<()> {
2184 Ok(())
2185 }
2186
2187 #[allow(clippy::unused_async)]
2188 async fn reopen(&self) -> Result<()> {
2189 Ok(())
2190 }
2191});
2192
2193#[derive(Debug, Serialize, Deserialize)]
2195struct RoomMember {
2196 user_id: OwnedUserId,
2197 membership: MembershipState,
2198}
2199
2200impl From<&SyncStateEvent<RoomMemberEventContent>> for RoomMember {
2201 fn from(event: &SyncStateEvent<RoomMemberEventContent>) -> Self {
2202 Self { user_id: event.state_key().clone(), membership: event.membership().clone() }
2203 }
2204}
2205
2206impl From<&StrippedRoomMemberEvent> for RoomMember {
2207 fn from(event: &StrippedRoomMemberEvent) -> Self {
2208 Self { user_id: event.state_key.clone(), membership: event.content.membership.clone() }
2209 }
2210}
2211
2212#[cfg(test)]
2213mod migration_tests {
2214 use std::assert_matches;
2215
2216 use matrix_sdk_base::store::{QueuedRequestKind, SerializableEventContent};
2217 use ruma::{
2218 OwnedRoomId, OwnedTransactionId, TransactionId,
2219 events::room::message::RoomMessageEventContent, room_id,
2220 };
2221 use serde::{Deserialize, Serialize};
2222
2223 use crate::state_store::PersistedQueuedRequest;
2224
2225 #[derive(Serialize, Deserialize)]
2226 struct OldPersistedQueuedRequest {
2227 room_id: OwnedRoomId,
2228 event: SerializableEventContent,
2229 transaction_id: OwnedTransactionId,
2230 is_wedged: bool,
2231 }
2232
2233 #[test]
2237 fn test_migrating_persisted_queue_event_serialization() {
2238 let room_a_id = room_id!("!room_a:dummy.local");
2239 let transaction_id = TransactionId::new();
2240 let content =
2241 SerializableEventContent::new(&RoomMessageEventContent::text_plain("Hello").into())
2242 .unwrap();
2243
2244 let old_persisted_queue_event = OldPersistedQueuedRequest {
2245 room_id: room_a_id.to_owned(),
2246 event: content,
2247 transaction_id: transaction_id.clone(),
2248 is_wedged: true,
2249 };
2250
2251 let serialized_persisted = serde_json::to_vec(&old_persisted_queue_event).unwrap();
2252
2253 let new_persisted: PersistedQueuedRequest =
2255 serde_json::from_slice(&serialized_persisted).unwrap();
2256
2257 assert_eq!(new_persisted.is_wedged, Some(true));
2258 assert!(new_persisted.error.is_none());
2259
2260 assert!(new_persisted.event.is_some());
2261 assert!(new_persisted.kind.is_none());
2262
2263 let queued = new_persisted.into_queued_request().unwrap();
2264 assert_matches!(queued.kind, QueuedRequestKind::Event { .. });
2265 assert_eq!(queued.transaction_id, transaction_id);
2266 assert!(queued.error.is_some());
2267 }
2268}
2269
2270#[cfg(all(test, target_family = "wasm"))]
2271mod tests {
2272 #[cfg(target_family = "wasm")]
2273 wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_browser);
2274
2275 use matrix_sdk_base::statestore_integration_tests;
2276 use uuid::Uuid;
2277
2278 use super::{IndexeddbStateStore, Result};
2279
2280 async fn get_store() -> Result<IndexeddbStateStore> {
2281 let db_name = format!("test-state-plain-{}", Uuid::new_v4().as_hyphenated());
2282 Ok(IndexeddbStateStore::builder().name(db_name).build().await?)
2283 }
2284
2285 statestore_integration_tests!();
2286}
2287
2288#[cfg(all(test, target_family = "wasm"))]
2289mod encrypted_tests {
2290 #[cfg(target_family = "wasm")]
2291 wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_browser);
2292
2293 use matrix_sdk_base::statestore_integration_tests;
2294 use uuid::Uuid;
2295
2296 use super::{IndexeddbStateStore, Result};
2297
2298 async fn get_store() -> Result<IndexeddbStateStore> {
2299 let db_name = format!("test-state-encrypted-{}", Uuid::new_v4().as_hyphenated());
2300 let passphrase = format!("some_passphrase-{}", Uuid::new_v4().as_hyphenated());
2301 Ok(IndexeddbStateStore::builder().name(db_name).passphrase(passphrase).build().await?)
2302 }
2303
2304 statestore_integration_tests!();
2305}