refactor(send queue): generalize stored send queue data

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

@@ -33,7 +33,9 @@ use ruma::{
 };
 use serde_json::{json, value::Value as JsonValue};
 
-use super::{DependentQueuedRequestKind, DynStateStore, ServerCapabilities};
+use super::{
+    send_queue::SentRequestKey, DependentQueuedRequestKind, DynStateStore, ServerCapabilities,
+};
 use crate::{
     deserialized_responses::MemberEvent,
     store::{ChildTransactionId, QueueWedgeError, Result, SerializableEventContent, StateStoreExt},
@@ -1210,7 +1212,7 @@ impl StateStoreIntegrationTests for DynStateStore {
         let event0 =
             SerializableEventContent::new(&RoomMessageEventContent::text_plain("msg0").into())
                 .unwrap();
-        self.save_send_queue_request(room_id, txn0.clone(), event0).await.unwrap();
+        self.save_send_queue_request(room_id, txn0.clone(), event0.into()).await.unwrap();
 
         // Reading it will work.
         let pending = self.load_send_queue_requests(room_id).await.unwrap();
@@ -1234,7 +1236,7 @@ impl StateStoreIntegrationTests for DynStateStore {
             )
             .unwrap();
 
-            self.save_send_queue_request(room_id, txn, event).await.unwrap();
+            self.save_send_queue_request(room_id, txn, event.into()).await.unwrap();
         }
 
         // Reading all the events should work.
@@ -1284,7 +1286,7 @@ impl StateStoreIntegrationTests for DynStateStore {
             &RoomMessageEventContent::text_plain("wow that's a cool test").into(),
         )
         .unwrap();
-        self.update_send_queue_request(room_id, txn2, event0).await.unwrap();
+        self.update_send_queue_request(room_id, txn2, event0.into()).await.unwrap();
 
         // And it is reflected.
         let pending = self.load_send_queue_requests(room_id).await.unwrap();
@@ -1332,7 +1334,7 @@ impl StateStoreIntegrationTests for DynStateStore {
             let event =
                 SerializableEventContent::new(&RoomMessageEventContent::text_plain("room2").into())
                     .unwrap();
-            self.save_send_queue_request(room_id2, txn.clone(), event).await.unwrap();
+            self.save_send_queue_request(room_id2, txn.clone(), event.into()).await.unwrap();
         }
 
         // Add and remove one event for room3.
@@ -1342,7 +1344,7 @@ impl StateStoreIntegrationTests for DynStateStore {
             let event =
                 SerializableEventContent::new(&RoomMessageEventContent::text_plain("room3").into())
                     .unwrap();
-            self.save_send_queue_request(room_id3, txn.clone(), event).await.unwrap();
+            self.save_send_queue_request(room_id3, txn.clone(), event.into()).await.unwrap();
 
             self.remove_send_queue_request(room_id3, &txn).await.unwrap();
         }
@@ -1363,7 +1365,7 @@ impl StateStoreIntegrationTests for DynStateStore {
         let event0 =
             SerializableEventContent::new(&RoomMessageEventContent::text_plain("hey").into())
                 .unwrap();
-        self.save_send_queue_request(room_id, txn0.clone(), event0).await.unwrap();
+        self.save_send_queue_request(room_id, txn0.clone(), event0.into()).await.unwrap();
 
         // No dependents, to start with.
         assert!(self.load_dependent_queued_requests(room_id).await.unwrap().is_empty());
@@ -1384,21 +1386,27 @@ impl StateStoreIntegrationTests for DynStateStore {
         assert_eq!(dependents.len(), 1);
         assert_eq!(dependents[0].parent_transaction_id, txn0);
         assert_eq!(dependents[0].own_transaction_id, child_txn);
-        assert!(dependents[0].event_id.is_none());
+        assert!(dependents[0].parent_key.is_none());
         assert_matches!(dependents[0].kind, DependentQueuedRequestKind::RedactEvent);
 
         // Update the event id.
         let event_id = owned_event_id!("$1");
-        let num_updated =
-            self.update_dependent_queued_request(room_id, &txn0, event_id.clone()).await.unwrap();
+        let num_updated = self
+            .update_dependent_queued_request(
+                room_id,
+                &txn0,
+                SentRequestKey::Event(event_id.clone()),
+            )
+            .await
+            .unwrap();
         assert_eq!(num_updated, 1);
 
         // It worked.
         let dependents = self.load_dependent_queued_requests(room_id).await.unwrap();
         assert_eq!(dependents.len(), 1);
         assert_eq!(dependents[0].parent_transaction_id, txn0);
         assert_eq!(dependents[0].own_transaction_id, child_txn);
-        assert_eq!(dependents[0].event_id.as_ref(), Some(&event_id));
+        assert_eq!(dependents[0].parent_key.as_ref(), Some(&SentRequestKey::Event(event_id)));
         assert_matches!(dependents[0].kind, DependentQueuedRequestKind::RedactEvent);
 
         // Now remove it.
@@ -1417,7 +1425,7 @@ impl StateStoreIntegrationTests for DynStateStore {
         let event1 =
             SerializableEventContent::new(&RoomMessageEventContent::text_plain("hey2").into())
                 .unwrap();
-        self.save_send_queue_request(room_id, txn1.clone(), event1).await.unwrap();
+        self.save_send_queue_request(room_id, txn1.clone(), event1.into()).await.unwrap();
 
         self.save_dependent_queued_request(
             room_id,

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

@@ -36,7 +36,7 @@ use ruma::{
 use tracing::{debug, instrument, trace, warn};
 
 use super::{
-    send_queue::{ChildTransactionId, QueuedRequest, SerializableEventContent},
+    send_queue::{ChildTransactionId, QueuedRequest, SentRequestKey},
     traits::{ComposerDraft, ServerCapabilities},
     DependentQueuedRequest, DependentQueuedRequestKind, QueuedRequestKind, Result, RoomInfo,
     StateChanges, StateStore, StoreError,
@@ -806,23 +806,22 @@ impl StateStore for MemoryStore {
         &self,
         room_id: &RoomId,
         transaction_id: OwnedTransactionId,
-        content: SerializableEventContent,
+        kind: QueuedRequestKind,
     ) -> Result<(), Self::Error> {
-        self.send_queue_events.write().unwrap().entry(room_id.to_owned()).or_default().push(
-            QueuedRequest {
-                kind: QueuedRequestKind::Event { content },
-                transaction_id,
-                error: None,
-            },
-        );
+        self.send_queue_events
+            .write()
+            .unwrap()
+            .entry(room_id.to_owned())
+            .or_default()
+            .push(QueuedRequest { kind, transaction_id, error: None });
         Ok(())
     }
 
     async fn update_send_queue_request(
         &self,
         room_id: &RoomId,
         transaction_id: &TransactionId,
-        content: SerializableEventContent,
+        kind: QueuedRequestKind,
     ) -> Result<bool, Self::Error> {
         if let Some(entry) = self
             .send_queue_events
@@ -833,7 +832,7 @@ impl StateStore for MemoryStore {
             .iter_mut()
             .find(|item| item.transaction_id == transaction_id)
         {
-            entry.kind = QueuedRequestKind::Event { content };
+            entry.kind = kind;
             entry.error = None;
             Ok(true)
         } else {
@@ -907,7 +906,7 @@ impl StateStore for MemoryStore {
                 kind: content,
                 parent_transaction_id: parent_transaction_id.to_owned(),
                 own_transaction_id,
-                event_id: None,
+                parent_key: None,
             },
         );
         Ok(())
@@ -917,13 +916,13 @@ impl StateStore for MemoryStore {
         &self,
         room: &RoomId,
         parent_txn_id: &TransactionId,
-        event_id: OwnedEventId,
+        sent_parent_key: SentRequestKey,
     ) -> Result<usize, Self::Error> {
         let mut dependent_send_queue_events = self.dependent_send_queue_events.write().unwrap();
         let dependents = dependent_send_queue_events.entry(room.to_owned()).or_default();
         let mut num_updated = 0;
         for d in dependents.iter_mut().filter(|item| item.parent_transaction_id == parent_txn_id) {
-            d.event_id = Some(event_id.clone());
+            d.parent_key = Some(sent_parent_key.clone());
             num_updated += 1;
         }
         Ok(num_updated)

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

@@ -76,7 +76,7 @@ pub use self::{
     memory_store::MemoryStore,
     send_queue::{
         ChildTransactionId, DependentQueuedRequest, DependentQueuedRequestKind, QueueWedgeError,
-        QueuedRequest, QueuedRequestKind, SerializableEventContent,
+        QueuedRequest, QueuedRequestKind, SentRequestKey, SerializableEventContent,
     },
     traits::{
         ComposerDraft, ComposerDraftType, DynStateStore, IntoStateStore, ServerCapabilities,

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

@@ -78,6 +78,12 @@ pub enum QueuedRequestKind {
     },
 }
 
+impl From<SerializableEventContent> for QueuedRequestKind {
+    fn from(content: SerializableEventContent) -> Self {
+        Self::Event { content }
+    }
+}
+
 /// A request to be sent with a send queue.
 #[derive(Clone)]
 pub struct QueuedRequest {
@@ -150,18 +156,15 @@ pub enum QueueWedgeError {
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub enum DependentQueuedRequestKind {
     /// The event should be edited.
-    #[serde(rename = "Edit")]
     EditEvent {
         /// The new event for the content.
         new_content: SerializableEventContent,
     },
 
     /// The event should be redacted/aborted/removed.
-    #[serde(rename = "Redact")]
     RedactEvent,
 
     /// The event should be reacted to, with the given key.
-    #[serde(rename = "React")]
     ReactEvent {
         /// Key used for the reaction.
         key: String,
@@ -207,6 +210,23 @@ impl From<ChildTransactionId> for OwnedTransactionId {
     }
 }
 
+/// A unique key (identifier) indicating that a transaction has been
+/// successfully sent to the server.
+///
+/// The owning child transactions can now be resolved.
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+pub enum SentRequestKey {
+    /// The parent transaction returned an event when it succeeded.
+    Event(OwnedEventId),
+}
+
+impl SentRequestKey {
+    /// Converts the current parent key into an event id, if possible.
+    pub fn into_event_id(self) -> Option<OwnedEventId> {
+        as_variant!(self, Self::Event)
+    }
+}
+
 /// A request to be sent, depending on a [`QueuedRequest`] to be sent first.
 ///
 /// Depending on whether the parent request has been sent or not, this will
@@ -231,11 +251,7 @@ pub struct DependentQueuedRequest {
 
     /// If the parent request has been sent, the parent's request identifier
     /// returned by the server once the local echo has been sent out.
-    ///
-    /// Note: this is the event id used for the depended-on event after it's
-    /// been sent, not for a possible event that could have been sent
-    /// because of this [`DependentQueuedRequest`].
-    pub event_id: Option<OwnedEventId>,
+    pub parent_key: Option<SentRequestKey>,
 }
 
 #[cfg(not(tarpaulin_include))]

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

@@ -41,8 +41,9 @@ use ruma::{
 use serde::{Deserialize, Serialize};
 
 use super::{
-    ChildTransactionId, DependentQueuedRequest, DependentQueuedRequestKind, QueueWedgeError,
-    QueuedRequest, SerializableEventContent, StateChanges, StoreError,
+    send_queue::SentRequestKey, ChildTransactionId, DependentQueuedRequest,
+    DependentQueuedRequestKind, QueueWedgeError, QueuedRequest, QueuedRequestKind, StateChanges,
+    StoreError,
 };
 use crate::{
     deserialized_responses::{RawAnySyncOrStrippedState, RawMemberEvent, RawSyncOrStrippedState},
@@ -356,7 +357,7 @@ pub trait StateStore: AsyncTraitDeps {
         &self,
         room_id: &RoomId,
         transaction_id: OwnedTransactionId,
-        content: SerializableEventContent,
+        request: QueuedRequestKind,
     ) -> Result<(), Self::Error>;
 
     /// Updates a send queue request with the given content, and resets its
@@ -374,7 +375,7 @@ pub trait StateStore: AsyncTraitDeps {
         &self,
         room_id: &RoomId,
         transaction_id: &TransactionId,
-        content: SerializableEventContent,
+        content: QueuedRequestKind,
     ) -> Result<bool, Self::Error>;
 
     /// Remove a request previously inserted with
@@ -416,15 +417,19 @@ pub trait StateStore: AsyncTraitDeps {
         content: DependentQueuedRequestKind,
     ) -> Result<(), Self::Error>;
 
-    /// Update a set of dependent send queue requests with an event id,
-    /// effectively marking them as ready.
+    /// Update a set of dependent send queue requests with a key identifying the
+    /// homeserver's response, effectively marking them as ready.
+    ///
+    /// ⚠ Beware! There's no verification applied that the parent key type is
+    /// compatible with the dependent event type. The invalid state may be
+    /// lazily filtered out in `load_dependent_queued_requests`.
     ///
     /// Returns the number of updated requests.
     async fn update_dependent_queued_request(
         &self,
         room_id: &RoomId,
         parent_txn_id: &TransactionId,
-        event_id: OwnedEventId,
+        sent_parent_key: SentRequestKey,
     ) -> Result<usize, Self::Error>;
 
     /// Remove a specific dependent send queue request by id.
@@ -635,7 +640,7 @@ impl<T: StateStore> StateStore for EraseStateStoreError<T> {
         &self,
         room_id: &RoomId,
         transaction_id: OwnedTransactionId,
-        content: SerializableEventContent,
+        content: QueuedRequestKind,
     ) -> Result<(), Self::Error> {
         self.0.save_send_queue_request(room_id, transaction_id, content).await.map_err(Into::into)
     }
@@ -644,7 +649,7 @@ impl<T: StateStore> StateStore for EraseStateStoreError<T> {
         &self,
         room_id: &RoomId,
         transaction_id: &TransactionId,
-        content: SerializableEventContent,
+        content: QueuedRequestKind,
     ) -> Result<bool, Self::Error> {
         self.0.update_send_queue_request(room_id, transaction_id, content).await.map_err(Into::into)
     }
@@ -697,10 +702,10 @@ impl<T: StateStore> StateStore for EraseStateStoreError<T> {
         &self,
         room_id: &RoomId,
         parent_txn_id: &TransactionId,
-        event_id: OwnedEventId,
+        sent_parent_key: SentRequestKey,
     ) -> Result<usize, Self::Error> {
         self.0
-            .update_dependent_queued_request(room_id, parent_txn_id, event_id)
+            .update_dependent_queued_request(room_id, parent_txn_id, sent_parent_key)
             .await
             .map_err(Into::into)
     }

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

@@ -46,7 +46,7 @@ use super::{
 };
 use crate::IndexeddbStateStoreError;
 
-const CURRENT_DB_VERSION: u32 = 11;
+const CURRENT_DB_VERSION: u32 = 12;
 const CURRENT_META_DB_VERSION: u32 = 2;
 
 /// Sometimes Migrations can't proceed without having to drop existing
@@ -235,6 +235,9 @@ pub async fn upgrade_inner_db(
             if old_version < 11 {
                 db = migrate_to_v11(db).await?;
             }
+            if old_version < 12 {
+                db = migrate_to_v12(db).await?;
+            }
         }
 
         db.close();
@@ -771,6 +774,26 @@ async fn migrate_to_v11(db: IdbDatabase) -> Result<IdbDatabase> {
     apply_migration(db, 11, migration).await
 }
 
+/// The format of data serialized into the send queue and dependent send queue
+/// tables have changed, clear both.
+async fn migrate_to_v12(db: IdbDatabase) -> Result<IdbDatabase> {
+    let store_keys = &[keys::DEPENDENT_SEND_QUEUE, keys::ROOM_SEND_QUEUE];
+    let tx = db.transaction_on_multi_with_mode(store_keys, IdbTransactionMode::Readwrite)?;
+
+    for store_name in store_keys {
+        let store = tx.object_store(store_name)?;
+        store.clear()?;
+    }
+
+    tx.await.into_result()?;
+
+    let name = db.name();
+    db.close();
+
+    // Update the version of the database.
+    Ok(IdbDatabase::open_u32(&name, 12)?.await?)
+}
+
 #[cfg(all(test, target_arch = "wasm32"))]
 mod tests {
     wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_browser);