state store: change schema for send_queue_events table

It turns out that so as to be able to read the room ids, they need to be
*values*, not only *keys* (since keys are one-way hashed). This means we
need to duplicate the room_id field in indexeddb/sqlite, so each entry
contains both the room_id as a key (for queries) and as a value (to
return it).

Since there's no meaningful migration we can apply, the way to go is to
drop the pending events table and recreate it from the ground up. It is
assumed that no one has used the store on indexeddb; otherwise the
workaround would be to drop and recreate it.

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

@@ -1385,6 +1385,38 @@ impl StateStoreIntegrationTests for DynStateStore {
         for i in 0..3 {
             assert_ne!(pending[i].transaction_id, txn0);
         }
+
+        // Now add one event for two other rooms, remove one of the events, and then
+        // query all the rooms which have outstanding unsent events.
+
+        // Add one event for room2.
+        let room_id2 = room_id!("!test_send_queue_two:localhost");
+        {
+            let txn = TransactionId::new();
+            let event =
+                SerializableEventContent::new(RoomMessageEventContent::text_plain("room2").into())
+                    .unwrap();
+            self.save_send_queue_event(room_id2, txn.clone(), event).await.unwrap();
+        }
+
+        // Add and remove one event for room3.
+        {
+            let room_id3 = room_id!("!test_send_queue_three:localhost");
+            let txn = TransactionId::new();
+            let event =
+                SerializableEventContent::new(RoomMessageEventContent::text_plain("room3").into())
+                    .unwrap();
+            self.save_send_queue_event(room_id3, txn.clone(), event).await.unwrap();
+
+            self.remove_send_queue_event(room_id3, &txn).await.unwrap();
+        }
+
+        // Query all the rooms which have unsent events. Per the previous steps,
+        // it should be room1 and room2, not room3.
+        let outstanding_rooms = self.load_rooms_with_unsent_events().await.unwrap();
+        assert_eq!(outstanding_rooms.len(), 2);
+        assert!(outstanding_rooms.iter().any(|room| room == room_id));
+        assert!(outstanding_rooms.iter().any(|room| room == room_id2));
     }
 }
 

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

@@ -890,12 +890,20 @@ impl StateStore for MemoryStore {
         room_id: &RoomId,
         transaction_id: &TransactionId,
     ) -> Result<(), Self::Error> {
-        self.send_queue_events
-            .write()
-            .unwrap()
-            .entry(room_id.to_owned())
-            .or_default()
-            .retain(|item| item.transaction_id != transaction_id);
+        let mut q = self.send_queue_events.write().unwrap();
+
+        let entry = q.get_mut(room_id);
+        if let Some(entry) = entry {
+            // Find the event by id in its room queue, and remove it if present.
+            if let Some(pos) = entry.iter().position(|item| item.transaction_id == transaction_id) {
+                entry.remove(pos);
+                // And if this was the last event before removal, remove the entire room entry.
+                if entry.is_empty() {
+                    q.remove(room_id);
+                }
+            }
+        }
+
         Ok(())
     }
 
@@ -925,6 +933,10 @@ impl StateStore for MemoryStore {
         }
         Ok(())
     }
+
+    async fn load_rooms_with_unsent_events(&self) -> Result<Vec<OwnedRoomId>, Self::Error> {
+        Ok(self.send_queue_events.read().unwrap().keys().cloned().collect())
+    }
 }
 
 #[cfg(test)]

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

@@ -34,7 +34,8 @@ use ruma::{
         StateEventType, StaticEventContent, StaticStateEventContent,
     },
     serde::Raw,
-    EventId, MxcUri, OwnedEventId, OwnedTransactionId, OwnedUserId, RoomId, TransactionId, UserId,
+    EventId, MxcUri, OwnedEventId, OwnedRoomId, OwnedTransactionId, OwnedUserId, RoomId,
+    TransactionId, UserId,
 };
 use serde::{Deserialize, Serialize};
 
@@ -429,6 +430,9 @@ pub trait StateStore: AsyncTraitDeps {
         transaction_id: &TransactionId,
         wedged: bool,
     ) -> Result<(), Self::Error>;
+
+    /// Loads all the rooms which have any pending events in their send queue.
+    async fn load_rooms_with_unsent_events(&self) -> Result<Vec<OwnedRoomId>, Self::Error>;
 }
 
 #[repr(transparent)]
@@ -688,6 +692,10 @@ impl<T: StateStore> StateStore for EraseStateStoreError<T> {
             .await
             .map_err(Into::into)
     }
+
+    async fn load_rooms_with_unsent_events(&self) -> Result<Vec<OwnedRoomId>, Self::Error> {
+        self.0.load_rooms_with_unsent_events().await.map_err(Into::into)
+    }
 }
 
 /// Convenience functionality for state stores.
@@ -1053,7 +1061,7 @@ impl SerializableEventContent {
 }
 
 /// An event to be sent with a send queue.
