WIP move to ruma master. Waiting for value->canonicaljsonobject helper

Cargo.toml

@@ -14,23 +14,22 @@ edition = "2018"
 [dependencies]
 # Used to handle requests
 # TODO: This can become optional as soon as proper configs are supported
-rocket = { git = "https://github.com/SergioBenitez/Rocket.git", rev = "1f1f44f336e5a172361fc1860461bb03667b1ed2", features = ["tls"] } # Used to handle requests
+#rocket = { git = "https://github.com/SergioBenitez/Rocket.git", rev = "8d779caa22c63b15a6c3ceb75d8f6d4971b2eb67", default-features = false, features = ["tls"] } # Used to handle requests
-#rocket = { git = "https://github.com/timokoesters/Rocket.git", branch = "empty_parameters", default-features = false, features = ["tls"] }
+rocket = { git = "https://github.com/timokoesters/Rocket.git", branch = "empty_parameters", default-features = false, features = ["tls"] }
 
 # Used for matrix spec type definitions and helpers
-ruma = { git = "https://github.com/ruma/ruma", features = ["rand", "client-api", "federation-api", "unstable-pre-spec", "unstable-synapse-quirks", "unstable-exhaustive-types"], rev = "e8882fe8142d7b55ed4c8ccc6150946945f9e237" }
+ruma = { git = "https://github.com/ruma/ruma", features = ["rand", "client-api", "federation-api", "unstable-pre-spec", "unstable-synapse-quirks", "unstable-exhaustive-types"], rev = "f85d98f557b457fdf9846b851055a5fbab2eeee2" }
-# ruma = { git = "https://github.com/DevinR528/ruma", features = ["rand", "client-api", "federation-api", "unstable-exhaustive-types", "unstable-pre-spec", "unstable-synapse-quirks"], branch = "unstable-join" }
+#ruma = { git = "https://github.com/timokoesters/ruma", features = ["rand", "client-api", "federation-api", "unstable-exhaustive-types", "unstable-pre-spec", "unstable-synapse-quirks"], branch = "timo-fed-fixes" }
-# ruma = { path = "../ruma/ruma", features = ["unstable-exhaustive-types", "rand", "client-api", "federation-api", "unstable-pre-spec", "unstable-synapse-quirks"] }
+#ruma = { path = "../ruma/ruma", features = ["unstable-exhaustive-types", "rand", "client-api", "federation-api", "unstable-pre-spec", "unstable-synapse-quirks"] }
 
 # Used when doing state resolution
-# state-res = { git = "https://github.com/timokoesters/state-res", branch = "spec-comp", features = ["unstable-pre-spec"] }
+state-res = { git = "https://github.com/timokoesters/state-res", branch = "spec-comp-ruma-master", features = ["unstable-pre-spec"] }
-state-res = { git = "https://github.com/ruma/state-res", branch = "timo-spec-comp", features = ["unstable-pre-spec", "gen-eventid"] }
+#state-res = { path = "../state-res", features = ["unstable-pre-spec"] }
-# state-res = { path = "../../state-res", features = ["unstable-pre-spec", "gen-eventid"] }
 
-# Used for long polling and federation sender, should be the same as rocket::tokio
+# Used for long polling
-tokio = { version = "0.2.23" }
+tokio = "0.2.22"
 # Used for storing data permanently
-sled = { version = "0.34.6", default-features = false }
+sled = { version = "0.34.4", default-features = false }
 # Used for emitting log entries
 log = "0.4.11"
 # Used for rocket<->ruma conversions
@@ -40,25 +39,25 @@ directories = "3.0.1"
 # Used for number types for ruma
 js_int = "0.1.9"
 # Used for ruma wrapper
-serde_json = { version = "1.0.60", features = ["raw_value"] }
+serde_json = { version = "1.0.57", features = ["raw_value"] }
 # Used for pdu definition
-serde = "1.0.117"
+serde = "1.0.116"
 # Used for secure identifiers
 rand = "0.7.3"
 # Used to hash passwords
-rust-argon2 = "0.8.3"
+rust-argon2 = "0.8.2"
 # Used to send requests
-reqwest = "0.10.9"
+reqwest = "0.10.8"
 # Used for conduit::Error type
-thiserror = "1.0.22"
+thiserror = "1.0.20"
 # Used to generate thumbnails for images
-image = { version = "0.23.12", default-features = false, features = ["jpeg", "png", "gif"] }
+image = { version = "0.23.9", default-features = false, features = ["jpeg", "png", "gif"] }
 # Used to encode server public key
-base64 = "0.13.0"
+base64 = "0.12.3"
 # Used when hashing the state
-ring = "0.16.19"
+ring = "0.16.15"
 # Used when querying the SRV record of other servers
-trust-dns-resolver = "0.19.6"
+trust-dns-resolver = "0.19.5"
 
 [features]
 default = ["conduit_bin"]

src/client_server/account.rs

@@ -2,7 +2,6 @@ use std::{collections::BTreeMap, convert::TryInto};
 
 use super::{State, DEVICE_ID_LENGTH, SESSION_ID_LENGTH, TOKEN_LENGTH};
 use crate::{pdu::PduBuilder, utils, ConduitResult, Database, Error, Ruma};
-use log::info;
 use ruma::{
     api::client::{
         error::ErrorKind,
@@ -15,10 +14,13 @@ use ruma::{
         },
     },
     events::{
-        room::{
+        room::canonical_alias,
-            canonical_alias, guest_access, history_visibility, join_rules, member, message, name,
+        room::guest_access,
-            topic,
+        room::history_visibility,
-        },
+        room::join_rules,
+        room::member,
+        room::name,
+        room::{message, topic},
         EventType,
     },
     RoomAliasId, RoomId, RoomVersionId, UserId,
@@ -239,7 +241,6 @@ pub async fn register_route(
             &room_id,
             &db.globals,
             &db.sending,
-            &db.admin,
             &db.account_data,
         )?;
 
@@ -263,7 +264,6 @@ pub async fn register_route(
             &room_id,
             &db.globals,
             &db.sending,
-            &db.admin,
             &db.account_data,
         )?;
 
@@ -300,7 +300,6 @@ pub async fn register_route(
             &room_id,
             &db.globals,
             &db.sending,
-            &db.admin,
             &db.account_data,
         )?;
 
@@ -320,7 +319,6 @@ pub async fn register_route(
             &room_id,
             &db.globals,
             &db.sending,
-            &db.admin,
             &db.account_data,
         )?;
 
@@ -342,7 +340,6 @@ pub async fn register_route(
             &room_id,
             &db.globals,
             &db.sending,
-            &db.admin,
             &db.account_data,
         )?;
 
@@ -362,7 +359,6 @@ pub async fn register_route(
             &room_id,
             &db.globals,
             &db.sending,
-            &db.admin,
             &db.account_data,
         )?;
 
@@ -384,7 +380,6 @@ pub async fn register_route(
             &room_id,
             &db.globals,
             &db.sending,
-            &db.admin,
             &db.account_data,
         )?;
 
@@ -403,7 +398,6 @@ pub async fn register_route(
             &room_id,
             &db.globals,
             &db.sending,
-            &db.admin,
             &db.account_data,
         )?;
 
@@ -428,7 +422,6 @@ pub async fn register_route(
             &room_id,
             &db.globals,
             &db.sending,
-            &db.admin,
             &db.account_data,
         )?;
 
@@ -454,7 +447,6 @@ pub async fn register_route(
             &room_id,
             &db.globals,
             &db.sending,
-            &db.admin,
             &db.account_data,
         )?;
         db.rooms.build_and_append_pdu(
@@ -476,7 +468,6 @@ pub async fn register_route(
             &room_id,
             &db.globals,
             &db.sending,
-            &db.admin,
             &db.account_data,
         )?;
 
@@ -492,7 +483,6 @@ pub async fn register_route(
                             body: "Thanks for trying out Conduit! This software is still in development, so expect many bugs and missing features. If you have federation enabled, you can join the Conduit chat room by typing <code>/join #conduit:matrix.org</code>. <strong>Important: Please don't join any other Matrix rooms over federation without permission from the room's admins.</strong> Some actions might trigger bugs in other server implementations, breaking the chat for everyone else.".to_owned(),
                         }),
                         relates_to: None,
-                        new_content: None,
                     },
                 ))
                 .expect("event is valid, we just created it"),
@@ -504,13 +494,10 @@ pub async fn register_route(
             &room_id,
             &db.globals,
             &db.sending,
-            &db.admin,
             &db.account_data,
         )?;
     }
 
