feat(send queue): send attachments with the send queue

bindings/matrix-sdk-ffi/src/error.rs

@@ -183,6 +183,12 @@ pub enum QueueWedgeError {
     /// session before sending.
     CrossVerificationRequired,
 
+    /// Some media content to be sent has disappeared from the cache.
+    MissingMediaContent,
+
+    /// Some mime type couldn't be parsed.
+    InvalidMimeType { mime_type: String },
+
     /// Other errors.
     GenericApiError { msg: String },
 }
@@ -201,10 +207,17 @@ impl Display for QueueWedgeError {
             QueueWedgeError::CrossVerificationRequired => {
                 f.write_str("Own verification is required")
             }
+            QueueWedgeError::MissingMediaContent => {
+                f.write_str("Media to be sent disappeared from local storage")
+            }
+            QueueWedgeError::InvalidMimeType { mime_type } => {
+                write!(f, "Invalid mime type '{mime_type}' for media upload")
+            }
             QueueWedgeError::GenericApiError { msg } => f.write_str(msg),
         }
     }
 }
+
 impl From<SdkQueueWedgeError> for QueueWedgeError {
     fn from(value: SdkQueueWedgeError) -> Self {
         match value {
@@ -223,6 +236,10 @@ impl From<SdkQueueWedgeError> for QueueWedgeError {
                 users: users.iter().map(ruma::OwnedUserId::to_string).collect(),
             },
             SdkQueueWedgeError::CrossVerificationRequired => Self::CrossVerificationRequired,
+            SdkQueueWedgeError::MissingMediaContent => Self::MissingMediaContent,
+            SdkQueueWedgeError::InvalidMimeType { mime_type } => {
+                Self::InvalidMimeType { mime_type }
+            }
             SdkQueueWedgeError::GenericApiError { msg } => Self::GenericApiError { msg },
         }
     }

crates/matrix-sdk-base/src/media.rs

@@ -14,6 +14,7 @@ use ruma::{
     },
     MxcUri, UInt,
 };
+use serde::{Deserialize, Serialize};
 
 const UNIQUE_SEPARATOR: &str = "_";
 
@@ -25,7 +26,7 @@ pub trait UniqueKey {
 }
 
 /// The requested format of a media file.
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Serialize, Deserialize)]
 pub enum MediaFormat {
     /// The file that was uploaded.
     File,
@@ -44,7 +45,7 @@ impl UniqueKey for MediaFormat {
 }
 
 /// The requested size of a media thumbnail.
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct MediaThumbnailSize {
     /// The desired resizing method.
     pub method: Method,
@@ -65,7 +66,7 @@ impl UniqueKey for MediaThumbnailSize {
 }
 
 /// The desired settings of a media thumbnail.
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct MediaThumbnailSettings {
     /// The desired size of the thumbnail.
     pub size: MediaThumbnailSize,
@@ -110,7 +111,7 @@ impl UniqueKey for MediaSource {
 }
 
 /// A request for media data.
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct MediaRequest {
     /// The source of the media file.
     pub source: MediaSource,

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,29 @@ 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_matches!(dependents[0].parent_key.as_ref(), Some(&SentRequestKey::Event(ref eid)) => {
+            assert_eq!(*eid, event_id);
+        });
         assert_matches!(dependents[0].kind, DependentQueuedRequestKind::RedactEvent);
 
         // Now remove it.
@@ -1417,7 +1427,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

@@ -75,8 +75,9 @@ pub use self::integration_tests::StateStoreIntegrationTests;
 pub use self::{
     memory_store::MemoryStore,
     send_queue::{
-        ChildTransactionId, DependentQueuedRequest, DependentQueuedRequestKind, QueueWedgeError,
-        QueuedRequest, QueuedRequestKind, SerializableEventContent,
+        ChildTransactionId, DependentQueuedRequest, DependentQueuedRequestKind,
+        FinishUploadThumbnailInfo, QueueWedgeError, QueuedRequest, QueuedRequestKind,
+        SentRequestKey, SerializableEventContent,
     },
     traits::{
         ComposerDraft, ComposerDraftType, DynStateStore, IntoStateStore, ServerCapabilities,