improvement: better appservice compatibility and optimizations

src/appservice_server.rs

@@ -1,6 +1,6 @@
 use crate::{utils, Error, Result};
 use http::header::{HeaderValue, CONTENT_TYPE};
-use log::warn;
+use log::{info, warn};
 use ruma::api::OutgoingRequest;
 use std::{
     convert::{TryFrom, TryInto},

src/client_server/state.rs

@@ -234,7 +234,7 @@ pub async fn get_state_events_for_empty_key_route(
         .1;
 
     Ok(get_state_events_for_empty_key::Response {
-        content: serde_json::value::to_raw_value(&event)
+        content: serde_json::value::to_raw_value(&event.content)
             .map_err(|_| Error::bad_database("Invalid event content in database"))?,
     }
     .into())

src/client_server/sync.rs

@@ -95,15 +95,7 @@ pub async fn sync_events_route(
 
         // Database queries:
 
-        let current_state = db.rooms.room_state_full(&room_id)?;
+        let current_state_hash = db.rooms.current_state_hash(&room_id)?;
-        let current_members = current_state
-            .iter()
-            .filter(|(key, _)| key.0 == EventType::RoomMember)
-            .map(|(key, value)| (&key.1, value)) // Only keep state key
-            .collect::<Vec<_>>();
-        let encrypted_room = current_state
-            .get(&(EventType::RoomEncryption, "".to_owned()))
-            .is_some();
 
         // These type is Option<Option<_>>. The outer Option is None when there is no event between
         // since and the current room state, meaning there should be no updates.
@@ -115,6 +107,22 @@ pub async fn sync_events_route(
             .as_ref()
             .map(|pdu| db.rooms.pdu_state_hash(&pdu.as_ref().ok()?.0).ok()?);
 
+        let (
+            heroes,
+            joined_member_count,
+            invited_member_count,
+            joined_since_last_sync,
+            state_events,
+        ) = if since_state_hash != None && Some(&current_state_hash) != since_state_hash.as_ref() {
+            let current_state = db.rooms.room_state_full(&room_id)?;
+            let current_members = current_state
+                .iter()
+                .filter(|(key, _)| key.0 == EventType::RoomMember)
+                .map(|(key, value)| (&key.1, value)) // Only keep state key
+                .collect::<Vec<_>>();
+            let encrypted_room = current_state
+                .get(&(EventType::RoomEncryption, "".to_owned()))
+                .is_some();
             let since_state = since_state_hash.as_ref().map(|state_hash| {
                 state_hash
                     .as_ref()
@@ -233,13 +241,6 @@ pub async fn sync_events_route(
                 );
             }
 
-        // Look for device list updates in this room
-        device_list_updates.extend(
-            db.users
-                .keys_changed(&room_id.to_string(), since, None)
-                .filter_map(|r| r.ok()),
-        );
-
             let (joined_member_count, invited_member_count, heroes) = if send_member_count {
                 let joined_member_count = db.rooms.room_members(&room_id).count();
                 let invited_member_count = db.rooms.room_members_invited(&room_id).count();
@@ -262,10 +263,13 @@ pub async fn sync_events_route(
                             >(pdu.content.clone())
                             .expect("Raw::from_value always works")
                             .deserialize()
-                        .map_err(|_| Error::bad_database("Invalid member event in database."))?;
+                            .map_err(|_| {
+                                Error::bad_database("Invalid member event in database.")
+                            })?;
 
                             if let Some(state_key) = &pdu.state_key {
-                            let user_id = UserId::try_from(state_key.clone()).map_err(|_| {
+                                let user_id =
+                                    UserId::try_from(state_key.clone()).map_err(|_| {
                                         Error::bad_database("Invalid UserId in member PDU.")
                                     })?;
 
@@ -305,6 +309,53 @@ pub async fn sync_events_route(
                 (None, None, Vec::new())
             };
 
+            let state_events = if joined_since_last_sync {
+                db.rooms
+                    .room_state_full(&room_id)?
+                    .into_iter()
+                    .map(|(_, pdu)| pdu.to_sync_state_event())
+                    .collect()
+            } else {
+                match since_state {
+                    None => Vec::new(),
+                    Some(Some(since_state)) => current_state
+                        .iter()
+                        .filter(|(key, value)| {
+                            since_state.get(key).map(|e| &e.event_id) != Some(&value.event_id)
+                        })
+                        .filter(|(_, value)| {
+                            !timeline_pdus.iter().any(|(_, timeline_pdu)| {
+                                timeline_pdu.kind == value.kind
+                                    && timeline_pdu.state_key == value.state_key
+                            })
+                        })
+                        .map(|(_, pdu)| pdu.to_sync_state_event())
+                        .collect(),
+                    Some(None) => current_state
+                        .iter()
+                        .map(|(_, pdu)| pdu.to_sync_state_event())
+                        .collect(),
+                }
+            };
+
+            (
+                heroes,
+                joined_member_count,
+                invited_member_count,
+                joined_since_last_sync,
+                state_events,
+            )
+        } else {
+            (Vec::new(), None, None, false, Vec::new())
+        };
+
+        // Look for device list updates in this room
+        device_list_updates.extend(
+            db.users
+                .keys_changed(&room_id.to_string(), since, None)
+                .filter_map(|r| r.ok()),
+        );
+
         let notification_count = if send_notification_counts {
             if let Some(last_read) = db.rooms.edus.private_read_get(&room_id, &sender_user)? {
                 Some(
@@ -385,34 +436,7 @@ pub async fn sync_events_route(
                 events: room_events,
             },
             state: sync_events::State {
-                events: if joined_since_last_sync {
+                events: state_events,
-                    db.rooms
-                        .room_state_full(&room_id)?
-                        .into_iter()
-                        .map(|(_, pdu)| pdu.to_sync_state_event())
-                        .collect()
-                } else {
-                    match since_state {
-                        None => Vec::new(),
-                        Some(Some(since_state)) => current_state
-                            .iter()
-                            .filter(|(key, value)| {
-                                since_state.get(key).map(|e| &e.event_id) != Some(&value.event_id)
-                            })
-                            .filter(|(_, value)| {
-                                !timeline_pdus.iter().any(|(_, timeline_pdu)| {
-                                    timeline_pdu.kind == value.kind
-                                        && timeline_pdu.state_key == value.state_key
-                                })
-                            })
-                            .map(|(_, pdu)| pdu.to_sync_state_event())
-                            .collect(),
-                        Some(None) => current_state
-                            .iter()
-                            .map(|(_, pdu)| pdu.to_sync_state_event())
-                            .collect(),
-                    }
-                },
             },
             ephemeral: sync_events::Ephemeral { events: edus },
         };

src/database.rs

@@ -27,7 +27,7 @@ pub struct Config {
     server_name: Box<ServerName>,
     database_path: String,
     #[serde(default = "default_cache_capacity")]
-    cache_capacity: u64,
+    cache_capacity: u32,
     #[serde(default = "default_max_request_size")]
     max_request_size: u32,
     #[serde(default = "default_max_concurrent_requests")]
@@ -48,7 +48,7 @@ fn true_fn() -> bool {
     true
 }
 
-fn default_cache_capacity() -> u64 {
+fn default_cache_capacity() -> u32 {
     1024 * 1024 * 1024
 }
 
@@ -93,8 +93,7 @@ impl Database {
     pub async fn load_or_create(config: Config) -> Result<Self> {
         let db = sled::Config::default()
             .path(&config.database_path)
-            .cache_capacity(config.cache_capacity)
+            .cache_capacity(config.cache_capacity as u64)
-            .print_profile_on_drop(false)
             .open()?;
 
         info!("Opened sled database at {}", config.database_path);

src/database/rooms.rs

@@ -1010,6 +1010,10 @@ impl Rooms {
                         .filter_map(|r| r.ok())
                         .any(|room_alias| aliases.is_match(room_alias.as_str()))
                 }) || rooms.map_or(false, |rooms| rooms.contains(&room_id.as_str().into()))
+                    || self
+                        .room_members(&room_id)
+                        .filter_map(|r| r.ok())
+                        .any(|member| users.iter().any(|regex| regex.is_match(member.as_str())))
                 {
                     sending.send_pdu_appservice(&appservice.0, &pdu_id)?;
                 }

src/database/sending.rs

@@ -8,7 +8,7 @@ use std::{
 
 use crate::{appservice_server, server_server, utils, Error, PduEvent, Result};
 use federation::transactions::send_transaction_message;
-use log::info;
+use log::{error, info};
 use rocket::futures::stream::{FuturesUnordered, StreamExt};
 use ruma::{
     api::{appservice, federation, OutgoingRequest},
@@ -131,6 +131,7 @@ impl Sending {
                                 };
                                 prefix.extend_from_slice(server.as_bytes());
                                 prefix.push(0xff);
+
                                 last_failed_try.insert(server.clone(), match last_failed_try.get(&server) {
                                     Some(last_failed) => {
                                         (last_failed.0+1, Instant::now())

src/main.rs

@@ -15,6 +15,7 @@ pub use database::Database;
 pub use error::{ConduitLogger, Error, Result};
 pub use pdu::PduEvent;
 pub use rocket::State;
+use ruma::api::client::error::ErrorKind;
 pub use ruma_wrapper::{ConduitResult, Ruma, RumaResponse};
 
 use log::LevelFilter;
@@ -154,7 +155,13 @@ fn setup_rocket() -> rocket::Rocket {
                 server_server::get_profile_information_route,
             ],
         )
-        .register(catchers![not_found_catcher])
+        .register(catchers![
+            not_found_catcher,
+            forbidden_catcher,
+            unknown_token_catcher,
+            missing_token_catcher,
+            bad_json_catcher
+        ])
         .attach(AdHoc::on_attach("Config", |rocket| async {
             let config = rocket
                 .figment()
@@ -186,3 +193,26 @@ async fn main() {
 fn not_found_catcher(_req: &'_ Request<'_>) -> String {
     "404 Not Found".to_owned()
 }
+
+#[catch(580)]
+fn forbidden_catcher() -> Result<()> {
+    Err(Error::BadRequest(ErrorKind::Forbidden, "Forbidden."))
+}
+
+#[catch(581)]
+fn unknown_token_catcher() -> Result<()> {
+    Err(Error::BadRequest(
+        ErrorKind::UnknownToken { soft_logout: false },
+        "Unknown token.",
+    ))
+}
+
+#[catch(582)]
+fn missing_token_catcher() -> Result<()> {
+    Err(Error::BadRequest(ErrorKind::MissingToken, "Missing token."))
+}
+
+#[catch(583)]
+fn bad_json_catcher() -> Result<()> {
+    Err(Error::BadRequest(ErrorKind::BadJson, "Bad json."))
+}

src/ruma_wrapper.rs

@@ -45,7 +45,7 @@ where
         http::request::Request<std::vec::Vec<u8>>,
     >>::Error: std::fmt::Debug,
 {
-    type Error = (); // TODO: Better error handling
+    type Error = ();
     type Owned = Data;
     type Borrowed = Self::Owned;
 
@@ -82,7 +82,9 @@ where
                         registration
                             .get("as_token")
                             .and_then(|as_token| as_token.as_str())
-                            .map_or(false, |as_token| token.as_deref() == Some(as_token))
+                            .map_or(false, |as_token| {
+                                dbg!(token.as_deref()) == dbg!(Some(as_token))
+                            })
                     }) {
                 match T::METADATA.authentication {
                     AuthScheme::AccessToken | AuthScheme::QueryOnlyAccessToken => {
@@ -105,7 +107,8 @@ where
                         );
 
                         if !db.users.exists(&user_id).unwrap() {
-                            return Failure((Status::Unauthorized, ()));
+                            // Forbidden
+                            return Failure((Status::raw(580), ()));
                         }
 
                         // TODO: Check if appservice is allowed to be that user
@@ -119,15 +122,15 @@ where
                     AuthScheme::AccessToken | AuthScheme::QueryOnlyAccessToken => {
                         if let Some(token) = token {
                             match db.users.find_from_token(&token).unwrap() {
-                                // TODO: M_UNKNOWN_TOKEN
+                                // Unknown Token
-                                None => return Failure((Status::Unauthorized, ())),
+                                None => return Failure((Status::raw(581), ())),
                                 Some((user_id, device_id)) => {
                                     (Some(user_id), Some(device_id.into()), false)
                                 }
                             }
                         } else {
-                            // TODO: M_MISSING_TOKEN
+                            // Missing Token
-                            return Failure((Status::Unauthorized, ()));
+                            return Failure((Status::raw(582), ()));
                         }
                     }
                     AuthScheme::ServerSignatures => (None, None, false),
@@ -163,7 +166,7 @@ where
                 }),
                 Err(e) => {
                     warn!("{:?}", e);
-                    Failure((Status::BadRequest, ()))
+                    Failure((Status::raw(583), ()))
                 }
             }
         })

src/server_server.rs

@@ -557,7 +557,7 @@ pub async fn send_transaction_message_route<'a>(
         pdu_id.push(0xff);
         pdu_id.extend_from_slice(&count.to_be_bytes());
 
-        db.rooms.append_to_state(&pdu_id, &pdu, &db.globals)?;
+        let next_room_state = db.rooms.append_to_state(&pdu_id, &pdu, &db.globals)?;
 
         db.rooms.append_pdu(
             &pdu,
@@ -569,6 +569,8 @@ pub async fn send_transaction_message_route<'a>(
             &db.admin,
         )?;
 
+        db.rooms.set_room_state(&room_id, &next_room_state)?;
+
         for appservice in db.appservice.iter_all().filter_map(|r| r.ok()) {
             db.sending.send_pdu_appservice(&appservice.0, &pdu_id)?;
         }