fix(appservices): push inivte over federation events

src/api/ruma_wrapper/axum.rs

@@ -107,10 +107,7 @@ where
                         "Unknown access token.",
                     ))
                 }
-                (
+                (AuthScheme::AccessToken, Token::Appservice(info)) => {
-                    AuthScheme::AccessToken | AuthScheme::AccessTokenOptional,
-                    Token::Appservice(info),
-                ) => {
                     let user_id = query_params
                         .user_id
                         .map_or_else(
@@ -135,9 +132,12 @@ where
                     // TODO: Check if appservice is allowed to be that user
                     (Some(user_id), None, None, true)
                 }
-                (AuthScheme::None | AuthScheme::AppserviceToken, Token::Appservice(_)) => {
+                (
-                    (None, None, None, true)
+                    AuthScheme::None
-                }
+                    | AuthScheme::AppserviceToken
+                    | AuthScheme::AccessTokenOptional,
+                    Token::Appservice(_),
+                ) => (None, None, None, true),
                 (AuthScheme::AccessToken, Token::None) => {
                     return Err(Error::BadRequest(
                         ErrorKind::MissingToken,

src/api/server_server.rs

@@ -43,7 +43,7 @@ use ruma::{
     to_device::DeviceIdOrAllDevices,
     uint, user_id, CanonicalJsonObject, CanonicalJsonValue, EventId, MilliSecondsSinceUnixEpoch,
     OwnedEventId, OwnedRoomId, OwnedServerName, OwnedServerSigningKeyId, OwnedUserId, RoomId,
-    ServerName,
+    ServerName, UserId,
 };
 use serde_json::value::{to_raw_value, RawValue as RawJsonValue};
 use std::{
@@ -1715,7 +1715,10 @@ pub async fn create_invite_route(
     let mut event: JsonObject = serde_json::from_str(body.event.get())
         .map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Invalid invite event bytes."))?;
 
-    event.insert("event_id".to_owned(), "$dummy".into());
+    event.insert(
+        "event_id".to_owned(),
+        serde_json::Value::String(EventId::new(services().globals.server_name()).to_string()),
+    );
 
     let pdu: PduEvent = serde_json::from_value(event.into()).map_err(|e| {
         warn!("Invalid invite event: {}", e);
@@ -1738,6 +1741,22 @@ pub async fn create_invite_route(
             Some(invite_state),
             true,
         )?;
+
+        if let Some(Ok(user)) = pdu.state_key.as_ref().map(UserId::parse) {
+            let lock = services().appservice.read().await;
+            for id in lock
+                .iter()
+                .filter(|(_, info)| info.users.is_match(user.as_str()))
+                .map(|(id, _)| id)
+            {
+                if let Err(e) = services()
+                    .sending
+                    .send_event_appservice(id.to_string(), pdu.clone())
+                {
+                    error!("Failed to send invite event to appservice: {}", e);
+                }
+            }
+        }
     }
 
     Ok(create_invite::v2::Response {

src/service/sending/mod.rs

@@ -1,6 +1,7 @@
 mod data;
 
 pub use data::Data;
+use serde_json::value::to_raw_value;
 
 use std::{
     collections::{BTreeMap, HashMap, HashSet},
@@ -79,8 +80,9 @@ impl OutgoingKind {
     }
 }
 
-#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+#[derive(Clone, Debug, PartialEq, Eq)]
 pub enum SendingEventType {
+    Event(Box<PduEvent>),
     Pdu(Vec<u8>), // pduid
     Edu(Vec<u8>), // pdu json
 }
@@ -440,6 +442,18 @@ impl Service {
         Ok(())
     }
 
+    #[tracing::instrument(skip(self))]
+    pub fn send_event_appservice(&self, appservice_id: String, event: PduEvent) -> Result<()> {
+        let outgoing_kind = OutgoingKind::Appservice(appservice_id);
+        let event = SendingEventType::Event(Box::new(event));
+        let keys = self.db.queue_requests(&[(&outgoing_kind, event.clone())])?;
+        self.sender
+            .send((outgoing_kind, event, keys.into_iter().next().unwrap()))
+            .unwrap();
+
+        Ok(())
+    }
+
     /// Cleanup event data
     /// Used for instance after we remove an appservice registration
     ///
@@ -476,6 +490,9 @@ impl Service {
                                 })?
                                 .to_room_event())
                         }
+                        SendingEventType::Event(event) => {
+                            pdu_jsons.push(event.to_room_event());
+                        }
                         SendingEventType::Edu(_) => {
                             // Appservices don't need EDUs (?)
                         }
@@ -504,6 +521,7 @@ impl Service {
                                 .iter()
                                 .map(|e| match e {
                                     SendingEventType::Edu(b) | SendingEventType::Pdu(b) => &**b,
+                                    SendingEventType::Event(b) => b.event_id.as_bytes(),
                                 })
                                 .collect::<Vec<_>>(),
                         )))
@@ -541,6 +559,7 @@ impl Service {
                                     })?,
                             );
                         }
+                        SendingEventType::Event(event) => pdus.push(*event.clone()),
                         SendingEventType::Edu(_) => {
                             // Push gateways don't need EDUs (?)
                         }
@@ -631,6 +650,11 @@ impl Service {
                                 edu_jsons.push(raw);
                             }
                         }
+                        SendingEventType::Event(event) => {
+                            pdu_jsons.push(
+                                to_raw_value(event).expect("To raw value always succeeds for pdus"),
+                            );
+                        }
                     }
                 }
 
@@ -649,6 +673,7 @@ impl Service {
                                     .iter()
                                     .map(|e| match e {
                                         SendingEventType::Edu(b) | SendingEventType::Pdu(b) => &**b,
+                                        SendingEventType::Event(b) => b.event_id.as_bytes(),
                                     })
                                     .collect::<Vec<_>>(),
                             ),