feat(sdk): Add the ability to send out custom message events

crates/matrix-sdk-crypto/src/machine.rs

@@ -43,10 +43,12 @@ use ruma::{
         },
         room_key::RoomKeyToDeviceEventContent,
         secret::request::SecretName,
-        AnyMessageEventContent, AnyRoomEvent, AnyToDeviceEvent, SyncMessageEvent, ToDeviceEvent,
+        AnyMessageEventContent, AnyRoomEvent, AnyToDeviceEvent, EventContent, SyncMessageEvent,
+        ToDeviceEvent,
     },
     DeviceId, DeviceIdBox, DeviceKeyAlgorithm, EventEncryptionAlgorithm, RoomId, UInt, UserId,
 };
+use serde_json::Value;
 use tracing::{debug, error, info, trace, warn};
 
 #[cfg(feature = "sled_cryptostore")]
@@ -642,6 +644,9 @@ impl OlmMachine {
     /// [`should_share_group_session`] method if a new group session needs to
     /// be shared.
     ///
+    /// **Note**: This method doesn't support encrypting custom events, see the
+    /// [`encrypt_raw()`] method to do so.
+    ///
     /// # Arguments
     ///
     /// * `room_id` - The id of the room for which the message should be
@@ -656,12 +661,45 @@ impl OlmMachine {
     ///
     /// [`should_share_group_session`]: #method.should_share_group_session
     /// [`share_group_session`]: #method.share_group_session
+    /// [`encrypt_raw()`]: #method.encrypt_raw
     pub async fn encrypt(
         &self,
         room_id: &RoomId,
         content: AnyMessageEventContent,
     ) -> MegolmResult<EncryptedEventContent> {
-        self.group_session_manager.encrypt(room_id, content).await
+        let event_type = content.event_type().to_owned();
+        let content = serde_json::to_value(content)?;
+
+        self.group_session_manager.encrypt(room_id, content, &event_type).await
+    }
+
+    /// Encrypt a json [`Value`] content for the given room.
+    ///
+    /// This method is equivalent to the [`encrypt()`] method but allows custom
+    /// events to be encrypted.
+    ///
+    /// # Arguments
+    ///
+    /// * `room_id` - The id of the room for which the message should be
+    /// encrypted.
+    ///
+    /// * `content` - The plaintext content of the message that should be
+    /// encrypted as a json [`Value`].
+    ///
+    /// * `event_type` - The plaintext type of the event.
+    ///
+    /// # Panics
+    ///
+    /// Panics if a group session for the given room wasn't shared beforehand.
+    ///
+    /// [`encrypt()`]: #method.encrypt
+    pub async fn encrypt_raw(
+        &self,
+        room_id: &RoomId,
+        content: Value,
+        event_type: &str,
+    ) -> MegolmResult<EncryptedEventContent> {
+        self.group_session_manager.encrypt(room_id, content, event_type).await
     }
 
     /// Invalidate the currently active outbound group session for the given

crates/matrix-sdk-crypto/src/olm/group_sessions/mod.rs

@@ -136,11 +136,11 @@ mod test {
     };
 
     use super::EncryptionSettings;
-    use crate::ReadOnlyAccount;
+    use crate::{MegolmError, ReadOnlyAccount};
 
     #[tokio::test]
     #[cfg(target_os = "linux")]
-    async fn expiration() {
+    async fn expiration() -> Result<(), MegolmError> {
         let settings = EncryptionSettings { rotation_period_msgs: 1, ..Default::default() };
 
         let account = ReadOnlyAccount::new(&user_id!("@alice:example.org"), "DEVICEID".into());
@@ -151,9 +151,12 @@ mod test {
 
         assert!(!session.expired());
         let _ = session
-            .encrypt(AnyMessageEventContent::RoomMessage(MessageEventContent::text_plain(
+            .encrypt(
-                "Test message",
+                serde_json::to_value(AnyMessageEventContent::RoomMessage(
-            )))
+                    MessageEventContent::text_plain("Test message"),
+                ))?,
+                "m.room.message",
+            )
             .await;
         assert!(session.expired());
 
@@ -170,5 +173,7 @@ mod test {
         assert!(!session.expired());
         session.creation_time = Arc::new(Instant::now() - Duration::from_secs(60 * 60));
         assert!(session.expired());
+
+        Ok(())
     }
 }

crates/matrix-sdk-crypto/src/olm/group_sessions/outbound.rs

@@ -41,12 +41,12 @@ use ruma::{
             history_visibility::HistoryVisibility,
         },
         room_key::RoomKeyToDeviceEventContent,
-        AnyMessageEventContent, AnyToDeviceEventContent, EventContent,
+        AnyToDeviceEventContent,
     },
     DeviceId, DeviceIdBox, DeviceKeyAlgorithm, EventEncryptionAlgorithm, RoomId, UserId,
 };
 use serde::{Deserialize, Serialize};
-use serde_json::json;
+use serde_json::{json, Value};
 use tracing::{debug, error, trace};
 
 use super::{
@@ -268,20 +268,23 @@ impl OutboundGroupSession {
     /// # Arguments
     ///
     /// * `content` - The plaintext content of the message that should be
-    /// encrypted.
+    /// encrypted in raw json [`Value`] form.
+    ///
+    /// * `event_type` - The plaintext type of the event, the outer type of the
+    /// event will become `m.room.encrypted`.
     ///
     /// # Panics
     ///
     /// Panics if the content can't be serialized.
-    pub async fn encrypt(&self, content: AnyMessageEventContent) -> EncryptedEventContent {
+    pub async fn encrypt(&self, content: Value, event_type: &str) -> EncryptedEventContent {
         let json_content = json!({
             "content": content,
             "room_id": &*self.room_id,
-            "type": content.event_type(),
+            "type": event_type,
         });
 
         let plaintext = json_content.to_string();
-        let relation = content.relation();
+        let relation = serde_json::from_value(content).ok();
 
         let ciphertext = self.encrypt_helper(plaintext).await;
 

crates/matrix-sdk-crypto/src/olm/mod.rs

@@ -75,7 +75,10 @@ pub(crate) mod test {
     };
     use serde_json::json;
 
-    use crate::olm::{InboundGroupSession, ReadOnlyAccount, Session};
+    use crate::{
+        olm::{InboundGroupSession, ReadOnlyAccount, Session},
+        MegolmError,
+    };
 
     fn alice_id() -> UserId {
         user_id!("@alice:example.org")
@@ -223,7 +226,7 @@ pub(crate) mod test {
     }
 
     #[tokio::test]
-    async fn edit_decryption() {
+    async fn edit_decryption() -> Result<(), MegolmError> {
         let alice = ReadOnlyAccount::new(&alice_id(), &alice_device_id());
         let room_id = room_id!("!test:localhost");
         let event_id = event_id!("$1234adfad:asdf");
@@ -247,15 +250,18 @@ pub(crate) mod test {
             &room_id,
             outbound.session_key().await,
             None,
-        )
+        )?;
-        .unwrap();
 
         assert_eq!(0, inbound.first_known_index());
 
         assert_eq!(outbound.session_id(), inbound.session_id());
 
-        let encrypted_content =
+        let encrypted_content = outbound
-            outbound.encrypt(AnyMessageEventContent::RoomMessage(content)).await;
+            .encrypt(
+                serde_json::to_value(AnyMessageEventContent::RoomMessage(content))?,
+                "m.room.message",
+            )
+            .await;
 
         let event = json!({
             "sender": alice.user_id(),
@@ -276,15 +282,17 @@ pub(crate) mod test {
                 panic!("Invalid event type")
             };
 
-        let decrypted = inbound.decrypt(&event).await.unwrap().0;
+        let decrypted = inbound.decrypt(&event).await?.0;
 
         if let AnySyncRoomEvent::Message(AnySyncMessageEvent::RoomMessage(e)) =
-            decrypted.deserialize().unwrap()
+            decrypted.deserialize()?
         {
             assert_matches!(e.content.relates_to, Some(Relation::Replacement(_)));
         } else {
             panic!("Invalid event type")
         }
+
+        Ok(())
     }
 
     #[tokio::test]

crates/matrix-sdk-crypto/src/session_manager/group_sessions.rs

@@ -23,12 +23,13 @@ use matrix_sdk_common::{executor::spawn, uuid::Uuid};
 use ruma::{
     events::{
         room::{encrypted::EncryptedEventContent, history_visibility::HistoryVisibility},
-        AnyMessageEventContent, AnyToDeviceEventContent, EventType,
+        AnyToDeviceEventContent, EventType,
     },
     serde::Raw,
     to_device::DeviceIdOrAllDevices,
     DeviceId, DeviceIdBox, RoomId, UserId,
 };
+use serde_json::Value;
 use tracing::{debug, info, trace};
 
 use crate::{
@@ -155,7 +156,8 @@ impl GroupSessionManager {
     pub async fn encrypt(
         &self,
         room_id: &RoomId,
-        content: AnyMessageEventContent,
+        content: Value,
+        event_type: &str,
     ) -> MegolmResult<EncryptedEventContent> {
         let session = if let Some(s) = self.sessions.get(room_id) {
             s
@@ -167,7 +169,7 @@ impl GroupSessionManager {
             panic!("Session expired");
         }
 
-        let content = session.encrypt(content).await;
+        let content = session.encrypt(content, event_type).await;
 
         let mut changes = Changes::default();
         changes.outbound_group_sessions.push(session);

crates/matrix-sdk/src/room/joined.rs

@@ -38,7 +38,7 @@ use ruma::{
             EncryptedFile,
         },
         tag::TagInfo,
-        AnyMessageEventContent, AnyStateEventContent,
+        AnyMessageEventContent, AnyStateEventContent, EventContent,
     },
     receipt::ReceiptType,
     serde::Raw,
@@ -337,13 +337,25 @@ impl Joined {
     /// If the encryption feature is enabled this method will transparently
     /// encrypt the room message if this room is encrypted.
     ///
+    /// **Note**: This method does not support sending custom events, the
+    /// [`send_raw()`] method can be used for that instead.
+    ///
     /// # Arguments
     ///
     /// * `content` - The content of the message event.
     ///
-    /// * `txn_id` - A unique `Uuid` that can be attached to a `MessageEvent`
+    /// * `txn_id` - A locally-unique ID describing a message transaction with
-    /// held in its unsigned field as `transaction_id`. If not given one is
+    ///   the homeserver. Unless you're doing something special, you can pass in
-    /// created for the message.
+    ///   `None` which will create a suitable one for you automatically.
+    ///     * On the sending side, this field is used for re-trying earlier
+    ///       failed transactions. Subsequent messages *must never* re-use an
+    ///       earlier transaction ID.
+    ///     * On the receiving side, the field is used for recognizing our own
+    ///       messages when they arrive down the sync: the server includes the
+    ///       ID in the [`Unsigned`] field [`transaction_id`] of the
+    ///       corresponding [`SyncMessageEvent`], but only for the *sending*
+    ///       device. Other devices will not see it. This is then used to ignore
+    ///       events sent by our own device and/or to implement local echo.
     ///
     /// # Example
     /// ```no_run
@@ -376,31 +388,117 @@ impl Joined {
     /// }
     /// # matrix_sdk::Result::Ok(()) });
     /// ```
+    ///
+    /// [`send_raw()`]: #method.send_raw
+    /// [`SyncMessageEvent`]: ruma::events::SyncMessageEvent
+    /// [`Unsigned`]: ruma::events::Unsigned
+    /// [`transaction_id`]: ruma::events::Unsigned#structfield.transaction_id
     pub async fn send(
         &self,
         content: impl Into<AnyMessageEventContent>,
         txn_id: Option<Uuid>,
     ) -> Result<send_message_event::Response> {
+        let content = content.into();
+        let event_type = content.event_type().to_owned();
+        let content = serde_json::to_value(content)?;
+
+        self.send_raw(content, &event_type, txn_id).await
+    }
+
+    /// Send a room message to this room from a json `Value`.
+    ///
+    /// Returns the parsed response from the server.
+    ///
+    /// If the encryption feature is enabled this method will transparently
+    /// encrypt the room message if this room is encrypted.
+    ///
+    /// This method is equivalent to the [`send()`] method but allows sending
+    /// custom events.
+    ///
+    /// # Arguments
+    ///
+    /// * `content` - The content of the event as a json `Value`.
+    ///
+    /// * `event_type` - The type of the event.
+    ///
+    /// * `txn_id` - A locally-unique ID describing a message transaction with
+    ///   the homeserver. Unless you're doing something special, you can pass in
+    ///   `None` which will create a suitable one for you automatically.
+    ///     * On the sending side, this field is used for re-trying earlier
+    ///       failed transactions. Subsequent messages *must never* re-use an
+    ///       earlier transaction ID.
+    ///     * On the receiving side, the field is used for recognizing our own
+    ///       messages when they arrive down the sync: the server includes the
+    ///       ID in the [`Unsigned`] field [`transaction_id`] of the
+    ///       corresponding [`SyncMessageEvent`], but only for the *sending*
+    ///       device. Other devices will not see it. This is then used to ignore
+    ///       events sent by our own device and/or to implement local echo.
+    ///
+    /// # Example
+    /// ```no_run
+    /// # use std::sync::{Arc, RwLock};
+    /// # use matrix_sdk::{Client, config::SyncSettings};
+    /// # use url::Url;
+    /// # use futures::executor::block_on;
+    /// # use matrix_sdk::ruma::room_id;
+    /// # use std::convert::TryFrom;
+    /// # block_on(async {
+    /// # let homeserver = Url::parse("http://localhost:8080")?;
+    /// # let mut client = Client::new(homeserver)?;
+    /// # let room_id = room_id!("!test:localhost");
+    /// use serde_json::json;
+    ///
+    /// let content = json!({
+    ///     "body": "Hello world",
+    /// });
+    ///
+    /// if let Some(room) = client.get_joined_room(&room_id) {
+    ///     room.send_raw(content, "m.room.message", None).await?;
+    /// }
+    /// # matrix_sdk::Result::Ok(()) });
+    /// ```
+    ///
+    /// [`send()`]: #method.send
+    /// [`SyncMessageEvent`]: ruma::events::SyncMessageEvent
+    /// [`Unsigned`]: ruma::events::Unsigned
+    /// [`transaction_id`]: ruma::events::Unsigned#structfield.transaction_id
+    pub async fn send_raw(
+        &self,
+        content: Value,
+        event_type: &str,
+        txn_id: Option<Uuid>,
+    ) -> Result<send_message_event::Response> {
+        let txn_id = txn_id.unwrap_or_else(Uuid::new_v4).to_string();
+
         #[cfg(not(feature = "encryption"))]
-        let content: AnyMessageEventContent = content.into();
+        let content = Raw::from_json(serde_json::value::to_raw_value(&content)?);
 
         #[cfg(feature = "encryption")]
-        let content = if self.is_encrypted() {
+        let (content, event_type) = if self.is_encrypted() {
             if !self.are_members_synced() {
                 self.request_members().await?;
                 // TODO query keys here?
             }
 
             self.preshare_group_session().await?;
-            AnyMessageEventContent::RoomEncrypted(
+
-                self.client.base_client.encrypt(self.inner.room_id(), content).await?,
+            let olm =
-            )
+                self.client.base_client.olm_machine().await.expect("Olm machine wasn't started");
+
+            let encrypted_content =
+                olm.encrypt_raw(self.inner.room_id(), content, event_type).await?;
+
+            (AnyMessageEventContent::RoomEncrypted(encrypted_content).into(), "m.room.encrypted")
         } else {
-            content.into()
+            (Raw::from_json(serde_json::value::to_raw_value(&content)?), event_type)
         };
 
-        let txn_id = txn_id.unwrap_or_else(Uuid::new_v4).to_string();
+        let request = send_message_event::Request::new_raw(
-        let request = send_message_event::Request::new(self.inner.room_id(), &txn_id, &content);
+            self.inner.room_id(),
+            &txn_id,
+            event_type,
+            content,
+        );
 
         let response = self.client.send(request, None).await?;
         Ok(response)