Skip to main content

matrix_sdk_indexeddb/state_store/
mod.rs

1// Copyright 2021 The Matrix.org Foundation C.I.C.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use 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    /// Room profiles.
166    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    /// Table used to save send queue events.
175    pub const ROOM_SEND_QUEUE: &str = "room_send_queue";
176    /// Table used to save dependent send queue events.
177    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    /// All names of the current state stores for convenience.
193    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    // static keys
216
217    pub const STORE_KEY: &str = "store_key";
218}
219
220pub use keys::ALL_STORES;
221use matrix_sdk_base::store::QueueWedgeError;
222
223/// Encrypt (if needs be) then JSON-serialize a value.
224fn 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
234/// Deserialize a JSON value and then decrypt it (if needs be).
235fn 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/// Builder for [`IndexeddbStateStore`].
277#[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    /// Set the name for the indexeddb store to use, `state` is none given.
294    pub fn name(mut self, value: String) -> Self {
295        self.name = Some(value);
296        self
297    }
298
299    /// Set the password the indexeddb should be encrypted with.
300    ///
301    /// If not given, the DB is not encrypted.
302    pub fn passphrase(mut self, value: String) -> Self {
303        self.passphrase = Some(value);
304        self
305    }
306
307    /// The strategy to use when a merge conflict is found.
308    ///
309    /// See [`MigrationConflictStrategy`] for details.
310    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    /// Generate a IndexeddbStateStoreBuilder with default parameters
347    pub fn builder() -> IndexeddbStateStoreBuilder {
348        IndexeddbStateStoreBuilder::new()
349    }
350
351    /// The version of the database containing the data.
352    pub fn version(&self) -> u32 {
353        self.inner.version() as u32
354    }
355
356    /// The version of the database containing the metadata.
357    pub fn meta_version(&self) -> u32 {
358        self.meta.version() as u32
359    }
360
361    /// Whether this database has any migration backups
362    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    /// What's the database name of the latest backup<
375    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    /// Encrypt (if needs be) then JSON-serialize a value.
394    fn serialize_value(&self, event: &impl Serialize) -> Result<JsValue> {
395        serialize_value(self.store_cipher.as_deref(), event)
396    }
397
398    /// Deserialize a JSON value and then decrypt it (if needs be).
399    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    /// Get user IDs for the given room with the given memberships and stripped
418    /// state.
419    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            // It should be faster to just get all user IDs in this case.
433            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        // Use the key (prefix) for the table name as well, to keep encoded keys
474        // compatible for the sync token and filters, which were in separate
475        // tables initially.
476        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/// A superset of [`QueuedRequest`] that also contains the room id, since we
526/// want to return them.
527#[derive(Serialize, Deserialize)]
528struct PersistedQueuedRequest {
529    /// In which room is this event going to be sent.
530    pub room_id: OwnedRoomId,
531
532    // All these fields are the same as in [`QueuedRequest`].
533    /// Kind. Optional because it might be missing from previous formats.
534    kind: Option<QueuedRequestKind>,
535    transaction_id: OwnedTransactionId,
536
537    pub error: Option<QueueWedgeError>,
538
539    priority: Option<usize>,
540
541    /// The time the original message was first attempted to be sent at.
542    #[serde(default = "created_now")]
543    created_at: MilliSecondsSinceUnixEpoch,
544
545    // Migrated fields: keep these private, they're not used anymore elsewhere
546    // in the code base.
547    /// Deprecated (from old format), now replaced with error field.
548    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                // Migrate to a generic error.
564                Some(QueueWedgeError::GenericApiError {
565                    msg: "local echo failed to send in a previous session".into(),
566                })
567            }
568            _ => self.error,
569        };
570
571        // By default, events without a priority have a priority of 0.
572        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// Small hack to have the following macro invocation act as the appropriate
597// trait impl block on wasm, but still be compiled on non-wasm as a regular impl
598// block otherwise.
599//
600// The trait impl doesn't compile on non-wasm due to unfulfilled trait bounds,
601// this hack allows us to still have most of rust-analyzer's IDE functionality
602// within the impl block without having to set it up to check things against the
603// wasm target (which would disable many other parts of the codebase).
604#[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            // nothing to do, quit early
820            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                            // Add the receipt to the room event receipts
1037                            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                        // If the profile was marked as unavailable we can
1129                        // unmark it. Both stores use the same key for a user.
1130                        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                // Use the same key as `GLOBAL_PROFILES`, so that a user can be
1148                // looked up in both stores with a single encoding.
1149                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        // All the stores which use a RoomId as their key (and nothing
1569        // additional).
1570        let direct_stores = [keys::ROOM_INFOS, keys::ROOM_SEND_QUEUE, keys::DEPENDENT_SEND_QUEUE];
1571
1572        // All the stores which use a RoomId as the first part of their key, but
1573        // may have some additional data in the key.
1574        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        // We store an encoded vector of the queued requests, with their
1643        // transaction ids.
1644
1645        // Reload the previous vector for this room, or create an empty one.
1646        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        // Push the new request.
1654        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        // Save the new vector into db.
1666        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        // We store an encoded vector of the queued requests, with their
1690        // transaction ids.
1691
1692        // Reload the previous vector for this room, or create an empty one.
1693        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        // Modify the one request.
1701        if let Some(entry) = prev.iter_mut().find(|entry| entry.transaction_id == transaction_id) {
1702            entry.kind = Some(kind);
1703            // Reset the error state.
1704            entry.error = None;
1705            // Remove migrated fields.
1706            entry.is_wedged = None;
1707            entry.event = None;
1708
1709            // Save the new vector into db.
1710            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        // We store an encoded vector of the queued requests, with their
1735        // transaction ids.
1736
1737        // Reload the previous vector for this room.
1738        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        // We store an encoded vector of the queued requests, with their
1761        // transaction ids.
1762        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        // Inverted stable ordering on priority.
1777        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        // We store an encoded vector of the dependent requests. Reload the
1854        // previous vector for this room, or create an empty one.
1855        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        // Push the new request.
1863        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        // Save the new vector into db.
1872        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        // We store an encoded vector of the dependent requests. Reload the
1896        // previous vector for this room, or create an empty one.
1897        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        // Modify the dependent request, if found.
1905        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        // We store an encoded vector of the dependent requests. Reload the
1939        // previous vector for this room, or create an empty one.
1940        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        // Modify all requests that match.
1948        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        // We store an encoded vector of the dependent requests. Reload the
1978        // previous vector for this room.
1979        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        // We store an encoded vector of the dependent requests.
2005        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            // See if there's a previous subscription.
2036            if let Some(previous_value) = obj.get(&encoded_key).await? {
2037                let previous: PersistedThreadSubscription =
2038                    self.deserialize_value(&previous_value)?;
2039
2040                // If the previous status is the same as the new one, don't do
2041                // anything.
2042                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            // We didn't have a previous subscription for this thread.
2080            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        // Only query user IDs, don't load any profiles.
2158        let mut missing = Vec::new();
2159        for user_id in user_ids {
2160            // A user's key is the same in both stores, see `save_changes`.
2161            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/// A room member.
2194#[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    // We now persist an error when an event failed to send instead of just a
2234    // boolean. To support that, `PersistedQueueEvent` changed a bool to
2235    // Option<bool>, ensures that this work properly.
2236    #[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        // Load it with the new version.
2254        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}