appservice-api: conversion from incoming transaction to sync response

ruma-appservice-api/Cargo.toml

@@ -13,15 +13,18 @@ edition = "2018"
 
 [dependencies]
 ruma-api = { version = "=0.17.0-alpha.4", path = "../ruma-api" }
+ruma-client-api = { version = "=0.10.0-alpha.3", path = "../ruma-client-api", features = ["client"], optional = true }
 ruma-common = { version = "0.5.0", path = "../ruma-common" }
 ruma-events = { version = "=0.22.0-alpha.3", path = "../ruma-events" }
 ruma-identifiers = { version = "0.19.0", path = "../ruma-identifiers" }
 ruma-serde = { version = "0.3.1", path = "../ruma-serde" }
 serde = { version = "1.0.118", features = ["derive"] }
 serde_json = "1.0.61"
+tracing = { version = "0.1.25", optional = true }
 
 [features]
 unstable-exhaustive-types = []
+helper = ["ruma-client-api", "tracing"]
 client = []
 server = []
 

ruma-appservice-api/src/event/push_events/v1.rs

@@ -41,6 +41,52 @@ impl IncomingRequest {
     pub fn new(txn_id: String, events: Vec<Raw<AnyEvent>>) -> IncomingRequest {
         IncomingRequest { txn_id, events }
     }
+
+    /// Consumes the `IncomingRequest` and tries to convert it to a `sync_events::Response`
+    ///
+    /// This is a helper conversion in cases where it's easier to work with `sync_events::Response`
+    /// instead of the original `push_events::IncomingRequest`. It puts all events with a `room_id`
+    /// into the `JoinedRoom`'s `timeline`. The rationale behind that is that incoming Appservice
+    /// transactions from the homeserver are not necessarily bound to a specific user but can cover
+    /// a multitude of namespaces, and as such the Appservice basically only "observes joined
+    /// rooms".
+    ///
+    /// Note: Currently homeservers only push PDUs to appservices, no EDUs. There's the open
+    /// [MSC2409] regarding supporting EDUs in the future, though it seems to be planned to put
+    /// EDUs into a different JSON key than `events` to stay backwards compatible.
+    ///
+    /// [MSC2409]: https://github.com/matrix-org/matrix-doc/pull/2409
+    #[cfg(feature = "helper")]
+    pub fn try_into_sync_response(
+        self,
+        next_batch: impl Into<String>,
+    ) -> Result<ruma_client_api::r0::sync::sync_events::Response, serde_json::Error> {
+        use ruma_client_api::r0::sync::sync_events;
+        use ruma_identifiers::RoomId;
+        use serde::Deserialize;
+        use tracing::warn;
+
+        #[derive(Debug, Deserialize)]
+        struct EventDeHelper {
+            room_id: Option<RoomId>,
+        }
+
+        let mut response = sync_events::Response::new(next_batch.into());
+
+        for raw_event in self.events {
+            let helper: EventDeHelper = serde_json::from_str(raw_event.json().get())?;
+            let event_json = Raw::into_json(raw_event);
+
+            if let Some(room_id) = helper.room_id {
+                let join = response.rooms.join.entry(room_id).or_default();
+                join.timeline.events.push(Raw::from_json(event_json));
+            } else {
+                warn!("Event without room_id: {}", event_json);
+            }
+        }
+
+        Ok(response)
+    }
 }
 
 impl Response {
@@ -50,6 +96,54 @@ impl Response {
     }
 }
 
+#[cfg(feature = "helper")]
+#[cfg(test)]
+mod helper_tests {
+    use super::{AnyEvent, IncomingRequest, Raw};
+    use ruma_client_api::r0::sync::sync_events;
+    use ruma_identifiers::room_id;
+    use serde_json::json;
+
+    #[test]
+    fn convert_incoming_request_to_sync_response() {
+        let txn_id = "any_txn_id".to_owned();
+        let state_event: AnyEvent = serde_json::from_value(json!({
+            "content": {},
+            "event_id": "$h29iv0s8:example.com",
+            "origin_server_ts": 1,
+            "room_id": "!roomid:room.com",
+            "sender": "@carl:example.com",
+            "state_key": "",
+            "type": "m.room.name"
+        }))
+        .unwrap();
+        let room_event: AnyEvent = serde_json::from_value(json!({
+            "type": "m.room.message",
+            "event_id": "$143273582443PhrSn:example.com",
+            "origin_server_ts": 1,
+            "room_id": "!roomid:room.com",
+            "sender": "@user:example.com",
+            "content": {
+                "body": "test",
+                "msgtype": "m.audio",
+                "url": "mxc://example.com/AuDi0",
+            }
+        }))
+        .unwrap();
+
+        let events = vec![Raw::from(state_event), Raw::from(room_event)];
+        let incoming_request = IncomingRequest { txn_id: txn_id.clone(), events };
+
+        let response: sync_events::Response =
+            incoming_request.try_into_sync_response(txn_id).unwrap();
+
+        let response_rooms_join =
+            response.rooms.join.get(&room_id!("!roomid:room.com")).expect("joined room response");
+
+        assert_eq!(response_rooms_join.timeline.events.len(), 2);
+    }
+}
+
 #[cfg(feature = "server")]
 #[cfg(test)]
 mod tests {

ruma/Cargo.toml

@@ -93,6 +93,10 @@ compat = [
     "ruma-client-api/compat",
 ]
 
+# Helper features that aren't exactly part of the spec but could be helpful
+# for crate consumers
+appservice-api-helper = ["ruma-appservice-api/helper"]
+
 # unstable: by using any of these, you opt out of all semver guarantees Ruma
 #           otherwise provides!
 unstable-exhaustive-types = [