diff --git a/crates/matrix-sdk-base/src/store/integration_tests.rs b/crates/matrix-sdk-base/src/store/integration_tests.rs index 569c9c1d7..23a70e635 100644 --- a/crates/matrix-sdk-base/src/store/integration_tests.rs +++ b/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)); } } diff --git a/crates/matrix-sdk-base/src/store/memory_store.rs b/crates/matrix-sdk-base/src/store/memory_store.rs index b27220e43..9c5d92188 100644 --- a/crates/matrix-sdk-base/src/store/memory_store.rs +++ b/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, Self::Error> { + Ok(self.send_queue_events.read().unwrap().keys().cloned().collect()) + } } #[cfg(test)] diff --git a/crates/matrix-sdk-base/src/store/traits.rs b/crates/matrix-sdk-base/src/store/traits.rs index b8214eeb2..c640a50db 100644 --- a/crates/matrix-sdk-base/src/store/traits.rs +++ b/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, Self::Error>; } #[repr(transparent)] @@ -688,6 +692,10 @@ impl StateStore for EraseStateStoreError { .await .map_err(Into::into) } + + async fn load_rooms_with_unsent_events(&self) -> Result, 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, diff --git a/crates/matrix-sdk-indexeddb/src/state_store/mod.rs b/crates/matrix-sdk-indexeddb/src/state_store/mod.rs index 71abe53f8..a8a469925 100644 --- a/crates/matrix-sdk-indexeddb/src/state_store/mod.rs +++ b/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::>(&val), + |val| self.deserialize_value::>(&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::>(&val)?; + let mut prev = self.deserialize_value::>(&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::>(&val), + |val| self.deserialize_value::>(&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::>(&val)?; + let mut prev = self.deserialize_value::>(&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> { + 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::, _>>()?; + + Ok(all_entries.into_iter().collect()) + } }); /// A room member. diff --git a/crates/matrix-sdk-sqlite/migrations/state_store/004_send_queue_with_roomid_value.sql b/crates/matrix-sdk-sqlite/migrations/state_store/004_send_queue_with_roomid_value.sql new file mode 100644 index 000000000..6516a5601 --- /dev/null +++ b/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") +); diff --git a/crates/matrix-sdk-sqlite/src/state_store.rs b/crates/matrix-sdk-sqlite/src/state_store.rs index 5fc8cea5a..904633217 100644 --- a/crates/matrix-sdk-sqlite/src/state_store.rs +++ b/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, 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> = 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::, _>>()? + .into_iter() + .collect()) + } } #[derive(Debug, Clone, Serialize, Deserialize)]