-#[derive(Clone, Serialize, Deserialize)]
+#[derive(Clone)]
 pub struct QueuedEvent {
     /// The content of the message-like event we'd like to send.
     pub event: SerializableEventContent,

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

@@ -44,8 +44,8 @@ use ruma::{
         GlobalAccountDataEventType, RoomAccountDataEventType, StateEventType, SyncStateEvent,
     },
     serde::Raw,
-    CanonicalJsonObject, EventId, MxcUri, OwnedEventId, OwnedTransactionId, OwnedUserId, RoomId,
-    RoomVersionId, TransactionId, UserId,
+    CanonicalJsonObject, EventId, MxcUri, OwnedEventId, OwnedRoomId, OwnedTransactionId,
+    OwnedUserId, RoomId, RoomVersionId, TransactionId, UserId,
 };
 use serde::{de::DeserializeOwned, Deserialize, Serialize};
 use tracing::{debug, warn};
@@ -408,6 +408,19 @@ impl IndexeddbStateStore {
     }
 }
 
+/// A superset of [`QueuedEvent`] that also contains the room id, since we want
+/// to return them.
+#[derive(Serialize, Deserialize)]
+struct PersistedQueuedEvent {
+    /// In which room is this event going to be sent.
+    pub room_id: OwnedRoomId,
+
+    // All these fields are the same as in [`QueuedEvent`].
+    event: SerializableEventContent,
+    transaction_id: OwnedTransactionId,
+    is_wedged: bool,
+}
+
 // Small hack to have the following macro invocation act as the appropriate
 // trait impl block on wasm, but still be compiled on non-wasm as a regular
 // impl block otherwise.
