improvement: better federation joins

Cargo.lock

@@ -1634,7 +1634,7 @@ dependencies = [
 [[package]]
 name = "ruma"
 version = "0.0.1"
-source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#63341000fbabce9b230b6665ce65c617944408fa"
+source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#8d763abaecb13f4799a31ecf1e0da77d2bc956a6"
 dependencies = [
  "ruma-api",
  "ruma-appservice-api",
@@ -1650,7 +1650,7 @@ dependencies = [
 [[package]]
 name = "ruma-api"
 version = "0.17.0-alpha.1"
-source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#63341000fbabce9b230b6665ce65c617944408fa"
+source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#8d763abaecb13f4799a31ecf1e0da77d2bc956a6"
 dependencies = [
  "http",
  "percent-encoding",
@@ -1665,7 +1665,7 @@ dependencies = [
 [[package]]
 name = "ruma-api-macros"
 version = "0.17.0-alpha.1"
-source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#63341000fbabce9b230b6665ce65c617944408fa"
+source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#8d763abaecb13f4799a31ecf1e0da77d2bc956a6"
 dependencies = [
  "proc-macro-crate",
  "proc-macro2",
@@ -1676,7 +1676,7 @@ dependencies = [
 [[package]]
 name = "ruma-appservice-api"
 version = "0.2.0-alpha.1"
-source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#63341000fbabce9b230b6665ce65c617944408fa"
+source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#8d763abaecb13f4799a31ecf1e0da77d2bc956a6"
 dependencies = [
  "ruma-api",
  "ruma-common",
@@ -1689,7 +1689,7 @@ dependencies = [
 [[package]]
 name = "ruma-client-api"
 version = "0.10.0-alpha.1"
-source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#63341000fbabce9b230b6665ce65c617944408fa"
+source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#8d763abaecb13f4799a31ecf1e0da77d2bc956a6"
 dependencies = [
  "assign",
  "http",
@@ -1708,7 +1708,7 @@ dependencies = [
 [[package]]
 name = "ruma-common"
 version = "0.2.0"
-source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#63341000fbabce9b230b6665ce65c617944408fa"
+source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#8d763abaecb13f4799a31ecf1e0da77d2bc956a6"
 dependencies = [
  "js_int",
  "ruma-api",
@@ -1722,7 +1722,7 @@ dependencies = [
 [[package]]
 name = "ruma-events"
 version = "0.22.0-alpha.1"
-source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#63341000fbabce9b230b6665ce65c617944408fa"
+source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#8d763abaecb13f4799a31ecf1e0da77d2bc956a6"
 dependencies = [
  "js_int",
  "ruma-common",
@@ -1737,7 +1737,7 @@ dependencies = [
 [[package]]
 name = "ruma-events-macros"
 version = "0.22.0-alpha.1"
-source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#63341000fbabce9b230b6665ce65c617944408fa"
+source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#8d763abaecb13f4799a31ecf1e0da77d2bc956a6"
 dependencies = [
  "proc-macro-crate",
  "proc-macro2",
@@ -1748,7 +1748,7 @@ dependencies = [
 [[package]]
 name = "ruma-federation-api"
 version = "0.0.3"
-source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#63341000fbabce9b230b6665ce65c617944408fa"
+source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#8d763abaecb13f4799a31ecf1e0da77d2bc956a6"
 dependencies = [
  "js_int",
  "ruma-api",
@@ -1763,7 +1763,7 @@ dependencies = [
 [[package]]
 name = "ruma-identifiers"
 version = "0.17.4"
-source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#63341000fbabce9b230b6665ce65c617944408fa"
+source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#8d763abaecb13f4799a31ecf1e0da77d2bc956a6"
 dependencies = [
  "rand",
  "ruma-identifiers-macros",
@@ -1775,7 +1775,7 @@ dependencies = [
 [[package]]
 name = "ruma-identifiers-macros"
 version = "0.17.4"
-source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#63341000fbabce9b230b6665ce65c617944408fa"
+source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#8d763abaecb13f4799a31ecf1e0da77d2bc956a6"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -1786,7 +1786,7 @@ dependencies = [
 [[package]]
 name = "ruma-identifiers-validation"
 version = "0.1.1"
-source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#63341000fbabce9b230b6665ce65c617944408fa"
+source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#8d763abaecb13f4799a31ecf1e0da77d2bc956a6"
 dependencies = [
  "serde",
  "strum",
@@ -1795,7 +1795,7 @@ dependencies = [
 [[package]]
 name = "ruma-serde"
 version = "0.2.3"
-source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#63341000fbabce9b230b6665ce65c617944408fa"
+source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#8d763abaecb13f4799a31ecf1e0da77d2bc956a6"
 dependencies = [
  "form_urlencoded",
  "itoa",
@@ -1807,7 +1807,7 @@ dependencies = [
 [[package]]
 name = "ruma-signatures"
 version = "0.6.0-dev.1"
-source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#63341000fbabce9b230b6665ce65c617944408fa"
+source = "git+https://github.com/timokoesters/ruma?branch=timo-fed-fixes#8d763abaecb13f4799a31ecf1e0da77d2bc956a6"
 dependencies = [
  "base64",
  "ring",
@@ -2072,7 +2072,7 @@ checksum = "7345c971d1ef21ffdbd103a75990a15eb03604fc8b8852ca8cb418ee1a099028"
 [[package]]
 name = "state-res"
 version = "0.1.0"
-source = "git+https://github.com/timokoesters/state-res?branch=spec-comp#0081081604b051d412a2365b68357e064c33320c"
+source = "git+https://github.com/timokoesters/state-res?branch=spec-comp#a9186476b748c901fbf4356414247a0b3ac01b5f"
 dependencies = [
  "itertools",
  "js_int",

src/client_server/membership.rs

@@ -2,7 +2,7 @@ use super::State;
 use crate::{
     client_server,
     pdu::{PduBuilder, PduEvent},
-    server_server, utils, ConduitResult, Database, Error, Ruma,
+    server_server, utils, ConduitResult, Database, Error, Result, Ruma,
 };
 use log::warn;
 use ruma::{
@@ -17,11 +17,14 @@ use ruma::{
         },
         federation,
     },
+    events::pdu::Pdu,
     events::{room::member, EventType},
     EventId, Raw, RoomId, RoomVersionId, UserId,
 };
 use state_res::StateEvent;
-use std::{collections::BTreeMap, convert::TryFrom, sync::Arc};
+use std::{
+    collections::BTreeMap, collections::HashMap, collections::HashSet, convert::TryFrom, sync::Arc,
+};
 
 #[cfg(feature = "conduit_bin")]
 use rocket::{get, post};
@@ -482,12 +485,7 @@ async fn join_room_by_id_helper(
         )
         .await?;
 
-        let mut event_map = send_join_response
+        let add_event_id = |pdu: &Raw<Pdu>| {
-            .room_state
-            .state
-            .iter()
-            .chain(send_join_response.room_state.auth_chain.iter())
-            .map(|pdu| {
             let mut value = serde_json::from_str(pdu.json().get())
                 .expect("converting raw jsons to values always works");
             let event_id = EventId::try_from(&*format!(
@@ -502,16 +500,34 @@ async fn join_room_by_id_helper(
                 .ok_or_else(|| Error::BadServerResponse("PDU is not an object."))?
                 .insert("event_id".to_owned(), event_id.to_string().into());
 
-                dbg!(&value);
+            Ok((event_id, value))
+        };
 
+        let room_state = send_join_response.room_state.state.iter().map(add_event_id);
+
+        let state_events = room_state
+            .clone()
+            .map(|pdu: Result<(EventId, serde_json::Value)>| Ok(pdu?.0))
+            .collect::<Result<HashSet<EventId>>>()?;
+
+        let auth_chain = send_join_response
+            .room_state
+            .auth_chain
+            .iter()
+            .map(add_event_id);
+
+        let mut event_map = room_state
+            .chain(auth_chain)
+            .map(|r| {
+                let (event_id, value) = r?;
                 serde_json::from_value::<StateEvent>(value)
-                    .map(|ev| (dbg!(&ev).event_id().clone(), Arc::new(ev)))
+                    .map(|ev| (event_id, Arc::new(ev)))
                     .map_err(|e| {
                         warn!("{}", e);
                         Error::BadServerResponse("Invalid PDU bytes in send_join response.")
                     })
             })
-            .collect::<Result<BTreeMap<EventId, Arc<StateEvent>>, _>>()?;
+            .collect::<Result<BTreeMap<EventId, Arc<StateEvent>>>>()?;
 
         let control_events = event_map
             .values()
@@ -575,6 +591,8 @@ async fn join_room_by_id_helper(
         )
         .expect("iterative auth check failed on resolved events");
 
+        let mut state = HashMap::new();
+
         // filter the events that failed the auth check keeping the remaining events
         // sorted correctly
         for ev_id in sorted_event_ids
@@ -587,9 +605,22 @@ async fn join_room_by_id_helper(
                 .expect("Found event_id in sorted events that is not in resolved state");
 
             // We do not rebuild the PDU in this case only insert to DB
+            let pdu_id =
                 db.rooms
-                .append_pdu(PduEvent::from(&**pdu), &db.globals, &db.account_data)?;
+                    .append_pdu(&PduEvent::from(&**pdu), &db.globals, &db.account_data)?;
+
+            if state_events.contains(ev_id) {
+                state.insert(
+                    (
+                        pdu.kind(),
+                        pdu.state_key().expect("State events have a state key"),
+                    ),
+                    pdu_id,
+                );
             }
+        }
+
+        db.rooms.force_state(room_id, state)?;
     } else {
         let event = member::MemberEventContent {
             membership: member::MembershipState::Join,

src/database/rooms.rs

@@ -220,6 +220,31 @@ impl Rooms {
             .is_some())
     }
 
+    /// Returns the full room state.
+    pub fn force_state(
+        &self,
+        room_id: &RoomId,
+        state: HashMap<(EventType, String), Vec<u8>>,
+    ) -> Result<()> {
+        let state_hash =
+            self.calculate_hash(&state.values().map(|pdu_id| &**pdu_id).collect::<Vec<_>>())?;
+        let mut prefix = state_hash.clone();
+        prefix.push(0xff);
+
+        for ((event_type, state_key), pdu_id) in state {
+            let mut state_id = prefix.clone();
+            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)?;
+        }
+
+        self.roomid_statehash
+            .insert(room_id.as_bytes(), &*state_hash)?;
+
+        Ok(())
+    }
+
     /// Returns the full room state.
     pub fn room_state_full(
         &self,
@@ -446,10 +471,10 @@ impl Rooms {
     /// Creates a new persisted data unit and adds it to a room.
     pub fn append_pdu(
         &self,
-        pdu: PduEvent,
+        pdu: &PduEvent,
         globals: &super::globals::Globals,
         account_data: &super::account_data::AccountData,
-    ) -> Result<EventId> {
+    ) -> Result<Vec<u8>> {
         let mut pdu_json = serde_json::to_value(&pdu).expect("event is valid, we just created it");
         ruma::signatures::hash_and_sign_event(
             globals.server_name().as_str(),
@@ -473,10 +498,6 @@ impl Rooms {
         self.eventid_pduid
             .insert(pdu.event_id.as_bytes(), &*pdu_id)?;
 
-        if pdu.state_key.is_some() {
-            self.append_to_state(&pdu_id, &pdu)?;
-        }
-
         match pdu.kind {
             EventType::RoomRedaction => {
                 if let Some(redact_id) = &pdu.redacts {
@@ -484,23 +505,22 @@ impl Rooms {
                 }
             }
             EventType::RoomMember => {
-                if let Some(state_key) = pdu.state_key {
+                if let Some(state_key) = &pdu.state_key {
                     // if the state_key fails
-                    let target_user_id = UserId::try_from(state_key)
+                    let target_user_id = UserId::try_from(state_key.clone())
                         .expect("This state_key was previously validated");
                     // Update our membership info, we do this here incase a user is invited
                     // and immediately leaves we need the DB to record the invite event for auth
                     self.update_membership(
                         &pdu.room_id,
                         &target_user_id,
-                        serde_json::from_value::<member::MemberEventContent>(pdu.content).map_err(
+                        serde_json::from_value::<member::MemberEventContent>(pdu.content.clone())
-                            |_| {
+                            .map_err(|_| {
                             Error::BadRequest(
                                 ErrorKind::InvalidParam,
                                 "Invalid redaction event content.",
                             )
-                            },
+                        })?,
-                        )?,
                         &pdu.sender,
                         account_data,
                         globals,
@@ -528,7 +548,7 @@ impl Rooms {
         self.edus
             .private_read_set(&pdu.room_id, &pdu.sender, index, &globals)?;
 
-        Ok(pdu.event_id)
+        Ok(pdu_id)
     }
 
     /// Generates a new StateHash and associates it with the incoming event.
@@ -789,7 +809,13 @@ impl Rooms {
         ))
         .expect("ruma's reference hashes are valid event ids");
 
-        self.append_pdu(pdu, globals, account_data)
+        let pdu_id = self.append_pdu(&pdu, globals, account_data)?;
+
+        if pdu.state_key.is_some() {
+            self.append_to_state(&pdu_id, &pdu)?;
+        }
+
+        Ok(pdu.event_id)
     }
 
     /// Returns an iterator over all PDUs in a room.
@@ -953,19 +979,17 @@ impl Rooms {
                     self.roomuseroncejoinedids.insert(&userroom_id, &[])?;
 
                     // Check if the room has a predecessor
-                    if let Some(predecessor) = serde_json::from_value::<
+                    if let Some(predecessor) = self
+                        .room_state_get(&room_id, &EventType::RoomCreate, "")?
+                        .and_then(|create| {
+                            serde_json::from_value::<
                                 Raw<ruma::events::room::create::CreateEventContent>,
-                    >(
+                            >(create.content)
-                        self.room_state_get(&room_id, &EventType::RoomCreate, "")?
-                            .ok_or_else(|| {
-                                Error::bad_database("Found room without m.room.create event.")
-                            })?
-                            .content,
-                    )
                             .expect("Raw::from_value always works")
                             .deserialize()
-                    .map_err(|_| Error::bad_database("Invalid room event in database."))?
+                            .ok()
-                    .predecessor
+                        })
+                        .and_then(|content| content.predecessor)
                     {
                         // Copy user settings from predecessor to the current room:
                         // - Push rules

src/server_server.rs

@@ -1,4 +1,4 @@
-use crate::{client_server, ConduitResult, Database, Error, Result, Ruma};
+use crate::{client_server, ConduitResult, Database, Error, PduEvent, Result, Ruma};
 use http::header::{HeaderValue, AUTHORIZATION};
 use rocket::{get, post, put, response::content::Json, State};
 use ruma::{
@@ -270,9 +270,11 @@ pub fn send_transaction_message_route<'a>(
     db: State<'a, Database>,
     body: Ruma<send_transaction_message::v1::Request<'_>>,
 ) -> ConduitResult<send_transaction_message::v1::Response> {
-    dbg!(&*body);
+    //dbg!(&*body);
     for pdu in &body.pdus {
-        let mut value = serde_json::to_value(pdu).expect("all ruma pdus are json values");
+        let mut value = serde_json::from_str(pdu.json().get())
+            .expect("converting raw jsons to values always works");
+
         let event_id = EventId::try_from(&*format!(
             "${}",
             ruma::signatures::reference_hash(&value).expect("ruma can calculate reference hashes")
@@ -284,11 +286,11 @@ pub fn send_transaction_message_route<'a>(
             .expect("ruma pdus are json objects")
             .insert("event_id".to_owned(), event_id.to_string().into());
 
-        db.rooms.append_pdu(
+        let pdu =
-            serde_json::from_value(value).expect("all ruma pdus are conduit pdus"),
+            serde_json::from_value::<PduEvent>(value).expect("all ruma pdus are conduit pdus");
-            &db.globals,
+        if db.rooms.exists(&pdu.room_id)? {
-            &db.account_data,
+            db.rooms.append_pdu(&pdu, &db.globals, &db.account_data)?;
-        )?;
+        }
     }
     Ok(send_transaction_message::v1::Response {
         pdus: BTreeMap::new(),