-    info!("{} registered on this server", user_id);
-
     db.flush().await?;
 
     Ok(register::Response {
@@ -679,7 +666,6 @@ pub async fn deactivate_route(
             &room_id,
             &db.globals,
             &db.sending,
-            &db.admin,
             &db.account_data,
         )?;
     }
@@ -687,8 +673,6 @@ pub async fn deactivate_route(
     // Remove devices and mark account as deactivated
     db.users.deactivate_account(&sender_user)?;
 
-    info!("{} deactivated their account", sender_user);
-
     db.flush().await?;
 
     Ok(deactivate::Response {

src/client_server/backup.rs

@@ -253,17 +253,14 @@ pub async fn get_backup_key_session_route(
 ) -> ConduitResult<get_backup_key_session::Response> {
     let sender_user = body.sender_user.as_ref().expect("user is authenticated");
 
-    let key_data = db
+    if let Some(key_data) =
-        .key_backups
+        db.key_backups
-        .get_session(&sender_user, &body.version, &body.room_id, &body.session_id)?
+            .get_session(&sender_user, &body.version, &body.room_id, &body.session_id)?
-        .ok_or_else(|| {
+    {
-            Error::BadRequest(
+        Ok(get_backup_key_session::Response { key_data }.into())
-                ErrorKind::NotFound,
+    } else {
-                "Backup key not found for this user's session.",
+        Err(Error::BadRequest(ErrorKind::NotFound, "Key not found.").into())
-            )
+    }
-        })?;
-
-    Ok(get_backup_key_session::Response { key_data }.into())
 }
 
 #[cfg_attr(

src/client_server/capabilities.rs

@@ -1,5 +1,8 @@
 use crate::ConduitResult;
-use ruma::{api::client::r0::capabilities::get_capabilities, RoomVersionId};
+use ruma::{
+    api::client::r0::capabilities::get_capabilities::{self, ChangePasswordCapability},
+    RoomVersionId,
+};
 use std::collections::BTreeMap;
 
 #[cfg(feature = "conduit_bin")]
@@ -22,7 +25,7 @@ pub async fn get_capabilities_route() -> ConduitResult<get_capabilities::Respons
 
     Ok(get_capabilities::Response {
         capabilities: get_capabilities::Capabilities {
-            change_password: get_capabilities::ChangePasswordCapability::default(), // enabled by default
+            change_password: ChangePasswordCapability { enabled: true },
             room_versions: get_capabilities::RoomVersionsCapability {
                 default: RoomVersionId::Version6,
                 available,

src/client_server/config.rs

@@ -6,7 +6,7 @@ use ruma::{
         r0::config::{get_global_account_data, set_global_account_data},
     },
     events::{custom::CustomEventContent, BasicEvent},
-    serde::Raw,
+    Raw,
 };
 
 #[cfg(feature = "conduit_bin")]

src/client_server/directory.rs

@@ -1,6 +1,5 @@
 use super::State;
 use crate::{server_server, ConduitResult, Database, Error, Result, Ruma};
-use log::info;
 use ruma::{
     api::{
         client::{
@@ -15,13 +14,14 @@ use ruma::{
         },
         federation,
     },
-    directory::{Filter, IncomingFilter, IncomingRoomNetwork, PublicRoomsChunk, RoomNetwork},
+    directory::Filter,
+    directory::RoomNetwork,
+    directory::{IncomingFilter, IncomingRoomNetwork, PublicRoomsChunk},
     events::{
         room::{avatar, canonical_alias, guest_access, history_visibility, name, topic},
         EventType,
     },
-    serde::Raw,
+    Raw, ServerName,
-    ServerName,
 };
 
 #[cfg(feature = "conduit_bin")]
@@ -82,19 +82,8 @@ pub async fn set_room_visibility_route(
     db: State<'_, Database>,
     body: Ruma<set_room_visibility::Request<'_>>,
 ) -> ConduitResult<set_room_visibility::Response> {
-    let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+    match body.visibility {
-
+        room::Visibility::Public => db.rooms.set_public(&body.room_id, true)?,
-    match &body.visibility {
-        room::Visibility::_Custom(_s) => {
-            return Err(Error::BadRequest(
-                ErrorKind::InvalidParam,
-                "Room visibility type is not supported.",
-            ));
-        }
-        room::Visibility::Public => {
-            db.rooms.set_public(&body.room_id, true)?;
-            info!("{} made {} public", sender_user, body.room_id);
-        }
         room::Visibility::Private => db.rooms.set_public(&body.room_id, false)?,
     }
 
@@ -301,9 +290,7 @@ pub async fn get_public_rooms_filtered_helper(
                                 .url,
                             )
                         })
-                        .transpose()?
+                        .transpose()?,
-                        // url is now an Option<String> so we must flatten
-                        .flatten(),
                 };
                 Ok(chunk)
             })

src/client_server/filter.rs

@@ -1,5 +1,5 @@
 use crate::{utils, ConduitResult};
-use ruma::api::client::r0::filter::{self, create_filter, get_filter};
+use ruma::api::client::r0::filter::{create_filter, get_filter, IncomingFilterDefinition};
 
 #[cfg(feature = "conduit_bin")]
 use rocket::{get, post};
@@ -7,14 +7,7 @@ use rocket::{get, post};
 #[cfg_attr(feature = "conduit_bin", get("/_matrix/client/r0/user/<_>/filter/<_>"))]
 pub async fn get_filter_route() -> ConduitResult<get_filter::Response> {
     // TODO
-    Ok(get_filter::Response::new(filter::IncomingFilterDefinition {
+    Ok(get_filter::Response::new(IncomingFilterDefinition::default()).into())
-        event_fields: None,
-        event_format: filter::EventFormat::default(),
-        account_data: filter::IncomingFilter::default(),
-        room: filter::IncomingRoomFilter::default(),
-        presence: filter::IncomingFilter::default(),
-    })
-    .into())
 }
 
 #[cfg_attr(feature = "conduit_bin", post("/_matrix/client/r0/user/<_>/filter"))]

src/client_server/media.rs

@@ -39,7 +39,7 @@ pub async fn create_content_route(
     db.media.create(
         mxc.clone(),
         &body.filename.as_deref(),
-        &body.content_type.as_deref(),
+        body.content_type.as_deref(),
         &body.file,
     )?;
 
@@ -85,7 +85,7 @@ pub async fn get_content_route(
         db.media.create(
             mxc,
             &get_content_response.content_disposition.as_deref(),
-            &get_content_response.content_type.as_deref(),
+            get_content_response.content_type.as_deref(),
             &get_content_response.file,
         )?;
 
@@ -135,7 +135,7 @@ pub async fn get_content_thumbnail_route(
         db.media.upload_thumbnail(
             mxc,
             &None,
-            &get_thumbnail_response.content_type,
+            get_thumbnail_response.content_type.as_deref(),
             body.width.try_into().expect("all UInts are valid u32s"),
             body.height.try_into().expect("all UInts are valid u32s"),
             &get_thumbnail_response.file,

src/client_server/membership.rs

@@ -18,13 +18,12 @@ use ruma::{
         federation,
     },
     events::{pdu::Pdu, room::member, EventType},
-    serde::{to_canonical_value, CanonicalJsonObject, Raw},
+    EventId, Raw, RoomId, RoomVersionId, ServerName, UserId,
-    EventId, RoomId, RoomVersionId, ServerName, UserId,
 };
 use state_res::StateEvent;
 use std::{
     collections::{BTreeMap, HashMap, HashSet},
-    convert::TryFrom,
+    convert::{TryFrom, TryInto},
     iter,
     sync::Arc,
 };
@@ -126,7 +125,6 @@ pub async fn leave_room_route(
         &body.room_id,
         &db.globals,
         &db.sending,
-        &db.admin,
         &db.account_data,
     )?;
 
@@ -165,7 +163,6 @@ pub async fn invite_user_route(
             &body.room_id,
             &db.globals,
             &db.sending,
-            &db.admin,
             &db.account_data,
         )?;
 
@@ -220,7 +217,6 @@ pub async fn kick_user_route(
         &body.room_id,
         &db.globals,
         &db.sending,
-        &db.admin,
         &db.account_data,
     )?;
 
@@ -279,7 +275,6 @@ pub async fn ban_user_route(
         &body.room_id,
         &db.globals,
         &db.sending,
-        &db.admin,
         &db.account_data,
     )?;
 
@@ -330,7 +325,6 @@ pub async fn unban_user_route(
         &body.room_id,
         &db.globals,
         &db.sending,
-        &db.admin,
         &db.account_data,
     )?;
 
@@ -478,25 +472,34 @@ async fn join_room_by_id_helper(
 
         let (make_join_response, remote_server) = make_join_response_and_server?;
 
-        let mut join_event_stub =
+        let room_version = make_join_response
-            serde_json::from_str::<CanonicalJsonObject>(make_join_response.event.json().get())
+            .room_version
+            .unwrap_or(RoomVersionId::Version2);
+
+        let mut join_event_stub_value =
+            serde_json::from_str::<serde_json::Value>(make_join_response.event.json().get())
                 .map_err(|_| {
                     Error::BadServerResponse("Invalid make_join event json received from server.")
                 })?;
 
+        let join_event_stub =
+            join_event_stub_value
+                .as_object_mut()
+                .ok_or(Error::BadServerResponse(
+                    "Invalid make join event object received from server.",
+                ))?;
+
         join_event_stub.insert(
             "origin".to_owned(),
-            to_canonical_value(db.globals.server_name())
+            db.globals.server_name().to_owned().to_string().into(),
-                .map_err(|_| Error::bad_database("Invalid server name found"))?,
         );
         join_event_stub.insert(
             "origin_server_ts".to_owned(),
-            to_canonical_value(utils::millis_since_unix_epoch())
+            utils::millis_since_unix_epoch().into(),
-                .expect("Timestamp is valid js_int value"),
         );
         join_event_stub.insert(
             "content".to_owned(),
-            to_canonical_value(member::MemberEventContent {
+            serde_json::to_value(member::MemberEventContent {
                 membership: member::MembershipState::Join,
                 displayname: db.users.displayname(&sender_user)?,
                 avatar_url: db.users.avatar_url(&sender_user)?,
@@ -506,34 +509,40 @@ async fn join_room_by_id_helper(
             .expect("event is valid, we just created it"),
         );
 
-        // We don't leave the event id in the pdu because that's only allowed in v1 or v2 rooms
-        join_event_stub.remove("event_id");
-
-        // In order to create a compatible ref hash (EventID) the `hashes` field needs to be present
-        ruma::signatures::hash_and_sign_event(
-            db.globals.server_name().as_str(),
-            db.globals.keypair(),
-            &mut join_event_stub,
-            &RoomVersionId::Version6,
-        )
-        .expect("event is valid, we just created it");
-
         // Generate event id
         let event_id = EventId::try_from(&*format!(
             "${}",
-            ruma::signatures::reference_hash(&join_event_stub, &RoomVersionId::Version6)
+            ruma::signatures::reference_hash(
-                .expect("ruma can calculate reference hashes")
+                &join_event_stub_value
+                    .try_into()
+                    .map_err(|_| Error::BadRequest(
+                        ErrorKind::InvalidParam,
+                        "Join event is not valid json."
+                    ))?,
+                &room_version
+            )
+            .expect("ruma can calculate reference hashes")
         ))
         .expect("ruma's reference hashes are valid event ids");
 
+        // We don't leave the event id into the pdu because that's only allowed in v1 or v2 rooms
+        let join_event_stub = join_event_stub_value.as_object_mut().unwrap();
+        join_event_stub.remove("event_id");
+
+        ruma::signatures::hash_and_sign_event(
+            db.globals.server_name().as_str(),
+            db.globals.keypair(),
+            &mut join_event_stub_value,
+            &room_version,
+        )
+        .expect("event is valid, we just created it");
+
         // Add event_id back
-        join_event_stub.insert(
+        let join_event_stub = join_event_stub_value.as_object_mut().unwrap();
-            "event_id".to_owned(),
+        join_event_stub.insert("event_id".to_owned(), event_id.to_string().into());
-            to_canonical_value(&event_id).expect("EventId is a valid CanonicalJsonValue"),
-        );
 
         // It has enough fields to be called a proper event now
-        let join_event = join_event_stub;
+        let join_event = join_event_stub_value;
 
         let send_join_response = server_server::send_request(
             &db.globals,
@@ -541,26 +550,25 @@ async fn join_room_by_id_helper(
             federation::membership::create_join_event::v2::Request {
                 room_id,
                 event_id: &event_id,
-                pdu: PduEvent::convert_to_outgoing_federation_event(join_event.clone()),
+                pdu_stub: PduEvent::convert_to_outgoing_federation_event(join_event.clone()),
             },
         )
         .await?;
 
-        let add_event_id = |pdu: &Raw<Pdu>| -> Result<(EventId, CanonicalJsonObject)> {
+        let add_event_id = |pdu: &Raw<Pdu>| {
-            let mut value = serde_json::from_str(pdu.json().get())
+            let mut value = serde_json::from_str::<serde_json::Value>(pdu.json().get())
                 .expect("converting raw jsons to values always works");
             let event_id = EventId::try_from(&*format!(
                 "${}",
-                ruma::signatures::reference_hash(&value, &RoomVersionId::Version6)
+                ruma::signatures::reference_hash(&value, &room_version)
                     .expect("ruma can calculate reference hashes")
             ))
             .expect("ruma's reference hashes are valid event ids");
 
-            value.insert(
+            value
-                "event_id".to_owned(),
+                .as_object_mut()
-                to_canonical_value(&event_id)
+                .ok_or_else(|| Error::BadServerResponse("PDU is not an object."))?
-                    .expect("a valid EventId can be converted to CanonicalJsonValue"),
+                .insert("event_id".to_owned(), event_id.to_string().into());
-            );
 
             Ok((event_id, value))
         };
@@ -569,7 +577,7 @@ async fn join_room_by_id_helper(
 
         let state_events = room_state
             .clone()
-            .map(|pdu: Result<(EventId, CanonicalJsonObject)>| Ok(pdu?.0))
+            .map(|pdu: Result<(EventId, serde_json::Value)>| Ok(pdu?.0))
             .chain(iter::once(Ok(event_id.clone()))) // Add join event we just created
             .collect::<Result<HashSet<EventId>>>()?;
 
@@ -584,11 +592,11 @@ async fn join_room_by_id_helper(
             .chain(iter::once(Ok((event_id, join_event)))) // Add join event we just created
             .map(|r| {
                 let (event_id, value) = r?;
-                state_res::StateEvent::from_id_canon_obj(event_id.clone(), value.clone())
+                serde_json::from_value::<StateEvent>(value.clone())
                     .map(|ev| (event_id, Arc::new(ev)))
                     .map_err(|e| {
-                        warn!("{:?}: {}", value, e);
+                        warn!("{}: {}", value, e);
-                        Error::BadServerResponse("Invalid PDU in send_join response.")
+                        Error::BadServerResponse("Invalid PDU bytes in send_join response.")
                     })
             })
             .collect::<Result<BTreeMap<EventId, Arc<StateEvent>>>>()?;
@@ -596,7 +604,7 @@ async fn join_room_by_id_helper(
         let control_events = event_map
             .values()
             .filter(|pdu| pdu.is_power_event())
-            .map(|pdu| pdu.event_id())
+            .map(|pdu| pdu.event_id().clone())
             .collect::<Vec<_>>();
 
         // These events are not guaranteed to be sorted but they are resolved according to spec
@@ -624,9 +632,7 @@ async fn join_room_by_id_helper(
         .expect("iterative auth check failed on resolved events");
 
         // This removes the control events that failed auth, leaving the resolved
-        // to be mainline sorted. In the actual `state_res::StateResolution::resolve`
+        // to be mainline sorted
-        // function both are removed since these are all events we don't know of
-        // we must keep track of everything to add to our DB.
         let events_to_sort = event_map
             .keys()
             .filter(|id| {
@@ -676,12 +682,12 @@ async fn join_room_by_id_helper(
             pdu_id.extend_from_slice(&count.to_be_bytes());
             db.rooms.append_pdu(
                 &PduEvent::from(&**pdu),
-                &utils::to_canonical_object(&**pdu).expect("Pdu is valid canonical object"),
+                &serde_json::to_value(&**pdu).expect("PDU is valid value"),
                 count,
                 pdu_id.clone().into(),
                 &db.globals,
                 &db.account_data,
-                &db.admin,
+                &db.sending,
             )?;
 
             if state_events.contains(ev_id) {
@@ -711,7 +717,6 @@ async fn join_room_by_id_helper(
             &room_id,
             &db.globals,
             &db.sending,
-            &db.admin,
             &db.account_data,
         )?;
     }

src/client_server/message.rs

@@ -67,7 +67,6 @@ pub async fn send_message_event_route(
         &body.room_id,
         &db.globals,
         &db.sending,
-        &db.admin,
         &db.account_data,
     )?;
 

src/client_server/presence.rs

@@ -32,7 +32,7 @@ pub async fn set_presence_route(
                             .try_into()
                             .expect("time is valid"),
                     ),
-                    presence: body.presence.clone(),
+                    presence: body.presence,
                     status_msg: body.status_msg.clone(),
                 },
                 sender: sender_user.clone(),

src/client_server/profile.rs

@@ -8,7 +8,7 @@ use ruma::{
         },
     },
     events::EventType,
-    serde::Raw,
+    Raw,
 };
 
 #[cfg(feature = "conduit_bin")]
@@ -65,7 +65,6 @@ pub async fn set_displayname_route(
             &room_id,
             &db.globals,
             &db.sending,
-            &db.admin,
             &db.account_data,
         )?;
 
@@ -161,7 +160,6 @@ pub async fn set_avatar_url_route(
             &room_id,
             &db.globals,
             &db.sending,
-            &db.admin,
             &db.account_data,
         )?;
 

src/client_server/redact.rs

@@ -33,7 +33,6 @@ pub async fn redact_event_route(
         &body.room_id,
         &db.globals,
         &db.sending,
-        &db.admin,
         &db.account_data,
     )?;
 

src/client_server/room.rs

@@ -1,6 +1,5 @@
 use super::State;
 use crate::{pdu::PduBuilder, ConduitResult, Database, Error, Ruma};
-use log::info;
 use ruma::{
     api::client::{
         error::ErrorKind,
@@ -10,8 +9,7 @@ use ruma::{
         room::{guest_access, history_visibility, join_rules, member, name, topic},
         EventType,
     },
-    serde::Raw,
+    Raw, RoomAliasId, RoomId, RoomVersionId,
-    RoomAliasId, RoomId, RoomVersionId,
 };
 use std::{cmp::max, collections::BTreeMap, convert::TryFrom};
 
@@ -67,7 +65,6 @@ pub async fn create_room_route(
         &room_id,
         &db.globals,
         &db.sending,
-        &db.admin,
         &db.account_data,
     )?;
 
@@ -91,7 +88,6 @@ pub async fn create_room_route(
         &room_id,
         &db.globals,
         &db.sending,
-        &db.admin,
         &db.account_data,
     )?;
 
@@ -135,21 +131,16 @@ pub async fn create_room_route(
         &room_id,
         &db.globals,
         &db.sending,
-        &db.admin,
         &db.account_data,
     )?;
 
     // 4. Events set by preset
 
     // Figure out preset. We need it for preset specific events
-    let preset = body
+    let preset = body.preset.unwrap_or_else(|| match body.visibility {
-        .preset
+        room::Visibility::Private => create_room::RoomPreset::PrivateChat,
-        .clone()
+        room::Visibility::Public => create_room::RoomPreset::PublicChat,
-        .unwrap_or_else(|| match &body.visibility {
+    });
-            room::Visibility::Private => create_room::RoomPreset::PrivateChat,
-            room::Visibility::Public => create_room::RoomPreset::PublicChat,
-            room::Visibility::_Custom(s) => create_room::RoomPreset::_Custom(s.into()),
-        });
 
     // 4.1 Join Rules
     db.rooms.build_and_append_pdu(
@@ -174,7 +165,6 @@ pub async fn create_room_route(
         &room_id,
         &db.globals,
         &db.sending,
-        &db.admin,
         &db.account_data,
     )?;
 
@@ -194,7 +184,6 @@ pub async fn create_room_route(
         &room_id,
         &db.globals,
         &db.sending,
-        &db.admin,
         &db.account_data,
     )?;
 
@@ -222,7 +211,6 @@ pub async fn create_room_route(
         &room_id,
         &db.globals,
         &db.sending,
-        &db.admin,
         &db.account_data,
     )?;
 
@@ -244,7 +232,6 @@ pub async fn create_room_route(
             &room_id,
             &db.globals,
             &db.sending,
-            &db.admin,
             &db.account_data,
         )?;
     }
@@ -268,7 +255,6 @@ pub async fn create_room_route(
             &room_id,
             &db.globals,
             &db.sending,
-            &db.admin,
             &db.account_data,
         )?;
     }
@@ -289,7 +275,6 @@ pub async fn create_room_route(
             &room_id,
             &db.globals,
             &db.sending,
-            &db.admin,
             &db.account_data,
         )?;
     }
@@ -315,7 +300,6 @@ pub async fn create_room_route(
             &room_id,
             &db.globals,
             &db.sending,
-            &db.admin,
             &db.account_data,
         )?;
     }
@@ -329,8 +313,6 @@ pub async fn create_room_route(
         db.rooms.set_public(&room_id, true)?;
     }
 
-    info!("{} created a room", sender_user);
-
     db.flush().await?;
 
     Ok(create_room::Response::new(room_id).into())
@@ -405,7 +387,6 @@ pub async fn upgrade_room_route(
         &body.room_id,
         &db.globals,
         &db.sending,
-        &db.admin,
         &db.account_data,
     )?;
 
@@ -448,7 +429,6 @@ pub async fn upgrade_room_route(
         &replacement_room,
         &db.globals,
         &db.sending,
-        &db.admin,
         &db.account_data,
     )?;
 
@@ -472,7 +452,6 @@ pub async fn upgrade_room_route(
         &replacement_room,
         &db.globals,
         &db.sending,
-        &db.admin,
         &db.account_data,
     )?;
 
@@ -508,7 +487,6 @@ pub async fn upgrade_room_route(
             &replacement_room,
             &db.globals,
             &db.sending,
-            &db.admin,
             &db.account_data,
         )?;
     }
@@ -554,7 +532,6 @@ pub async fn upgrade_room_route(
         &body.room_id,
         &db.globals,
         &db.sending,
-        &db.admin,
         &db.account_data,
     )?;
 

src/client_server/session.rs

@@ -1,6 +1,5 @@
 use super::{State, DEVICE_ID_LENGTH, TOKEN_LENGTH};
 use crate::{utils, ConduitResult, Database, Error, Ruma};
-use log::info;
 use ruma::{
     api::client::{
         error::ErrorKind,
@@ -94,8 +93,6 @@ pub async fn login_route(
         body.initial_device_display_name.clone(),
     )?;
 
-    info!("{} logged in", user_id);
-
     db.flush().await?;
 
     Ok(login::Response {

src/client_server/state.rs

@@ -9,8 +9,9 @@ use ruma::{
         },
     },
     events::{
-        room::history_visibility::{HistoryVisibility, HistoryVisibilityEventContent},
+        room::history_visibility::HistoryVisibility,
-        AnyStateEventContent, EventContent, EventType,
+        room::history_visibility::HistoryVisibilityEventContent, AnyStateEventContent,
+        EventContent, EventType,
     },
     EventId, RoomId, UserId,
 };
@@ -102,11 +103,10 @@ pub async fn get_state_events_route(
 ) -> ConduitResult<get_state_events::Response> {
     let sender_user = body.sender_user.as_ref().expect("user is authenticated");
 
-    #[allow(clippy::blocks_in_if_conditions)]
     // Users not in the room should not be able to access the state unless history_visibility is
     // WorldReadable
-    if !db.rooms.is_joined(sender_user, &body.room_id)?
+    if !db.rooms.is_joined(sender_user, &body.room_id)? {
-        && !matches!(
+        if !matches!(
             db.rooms
                 .room_state_get(&body.room_id, &EventType::RoomHistoryVisibility, "")?
                 .map(|(_, event)| {
@@ -119,12 +119,12 @@ pub async fn get_state_events_route(
                         .map(|e| e.history_visibility)
                 }),
             Some(Ok(HistoryVisibility::WorldReadable))
-        )
+        ) {
-    {
+            return Err(Error::BadRequest(
-        return Err(Error::BadRequest(
+                ErrorKind::Forbidden,
-            ErrorKind::Forbidden,
+                "You don't have permission to view the room state.",
-            "You don't have permission to view the room state.",
+            ));
-        ));
+        }
     }
 
     Ok(get_state_events::Response {
@@ -148,11 +148,10 @@ pub async fn get_state_events_for_key_route(
 ) -> ConduitResult<get_state_events_for_key::Response> {
     let sender_user = body.sender_user.as_ref().expect("user is authenticated");
 
-    #[allow(clippy::blocks_in_if_conditions)]
     // Users not in the room should not be able to access the state unless history_visibility is
     // WorldReadable
-    if !db.rooms.is_joined(sender_user, &body.room_id)?
+    if !db.rooms.is_joined(sender_user, &body.room_id)? {
-        && !matches!(
+        if !matches!(
             db.rooms
                 .room_state_get(&body.room_id, &EventType::RoomHistoryVisibility, "")?
                 .map(|(_, event)| {
@@ -165,12 +164,12 @@ pub async fn get_state_events_for_key_route(
                         .map(|e| e.history_visibility)
                 }),
             Some(Ok(HistoryVisibility::WorldReadable))
-        )
+        ) {
-    {
+            return Err(Error::BadRequest(
-        return Err(Error::BadRequest(
+                ErrorKind::Forbidden,
-            ErrorKind::Forbidden,
+                "You don't have permission to view the room state.",
-            "You don't have permission to view the room state.",
+            ));
-        ));
+        }
     }
 
     let event = db
@@ -199,11 +198,10 @@ pub async fn get_state_events_for_empty_key_route(
 ) -> ConduitResult<get_state_events_for_empty_key::Response> {
     let sender_user = body.sender_user.as_ref().expect("user is authenticated");
 
-    #[allow(clippy::blocks_in_if_conditions)]
     // Users not in the room should not be able to access the state unless history_visibility is
     // WorldReadable
-    if !db.rooms.is_joined(sender_user, &body.room_id)?
+    if !db.rooms.is_joined(sender_user, &body.room_id)? {
-        && !matches!(
+        if !matches!(
             db.rooms
                 .room_state_get(&body.room_id, &EventType::RoomHistoryVisibility, "")?
                 .map(|(_, event)| {
@@ -216,12 +214,12 @@ pub async fn get_state_events_for_empty_key_route(
                         .map(|e| e.history_visibility)
                 }),
             Some(Ok(HistoryVisibility::WorldReadable))
-        )
+        ) {
-    {
+            return Err(Error::BadRequest(
-        return Err(Error::BadRequest(
+                ErrorKind::Forbidden,
-            ErrorKind::Forbidden,
+                "You don't have permission to view the room state.",
-            "You don't have permission to view the room state.",
+            ));
-        ));
+        }
     }
 
     let event = db
@@ -286,7 +284,6 @@ pub async fn send_state_event_for_key_helper(
         &room_id,
         &db.globals,
         &db.sending,
-        &db.admin,
         &db.account_data,
     )?;
 

src/client_server/sync.rs

@@ -3,8 +3,7 @@ use crate::{ConduitResult, Database, Error, Ruma};
 use ruma::{
     api::client::r0::sync::sync_events,
     events::{room::member::MembershipState, AnySyncEphemeralRoomEvent, EventType},
-    serde::Raw,
+    Raw, RoomId, UserId,
-    RoomId, UserId,
 };
 
 #[cfg(feature = "conduit_bin")]

src/database.rs

@@ -1,5 +1,4 @@
 pub mod account_data;
-pub mod admin;
 pub mod globals;
 pub mod key_backups;
 pub mod media;
@@ -13,41 +12,10 @@ use crate::{Error, Result};
 use directories::ProjectDirs;
 use futures::StreamExt;
 use log::info;
-use rocket::futures::{self, channel::mpsc};
+use rocket::{futures, Config};
-use ruma::{DeviceId, ServerName, UserId};
+use ruma::{DeviceId, UserId};
-use serde::Deserialize;
+use std::{convert::TryFrom, fs::remove_dir_all};
-use std::{convert::TryInto, fs::remove_dir_all};
 
-#[derive(Clone, Deserialize)]
-pub struct Config {
-    #[serde(default = "default_server_name")]
-    server_name: Box<ServerName>,
-    database_path: Option<String>,
-    #[serde(default = "default_cache_capacity")]
-    cache_capacity: u64,
-    #[serde(default = "default_max_request_size")]
-    max_request_size: u32,
-    #[serde(default)]
-    registration_disabled: bool,
-    #[serde(default)]
-    encryption_disabled: bool,
-    #[serde(default)]
-    federation_enabled: bool,
-}
-
-fn default_server_name() -> Box<ServerName> {
-    "localhost".try_into().expect("")
-}
-
-fn default_cache_capacity() -> u64 {
-    1024 * 1024 * 1024
-}
-
-fn default_max_request_size() -> u32 {
-    20 * 1024 * 1024 // Default to 20 MB
-}
-
-#[derive(Clone)]
 pub struct Database {
     pub globals: globals::Globals,
     pub users: users::Users,
@@ -58,7 +26,6 @@ pub struct Database {
     pub key_backups: key_backups::KeyBackups,
     pub transaction_ids: transaction_ids::TransactionIds,
     pub sending: sending::Sending,
-    pub admin: admin::Admin,
     pub _db: sled::Db,
 }
 
@@ -66,7 +33,9 @@ impl Database {
     /// Tries to remove the old database but ignores all errors.
     pub fn try_remove(server_name: &str) -> Result<()> {
         let mut path = ProjectDirs::from("xyz", "koesters", "conduit")
-            .ok_or_else(|| Error::bad_config("The OS didn't return a valid home directory path."))?
+            .ok_or(Error::BadConfig(
+                "The OS didn't return a valid home directory path.",
+            ))?
             .data_dir()
             .to_path_buf();
         path.push(server_name);
@@ -76,36 +45,42 @@ impl Database {
     }
 
     /// Load an existing database or create a new one.
-    pub fn load_or_create(config: Config) -> Result<Self> {
+    pub fn load_or_create(config: &Config) -> Result<Self> {
+        let server_name = config.get_str("server_name").unwrap_or("localhost");
+
         let path = config
-            .database_path
+            .get_str("database_path")
-            .clone()
+            .map(|x| Ok::<_, Error>(x.to_owned()))
-            .map(Ok::<_, Error>)
+            .unwrap_or_else(|_| {
-            .unwrap_or_else(|| {
                 let path = ProjectDirs::from("xyz", "koesters", "conduit")
-                    .ok_or_else(|| {
+                    .ok_or(Error::BadConfig(
-                        Error::bad_config("The OS didn't return a valid home directory path.")
+                        "The OS didn't return a valid home directory path.",
-                    })?
+                    ))?
                     .data_dir()
-                    .join(config.server_name.as_str());
+                    .join(server_name);
 
                 Ok(path
                     .to_str()
-                    .ok_or_else(|| Error::bad_config("Database path contains invalid unicode."))?
+                    .ok_or(Error::BadConfig("Database path contains invalid unicode."))?
                     .to_owned())
             })?;
 
         let db = sled::Config::default()
             .path(&path)
-            .cache_capacity(config.cache_capacity)
+            .cache_capacity(
-            .print_profile_on_drop(true)
+                u64::try_from(
+                    config
+                        .get_int("cache_capacity")
+                        .unwrap_or(1024 * 1024 * 1024),
+                )
+                .map_err(|_| Error::BadConfig("Cache capacity needs to be a u64."))?,
+            )
+            .print_profile_on_drop(false)
             .open()?;
 
         info!("Opened sled database at {}", path);
 
-        let (admin_sender, admin_receiver) = mpsc::unbounded();
+        Ok(Self {
-
-        let db = Self {
             globals: globals::Globals::load(db.open_tree("global")?, config)?,
             users: users::Users {
                 userid_password: db.open_tree("userid_password")?,
@@ -177,15 +152,8 @@ impl Database {
                 servernamepduids: db.open_tree("servernamepduids")?,
                 servercurrentpdus: db.open_tree("servercurrentpdus")?,
             },
-            admin: admin::Admin {
-                sender: admin_sender,
-            },
             _db: db,
-        };
+        })
-
-        db.admin.start_handler(db.clone(), admin_receiver);
-
-        Ok(db)
     }
 
     pub async fn watch(&self, user_id: &UserId, device_id: &DeviceId) {

src/database/account_data.rs

@@ -2,14 +2,12 @@ use crate::{utils, Error, Result};
 use ruma::{
     api::client::error::ErrorKind,
     events::{AnyEvent as EduEvent, EventType},
-    serde::Raw,
+    Raw, RoomId, UserId,
-    RoomId, UserId,
 };
 use serde::{de::DeserializeOwned, Serialize};
 use sled::IVec;
 use std::{collections::HashMap, convert::TryFrom};
 
-#[derive(Clone)]
 pub struct AccountData {
     pub(super) roomuserdataid_accountdata: sled::Tree, // RoomUserDataId = Room + User + Count + Type
 }

src/database/admin.rs

@@ -1,80 +0,0 @@
-use std::convert::{TryFrom, TryInto};
-
-use crate::pdu::PduBuilder;
-use log::warn;
-use rocket::futures::{channel::mpsc, stream::StreamExt};
-use ruma::{
-    events::{room::message, EventType},
-    UserId,
-};
-use tokio::select;
-
-pub enum AdminCommand {
-    SendTextMessage(message::TextMessageEventContent),
-}
-
-#[derive(Clone)]
-pub struct Admin {
-    pub sender: mpsc::UnboundedSender<AdminCommand>,
-}
-
-impl Admin {
-    pub fn start_handler(
-        &self,
-        db: super::Database,
-        mut receiver: mpsc::UnboundedReceiver<AdminCommand>,
-    ) {
-        tokio::spawn(async move {
-            // TODO: Use futures when we have long admin commands
-            //let mut futures = FuturesUnordered::new();
-
-            let conduit_user = UserId::try_from(format!("@conduit:{}", db.globals.server_name()))
-                .expect("@conduit:server_name is valid");
-
-            let conduit_room = db
-                .rooms
-                .id_from_alias(
-                    &format!("#admins:{}", db.globals.server_name())
-                        .try_into()
-                        .expect("#admins:server_name is a valid room alias"),
-                )
-                .unwrap();
-
-            if conduit_room.is_none() {
-                warn!("Conduit instance does not have an #admins room. Logging to that room will not work.");
-            }
-
-            loop {
-                select! {
-                    Some(event) = receiver.next() => {
-                        match event {
-                            AdminCommand::SendTextMessage(message) => {
-                                if let Some(conduit_room) = &conduit_room {
-                                    db.rooms.build_and_append_pdu(
-                                        PduBuilder {
-                                            event_type: EventType::RoomMessage,
-                                            content: serde_json::to_value(message).expect("event is valid, we just created it"),
-                                            unsigned: None,
-                                            state_key: None,
-                                            redacts: None,
-                                        },
-                                        &conduit_user,
-                                        &conduit_room,
-                                        &db.globals,
-                                        &db.sending,
-                                        &db.admin,
-                                        &db.account_data,
-                                    ).unwrap();
-                                }
-                            }
-                        }
-                    }
-                }
-            }
-        });
-    }
-
-    pub fn send(&self, command: AdminCommand) {
-        self.sender.unbounded_send(command).unwrap()
-    }
-}

src/database/globals.rs

@@ -1,7 +1,7 @@
-use crate::{database::Config, utils, Error, Result};
+use crate::{utils, Error, Result};
 use log::error;
 use ruma::ServerName;
-use std::sync::Arc;
+use std::{convert::TryInto, sync::Arc};
 
 pub const COUNTER: &str = "c";
 
@@ -10,11 +10,15 @@ pub struct Globals {
     pub(super) globals: sled::Tree,
     keypair: Arc<ruma::signatures::Ed25519KeyPair>,
     reqwest_client: reqwest::Client,
-    config: Config,
+    server_name: Box<ServerName>,
+    max_request_size: u32,
+    registration_disabled: bool,
+    encryption_disabled: bool,
+    federation_enabled: bool,
 }
 
 impl Globals {
-    pub fn load(globals: sled::Tree, config: Config) -> Result<Self> {
+    pub fn load(globals: sled::Tree, config: &rocket::Config) -> Result<Self> {
         let bytes = &*globals
             .update_and_fetch("keypair", utils::generate_keypair)?
             .expect("utils::generate_keypair always returns Some");
@@ -53,7 +57,20 @@ impl Globals {
             globals,
             keypair: Arc::new(keypair),
             reqwest_client: reqwest::Client::new(),
-            config,
+            server_name: config
+                .get_str("server_name")
+                .unwrap_or("localhost")
+                .to_string()
+                .try_into()
+                .map_err(|_| Error::BadConfig("Invalid server_name."))?,
+            max_request_size: config
+                .get_int("max_request_size")
+                .unwrap_or(20 * 1024 * 1024) // Default to 20 MB
+                .try_into()
+                .map_err(|_| Error::BadConfig("Invalid max_request_size."))?,
+            registration_disabled: config.get_bool("registration_disabled").unwrap_or(false),
+            encryption_disabled: config.get_bool("encryption_disabled").unwrap_or(false),
+            federation_enabled: config.get_bool("federation_enabled").unwrap_or(false),
         })
     }
 
@@ -85,22 +102,22 @@ impl Globals {
     }
 
     pub fn server_name(&self) -> &ServerName {
-        self.config.server_name.as_ref()
+        self.server_name.as_ref()
     }
 
     pub fn max_request_size(&self) -> u32 {
-        self.config.max_request_size
+        self.max_request_size
     }
 
     pub fn registration_disabled(&self) -> bool {
-        self.config.registration_disabled
+        self.registration_disabled
     }
 
     pub fn encryption_disabled(&self) -> bool {
-        self.config.encryption_disabled
+        self.encryption_disabled
     }
 
     pub fn federation_enabled(&self) -> bool {
-        self.config.federation_enabled
+        self.federation_enabled
     }
 }

src/database/key_backups.rs

@@ -8,7 +8,6 @@ use ruma::{
 };
 use std::{collections::BTreeMap, convert::TryFrom};
 
-#[derive(Clone)]
 pub struct KeyBackups {
     pub(super) backupid_algorithm: sled::Tree, // BackupId = UserId + Version(Count)
     pub(super) backupid_etag: sled::Tree,      // BackupId = UserId + Version(Count)

src/database/media.rs

@@ -9,7 +9,6 @@ pub struct FileMeta {
     pub file: Vec<u8>,
 }
 
-#[derive(Clone)]
 pub struct Media {
     pub(super) mediaid_file: sled::Tree, // MediaId = MXC + WidthHeight + Filename + ContentType
 }
@@ -20,7 +19,7 @@ impl Media {
         &self,
         mxc: String,
         filename: &Option<&str>,
-        content_type: &Option<&str>,
+        content_type: Option<&str>,
         file: &[u8],
     ) -> Result<()> {
         let mut key = mxc.as_bytes().to_vec();
@@ -30,12 +29,7 @@ impl Media {
         key.push(0xff);
         key.extend_from_slice(filename.as_ref().map(|f| f.as_bytes()).unwrap_or_default());
         key.push(0xff);
-        key.extend_from_slice(
+        key.extend_from_slice(content_type.map(|c| c.as_bytes()).unwrap_or(&[]));
-            content_type
-                .as_ref()
-                .map(|c| c.as_bytes())
-                .unwrap_or_default(),
-        );
 
         self.mediaid_file.insert(key, file)?;
 
@@ -47,7 +41,7 @@ impl Media {
         &self,
         mxc: String,
         filename: &Option<String>,
-        content_type: &Option<String>,
+        content_type: Option<&str>,
         width: u32,
         height: u32,
         file: &[u8],
@@ -59,12 +53,7 @@ impl Media {
         key.push(0xff);
         key.extend_from_slice(filename.as_ref().map(|f| f.as_bytes()).unwrap_or_default());
         key.push(0xff);
-        key.extend_from_slice(
+        key.extend_from_slice(content_type.map(|c| c.as_bytes()).unwrap_or(&[]));
-            content_type
-                .as_ref()
-                .map(|c| c.as_bytes())
-                .unwrap_or_default(),
-        );
 
         self.mediaid_file.insert(key, file)?;
 
@@ -85,12 +74,15 @@ impl Media {
 
             let content_type = parts
                 .next()
-                .map(|bytes| {
+                .ok_or_else(|| Error::bad_database("Media ID in db is invalid."))?;
-                    Ok::<_, Error>(utils::string_from_bytes(bytes).map_err(|_| {
+
-                        Error::bad_database("Content type in mediaid_file is invalid unicode.")
+            let content_type_parsed = if content_type.is_empty() {
-                    })?)
+                None
-                })
+            } else {
-                .transpose()?;
+                Some(utils::string_from_bytes(content_type).map_err(|_| {
+                    Error::bad_database("Content type in mediaid_file is invalid unicode.")
+                })?)
+            };
 
             let filename_bytes = parts
                 .next()
@@ -106,7 +98,7 @@ impl Media {
 
             Ok(Some(FileMeta {
                 filename,
-                content_type,
+                content_type: content_type_parsed,
                 file: file.to_vec(),
             }))
         } else {
@@ -162,12 +154,15 @@ impl Media {
 
             let content_type = parts
                 .next()
-                .map(|bytes| {
+                .ok_or_else(|| Error::bad_database("Media ID in db is invalid."))?;
-                    Ok::<_, Error>(utils::string_from_bytes(bytes).map_err(|_| {
+
-                        Error::bad_database("Content type in mediaid_file is invalid unicode.")
+            let content_type_parsed = if content_type.is_empty() {
-                    })?)
+                None
-                })
+            } else {
-                .transpose()?;
+                Some(utils::string_from_bytes(content_type).map_err(|_| {
+                    Error::bad_database("Content type in mediaid_file is invalid unicode.")
+                })?)
+            };
 
             let filename_bytes = parts
                 .next()
@@ -184,7 +179,7 @@ impl Media {
 
             Ok(Some(FileMeta {
                 filename,
-                content_type,
+                content_type: content_type_parsed,
                 file: file.to_vec(),
             }))
         } else if let Some(r) = self.mediaid_file.scan_prefix(&original_prefix).next() {
@@ -195,12 +190,15 @@ impl Media {
 
             let content_type = parts
                 .next()
-                .map(|bytes| {
+                .ok_or_else(|| Error::bad_database("Media ID in db is invalid."))?;
-                    Ok::<_, Error>(utils::string_from_bytes(bytes).map_err(|_| {
+
-                        Error::bad_database("Content type in mediaid_file is invalid unicode.")
+            let content_type_parsed = if content_type.is_empty() {
-                    })?)
+                None
-                })
+            } else {
-                .transpose()?;
+                Some(utils::string_from_bytes(content_type).map_err(|_| {
+                    Error::bad_database("Content type in mediaid_file is invalid unicode.")
+                })?)
+            };
 
             let filename_bytes = parts
                 .next()
@@ -286,7 +284,7 @@ impl Media {
 
                 Ok(Some(FileMeta {
                     filename,
-                    content_type,
+                    content_type: content_type_parsed,
                     file: thumbnail_bytes.to_vec(),
                 }))
             } else {

src/database/rooms.rs

@@ -15,8 +15,7 @@ use ruma::{
         },
         EventType,
     },
-    serde::{to_canonical_value, CanonicalJsonObject, Raw},
+    EventId, Raw, RoomAliasId, RoomId, RoomVersionId, ServerName, UserId,
-    EventId, RoomAliasId, RoomId, RoomVersionId, ServerName, UserId,
 };
 use sled::IVec;
 use state_res::{event_auth, Error as StateError, Requester, StateEvent, StateMap, StateStore};
@@ -28,24 +27,12 @@ use std::{
     sync::Arc,
 };
 
-use super::admin::AdminCommand;
-
 /// The unique identifier of each state group.
 ///
 /// This is created when a state group is added to the database by
 /// hashing the entire state.
 pub type StateHashId = IVec;
 
-/// An enum that represents the two valid states when searching
-/// for an events "parent".
-///
-/// An events parent is any event we are aware of that is part of
-/// the events `prev_events` array.
-pub(crate) enum ClosestParent {
-    Append,
-    Insert(u64),
-}
-
 #[derive(Clone)]
 pub struct Rooms {
     pub edus: edus::RoomEdus,
@@ -85,10 +72,7 @@ impl StateStore for Rooms {
             .get_pdu_id(event_id)
             .map_err(StateError::custom)?
             .ok_or_else(|| {
-                StateError::NotFound(format!(
+                StateError::NotFound("PDU via room_id and event_id not found in the db.".into())
-                    "PDU via room_id and event_id not found in the db: {}",
-                    event_id.as_str()
-                ))
             })?;
 
         serde_json::from_slice(
@@ -102,7 +86,7 @@ impl StateStore for Rooms {
         .and_then(|pdu: StateEvent| {
             // conduit's PDU's always contain a room_id but some
             // of ruma's do not so this must be an Option
-            if pdu.room_id() == room_id {
+            if pdu.room_id() == Some(room_id) {
                 Ok(Arc::new(pdu))
             } else {
                 Err(StateError::NotFound(
@@ -210,7 +194,7 @@ impl Rooms {
         Ok(self.pduid_statehash.get(pdu_id)?)
     }
 
-    /// Returns the last state hash key added to the db for the given room.
+    /// Returns the last state hash key added to the db.
     pub fn current_state_hash(&self, room_id: &RoomId) -> Result<Option<StateHashId>> {
         Ok(self.roomid_statehash.get(room_id.as_bytes())?)
     }
@@ -263,9 +247,7 @@ impl Rooms {
             .is_some())
     }
 
-    /// Force the creation of a new StateHash and insert it into the db.
+    /// Returns the full room state.
-    ///
-    /// Whatever `state` is supplied to `force_state` __is__ the current room state snapshot.
     pub fn force_state(
         &self,
         room_id: &RoomId,
@@ -278,7 +260,7 @@ impl Rooms {
 
         for ((event_type, state_key), pdu_id) in state {
             let mut state_id = prefix.clone();
-            state_id.extend_from_slice(&event_type.as_ref().as_bytes());
+            state_id.extend_from_slice(&event_type.as_str().as_bytes());
             state_id.push(0xff);
             state_id.extend_from_slice(&state_key.as_bytes());
             self.stateid_pduid.insert(state_id, pdu_id)?;
@@ -385,8 +367,8 @@ impl Rooms {
         })
     }
 
-    /// Returns the pdu as a `BTreeMap<String, CanonicalJsonValue>`.
+    /// Returns the pdu.
-    pub fn get_pdu_json_from_id(&self, pdu_id: &[u8]) -> Result<Option<CanonicalJsonObject>> {
+    pub fn get_pdu_json_from_id(&self, pdu_id: &[u8]) -> Result<Option<serde_json::Value>> {
         self.pduid_pdu.get(pdu_id)?.map_or(Ok(None), |pdu| {
             Ok(Some(
                 serde_json::from_slice(&pdu)
@@ -411,54 +393,6 @@ impl Rooms {
         }
     }
 
-    /// Recursively search for a PDU from our DB that is also in the
-    /// `prev_events` field of the incoming PDU.
-    ///
-    /// First we check if the last PDU inserted to the given room is a parent
-    /// if not we recursively check older `prev_events` to insert the incoming
-    /// event after.
-    pub(crate) fn get_latest_pduid_before(
-        &self,
-        room: &RoomId,
-        incoming_prev_ids: &[EventId],
-        their_state: &BTreeMap<EventId, Arc<StateEvent>>,
-    ) -> Result<Option<ClosestParent>> {
-        match self.pduid_pdu.scan_prefix(room.as_bytes()).last() {
-            Some(Ok(val))
-                if incoming_prev_ids.contains(
-                    &serde_json::from_slice::<PduEvent>(&val.1)
-                        .map_err(|_| {
-                            Error::bad_database("last DB entry contains invalid PDU bytes")
-                        })?
-                        .event_id,
-                ) =>
-            {
-                Ok(Some(ClosestParent::Append))
-            }
-            _ => {
-                let mut prev_ids = incoming_prev_ids.to_vec();
-                while let Some(id) = prev_ids.pop() {
-                    match self.get_pdu_id(&id)? {
-                        Some(pdu_id) => {
-                            return Ok(Some(ClosestParent::Insert(self.pdu_count(&pdu_id)?)));
-                        }
-                        None => {
-                            prev_ids.extend(their_state.get(&id).map_or(
-                                Err(Error::BadServerResponse(
-                                    "Failed to find previous event for PDU in state",
-                                )),
-                                // `prev_event_ids` will return an empty Vec instead of failing
-                                // so it works perfect for our use here
-                                |pdu| Ok(pdu.prev_event_ids()),
-                            )?);
-                        }
-                    }
-                }
-                Ok(None)
-            }
-        }
-    }
-
     /// Returns the leaf pdus of a room.
     pub fn get_pdu_leaves(&self, room_id: &RoomId) -> Result<Vec<EventId>> {
         let mut prefix = room_id.as_bytes().to_vec();
@@ -501,18 +435,15 @@ impl Rooms {
     }
 
     /// Creates a new persisted data unit and adds it to a room.
-    ///
-    /// By this point the incoming event should be fully authenticated, no auth happens
-    /// in `append_pdu`.
     pub fn append_pdu(
         &self,
         pdu: &PduEvent,
-        pdu_json: &CanonicalJsonObject,
+        pdu_json: &serde_json::Value,
         count: u64,
         pdu_id: IVec,
         globals: &super::globals::Globals,
         account_data: &super::account_data::AccountData,
-        admin: &super::admin::Admin,
+        sending: &super::sending::Sending,
     ) -> Result<()> {
         self.replace_pdu_leaves(&pdu.room_id, &pdu.event_id)?;
 
@@ -521,11 +452,7 @@ impl Rooms {
         self.edus
             .private_read_set(&pdu.room_id, &pdu.sender, count, &globals)?;
 
-        self.pduid_pdu.insert(
+        self.pduid_pdu.insert(&pdu_id, &*pdu_json.to_string())?;
-            &pdu_id,
-            &*serde_json::to_string(pdu_json)
-                .expect("CanonicalJsonObject is always a valid String"),
-        )?;
 
         self.eventid_pduid
             .insert(pdu.event_id.as_bytes(), &*pdu_id)?;
@@ -587,14 +514,28 @@ impl Rooms {
                         if let Some(command) = parts.next() {
                             let args = parts.collect::<Vec<_>>();
 
-                            admin.send(AdminCommand::SendTextMessage(
+                            self.build_and_append_pdu(
-                                message::TextMessageEventContent {
+                                PduBuilder {
-                                    body: format!("Command: {}, Args: {:?}", command, args),
+                                    event_type: EventType::RoomMessage,
-                                    formatted: None,
+                                    content: serde_json::to_value(
-                                    relates_to: None,
+                                        message::TextMessageEventContent {
-                                    new_content: None,
+                                            body: format!("Command: {}, Args: {:?}", command, args),
+                                            formatted: None,
+                                            relates_to: None,
+                                        },
+                                    )
+                                    .expect("event is valid, we just created it"),
+                                    unsigned: None,
+                                    state_key: None,
+                                    redacts: None,
                                 },
-                            ));
+                                &UserId::try_from(format!("@conduit:{}", globals.server_name()))
+                                    .expect("@conduit:server_name is valid"),
+                                &pdu.room_id,
+                                &globals,
+                                &sending,
+                                &account_data,
+                            )?;
                         }
                     }
                 }
@@ -625,7 +566,6 @@ impl Rooms {
                 self.stateid_pduid
                     .scan_prefix(&prefix)
                     .filter_map(|pdu| pdu.map_err(|e| error!("{}", e)).ok())
-                    // Chop the old state_hash out leaving behind the (EventType, StateKey)
                     .map(|(k, v)| (k.subslice(prefix.len(), k.len() - prefix.len()), v))
                     .collect::<HashMap<IVec, IVec>>()
             } else {
@@ -634,7 +574,7 @@ impl Rooms {
 
         if let Some(state_key) = &new_pdu.state_key {
             let mut new_state = old_state;
-            let mut pdu_key = new_pdu.kind.as_ref().as_bytes().to_vec();
+            let mut pdu_key = new_pdu.kind.as_str().as_bytes().to_vec();
             pdu_key.push(0xff);
             pdu_key.extend_from_slice(state_key.as_bytes());
             new_state.insert(pdu_key.into(), new_pdu_id.into());
@@ -665,7 +605,6 @@ impl Rooms {
     }
 
     /// Creates a new persisted data unit and adds it to a room.
-    #[allow(clippy::too_many_arguments)]
     pub fn build_and_append_pdu(
         &self,
         pdu_builder: PduBuilder,
@@ -673,7 +612,6 @@ impl Rooms {
         room_id: &RoomId,
         globals: &super::globals::Globals,
         sending: &super::sending::Sending,
-        admin: &super::admin::Admin,
         account_data: &super::account_data::AccountData,
     ) -> Result<EventId> {
         let PduBuilder {
@@ -761,7 +699,7 @@ impl Rooms {
                 }
                 EventType::RoomMember => {
                     let prev_event = self
-                        .get_pdu(prev_events.get(0).ok_or(Error::BadRequest(
+                        .get_pdu(prev_events.iter().next().ok_or(Error::BadRequest(
                             ErrorKind::Unknown,
                             "Membership can't be the first event",
                         ))?)?
@@ -777,7 +715,7 @@ impl Rooms {
                             sender: &sender,
                         },
                         prev_event,
-                        None, // TODO: third party invite
+                        None,
                         &auth_events
                             .iter()
                             .map(|((ty, key), pdu)| {
@@ -835,7 +773,7 @@ impl Rooms {
         }
 
         let mut pdu = PduEvent {
-            event_id: ruma::event_id!("$thiswillbefilledinlater"),
+            event_id: EventId::try_from("$thiswillbefilledinlater").expect("we know this is valid"),
             room_id: room_id.clone(),
             sender: sender.clone(),
             origin_server_ts: utils::millis_since_unix_epoch()
@@ -861,17 +799,17 @@ impl Rooms {
         };
 
         // Hash and sign
-        let mut pdu_json =
+        let mut pdu_json = serde_json::to_value(&pdu).expect("event is valid, we just created it");
-            utils::to_canonical_object(&pdu).expect("event is valid, we just created it");
+        pdu_json
-
+            .as_object_mut()
-        pdu_json.remove("event_id");
+            .expect("json is object")
+            .remove("event_id");
 
         // Add origin because synapse likes that (and it's required in the spec)
-        pdu_json.insert(
+        pdu_json
-            "origin".to_owned(),
+            .as_object_mut()
-            to_canonical_value(globals.server_name())
+            .expect("json is object")
-                .expect("server name is a valid CanonicalJsonValue"),
+            .insert("origin".to_owned(), globals.server_name().as_str().into());
-        );
 
         ruma::signatures::hash_and_sign_event(
             globals.server_name().as_str(),
@@ -889,10 +827,10 @@ impl Rooms {
         ))
         .expect("ruma's reference hashes are valid event ids");
 
-        pdu_json.insert(
+        pdu_json
-            "event_id".to_owned(),
+            .as_object_mut()
-            to_canonical_value(&pdu.event_id).expect("EventId is a valid CanonicalJsonValue"),
+            .expect("json is object")
-        );
+            .insert("event_id".to_owned(), pdu.event_id.to_string().into());
 
         // Increment the last index and use that
         // This is also the next_batch/since value
@@ -912,7 +850,7 @@ impl Rooms {
             pdu_id.clone().into(),
             globals,
             account_data,
-            admin,
+            sending,
         )?;
 
         for server in self
@@ -920,7 +858,7 @@ impl Rooms {
             .filter_map(|r| r.ok())
             .filter(|server| &**server != globals.server_name())
         {
-            sending.send_pdu(&server, &pdu_id)?;
+            sending.send_pdu(server, &pdu_id)?;
         }
 
         Ok(pdu.event_id)

src/database/rooms/edus.rs

@@ -6,8 +6,7 @@ use ruma::{
         AnyEvent as EduEvent, SyncEphemeralRoomEvent,
     },
     presence::PresenceState,
-    serde::Raw,
+    Raw, RoomId, UserId,
-    RoomId, UserId,
 };
 use std::{
     collections::HashMap,

src/database/sending.rs

@@ -2,13 +2,12 @@ use std::{collections::HashMap, convert::TryFrom, time::SystemTime};
 
 use crate::{server_server, utils, Error, PduEvent, Result};
 use federation::transactions::send_transaction_message;
-use log::{debug, warn};
+use log::debug;
 use rocket::futures::stream::{FuturesUnordered, StreamExt};
 use ruma::{api::federation, ServerName};
 use sled::IVec;
 use tokio::select;
 
-#[derive(Clone)]
 pub struct Sending {
     /// The state for a given state hash.
     pub(super) servernamepduids: sled::Tree, // ServernamePduId = ServerName + PduId
@@ -55,14 +54,13 @@ impl Sending {
                     ))
                 })
                 .filter_map(|r| r.ok())
-                .filter(|(_, pdu)| !pdu.is_empty()) // Skip reservation key
-                .take(50)
-            // This should not contain more than 50 anyway
             {
-                current_transactions
+                if !pdu.is_empty() {
-                    .entry(server)
+                    current_transactions
-                    .or_insert_with(Vec::new)
+                        .entry(server)
-                    .push(pdu);
+                        .or_insert_with(Vec::new)
+                        .push(pdu);
+                }
             }
 
             for (server, pdus) in current_transactions {
@@ -73,7 +71,7 @@ impl Sending {
             loop {
                 select! {
                     Some(server) = futures.next() => {
-                        debug!("sending response: {:?}", &server);
+                        debug!("response: {:?}", &server);
                         match server {
                             Ok((server, _response)) => {
                                 let mut prefix = server.as_bytes().to_vec();
@@ -97,9 +95,7 @@ impl Sending {
                                     .filter_map(|r| r.ok())
                                     .map(|k| {
                                         k.subslice(prefix.len(), k.len() - prefix.len())
-                                    })
+                                    }).collect::<Vec<_>>();
-                                    .take(50)
-                                    .collect::<Vec<_>>();
 
                                 if !new_pdus.is_empty() {
                                     for pdu_id in &new_pdus {
@@ -112,11 +108,9 @@ impl Sending {
                                     futures.push(Self::handle_event(server, new_pdus, &globals, &rooms));
                                 } else {
                                     servercurrentpdus.remove(&prefix).unwrap();
-                                    // servercurrentpdus with the prefix should be empty now
                                 }
                             }
-                            Err((server, e)) => {
+                            Err((_server, _e)) => {
-                                warn!("Couldn't send transaction to {}: {}", server, e)
                                 // TODO: exponential backoff
                             }
                         };
@@ -132,7 +126,7 @@ impl Sending {
                                         .expect("splitn will always return 1 or more elements"),
                                 )
                                 .map_err(|_| Error::bad_database("ServerName in servernamepduid bytes are invalid."))
-                                .and_then(|server_str| Box::<ServerName>::try_from(server_str)
+                                .and_then(|server_str|Box::<ServerName>::try_from(server_str)
                                     .map_err(|_| Error::bad_database("ServerName in servernamepduid is invalid.")))
                                 .ok()
                                 .and_then(|server| parts
@@ -163,7 +157,7 @@ impl Sending {
         });
     }
 
-    pub fn send_pdu(&self, server: &ServerName, pdu_id: &[u8]) -> Result<()> {
+    pub fn send_pdu(&self, server: Box<ServerName>, pdu_id: &[u8]) -> Result<()> {
         let mut key = server.as_bytes().to_vec();
         key.push(0xff);
         key.extend_from_slice(pdu_id);
@@ -184,27 +178,17 @@ impl Sending {
         let pdu_jsons = pdu_ids
             .iter()
             .map(|pdu_id| {
-                Ok::<_, (Box<ServerName>, Error)>(
+                Ok::<_, (Box<ServerName>, Error)>(PduEvent::convert_to_outgoing_federation_event(
-                    // TODO: check room version and remove event_id if needed
+                    rooms
-                    serde_json::from_str(
+                        .get_pdu_json_from_id(pdu_id)
-                        PduEvent::convert_to_outgoing_federation_event(
+                        .map_err(|e| (server.clone(), e))?
-                            rooms
+                        .ok_or_else(|| {
-                                .get_pdu_json_from_id(pdu_id)
+                            (
-                                .map_err(|e| (server.clone(), e))?
+                                server.clone(),
-                                .ok_or_else(|| {
+                                Error::bad_database("Event in servernamepduids not found in db."),
-                                    (
+                            )
-                                        server.clone(),
+                        })?,
-                                        Error::bad_database(
+                ))
-                                            "Event in servernamepduids not found in db.",
-                                        ),
-                                    )
-                                })?,
-                        )
-                        .json()
-                        .get(),
-                    )
-                    .expect("Raw<..> is always valid"),
-                )
             })
             .filter_map(|r| r.ok())
             .collect::<Vec<_>>();

src/database/transaction_ids.rs

@@ -2,7 +2,6 @@ use crate::Result;
 use ruma::{DeviceId, UserId};
 use sled::IVec;
 
-#[derive(Clone)]
 pub struct TransactionIds {
     pub(super) userdevicetxnid_response: sled::Tree, // Response can be empty (/sendToDevice) or the event id (/send)
 }

src/database/uiaa.rs

@@ -7,7 +7,6 @@ use ruma::{
     DeviceId, UserId,
 };
 
-#[derive(Clone)]
 pub struct Uiaa {
     pub(super) userdeviceid_uiaainfo: sled::Tree, // User-interactive authentication
 }

src/database/users.rs

@@ -10,12 +10,10 @@ use ruma::{
     },
     encryption::DeviceKeys,
     events::{AnyToDeviceEvent, EventType},
-    serde::Raw,
+    DeviceId, DeviceKeyAlgorithm, DeviceKeyId, Raw, UserId,
-    DeviceId, DeviceKeyAlgorithm, DeviceKeyId, UserId,
 };
 use std::{collections::BTreeMap, convert::TryFrom, mem, time::SystemTime};
 
-#[derive(Clone)]
 pub struct Users {
     pub(super) userid_password: sled::Tree,
     pub(super) userid_displayname: sled::Tree,

src/error.rs

@@ -1,14 +1,7 @@
-use std::{collections::HashMap, sync::RwLock, time::Duration, time::Instant};
-
 use log::error;
-use ruma::{
+use ruma::api::client::{error::ErrorKind, r0::uiaa::UiaaInfo};
-    api::client::{error::ErrorKind, r0::uiaa::UiaaInfo},
-    events::room::message,
-};
 use thiserror::Error;
 
-use crate::{database::admin::AdminCommand, Database};
-
 #[cfg(feature = "conduit_bin")]
 use {
     crate::RumaResponse,
@@ -60,11 +53,6 @@ impl Error {
         error!("BadDatabase: {}", message);
         Self::BadDatabase(message)
     }
-
-    pub fn bad_config(message: &'static str) -> Self {
-        error!("BadConfig: {}", message);
-        Self::BadConfig(message)
-    }
 }
 
 #[cfg(feature = "conduit_bin")]
@@ -107,46 +95,3 @@ where
         .respond_to(r)
     }
 }
-
-pub struct ConduitLogger {
-    pub db: Database,
-    pub last_logs: RwLock<HashMap<String, Instant>>,
-}
-
-impl log::Log for ConduitLogger {
-    fn enabled(&self, _metadata: &log::Metadata<'_>) -> bool {
-        true
-    }
-
-    fn log(&self, record: &log::Record<'_>) {
-        let output = format!("{} - {}", record.level(), record.args());
-
-        println!("{}", output);
-
-        if self.enabled(record.metadata())
-            && record
-                .module_path()
-                .map_or(false, |path| path.starts_with("conduit::"))
-        {
-            if self
-                .last_logs
-                .read()
-                .unwrap()
-                .get(&output)
-                .map_or(false, |i| i.elapsed() < Duration::from_secs(60 * 30))
-            {
-                return;
-            }
-
-            if let Ok(mut_last_logs) = &mut self.last_logs.try_write() {
-                mut_last_logs.insert(output.clone(), Instant::now());
-            }
-
-            self.db.admin.send(AdminCommand::SendTextMessage(
-                message::TextMessageEventContent::plain(output),
-            ));
-        }
-    }
-
-    fn flush(&self) {}
-}

src/main.rs

@@ -11,8 +11,7 @@ mod ruma_wrapper;
 mod utils;
 
 pub use database::Database;
-pub use error::{ConduitLogger, Error, Result};
+pub use error::{Error, Result};
-use log::LevelFilter;
 pub use pdu::PduEvent;
 pub use rocket::State;
 pub use ruma_wrapper::{ConduitResult, Ruma, RumaResponse};
@@ -20,9 +19,6 @@ pub use ruma_wrapper::{ConduitResult, Ruma, RumaResponse};
 use rocket::{fairing::AdHoc, routes};
 
 fn setup_rocket() -> rocket::Rocket {
-    // Force log level off, so we can use our own logger
-    std::env::set_var("ROCKET_LOG_LEVEL", "off");
-
     rocket::ignite()
         .mount(
             "/",
@@ -123,9 +119,9 @@ fn setup_rocket() -> rocket::Rocket {
                 client_server::get_pushers_route,
                 client_server::set_pushers_route,
                 client_server::upgrade_room_route,
-                server_server::get_server_version_route,
+                server_server::get_server_version,
-                server_server::get_server_keys_route,
+                server_server::get_server_keys,
-                server_server::get_server_keys_deprecated_route,
+                server_server::get_server_keys_deprecated,
                 server_server::get_public_rooms_route,
                 server_server::get_public_rooms_filtered_route,
                 server_server::send_transaction_message_route,
@@ -133,18 +129,10 @@ fn setup_rocket() -> rocket::Rocket {
                 server_server::get_profile_information_route,
             ],
         )
-        .attach(AdHoc::on_attach("Config", |rocket| async {
+        .attach(AdHoc::on_attach("Config", |mut rocket| async {
-            let data =
+            let data = Database::load_or_create(rocket.config().await).expect("valid config");
-                Database::load_or_create(rocket.figment().extract().expect("config is valid"))
-                    .expect("config is valid");
 
             data.sending.start_handler(&data.globals, &data.rooms);
-            log::set_boxed_logger(Box::new(ConduitLogger {
-                db: data.clone(),
-                last_logs: Default::default(),
-            }))
-            .unwrap();
-            log::set_max_level(LevelFilter::Info);
 
             Ok(rocket.manage(data))
         }))
@@ -152,5 +140,10 @@ fn setup_rocket() -> rocket::Rocket {
 
 #[rocket::main]
 async fn main() {
+    // Default log level
+    if std::env::var("ROCKET_LOG").is_err() {
+        std::env::set_var("ROCKET_LOG", "critical");
+    }
+
     setup_rocket().launch().await.unwrap();
 }

src/pdu.rs

@@ -5,17 +5,11 @@ use ruma::{
         pdu::EventHash, room::member::MemberEventContent, AnyEvent, AnyRoomEvent, AnyStateEvent,
         AnyStrippedStateEvent, AnySyncRoomEvent, AnySyncStateEvent, EventType, StateEvent,
     },
-    serde::{to_canonical_value, CanonicalJsonObject, CanonicalJsonValue, Raw},
+    EventId, Raw, RoomId, ServerKeyId, ServerName, UserId,
-    EventId, RoomId, RoomVersionId, ServerName, ServerSigningKeyId, UserId,
 };
 use serde::{Deserialize, Serialize};
 use serde_json::json;
-use std::{
+use std::{collections::BTreeMap, convert::TryInto, sync::Arc, time::UNIX_EPOCH};
-    collections::BTreeMap,
-    convert::{TryFrom, TryInto},
-    sync::Arc,
-    time::UNIX_EPOCH,
-};
 
 #[derive(Deserialize, Serialize, Debug)]
 pub struct PduEvent {
@@ -36,7 +30,7 @@ pub struct PduEvent {
     #[serde(default, skip_serializing_if = "serde_json::Map::is_empty")]
     pub unsigned: serde_json::Map<String, serde_json::Value>,
     pub hashes: EventHash,
-    pub signatures: BTreeMap<Box<ServerName>, BTreeMap<ServerSigningKeyId, String>>,
+    pub signatures: BTreeMap<Box<ServerName>, BTreeMap<ServerKeyId, String>>,
 }
 
 impl PduEvent {
@@ -205,35 +199,32 @@ impl PduEvent {
         serde_json::from_value(json).expect("Raw::from_value always works")
     }
 
-    /// This does not return a full `Pdu` it is only to satisfy ruma's types.
+    pub fn convert_to_outgoing_federation_event<T>(mut pdu_json: serde_json::Value) -> Raw<T> {
-    pub fn convert_to_outgoing_federation_event(
+        if let Some(unsigned) = pdu_json
-        mut pdu_json: CanonicalJsonObject,
+            .as_object_mut()
-    ) -> Raw<ruma::events::pdu::Pdu> {
+            .expect("json is object")
-        if let Some(CanonicalJsonValue::Object(unsigned)) = pdu_json.get_mut("unsigned") {
+            .get_mut("unsigned")
-            unsigned.remove("transaction_id");
+        {
+            unsigned
+                .as_object_mut()
+                .expect("unsigned is object")
+                .remove("transaction_id");
         }
 
-        pdu_json.remove("event_id");
+        pdu_json
+            .as_object_mut()
+            .expect("json is object")
+            .remove("event_id");
 
-        // TODO: another option would be to convert it to a canonical string to validate size
+        serde_json::from_value::<Raw<_>>(pdu_json).expect("Raw::from_value always works")
-        // and return a Result<Raw<...>>
-        // serde_json::from_str::<Raw<_>>(
-        //     ruma::serde::to_canonical_json_string(pdu_json).expect("CanonicalJson is valid serde_json::Value"),
-        // )
-        // .expect("Raw::from_value always works")
-
-        serde_json::from_value::<Raw<_>>(
-            serde_json::to_value(pdu_json).expect("CanonicalJson is valid serde_json::Value"),
-        )
-        .expect("Raw::from_value always works")
     }
 }
 
 impl From<&state_res::StateEvent> for PduEvent {
     fn from(pdu: &state_res::StateEvent) -> Self {
         Self {
-            event_id: pdu.event_id(),
+            event_id: pdu.event_id().clone(),
-            room_id: pdu.room_id().clone(),
+            room_id: pdu.room_id().unwrap().clone(),
             sender: pdu.sender().clone(),
             origin_server_ts: (pdu
                 .origin_server_ts()
@@ -259,56 +250,35 @@ impl From<&state_res::StateEvent> for PduEvent {
 impl PduEvent {
     pub fn convert_for_state_res(&self) -> Arc<state_res::StateEvent> {
         Arc::new(
-            // For consistency of eventId (just in case) we use the one
+            serde_json::from_value(json!({
-            // generated by conduit for everything.
+                "event_id": self.event_id,
-            state_res::StateEvent::from_id_value(
+                "room_id": self.room_id,
-                self.event_id.clone(),
+                "sender": self.sender,
-                json!({
+                "origin_server_ts": self.origin_server_ts,
-                    "event_id": self.event_id,
+                "type": self.kind,
-                    "room_id": self.room_id,
+                "content": self.content,
-                    "sender": self.sender,
+                "state_key": self.state_key,
-                    "origin_server_ts": self.origin_server_ts,
+                "prev_events": self.prev_events
-                    "type": self.kind,
+                    .iter()
-                    "content": self.content,
+                    // TODO How do we create one of these
-                    "state_key": self.state_key,
+                    .map(|id| (id, EventHash { sha256: "hello".into() }))
-                    "prev_events": self.prev_events,
+                    .collect::<Vec<_>>(),
-                    "depth": self.depth,
+                "depth": self.depth,
-                    "auth_events": self.auth_events,
+                "auth_events": self.auth_events
-                    "redacts": self.redacts,
+                    .iter()
-                    "unsigned": self.unsigned,
+                    // TODO How do we create one of these
-                    "hashes": self.hashes,
+                    .map(|id| (id, EventHash { sha256: "hello".into() }))
-                    "signatures": self.signatures,
+                    .collect::<Vec<_>>(),
-                }),
+                "redacts": self.redacts,
-            )
+                "unsigned": self.unsigned,
+                "hashes": self.hashes,
+                "signatures": self.signatures,
+            }))
             .expect("all conduit PDUs are state events"),
         )
     }
 }
 
-/// Generates a correct eventId for the incoming pdu.
-///
-/// Returns a tuple of the new `EventId` and the PDU with the eventId inserted as a `serde_json::Value`.
-pub(crate) fn process_incoming_pdu(
-    pdu: &Raw<ruma::events::pdu::Pdu>,
-) -> (EventId, CanonicalJsonObject) {
-    let mut value =
-        serde_json::from_str(pdu.json().get()).expect("A Raw<...> is always valid JSON");
-
-    let event_id = EventId::try_from(&*format!(
-        "${}",
-        ruma::signatures::reference_hash(&value, &RoomVersionId::Version6)
-            .expect("ruma can calculate reference hashes")
-    ))
-    .expect("ruma's reference hashes are valid event ids");
-
-    value.insert(
-        "event_id".to_owned(),
-        to_canonical_value(&event_id).expect("EventId is a valid CanonicalJsonValue"),
-    );
-
-    (event_id, value)
-}
-
 /// Build the start of a PDU in order to add it to the `Database`.
 #[derive(Debug, Deserialize)]
 pub struct PduBuilder {

src/push_rules.rs

@@ -1,18 +1,15 @@
 use ruma::{
     push::{
-        Action, ConditionalPushRule, ConditionalPushRuleInit, ContentPushRule, OverridePushRule,
+        Action, ConditionalPushRule, ConditionalPushRuleInit, PatternedPushRule,
-        PatternedPushRule, PatternedPushRuleInit, PushCondition, RoomMemberCountIs, Ruleset, Tweak,
+        PatternedPushRuleInit, PushCondition, RoomMemberCountIs, Ruleset, Tweak,
-        UnderridePushRule,
     },
     UserId,
 };
 
 pub fn default_pushrules(user_id: &UserId) -> Ruleset {
     let mut rules = Ruleset::default();
-
+    rules.content = vec![contains_user_name_rule(&user_id)];
-    rules.add(ContentPushRule(contains_user_name_rule(&user_id)));
+    rules.override_ = vec![
-
-    for rule in vec![
         master_rule(),
         suppress_notices_rule(),
         invite_for_me_rule(),
@@ -20,20 +17,14 @@ pub fn default_pushrules(user_id: &UserId) -> Ruleset {
         contains_display_name_rule(),
         tombstone_rule(),
         roomnotif_rule(),
-    ] {
+    ];
-        rules.add(OverridePushRule(rule));
+    rules.underride = vec![
-    }
-
-    for rule in vec![
         call_rule(),
         encrypted_room_one_to_one_rule(),
         room_one_to_one_rule(),
         message_rule(),
         encrypted_rule(),
-    ] {
+    ];
-        rules.add(UnderridePushRule(rule));
-    }
-
     rules
 }
 

src/ruma_wrapper.rs

@@ -1,13 +1,11 @@
 use crate::Error;
 use ruma::{
-    api::{AuthScheme, OutgoingRequest},
+    api::AuthScheme,
+    api::OutgoingRequest,
     identifiers::{DeviceId, UserId},
     Outgoing,
 };
-use std::{
+use std::{convert::TryFrom, convert::TryInto, ops::Deref};
-    convert::{TryFrom, TryInto},
-    ops::Deref,
-};
 
 #[cfg(feature = "conduit_bin")]
 use {
@@ -15,8 +13,7 @@ use {
     log::warn,
     rocket::{
         data::{
-            ByteUnit, Data, FromDataFuture, FromTransformedData, Transform, TransformFuture,
+            Data, FromDataFuture, FromTransformedData, Transform, TransformFuture, Transformed,
-            Transformed,
         },
         http::Status,
         outcome::Outcome::*,
@@ -67,7 +64,7 @@ where
                 .expect("database was loaded");
 
             let (sender_user, sender_device) = match T::METADATA.authentication {
-                AuthScheme::AccessToken | AuthScheme::QueryOnlyAccessToken => {
+                AuthScheme::AccessToken => {
                     // Get token from header or query value
                     let token = match request
                         .headers()
@@ -87,7 +84,9 @@ where
                         Some((user_id, device_id)) => (Some(user_id), Some(device_id.into())),
                     }
                 }
-                _ => (None, None),
+                AuthScheme::None => (None, None),
+                AuthScheme::ServerSignatures => (None, None),
+                AuthScheme::QueryOnlyAccessToken => (None, None),
             };
 
             let mut http_request = http::Request::builder()
@@ -98,12 +97,12 @@ where
             }
 
             let limit = db.globals.max_request_size();
-            let mut handle = data.open(ByteUnit::Byte(limit.into()));
+            let mut handle = data.open().take(limit.into());
             let mut body = Vec::new();
             handle.read_to_end(&mut body).await.unwrap();
 
             let http_request = http_request.body(body.clone()).unwrap();
-            log::debug!("{:?}", http_request);
+            log::info!("{:?}", http_request);
 
             match <T as Outgoing>::Incoming::try_from(http_request) {
                 Ok(t) => Success(Ruma {

src/server_server.rs

@@ -1,18 +1,14 @@
-use crate::{
+use crate::{client_server, ConduitResult, Database, Error, PduEvent, Result, Ruma};
-    client_server, database::rooms::ClosestParent, utils, ConduitResult, Database, Error, PduEvent,
-    Result, Ruma,
-};
 use get_profile_information::v1::ProfileField;
 use http::header::{HeaderValue, AUTHORIZATION, HOST};
-use log::{error, warn};
+use log::warn;
 use rocket::{get, post, put, response::content::Json, State};
 use ruma::{
     api::{
         federation::{
             directory::{get_public_rooms, get_public_rooms_filtered},
             discovery::{
-                get_server_keys, get_server_version::v1 as get_server_version, ServerSigningKeys,
+                get_server_keys, get_server_version::v1 as get_server_version, ServerKey, VerifyKey,
-                VerifyKey,
             },
             event::get_missing_events,
             query::get_profile_information,
@@ -21,13 +17,12 @@ use ruma::{
         OutgoingRequest,
     },
     directory::{IncomingFilter, IncomingRoomNetwork},
-    EventId, RoomId, ServerName, ServerSigningKeyId, UserId,
+    EventId, RoomVersionId, ServerName,
 };
 use std::{
     collections::BTreeMap,
     convert::TryFrom,
     fmt::Debug,
-    sync::Arc,
     time::{Duration, SystemTime},
 };
 use trust_dns_resolver::AsyncResolver;
@@ -63,12 +58,12 @@ where
     T: Debug,
 {
     if !globals.federation_enabled() {
-        return Err(Error::bad_config("Federation is disabled."));
+        return Err(Error::BadConfig("Federation is disabled."));
     }
 
-    let resolver = AsyncResolver::tokio_from_system_conf().await.map_err(|_| {
+    let resolver = AsyncResolver::tokio_from_system_conf()
-        Error::bad_config("Failed to set up trust dns resolver with system config.")
+        .await
-    })?;
+        .map_err(|_| Error::BadConfig("Failed to set up trust dns resolver with system config."))?;
 
     let mut host = None;
 
@@ -100,7 +95,7 @@ where
     let mut http_request = request
         .try_into_http_request(&actual_destination, Some(""))
         .map_err(|e| {
-            warn!("Failed to find destination {}: {}", actual_destination, e);
+            warn!("{}: {}", actual_destination, e);
             Error::BadServerResponse("Invalid destination")
         })?;
 
@@ -127,9 +122,7 @@ where
     request_map.insert("origin".to_owned(), globals.server_name().as_str().into());
     request_map.insert("destination".to_owned(), destination.as_str().into());
 
-    let mut request_json =
+    let mut request_json = request_map.into();
-        serde_json::from_value(request_map.into()).expect("valid JSON is valid BTreeMap");
-
     ruma::signatures::sign_json(
         globals.server_name().as_str(),
         globals.keypair(),
@@ -137,9 +130,6 @@ where
     )
     .expect("our request json is what ruma expects");
 
-    let request_json: serde_json::Map<String, serde_json::Value> =
-        serde_json::from_slice(&serde_json::to_vec(&request_json).unwrap()).unwrap();
-
     let signatures = request_json["signatures"]
         .as_object()
         .unwrap()
@@ -193,8 +183,6 @@ where
                 }
             }
 
-            let status = reqwest_response.status();
-
             let body = reqwest_response
                 .bytes()
                 .await
@@ -203,27 +191,17 @@ where
                     Vec::new().into()
                 }) // TODO: handle timeout
                 .into_iter()
-                .collect::<Vec<_>>();
+                .collect();
-
-            if status != 200 {
-                warn!(
-                    "Server returned bad response {} ({}): {} {:?}",
-                    destination,
-                    url,
-                    status,
-                    utils::string_from_bytes(&body)
-                );
-            }
 
             let response = T::IncomingResponse::try_from(
                 http_response
                     .body(body)
                     .expect("reqwest body is valid http body"),
             );
-            response.map_err(|_| {
+            response.map_err(|e| {
                 warn!(
-                    "Server returned invalid response bytes {} ({})",
+                    "Server returned bad response {} ({}): {:?}",
-                    destination, url
+                    destination, url, e
                 );
                 Error::BadServerResponse("Server returned bad response.")
             })
@@ -233,11 +211,9 @@ where
 }
 
 #[cfg_attr(feature = "conduit_bin", get("/_matrix/federation/v1/version"))]
-pub fn get_server_version_route(
+pub fn get_server_version(db: State<'_, Database>) -> ConduitResult<get_server_version::Response> {
-    db: State<'_, Database>,
-) -> ConduitResult<get_server_version::Response> {
     if !db.globals.federation_enabled() {
-        return Err(Error::bad_config("Federation is disabled."));
+        return Err(Error::BadConfig("Federation is disabled."));
     }
 
     Ok(get_server_version::Response {
@@ -250,7 +226,7 @@ pub fn get_server_version_route(
 }
 
 #[cfg_attr(feature = "conduit_bin", get("/_matrix/key/v2/server"))]
-pub fn get_server_keys_route(db: State<'_, Database>) -> Json<String> {
+pub fn get_server_keys(db: State<'_, Database>) -> Json<String> {
     if !db.globals.federation_enabled() {
         // TODO: Use proper types
         return Json("Federation is disabled.".to_owned());
@@ -258,17 +234,14 @@ pub fn get_server_keys_route(db: State<'_, Database>) -> Json<String> {
 
     let mut verify_keys = BTreeMap::new();
     verify_keys.insert(
-        ServerSigningKeyId::try_from(
+        format!("ed25519:{}", db.globals.keypair().version()),
-            format!("ed25519:{}", db.globals.keypair().version()).as_str(),
-        )
-        .expect("found invalid server signing keys in DB"),
         VerifyKey {
             key: base64::encode_config(db.globals.keypair().public_key(), base64::STANDARD_NO_PAD),
         },
     );
     let mut response = serde_json::from_slice(
         http::Response::try_from(get_server_keys::v2::Response {
-            server_key: ServerSigningKeys {
+            server_key: ServerKey {
                 server_name: db.globals.server_name().to_owned(),
                 verify_keys,
                 old_verify_keys: BTreeMap::new(),
@@ -280,20 +253,18 @@ pub fn get_server_keys_route(db: State<'_, Database>) -> Json<String> {
         .body(),
     )
     .unwrap();
-
     ruma::signatures::sign_json(
         db.globals.server_name().as_str(),
         db.globals.keypair(),
         &mut response,
     )
     .unwrap();
-
+    Json(response.to_string())
-    Json(ruma::serde::to_canonical_json_string(&response).expect("JSON is canonical"))
 }
 
 #[cfg_attr(feature = "conduit_bin", get("/_matrix/key/v2/server/<_>"))]
-pub fn get_server_keys_deprecated_route(db: State<'_, Database>) -> Json<String> {
+pub fn get_server_keys_deprecated(db: State<'_, Database>) -> Json<String> {
-    get_server_keys_route(db)
+    get_server_keys(db)
 }
 
 #[cfg_attr(
@@ -305,7 +276,7 @@ pub async fn get_public_rooms_filtered_route(
     body: Ruma<get_public_rooms_filtered::v1::Request<'_>>,
 ) -> ConduitResult<get_public_rooms_filtered::v1::Response> {
     if !db.globals.federation_enabled() {
-        return Err(Error::bad_config("Federation is disabled."));
+        return Err(Error::BadConfig("Federation is disabled."));
     }
 
     let response = client_server::get_public_rooms_filtered_helper(
@@ -352,7 +323,7 @@ pub async fn get_public_rooms_route(
     body: Ruma<get_public_rooms::v1::Request<'_>>,
 ) -> ConduitResult<get_public_rooms::v1::Response> {
     if !db.globals.federation_enabled() {
-        return Err(Error::bad_config("Federation is disabled."));
+        return Err(Error::BadConfig("Federation is disabled."));
     }
 
     let response = client_server::get_public_rooms_filtered_helper(
@@ -394,248 +365,54 @@ pub async fn get_public_rooms_route(
     feature = "conduit_bin",
     put("/_matrix/federation/v1/send/<_>", data = "<body>")
 )]
-pub async fn send_transaction_message_route<'a>(
+pub fn send_transaction_message_route<'a>(
     db: State<'a, Database>,
     body: Ruma<send_transaction_message::v1::Request<'_>>,
 ) -> ConduitResult<send_transaction_message::v1::Response> {
     if !db.globals.federation_enabled() {
-        return Err(Error::bad_config("Federation is disabled."));
+        return Err(Error::BadConfig("Federation is disabled."));
     }
 
-    for edu in &body.edus {
+    //dbg!(&*body);
-        match serde_json::from_str::<send_transaction_message::v1::Edu>(edu.json().get()) {
-            Ok(edu) => match edu.edu_type.as_str() {
-                "m.typing" => {
-                    if let Some(typing) = edu.content.get("typing") {
-                        if typing.as_bool().unwrap_or_default() {
-                            db.rooms.edus.typing_add(
-                                &UserId::try_from(edu.content["user_id"].as_str().unwrap())
-                                    .unwrap(),
-                                &RoomId::try_from(edu.content["room_id"].as_str().unwrap())
-                                    .unwrap(),
-                                3000 + utils::millis_since_unix_epoch(),
-                                &db.globals,
-                            )?;
-                        } else {
-                            db.rooms.edus.typing_remove(
-                                &UserId::try_from(edu.content["user_id"].as_str().unwrap())
-                                    .unwrap(),
-                                &RoomId::try_from(edu.content["room_id"].as_str().unwrap())
-                                    .unwrap(),
-                                &db.globals,
-                            )?;
-                        }
-                    }
-                }
-                "m.presence" => {}
-                "m.receipt" => {}
-                _ => {}
-            },
-            Err(_err) => {
-                continue;
-            }
-        }
-    }
-
-    // TODO: For RoomVersion6 we must check that Raw<..> is canonical do we anywhere?
-    // SPEC:
-    // Servers MUST strictly enforce the JSON format specified in the appendices.
-    // This translates to a 400 M_BAD_JSON error on most endpoints, or discarding of
-    // events over federation. For example, the Federation API's /send endpoint would
-    // discard the event whereas the Client Server API's /send/{eventType} endpoint
-    // would return a M_BAD_JSON error.
-    let mut resolved_map = BTreeMap::new();
     for pdu in &body.pdus {
-        // Ruma/PduEvent/StateEvent satisfies - 1. Is a valid event, otherwise it is dropped.
+        let mut value = serde_json::from_str::<serde_json::Value>(pdu.json().get())
+            .expect("converting raw jsons to values always works");
 
-        // state-res checks signatures - 2. Passes signature checks, otherwise event is dropped.
+        let event_id = EventId::try_from(&*format!(
+            "${}",
+            ruma::signatures::reference_hash(&value, &RoomVersionId::Version6)
+                .expect("ruma can calculate reference hashes")
+        ))
+        .expect("ruma's reference hashes are valid event ids");
 
-        // 3. Passes hash checks, otherwise it is redacted before being processed further.
+        value
-        // TODO: redact event if hashing fails
+            .as_object_mut()
-        let (event_id, value) = crate::pdu::process_incoming_pdu(pdu);
+            .expect("ruma pdus are json objects")
-
+            .insert("event_id".to_owned(), event_id.to_string().into());
-        let pdu = serde_json::from_value::<PduEvent>(
-            serde_json::to_value(&value).expect("CanonicalJsonObj is a valid JsonValue"),
-        )
-        .expect("all ruma pdus are conduit pdus");
-        let room_id = &pdu.room_id;
-
-        // If we have no idea about this room skip the PDU
-        if !db.rooms.exists(room_id)? {
-            resolved_map.insert(event_id, Err("Room is unknown to this server".into()));
-            continue;
-        }
-
-        // If it is not a state event, we can skip state-res... maybe
-        if value.get("state_key").is_none() {
-            if !db.rooms.is_joined(&pdu.sender, room_id)? {
-                warn!("Sender is not joined {}", pdu.kind);
-                resolved_map.insert(event_id, Err("User is not in this room".into()));
-                continue;
-            }
 
+        let pdu = serde_json::from_value::<PduEvent>(value.clone())
+            .expect("all ruma pdus are conduit pdus");
+        if db.rooms.exists(&pdu.room_id)? {
             let count = db.globals.next_count()?;
-            let mut pdu_id = room_id.as_bytes().to_vec();
+            let mut pdu_id = pdu.room_id.as_bytes().to_vec();
             pdu_id.push(0xff);
             pdu_id.extend_from_slice(&count.to_be_bytes());
-
             db.rooms.append_to_state(&pdu_id, &pdu)?;
-
             db.rooms.append_pdu(
                 &pdu,
                 &value,
                 count,
-                pdu_id.into(),
+                pdu_id.clone().into(),
                 &db.globals,
                 &db.account_data,
-                &db.admin,
+                &db.sending,
             )?;
-
-            resolved_map.insert(event_id, Ok::<(), String>(()));
-            continue;
         }
-
-        // We have a state event so we need info for state-res
-        let get_state_response = match send_request(
-            &db.globals,
-            body.body.origin.clone(),
-            ruma::api::federation::event::get_room_state::v1::Request {
-                room_id,
-                event_id: &event_id,
-            },
-        )
-        .await
-        {
-            Ok(res) => res,
-            // We can't hard fail because there are some valid errors, just
-            // keep checking PDU's
-            //
-            // As an example a possible error
-            // {"errcode":"M_FORBIDDEN","error":"Host not in room."}
-            Err(err) => {
-                resolved_map.insert(event_id, Err(err.to_string()));
-                continue;
-            }
-        };
-
-        let their_current_state = get_state_response
-            .pdus
-            .iter()
-            .chain(get_state_response.auth_chain.iter()) // add auth events
-            .map(|pdu| {
-                let (event_id, json) = crate::pdu::process_incoming_pdu(pdu);
-                (
-                    event_id.clone(),
-                    Arc::new(
-                        // When creating a StateEvent the event_id arg will be used
-                        // over any found in the json and it will not use ruma::reference_hash
-                        // to generate one
-                        state_res::StateEvent::from_id_canon_obj(event_id, json)
-                            .expect("valid pdu json"),
-                    ),
-                )
-            })
-            .collect::<BTreeMap<_, _>>();
-
-        let our_current_state = db.rooms.room_state_full(room_id)?;
-        // State resolution takes care of these checks
-        // 4. Passes authorization rules based on the event's auth events, otherwise it is rejected.
-        // 5. Passes authorization rules based on the state at the event, otherwise it is rejected.
-
-        // TODO: 6. Passes authorization rules based on the current state of the room, otherwise it is "soft failed".
-        match state_res::StateResolution::resolve(
-            room_id,
-            &ruma::RoomVersionId::Version6,
-            &[
-                our_current_state
-                    .iter()
-                    .map(|((ev, sk), v)| ((ev.clone(), sk.to_owned()), v.event_id.clone()))
-                    .collect::<BTreeMap<_, _>>(),
-                their_current_state
-                    .iter()
-                    .map(|(_id, v)| ((v.kind(), v.state_key()), v.event_id()))
-                    .collect::<BTreeMap<_, _>>(),
-            ],
-            Some(
-                our_current_state
-                    .iter()
-                    .map(|(_k, v)| (v.event_id.clone(), v.convert_for_state_res()))
-                    .chain(
-                        their_current_state
-                            .iter()
-                            .map(|(id, ev)| (id.clone(), ev.clone())),
-                    )
-                    .collect::<BTreeMap<_, _>>(),
-            ),
-            &db.rooms,
-        ) {
-            Ok(resolved) if resolved.values().any(|id| &event_id == id) => {
-                // If the event is older than the last event in pduid_pdu Tree then find the
-                // closest ancestor we know of and insert after the known ancestor by
-                // altering the known events pduid to = same roomID + same count bytes + 0x1
-                // pushing a single byte every time a simple append cannot be done.
-                match db.rooms.get_latest_pduid_before(
-                    room_id,
-                    &pdu.prev_events,
-                    &their_current_state,
-                )? {
-                    Some(ClosestParent::Append) => {
-                        let count = db.globals.next_count()?;
-                        let mut pdu_id = room_id.as_bytes().to_vec();
-                        pdu_id.push(0xff);
-                        pdu_id.extend_from_slice(&count.to_be_bytes());
-
-                        db.rooms.append_pdu(
-                            &pdu,
-                            &value,
-                            count,
-                            pdu_id.into(),
-                            &db.globals,
-                            &db.account_data,
-                            &db.admin,
-                        )?;
-                    }
-                    Some(ClosestParent::Insert(old_count)) => {
-                        let count = old_count;
-                        let mut pdu_id = room_id.as_bytes().to_vec();
-                        pdu_id.push(0xff);
-                        pdu_id.extend_from_slice(&count.to_be_bytes());
-                        // Create a new count that is after old_count but before
-                        // the pdu appended after
-                        pdu_id.push(1);
-
-                        db.rooms.append_pdu(
-                            &pdu,
-                            &value,
-                            count,
-                            pdu_id.into(),
-                            &db.globals,
-                            &db.account_data,
-                            &db.admin,
-                        )?;
-                    }
-                    _ => {
-                        error!("Not a sequential event or no parents found");
-                        continue;
-                    }
-                }
-
-                resolved_map.insert(event_id, Ok::<(), String>(()));
-            }
-            // If the eventId is not found in the resolved state auth has failed
-            Ok(_) => {
-                resolved_map.insert(
-                    event_id,
-                    Err("This event failed authentication, not found in resolved set".into()),
-                );
-            }
-            Err(e) => {
-                resolved_map.insert(event_id, Err(e.to_string()));
-            }
-        };
     }
-
+    Ok(send_transaction_message::v1::Response {
-    Ok(send_transaction_message::v1::Response { pdus: resolved_map }.into())
+        pdus: BTreeMap::new(),
+    }
+    .into())
 }
 
 #[cfg_attr(
@@ -647,7 +424,7 @@ pub fn get_missing_events_route<'a>(
     body: Ruma<get_missing_events::v1::Request<'_>>,
 ) -> ConduitResult<get_missing_events::v1::Response> {
     if !db.globals.federation_enabled() {
-        return Err(Error::bad_config("Federation is disabled."));
+        return Err(Error::BadConfig("Federation is disabled."));
     }
 
     let mut queued_events = body.latest_events.clone();
@@ -675,7 +452,7 @@ pub fn get_missing_events_route<'a>(
                 )
                 .map_err(|_| Error::bad_database("Invalid prev_events content in pdu in db."))?,
             );
-            events.push(serde_json::from_value(pdu).expect("Raw<..> is always valid"));
+            events.push(PduEvent::convert_to_outgoing_federation_event(pdu));
         }
         i += 1;
     }
@@ -692,15 +469,13 @@ pub fn get_profile_information_route<'a>(
     body: Ruma<get_profile_information::v1::Request<'_>>,
 ) -> ConduitResult<get_profile_information::v1::Response> {
     if !db.globals.federation_enabled() {
-        return Err(Error::bad_config("Federation is disabled."));
+        return Err(Error::BadConfig("Federation is disabled."));
     }
 
     let mut displayname = None;
     let mut avatar_url = None;
 
-    match &body.field {
+    match body.field {
-        // TODO: what to do with custom
-        Some(ProfileField::_Custom(_s)) => {}
         Some(ProfileField::DisplayName) => displayname = db.users.displayname(&body.user_id)?,
         Some(ProfileField::AvatarUrl) => avatar_url = db.users.avatar_url(&body.user_id)?,
         None => {
@@ -726,7 +501,7 @@ pub fn get_user_devices_route<'a>(
     body: Ruma<membership::v1::Request<'_>>,
 ) -> ConduitResult<get_profile_information::v1::Response> {
     if !db.globals.federation_enabled() {
-        return Err(Error::bad_config("Federation is disabled."));
+        return Err(Error::BadConfig("Federation is disabled."));
     }
 
     let mut displayname = None;

src/utils.rs

@@ -1,7 +1,6 @@
 use argon2::{Config, Variant};
 use cmp::Ordering;
 use rand::prelude::*;
-use ruma::serde::{try_from_json_map, CanonicalJsonError, CanonicalJsonObject};
 use sled::IVec;
 use std::{
     cmp,
@@ -90,24 +89,9 @@ pub fn common_elements(
                         }
                     }
                 }
+
                 false
             })
             .all(|b| b)
     }))
 }
-
-/// Fallible conversion from any value that implements `Serialize` to a `CanonicalJsonObject`.
-///
-/// `value` must serialize to an `serde_json::Value::Object`.
-pub fn to_canonical_object<T: serde::Serialize>(
-    value: T,
-) -> Result<CanonicalJsonObject, CanonicalJsonError> {
-    use serde::ser::Error;
-
-    match serde_json::to_value(value).map_err(CanonicalJsonError::SerDe)? {
-        serde_json::Value::Object(map) => try_from_json_map(map),
-        _ => Err(CanonicalJsonError::SerDe(serde_json::Error::custom(
-            "Value must be an object",
-        ))),
-    }
-}

tests/Complement.Dockerfile

@@ -14,27 +14,17 @@ RUN cargo build
 FROM valkum/docker-rust-ci:latest
 WORKDIR /workdir
 
-RUN curl -OL "https://github.com/caddyserver/caddy/releases/download/v2.2.1/caddy_2.2.1_linux_amd64.tar.gz"
+RUN curl -OL "https://github.com/caddyserver/caddy/releases/download/v2.1.1/caddy_2.1.1_linux_amd64.tar.gz"
-RUN tar xzf caddy_2.2.1_linux_amd64.tar.gz
+RUN tar xzf caddy_2.1.1_linux_amd64.tar.gz
 
 COPY --from=builder /workdir/target/debug/conduit /workdir/conduit
 
 COPY Rocket-example.toml Rocket.toml
 
 ENV SERVER_NAME=localhost
-ENV ROCKET_LOG=normal
 
+RUN sed -i "s/server_name = \"your.server.name\"/server_name = \"${SERVER_NAME}\"/g" Rocket.toml
 RUN sed -i "s/port = 14004/port = 8008/g" Rocket.toml
-RUN echo "federation_enabled = true" >> Rocket.toml
 
-# Enabled Caddy auto cert generation for complement provided CA.
-RUN echo '{"apps":{"http":{"https_port":8448,"servers":{"srv0":{"listen":[":8448"],"routes":[{"match":[{"host":["your.server.name"]}],"handle":[{"handler":"subroute","routes":[{"handle":[{"handler":"reverse_proxy","upstreams":[{"dial":"localhost:8008"}]}]}]}],"terminal":true}],"tls_connection_policies": [{"match": {"sni": ["your.server.name"]}}]}}},"pki": {"certificate_authorities": {"local": {"name": "Complement CA","root": {"certificate": "/ca/ca.crt","private_key": "/ca/ca.key"},"intermediate": {"certificate": "/ca/ca.crt","private_key": "/ca/ca.key"}}}},"tls":{"automation":{"policies":[{"subjects":["your.server.name"],"issuer":{"module":"internal"},"on_demand":true},{"issuer":{"module":"internal", "ca": "local"}}]}}}}' > caddy.json 
- 
 EXPOSE 8008 8448
-
+CMD /workdir/caddy reverse-proxy --from ${SERVER_NAME}:8448 --to localhost:8008 > /dev/null 2>&1 & /workdir/conduit
-CMD ([ -z "${COMPLEMENT_CA}" ] && echo "Error: Need Complement PKI support" && true) || \
-    sed -i "s/server_name = \"your.server.name\"/server_name = \"${SERVER_NAME}\"/g" Rocket.toml && \
-    sed -i "s/your.server.name/${SERVER_NAME}/g" caddy.json && \
-    /workdir/caddy start --config caddy.json > /dev/null && \
-    /workdir/conduit
-