@@ -1363,11 +1376,16 @@ impl_state_store!({
 
         let mut prev = prev.map_or_else(
             || Ok(Vec::new()),
-            |val| self.deserialize_value::<Vec<QueuedEvent>>(&val),
+            |val| self.deserialize_value::<Vec<PersistedQueuedEvent>>(&val),
         )?;
 
         // Push the new event.
-        prev.push(QueuedEvent { event: content, transaction_id, is_wedged: false });
+        prev.push(PersistedQueuedEvent {
+            room_id: room_id.to_owned(),
+            event: content,
+            transaction_id,
+            is_wedged: false,
+        });
 
         // Save the new vector into db.
         obj.put_key_val(&encoded_key, &self.serialize_value(&prev)?)?;
@@ -1394,7 +1412,7 @@ impl_state_store!({
 
         // Reload the previous vector for this room.
         if let Some(val) = obj.get(&encoded_key)?.await? {
-            let mut prev = self.deserialize_value::<Vec<QueuedEvent>>(&val)?;
+            let mut prev = self.deserialize_value::<Vec<PersistedQueuedEvent>>(&val)?;
             if let Some(pos) = prev.iter().position(|item| item.transaction_id == transaction_id) {
                 prev.remove(pos);
 
@@ -1424,10 +1442,17 @@ impl_state_store!({
 
         let prev = prev.map_or_else(
             || Ok(Vec::new()),
-            |val| self.deserialize_value::<Vec<QueuedEvent>>(&val),
+            |val| self.deserialize_value::<Vec<PersistedQueuedEvent>>(&val),
         )?;
 
-        Ok(prev)
+        Ok(prev
+            .into_iter()
+            .map(|item| QueuedEvent {
+                event: item.event,
+                transaction_id: item.transaction_id,
+                is_wedged: item.is_wedged,
+            })
+            .collect())
     }
 
     async fn update_send_queue_event_status(
@@ -1445,7 +1470,7 @@ impl_state_store!({
         let obj = tx.object_store(keys::ROOM_SEND_QUEUE)?;
 
         if let Some(val) = obj.get(&encoded_key)?.await? {
-            let mut prev = self.deserialize_value::<Vec<QueuedEvent>>(&val)?;
+            let mut prev = self.deserialize_value::<Vec<PersistedQueuedEvent>>(&val)?;
             if let Some(queued_event) =
                 prev.iter_mut().find(|item| item.transaction_id == transaction_id)
             {
@@ -1458,6 +1483,25 @@ impl_state_store!({
 
         Ok(())
     }
+
+    async fn load_rooms_with_unsent_events(&self) -> Result<Vec<OwnedRoomId>> {
+        let tx = self
+            .inner
+            .transaction_on_one_with_mode(keys::ROOM_SEND_QUEUE, IdbTransactionMode::Readwrite)?;
+
+        let obj = tx.object_store(keys::ROOM_SEND_QUEUE)?;
+
+        let all_entries = obj
+            .get_all()?
+            .await?
+            .into_iter()
+            .map(|item| {
+                self.deserialize_value(&item).map(|event: PersistedQueuedEvent| event.room_id)
+            })
+            .collect::<Result<HashSet<OwnedRoomId>, _>>()?;
+
+        Ok(all_entries.into_iter().collect())
+    }
 });
 
 /// A room member.

crates/matrix-sdk-sqlite/migrations/state_store/004_send_queue_with_roomid_value.sql

@@ -0,0 +1,22 @@
+-- Send queue events, keyed by room id and transaction id, but also include the room id in the
+-- value.
+DROP TABLE "send_queue_events";
+
+CREATE TABLE "send_queue_events" (
+    -- This is used as a key, thus hashed.
+    "room_id" BLOB NOT NULL,
+
+    -- This is used as a value (thus encrypted/decrypted).
+    "room_id_val" BLOB NOT NULL,
+
+    -- This is used as both a key and a value, thus neither encrypted/decrypted/hashed.
+    "transaction_id" BLOB NOT NULL,
+
+    -- Used as a value, thus encrypted/decrypted.
+    "content" BLOB NOT NULL,
+
+    -- In clear.
+    "wedged" BOOLEAN NOT NULL,
+
+    PRIMARY KEY ("room_id", "transaction_id")
+);

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

@@ -29,8 +29,8 @@ use ruma::{
         GlobalAccountDataEventType, RoomAccountDataEventType, StateEventType,
     },
     serde::Raw,
-    CanonicalJsonObject, EventId, OwnedEventId, OwnedTransactionId, OwnedUserId, RoomId,
-    RoomVersionId, TransactionId, UserId,
+    CanonicalJsonObject, EventId, OwnedEventId, OwnedRoomId, OwnedTransactionId, OwnedUserId,
+    RoomId, RoomVersionId, TransactionId, UserId,
 };
 use rusqlite::{OptionalExtension, Transaction};
 use serde::{de::DeserializeOwned, Deserialize, Serialize};
@@ -59,7 +59,7 @@ mod keys {
     pub const SEND_QUEUE: &str = "send_queue_events";
 }
 
-const DATABASE_VERSION: u8 = 4;
+const DATABASE_VERSION: u8 = 5;
 
 /// A sqlite based cryptostore.
 #[derive(Clone)]
@@ -224,6 +224,17 @@ impl SqliteStateStore {
             .await?;
         }
 
+        if from < 5 && to >= 5 {
+            conn.with_transaction(move |txn| {
+                // Create new table.
+                txn.execute_batch(include_str!(
+                    "../migrations/state_store/004_send_queue_with_roomid_value.sql"
+                ))?;
+                Result::<_, Error>::Ok(())
+            })
+            .await?;
+        }
+
         conn.set_kv("version", vec![to]).await?;
 
         Ok(())
@@ -1674,7 +1685,8 @@ impl StateStore for SqliteStateStore {
         transaction_id: OwnedTransactionId,
         content: SerializableEventContent,
     ) -> Result<(), Self::Error> {
-        let room_id = self.encode_key(keys::SEND_QUEUE, room_id);
+        let room_id_key = self.encode_key(keys::SEND_QUEUE, room_id);
+        let room_id_value = self.serialize_value(&room_id.to_owned())?;
 
         let content = self.serialize_json(&content)?;
 
@@ -1686,7 +1698,7 @@ impl StateStore for SqliteStateStore {
         self.acquire()
             .await?
             .with_transaction(move |txn| {
-                txn.prepare_cached("INSERT INTO send_queue_events (room_id, transaction_id, content, wedged) VALUES (?, ?, ?, false)")?.execute((room_id, transaction_id.to_string(), content))?;
+                txn.prepare_cached("INSERT INTO send_queue_events (room_id, room_id_val, transaction_id, content, wedged) VALUES (?, ?, ?, ?, false)")?.execute((room_id_key, room_id_value, transaction_id.to_string(), content))?;
                 Ok(())
             })
             .await
@@ -1767,6 +1779,29 @@ impl StateStore for SqliteStateStore {
             })
             .await
     }
+
+    async fn load_rooms_with_unsent_events(&self) -> Result<Vec<OwnedRoomId>, Self::Error> {
+        // If the values were not encrypted, we could use `SELECT DISTINCT` here, but we
+        // have to manually do the deduplication: indeed, for all X, encrypt(X)
+        // != encrypted(X), since we use a nonce in the encryption process.
+
+        let res: Vec<Vec<u8>> = self
+            .acquire()
+            .await?
+            .prepare("SELECT room_id_val FROM send_queue_events", |mut stmt| {
+                stmt.query(())?.mapped(|row| Ok(row.get(0)?)).collect()
+            })
+            .await?;
+
+        // So we collect the results into a `BTreeSet` to perform the deduplication, and
+        // then rejigger that into a vector.
+        Ok(res
+            .into_iter()
+            .map(|entry| self.deserialize_value(&entry))
+            .collect::<Result<BTreeSet<OwnedRoomId>, _>>()?
+            .into_iter()
+            .collect())
+    }
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]