Fix clippy warnings remove unused imports

Cargo.lock

@@ -2065,7 +2065,7 @@ checksum = "3015a7d0a5fd5105c91c3710d42f9ccf0abfb287d62206484dcc67f9569a6483"
 [[package]]
 name = "state-res"
 version = "0.1.0"
-source = "git+https://github.com/ruma/state-res?branch=event-trait#bfadbdf57e26f26c2ea5b2ed50ce3e5f6fb914cd"
+source = "git+https://github.com/ruma/state-res?branch=event-trait#e5d32e44adb66c5932a81d2c8a8d840abd17c870"
 dependencies = [
  "itertools",
  "maplit",

src/appservice_server.rs

@@ -25,7 +25,7 @@ where
 
     let mut parts = http_request.uri().clone().into_parts();
     let old_path_and_query = parts.path_and_query.unwrap().as_str().to_owned();
-    let symbol = if old_path_and_query.contains("?") {
+    let symbol = if old_path_and_query.contains('?') {
         "&"
     } else {
         "?"

src/client_server/membership.rs

@@ -675,7 +675,6 @@ async fn join_room_by_id_helper(
             .iter()
             .filter(|id| resolved_events.values().any(|rid| rid == *id))
         {
-            // this is a `state_res::StateEvent` that holds a `ruma::Pdu`
             let pdu = event_map
                 .get(ev_id)
                 .expect("Found event_id in sorted events that is not in resolved state");

src/database/globals.rs

@@ -9,13 +9,15 @@ use trust_dns_resolver::TokioAsyncResolver;
 
 pub const COUNTER: &str = "c";
 
+pub type DestinationCache = Arc<RwLock<HashMap<Box<ServerName>, (String, Option<String>)>>>;
+
 #[derive(Clone)]
 pub struct Globals {
     pub(super) globals: sled::Tree,
     config: Config,
     keypair: Arc<ruma::signatures::Ed25519KeyPair>,
     reqwest_client: reqwest::Client,
-    pub actual_destination_cache: Arc<RwLock<HashMap<Box<ServerName>, (String, Option<String>)>>>, // actual_destination, host
+    pub actual_destination_cache: DestinationCache, // actual_destination, host
     dns_resolver: TokioAsyncResolver,
 }
 

src/database/sending.rs

@@ -79,7 +79,7 @@ impl Sending {
                         match response {
                             Ok((server, is_appservice)) => {
                                 let mut prefix = if is_appservice {
-                                    "+".as_bytes().to_vec()
+                                    b"+".to_vec()
                                 } else {
                                     Vec::new()
                                 };
@@ -125,7 +125,7 @@ impl Sending {
                             Err((server, is_appservice, e)) => {
                                 info!("Couldn't send transaction to {}\n{}", server, e);
                                 let mut prefix = if is_appservice {
-                                    "+".as_bytes().to_vec()
+                                    b"+".to_vec()
                                 } else {
                                     Vec::new()
                                 };
@@ -156,7 +156,7 @@ impl Sending {
                                 .map_err(|_| Error::bad_database("ServerName in servernamepduid bytes are invalid."))
                                 .map(|server_str| {
                                     // Appservices start with a plus
-                                    if server_str.starts_with("+") {
+                                    if server_str.starts_with('+') {
                                         (server_str[1..].to_owned(), true)
                                     } else {
                                         (server_str, false)
@@ -185,7 +185,7 @@ impl Sending {
                                     }
 
                                     let mut prefix = if *is_appservice {
-                                        "+".as_bytes().to_vec()
+                                        b"+".to_vec()
                                     } else {
                                         Vec::new()
                                     };
@@ -219,7 +219,7 @@ impl Sending {
     }
 
     pub fn send_pdu_appservice(&self, appservice_id: &str, pdu_id: &[u8]) -> Result<()> {
-        let mut key = "+".as_bytes().to_vec();
+        let mut key = b"+".to_vec();
         key.extend_from_slice(appservice_id.as_bytes());
         key.push(0xff);
         key.extend_from_slice(pdu_id);
@@ -329,7 +329,7 @@ impl Sending {
         })?;
 
         // Appservices start with a plus
-        let (server, is_appservice) = if server.starts_with("+") {
+        let (server, is_appservice) = if server.starts_with('+') {
             (&server[1..], true)
         } else {
             (&*server, false)

src/pdu.rs

@@ -9,12 +9,7 @@ use ruma::{
 };
 use serde::{Deserialize, Serialize};
 use serde_json::json;
-use std::{
-    collections::BTreeMap,
-    convert::{TryFrom, TryInto},
-    sync::Arc,
-    time::UNIX_EPOCH,
-};
+use std::{collections::BTreeMap, convert::TryFrom, time::UNIX_EPOCH};
 
 #[derive(Clone, Deserialize, Serialize, Debug)]
 pub struct PduEvent {

src/server_server.rs

@@ -1,4 +1,4 @@
-use crate::{client_server, pdu, utils, ConduitResult, Database, Error, PduEvent, Result, Ruma};
+use crate::{client_server, utils, ConduitResult, Database, Error, PduEvent, Result, Ruma};
 use get_profile_information::v1::ProfileField;
 use http::header::{HeaderValue, AUTHORIZATION, HOST};
 use log::{error, info, warn};
@@ -11,14 +11,13 @@ use ruma::{
                 get_server_keys, get_server_version::v1 as get_server_version, ServerSigningKeys,
                 VerifyKey,
             },
-            event::{get_event, get_missing_events, get_room_state, get_room_state_ids},
+            event::{get_event, get_missing_events, get_room_state_ids},
             query::get_profile_information,
             transactions::send_transaction_message,
         },
         OutgoingRequest,
     },
     directory::{IncomingFilter, IncomingRoomNetwork},
-    serde::Raw,
     signatures::{CanonicalJsonObject, PublicKeyMap},
     EventId, RoomId, RoomVersionId, ServerName, ServerSigningKeyId, UserId,
 };
@@ -220,7 +219,7 @@ fn add_port_to_hostname(destination_str: String) -> String {
 /// Numbers in comments below refer to bullet points in linked section of specification
 async fn find_actual_destination(
     globals: &crate::database::globals::Globals,
-    destination: &Box<ServerName>,
+    destination: &ServerName,
 ) -> (String, Option<String>) {
     let mut host = None;
 
@@ -594,13 +593,14 @@ pub async fn send_transaction_message_route<'a>(
             continue;
         }
 
+        let server_name = body.body.origin.clone();
         let event = Arc::new(pdu.clone());
-
-        let previous = pdu
-            .prev_events
-            .first()
-            .map(|id| db.rooms.get_pdu(id).expect("todo").map(Arc::new))
-            .flatten();
+        // Fetch any unknown events or retrieve them from the DB
+        let previous =
+            match fetch_events(&db, server_name.clone(), &pub_key_map, &pdu.prev_events).await? {
+                mut evs if evs.len() == 1 => Some(Arc::new(evs.remove(0))),
+                _ => None,
+            };
 
         // 4. Passes authorization rules based on the event's auth events, otherwise it is rejected.
         // TODO: To me this sounds more like the auth_events should be get the pdu.auth_events not
@@ -616,14 +616,14 @@ pub async fn send_transaction_message_route<'a>(
 
         let mut event_map: state_res::EventMap<Arc<PduEvent>> = auth_events
             .iter()
-            .map(|(k, v)| (v.event_id().clone(), Arc::new(v.clone())))
+            .map(|(_k, v)| (v.event_id().clone(), Arc::new(v.clone())))
             .collect();
 
         if !state_res::event_auth::auth_check(
             &RoomVersionId::Version6,
             &event,
             previous.clone(),
-            auth_events
+            &auth_events
                 .into_iter()
                 .map(|(k, v)| (k, Arc::new(v)))
                 .collect(),
@@ -638,7 +638,6 @@ pub async fn send_transaction_message_route<'a>(
             continue;
         }
 
-        let server_name = body.body.origin.clone();
         let (state_at_event, incoming_auth_events): (StateMap<Arc<PduEvent>>, _) = match db
             .sending
             .send_federation_request(
@@ -652,8 +651,18 @@ pub async fn send_transaction_message_route<'a>(
             .await
         {
             Ok(res) => {
-                let state = fetch_events(&db, server_name.clone(), &pub_key_map, &res.pdu_ids)
-                    .await?
+                let state =
+                    fetch_events(&db, server_name.clone(), &pub_key_map, &res.pdu_ids).await?;
+                // Sanity check: there are no conflicting events in the state we received
+                let mut seen = BTreeSet::new();
+                for ev in &state {
+                    // If the key is already present
+                    if !seen.insert((&ev.kind, &ev.state_key)) {
+                        todo!("Server sent us an invalid state")
+                    }
+                }
+
+                let state = state
                     .into_iter()
                     .map(|pdu| ((pdu.kind.clone(), pdu.state_key.clone()), Arc::new(pdu)))
                     .collect();
@@ -677,7 +686,7 @@ pub async fn send_transaction_message_route<'a>(
             &RoomVersionId::Version6,
             &event,
             previous.clone(),
-            state_at_event.clone(), // TODO: could this be &state avoid .clone
+            &state_at_event,
             None, // TODO: third party invite
         )
         .map_err(|_e| Error::Conflict("Auth check failed"))?
@@ -760,7 +769,7 @@ pub async fn send_transaction_message_route<'a>(
             &RoomVersionId::Version6,
             &event,
             previous,
-            state_at_forks,
+            &state_at_forks,
             None,
         )
         .map_err(|_e| Error::Conflict("Auth check failed"))?