|
|
|
|
@ -1,7 +1,4 @@
@@ -1,7 +1,4 @@
|
|
|
|
|
use crate::{ |
|
|
|
|
client_server, database::rooms::ClosestParent, 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, warn}; |
|
|
|
|
@ -21,10 +18,10 @@ use ruma::{
@@ -21,10 +18,10 @@ use ruma::{
|
|
|
|
|
OutgoingRequest, |
|
|
|
|
}, |
|
|
|
|
directory::{IncomingFilter, IncomingRoomNetwork}, |
|
|
|
|
EventId, RoomId, ServerName, ServerSigningKeyId, UserId, |
|
|
|
|
EventId, RoomId, RoomVersionId, ServerName, ServerSigningKeyId, UserId, |
|
|
|
|
}; |
|
|
|
|
use std::{ |
|
|
|
|
collections::BTreeMap, |
|
|
|
|
collections::{BTreeMap, BTreeSet}, |
|
|
|
|
convert::TryFrom, |
|
|
|
|
fmt::Debug, |
|
|
|
|
sync::Arc, |
|
|
|
|
@ -390,6 +387,34 @@ pub async fn get_public_rooms_route(
@@ -390,6 +387,34 @@ pub async fn get_public_rooms_route(
|
|
|
|
|
.into()) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
#[derive(Clone, Debug, Ord, PartialOrd, Eq, PartialEq)] |
|
|
|
|
pub enum PrevEvents<T> { |
|
|
|
|
Sequential(T), |
|
|
|
|
Fork(Vec<T>), |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
impl<T> IntoIterator for PrevEvents<T> { |
|
|
|
|
type Item = T; |
|
|
|
|
type IntoIter = std::vec::IntoIter<Self::Item>; |
|
|
|
|
|
|
|
|
|
fn into_iter(self) -> Self::IntoIter { |
|
|
|
|
match self { |
|
|
|
|
Self::Sequential(item) => vec![item].into_iter(), |
|
|
|
|
Self::Fork(list) => list.into_iter(), |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
impl<T: Clone> PrevEvents<T> { |
|
|
|
|
pub fn new(id: &[T]) -> Self { |
|
|
|
|
match id { |
|
|
|
|
[] => panic!("All events must have previous event"), |
|
|
|
|
[single_id] => Self::Sequential(single_id.clone()), |
|
|
|
|
rest => Self::Fork(rest.to_vec()), |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
#[cfg_attr(
|
|
|
|
|
feature = "conduit_bin", |
|
|
|
|
put("/_matrix/federation/v1/send/<_>", data = "<body>") |
|
|
|
|
@ -445,16 +470,23 @@ pub async fn send_transaction_message_route<'a>(
@@ -445,16 +470,23 @@ pub async fn send_transaction_message_route<'a>(
|
|
|
|
|
// 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 pdu_idx = 0; |
|
|
|
|
// This is `for pdu in &body.pdus` but we need to do fancy continue/break so we need loop labels
|
|
|
|
|
'outer: loop { |
|
|
|
|
if pdu_idx == body.pdus.len() { |
|
|
|
|
break 'outer; |
|
|
|
|
} |
|
|
|
|
let pdu = &body.pdus[pdu_idx]; |
|
|
|
|
pdu_idx += 1; |
|
|
|
|
|
|
|
|
|
// Ruma/PduEvent/StateEvent satisfies - 1. Is a valid event, otherwise it is dropped.
|
|
|
|
|
// state-res checks signatures - 2. Passes signature checks, otherwise event is dropped.
|
|
|
|
|
|
|
|
|
|
// 3. Passes hash checks, otherwise it is redacted before being processed further.
|
|
|
|
|
|
|
|
|
|
// TODO: redact event if hashing fails
|
|
|
|
|
let (event_id, value) = crate::pdu::process_incoming_pdu(pdu); |
|
|
|
|
|
|
|
|
|
let pdu = serde_json::from_value::<PduEvent>( |
|
|
|
|
let mut pdu = serde_json::from_value::<PduEvent>( |
|
|
|
|
serde_json::to_value(&value).expect("CanonicalJsonObj is a valid JsonValue"), |
|
|
|
|
) |
|
|
|
|
.expect("all ruma pdus are conduit pdus"); |
|
|
|
|
@ -466,128 +498,469 @@ pub async fn send_transaction_message_route<'a>(
@@ -466,128 +498,469 @@ pub async fn send_transaction_message_route<'a>(
|
|
|
|
|
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 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()); |
|
|
|
|
// The events that must be resolved to catch up to the incoming event
|
|
|
|
|
let mut missing = vec![]; |
|
|
|
|
let mut seen = state_res::StateMap::new(); |
|
|
|
|
let mut seen_id = None; |
|
|
|
|
|
|
|
|
|
let mut prev_ids = vec![PrevEvents::new(&pdu.prev_events)]; |
|
|
|
|
|
|
|
|
|
// This is `while let Some(event_id) = prev_ids.pop_front()` but with a fancy continue
|
|
|
|
|
// in the case of a failed request to the server that sent the event
|
|
|
|
|
'inner: loop { |
|
|
|
|
match prev_ids.pop() { |
|
|
|
|
Some(PrevEvents::Sequential(id)) => match db |
|
|
|
|
.rooms |
|
|
|
|
.pdu_state_hash(&db.rooms.get_pdu_id(&id)?.unwrap_or_default())? |
|
|
|
|
{ |
|
|
|
|
// We found a common ancestor
|
|
|
|
|
Some(state_hash) => { |
|
|
|
|
seen_id = Some(id.clone()); |
|
|
|
|
seen = db.rooms.state_full(&state_hash)?; |
|
|
|
|
if let Some(pdu) = db.rooms.get_pdu(&id)? { |
|
|
|
|
if pdu.state_key.is_some() { |
|
|
|
|
// This becomes the state after the common event
|
|
|
|
|
seen.insert( |
|
|
|
|
(pdu.kind.clone(), pdu.state_key.clone().unwrap()), |
|
|
|
|
pdu, |
|
|
|
|
); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
break 'inner; |
|
|
|
|
} |
|
|
|
|
// We need to fill in information about this event's `prev_events` (parents)
|
|
|
|
|
None => { |
|
|
|
|
match send_request( |
|
|
|
|
&db.globals, |
|
|
|
|
body.body.origin.clone(), |
|
|
|
|
ruma::api::federation::event::get_event::v1::Request::new(&id), |
|
|
|
|
) |
|
|
|
|
.await |
|
|
|
|
{ |
|
|
|
|
Ok(res) => { |
|
|
|
|
let (_, val) = crate::pdu::process_incoming_pdu(&res.pdu); |
|
|
|
|
let prev_pdu = serde_json::from_value::<PduEvent>( |
|
|
|
|
serde_json::to_value(&val) |
|
|
|
|
.expect("CanonicalJsonObj is a valid JsonValue"), |
|
|
|
|
) |
|
|
|
|
.expect("all ruma pdus are conduit pdus"); |
|
|
|
|
|
|
|
|
|
// TODO: do we need this
|
|
|
|
|
assert_eq!(room_id, &prev_pdu.room_id); |
|
|
|
|
|
|
|
|
|
prev_ids.push(PrevEvents::new(&prev_pdu.prev_events)); |
|
|
|
|
missing.push(PrevEvents::Sequential(prev_pdu)); |
|
|
|
|
} |
|
|
|
|
// 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())); |
|
|
|
|
// We have to give up on this PDU
|
|
|
|
|
continue 'outer; |
|
|
|
|
} |
|
|
|
|
}; |
|
|
|
|
} |
|
|
|
|
}, |
|
|
|
|
Some(PrevEvents::Fork(ids)) => { |
|
|
|
|
error!( |
|
|
|
|
"prev_events > 1: {}", |
|
|
|
|
serde_json::to_string_pretty(&pdu).unwrap() |
|
|
|
|
); |
|
|
|
|
// Don't kill our server with state-res
|
|
|
|
|
// TODO: set this at a reasonable level this is for debug/wip purposes
|
|
|
|
|
if ids.len() > 5 { |
|
|
|
|
error!( |
|
|
|
|
"prev_events > 1: {}", |
|
|
|
|
serde_json::to_string_pretty(&pdu).unwrap() |
|
|
|
|
); |
|
|
|
|
resolved_map.insert( |
|
|
|
|
event_id, |
|
|
|
|
Err("Previous events are too large for state-res".into()), |
|
|
|
|
); |
|
|
|
|
continue 'outer; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
db.rooms.append_to_state(&pdu_id, &pdu)?; |
|
|
|
|
// We want this to stay unique incase the fork comes together?
|
|
|
|
|
let mut prev_fork_ids = BTreeSet::new(); |
|
|
|
|
let mut missing_fork = vec![]; |
|
|
|
|
for id in &ids { |
|
|
|
|
match db |
|
|
|
|
.rooms |
|
|
|
|
.pdu_state_hash(&db.rooms.get_pdu_id(&id)?.unwrap_or_default())? |
|
|
|
|
{ |
|
|
|
|
// We found a common ancestor
|
|
|
|
|
Some(state_hash) => { |
|
|
|
|
seen_id = Some(id.clone()); |
|
|
|
|
seen = db.rooms.state_full(&state_hash)?; |
|
|
|
|
if let Some(pdu) = db.rooms.get_pdu(&id)? { |
|
|
|
|
if pdu.state_key.is_some() { |
|
|
|
|
// This becomes the state after the common event
|
|
|
|
|
seen.insert( |
|
|
|
|
(pdu.kind.clone(), pdu.state_key.clone().unwrap()), |
|
|
|
|
pdu, |
|
|
|
|
); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
break 'inner; |
|
|
|
|
} |
|
|
|
|
None => match send_request( |
|
|
|
|
&db.globals, |
|
|
|
|
body.body.origin.clone(), |
|
|
|
|
ruma::api::federation::event::get_event::v1::Request::new(&id), |
|
|
|
|
) |
|
|
|
|
.await |
|
|
|
|
{ |
|
|
|
|
Ok(res) => { |
|
|
|
|
let (_, val) = crate::pdu::process_incoming_pdu(&res.pdu); |
|
|
|
|
let prev_pdu = serde_json::from_value::<PduEvent>( |
|
|
|
|
serde_json::to_value(&val) |
|
|
|
|
.expect("CanonicalJsonObj is a valid JsonValue"), |
|
|
|
|
) |
|
|
|
|
.expect("all ruma pdus are conduit pdus"); |
|
|
|
|
|
|
|
|
|
// TODO: do we need this
|
|
|
|
|
assert_eq!(room_id, &prev_pdu.room_id); |
|
|
|
|
|
|
|
|
|
for id in &prev_pdu.prev_events { |
|
|
|
|
prev_fork_ids.insert(id.clone()); |
|
|
|
|
} |
|
|
|
|
missing_fork.push(prev_pdu); |
|
|
|
|
} |
|
|
|
|
// We can't hard fail because there are some valid errors, just
|
|
|
|
|
Err(err) => { |
|
|
|
|
resolved_map.insert(event_id, Err(err.to_string())); |
|
|
|
|
// We have to give up on this PDU
|
|
|
|
|
continue 'outer; |
|
|
|
|
} |
|
|
|
|
}, |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
prev_ids.push(PrevEvents::new( |
|
|
|
|
&prev_fork_ids.into_iter().collect::<Vec<_>>(), |
|
|
|
|
)); |
|
|
|
|
missing.push(PrevEvents::new(&missing_fork)); |
|
|
|
|
} |
|
|
|
|
// All done finding missing events
|
|
|
|
|
None => { |
|
|
|
|
break 'inner; |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
db.rooms.append_pdu( |
|
|
|
|
&pdu, |
|
|
|
|
&value, |
|
|
|
|
count, |
|
|
|
|
pdu_id.into(), |
|
|
|
|
&db.globals, |
|
|
|
|
&db.account_data, |
|
|
|
|
&db.admin, |
|
|
|
|
)?; |
|
|
|
|
// We can treat this event as sequential and simply apply it against the current state of the room
|
|
|
|
|
// because we know that state
|
|
|
|
|
if missing.is_empty() { |
|
|
|
|
// Back to the original incoming event
|
|
|
|
|
// If it is a non state event we still must add it and associate a statehash with the pdu_id
|
|
|
|
|
if value.get("state_key").is_none() { |
|
|
|
|
// TODO: Some auth needs to be done for non state events
|
|
|
|
|
if !db.rooms.is_joined(&pdu.sender, room_id)? { |
|
|
|
|
error!("Sender is not joined {}", pdu.kind); |
|
|
|
|
resolved_map.insert(event_id, Err("Sender not found in room".into())); |
|
|
|
|
continue 'outer; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
resolved_map.insert(event_id, Ok::<(), String>(())); |
|
|
|
|
continue; |
|
|
|
|
append_state(&db, &pdu)?; |
|
|
|
|
resolved_map.insert(event_id, Ok::<(), String>(())); |
|
|
|
|
continue 'outer; |
|
|
|
|
} else { |
|
|
|
|
let incoming = pdu.convert_for_state_res(); |
|
|
|
|
match state_res::StateResolution::apply_event( |
|
|
|
|
room_id, |
|
|
|
|
&RoomVersionId::Version6, |
|
|
|
|
incoming.clone(), |
|
|
|
|
&seen |
|
|
|
|
.iter() |
|
|
|
|
.map(|(k, v)| (k.clone(), v.event_id.clone())) |
|
|
|
|
.collect::<BTreeMap<_, _>>(), |
|
|
|
|
Some( |
|
|
|
|
seen.iter() |
|
|
|
|
.map(|(_k, v)| (v.event_id.clone(), v.convert_for_state_res())) |
|
|
|
|
.collect::<BTreeMap<_, _>>(), |
|
|
|
|
), // TODO: make mut and keep around, this is all the auth events
|
|
|
|
|
&db.rooms, |
|
|
|
|
) { |
|
|
|
|
Ok(true) => { |
|
|
|
|
append_state(&db, &pdu)?; |
|
|
|
|
resolved_map.insert(event_id, Ok::<(), String>(())); |
|
|
|
|
continue 'outer; |
|
|
|
|
} |
|
|
|
|
Ok(false) => { |
|
|
|
|
resolved_map.insert(event_id, Err("Failed event auth".to_string())); |
|
|
|
|
error!("Failed sequential event auth for incoming"); |
|
|
|
|
continue 'outer; |
|
|
|
|
} |
|
|
|
|
Err(err) => { |
|
|
|
|
resolved_map.insert(event_id, Err(err.to_string())); |
|
|
|
|
error!("{}", err); |
|
|
|
|
continue 'outer; |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// 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; |
|
|
|
|
} |
|
|
|
|
}; |
|
|
|
|
// Well, now we have to actually do a bunch of work :(
|
|
|
|
|
// The steps are as follows
|
|
|
|
|
// 1. Rebuild the sending servers forward extremity, ignoring our own fork
|
|
|
|
|
// a) iterate "oldest" -> most recent authenticating each event with the state after the previous
|
|
|
|
|
// b) build a `snapshot_map` containing the state after the event for each missing event (EventId -> StateMap<EventId>)
|
|
|
|
|
// 2. Build our side of the fork (TODO do we have to re-auth these, is state at an event relative to the server its from)
|
|
|
|
|
// 3. resolve the two states (our current with the state after the most recent missing event)
|
|
|
|
|
|
|
|
|
|
let their_current_state = get_state_response |
|
|
|
|
.pdus |
|
|
|
|
// Now build up state
|
|
|
|
|
let mut state_snapshot = seen |
|
|
|
|
.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"), |
|
|
|
|
), |
|
|
|
|
) |
|
|
|
|
}) |
|
|
|
|
.map(|(k, v)| (k.clone(), v.event_id.clone())) |
|
|
|
|
.collect::<BTreeMap<_, _>>(); |
|
|
|
|
|
|
|
|
|
let our_current_state = db.rooms.room_state_full(room_id)?; |
|
|
|
|
// State resolution takes care of these checks
|
|
|
|
|
// TODO: So this is super memory inefficient we clone the state_snapshot every time
|
|
|
|
|
// we need it for state events and non
|
|
|
|
|
let mut snapshot_map = BTreeMap::new(); |
|
|
|
|
snapshot_map.insert(seen_id.clone().unwrap(), state_snapshot.clone()); |
|
|
|
|
|
|
|
|
|
let accum_event_map = seen |
|
|
|
|
.iter() |
|
|
|
|
.map(|(_, v)| (v.event_id.clone(), v.convert_for_state_res())) |
|
|
|
|
.chain( |
|
|
|
|
missing |
|
|
|
|
.iter() |
|
|
|
|
.cloned() |
|
|
|
|
.flatten() |
|
|
|
|
.map(|pdu| (pdu.event_id.clone(), pdu.convert_for_state_res())), |
|
|
|
|
) |
|
|
|
|
.collect::<BTreeMap<_, Arc<state_res::StateEvent>>>(); |
|
|
|
|
|
|
|
|
|
// 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.
|
|
|
|
|
for missing_pdu in missing.iter().rev() { |
|
|
|
|
match missing_pdu { |
|
|
|
|
PrevEvents::Sequential(missing_pdu) => { |
|
|
|
|
// For state events
|
|
|
|
|
if missing_pdu.state_key.is_some() { |
|
|
|
|
let missing_pdu = missing_pdu.convert_for_state_res(); |
|
|
|
|
match state_res::StateResolution::apply_event( |
|
|
|
|
room_id, |
|
|
|
|
&RoomVersionId::Version6, |
|
|
|
|
missing_pdu.clone(), |
|
|
|
|
&state_snapshot, |
|
|
|
|
Some(accum_event_map.clone()), // TODO: make mut and check on Ok(..) ?
|
|
|
|
|
&db.rooms, |
|
|
|
|
) { |
|
|
|
|
Ok(true) => { |
|
|
|
|
// TODO: do we need this
|
|
|
|
|
assert_eq!(room_id, missing_pdu.room_id()); |
|
|
|
|
|
|
|
|
|
// We can't add to DB yet since we don't know if it passes
|
|
|
|
|
// current room state
|
|
|
|
|
// append_state(&db, &PduEvent::from(&*missing_pdu))?;
|
|
|
|
|
|
|
|
|
|
// Only after the state is recorded in the DB can we update the state_snapshot
|
|
|
|
|
// This will update the state snapshot so it is correct next loop through
|
|
|
|
|
state_snapshot.insert( |
|
|
|
|
(missing_pdu.kind(), missing_pdu.state_key()), |
|
|
|
|
missing_pdu.event_id(), |
|
|
|
|
); |
|
|
|
|
// Keep track of the state after for resolution
|
|
|
|
|
snapshot_map.insert(missing_pdu.event_id(), state_snapshot.clone()); |
|
|
|
|
} |
|
|
|
|
Ok(false) => { |
|
|
|
|
error!( |
|
|
|
|
"apply missing: {}", |
|
|
|
|
serde_json::to_string_pretty(&*missing_pdu).unwrap() |
|
|
|
|
); |
|
|
|
|
continue; |
|
|
|
|
} |
|
|
|
|
Err(e) => { |
|
|
|
|
error!("{}", e); |
|
|
|
|
// This is not a fatal error but we do eventually need to handle
|
|
|
|
|
// events failing that are not the incoming events
|
|
|
|
|
// TODO: what to do when missing events fail (not incoming events)
|
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} else { |
|
|
|
|
// TODO: Some auth needs to be done for non state events
|
|
|
|
|
if !db |
|
|
|
|
.rooms |
|
|
|
|
.is_joined(&missing_pdu.sender, &missing_pdu.room_id)? |
|
|
|
|
{ |
|
|
|
|
error!("fork Sender is not joined {}", missing_pdu.kind); |
|
|
|
|
// TODO: we probably should not be getting events for different rooms
|
|
|
|
|
continue; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// 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) => { |
|
|
|
|
// TODO: a better way to signal non state events...
|
|
|
|
|
snapshot_map.insert(missing_pdu.event_id.clone(), state_snapshot.clone()); |
|
|
|
|
// Continue this inner for loop adding all the missing events
|
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
PrevEvents::Fork(pdus) => { |
|
|
|
|
let mut state_sets = BTreeSet::new(); |
|
|
|
|
for pdu in pdus { |
|
|
|
|
let mut state_at = state_snapshot.clone(); |
|
|
|
|
if pdu.state_key.is_some() { |
|
|
|
|
state_at.insert( |
|
|
|
|
(pdu.kind.clone(), pdu.state_key.clone().unwrap()), |
|
|
|
|
pdu.event_id.clone(), |
|
|
|
|
); |
|
|
|
|
} |
|
|
|
|
state_sets.insert(state_at); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
if state_sets.len() <= 1 { |
|
|
|
|
for missing_pdu in pdus.iter() { |
|
|
|
|
// For state events
|
|
|
|
|
if missing_pdu.state_key.is_some() { |
|
|
|
|
let missing_pdu = missing_pdu.convert_for_state_res(); |
|
|
|
|
match state_res::StateResolution::apply_event( |
|
|
|
|
room_id, |
|
|
|
|
&RoomVersionId::Version6, |
|
|
|
|
missing_pdu.clone(), |
|
|
|
|
&state_snapshot, |
|
|
|
|
Some(accum_event_map.clone()), // TODO: make mut and check on Ok(..) ?
|
|
|
|
|
&db.rooms, |
|
|
|
|
) { |
|
|
|
|
Ok(true) => { |
|
|
|
|
// TODO: do we need this
|
|
|
|
|
assert_eq!(room_id, missing_pdu.room_id()); |
|
|
|
|
|
|
|
|
|
state_snapshot.insert( |
|
|
|
|
(missing_pdu.kind(), missing_pdu.state_key()), |
|
|
|
|
missing_pdu.event_id(), |
|
|
|
|
); |
|
|
|
|
snapshot_map |
|
|
|
|
.insert(missing_pdu.event_id(), state_snapshot.clone()); |
|
|
|
|
} |
|
|
|
|
Ok(false) => { |
|
|
|
|
error!( |
|
|
|
|
"apply missing fork: {}", |
|
|
|
|
serde_json::to_string_pretty(&*missing_pdu).unwrap() |
|
|
|
|
); |
|
|
|
|
continue; |
|
|
|
|
} |
|
|
|
|
Err(e) => { |
|
|
|
|
error!("fork state-res: {}", e); |
|
|
|
|
// TODO: what to do when missing events fail (not incoming events)
|
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} else { |
|
|
|
|
// TODO: Some auth needs to be done for non state events
|
|
|
|
|
if !db |
|
|
|
|
.rooms |
|
|
|
|
.is_joined(&missing_pdu.sender, &missing_pdu.room_id)? |
|
|
|
|
{ |
|
|
|
|
error!("fork Sender is not joined {}", missing_pdu.kind); |
|
|
|
|
// TODO: we probably should not be getting events for different rooms
|
|
|
|
|
continue; |
|
|
|
|
} |
|
|
|
|
snapshot_map |
|
|
|
|
.insert(missing_pdu.event_id.clone(), state_snapshot.clone()); |
|
|
|
|
// Continue this inner for loop adding all the missing events
|
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} else { |
|
|
|
|
match state_res::StateResolution::resolve( |
|
|
|
|
room_id, |
|
|
|
|
&RoomVersionId::Version6, |
|
|
|
|
&state_sets.into_iter().collect::<Vec<_>>(), |
|
|
|
|
Some(accum_event_map.clone()), // TODO: make mut and check on Ok(..) ?
|
|
|
|
|
&db.rooms, |
|
|
|
|
) { |
|
|
|
|
Ok(resolved) => { |
|
|
|
|
for id in resolved |
|
|
|
|
.values() |
|
|
|
|
.filter(|id| pdus.iter().any(|pduid| pduid.event_id == **id)) |
|
|
|
|
{ |
|
|
|
|
snapshot_map.insert(id.clone(), resolved.clone()); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
Err(err) => { |
|
|
|
|
error!("{}", err); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// We know these events are valid for each state at the event
|
|
|
|
|
// we have already authed them
|
|
|
|
|
let known_fork = if let Some(id) = seen_id { |
|
|
|
|
db.rooms.state_from(room_id, &id)? |
|
|
|
|
} else { |
|
|
|
|
vec![] |
|
|
|
|
}; |
|
|
|
|
for pdu in known_fork.clone().into_iter().rev() { |
|
|
|
|
if pdu.state_key.is_some() { |
|
|
|
|
seen.insert((pdu.kind.clone(), pdu.state_key.clone().unwrap()), pdu); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
let current_room_state = seen |
|
|
|
|
.into_iter() |
|
|
|
|
.map(|(k, v)| (k, v.event_id)) |
|
|
|
|
.collect::<BTreeMap<_, _>>(); |
|
|
|
|
|
|
|
|
|
let mut state_sets = BTreeSet::new(); |
|
|
|
|
state_sets.insert(current_room_state.clone()); |
|
|
|
|
|
|
|
|
|
// The first item is the most recent
|
|
|
|
|
if let Some(fork) = missing.first() { |
|
|
|
|
for pdu in fork.clone().into_iter() { |
|
|
|
|
if let Some(map) = snapshot_map.get(&pdu.event_id) { |
|
|
|
|
state_sets.insert(map.clone()); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
let mut state_sets = state_sets.into_iter().collect::<Vec<_>>(); |
|
|
|
|
// If we have actual differences resolve
|
|
|
|
|
if state_sets.len() > 1 { |
|
|
|
|
// Set the incoming event to have parents from both forks
|
|
|
|
|
// ours/theirs
|
|
|
|
|
pdu.prev_events = missing |
|
|
|
|
.first() |
|
|
|
|
.cloned() |
|
|
|
|
.into_iter() |
|
|
|
|
.flatten() |
|
|
|
|
.map(|pdu| pdu.event_id) |
|
|
|
|
.chain(known_fork.first().cloned().map(|pdu| pdu.event_id)) |
|
|
|
|
.collect(); |
|
|
|
|
|
|
|
|
|
// The incoming event is a child of both forks now
|
|
|
|
|
for set in &mut state_sets { |
|
|
|
|
set.insert( |
|
|
|
|
(pdu.kind.clone(), pdu.state_key.clone().unwrap()), |
|
|
|
|
pdu.event_id.clone(), |
|
|
|
|
); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// If we have holes or a fork I am less sure what can be guaranteed about our state?
|
|
|
|
|
// Or what must be done to fix holes and forks?
|
|
|
|
|
match state_res::StateResolution::resolve( |
|
|
|
|
room_id, |
|
|
|
|
&RoomVersionId::Version6, |
|
|
|
|
&state_sets, |
|
|
|
|
Some(accum_event_map.clone()), // TODO: make mut and check on Ok(..) ?
|
|
|
|
|
&db.rooms, |
|
|
|
|
) { |
|
|
|
|
Ok(resolved) if resolved.values().any(|id| id == &event_id) => { |
|
|
|
|
for resolved_missing in missing |
|
|
|
|
.into_iter() |
|
|
|
|
.rev() // We want the oldest pdu's first
|
|
|
|
|
.flatten() |
|
|
|
|
.filter(|pdu| resolved.values().any(|res_id| res_id == &pdu.event_id)) |
|
|
|
|
{ |
|
|
|
|
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, |
|
|
|
|
&resolved_missing, |
|
|
|
|
&crate::utils::to_canonical_object(&resolved_missing) |
|
|
|
|
.expect("PDU is valid canonical JSON"), |
|
|
|
|
count, |
|
|
|
|
pdu_id.into(), |
|
|
|
|
&db.globals, |
|
|
|
|
@ -595,18 +968,58 @@ pub async fn send_transaction_message_route<'a>(
@@ -595,18 +968,58 @@ pub async fn send_transaction_message_route<'a>(
|
|
|
|
|
&db.admin, |
|
|
|
|
)?; |
|
|
|
|
} |
|
|
|
|
Some(ClosestParent::Insert(old_count)) => { |
|
|
|
|
let count = old_count; |
|
|
|
|
|
|
|
|
|
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()); |
|
|
|
|
|
|
|
|
|
// Since we know the state we force_state
|
|
|
|
|
// saving the incoming pdu's id with our new state
|
|
|
|
|
db.rooms.force_state_with_pdu( |
|
|
|
|
&*pdu_id, |
|
|
|
|
room_id, |
|
|
|
|
resolved |
|
|
|
|
.iter() |
|
|
|
|
.map(|(k, v)| { |
|
|
|
|
( |
|
|
|
|
k.clone(), |
|
|
|
|
serde_json::to_vec( |
|
|
|
|
&db.rooms |
|
|
|
|
.get_pdu(v) |
|
|
|
|
.expect("db err") |
|
|
|
|
.expect("we know of all the pdus"), |
|
|
|
|
) |
|
|
|
|
.expect("serde can serialize pdus"), |
|
|
|
|
) |
|
|
|
|
}) |
|
|
|
|
.collect(), |
|
|
|
|
)?; |
|
|
|
|
db.rooms.append_pdu( |
|
|
|
|
&pdu, |
|
|
|
|
&value, |
|
|
|
|
count, |
|
|
|
|
pdu_id.into(), |
|
|
|
|
&db.globals, |
|
|
|
|
&db.account_data, |
|
|
|
|
&db.admin, |
|
|
|
|
)?; |
|
|
|
|
resolved_map.insert(event_id, Ok::<(), String>(())); |
|
|
|
|
} |
|
|
|
|
Ok(resolved) => { |
|
|
|
|
for resolved_missing in missing |
|
|
|
|
.into_iter() |
|
|
|
|
.flatten() |
|
|
|
|
.filter(|pdu| resolved.values().any(|res_id| res_id == &pdu.event_id)) |
|
|
|
|
{ |
|
|
|
|
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()); |
|
|
|
|
// 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, |
|
|
|
|
&resolved_missing, |
|
|
|
|
&crate::utils::to_canonical_object(&resolved_missing) |
|
|
|
|
.expect("PDU is valid canonical JSON"), |
|
|
|
|
count, |
|
|
|
|
pdu_id.into(), |
|
|
|
|
&db.globals, |
|
|
|
|
@ -614,28 +1027,78 @@ pub async fn send_transaction_message_route<'a>(
@@ -614,28 +1027,78 @@ pub async fn send_transaction_message_route<'a>(
|
|
|
|
|
&db.admin, |
|
|
|
|
)?; |
|
|
|
|
} |
|
|
|
|
_ => { |
|
|
|
|
error!("Not a sequential event or no parents found"); |
|
|
|
|
continue; |
|
|
|
|
} |
|
|
|
|
resolved_map.insert(event_id, Err("Failed event auth".into())); |
|
|
|
|
error!( |
|
|
|
|
"auth failed: {}", |
|
|
|
|
serde_json::to_string_pretty(&pdu).unwrap() |
|
|
|
|
); |
|
|
|
|
} |
|
|
|
|
Err(err) => { |
|
|
|
|
resolved_map.insert(event_id, Err(err.to_string())); |
|
|
|
|
error!("{}", err); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
// If we have iterated through the incoming missing events sequentially we know that
|
|
|
|
|
// the original incoming event is the youngest child and so can be simply authed and appended
|
|
|
|
|
// to the state
|
|
|
|
|
} else { |
|
|
|
|
// If it is a non state event we still must add it and associate a statehash with the pdu_id
|
|
|
|
|
if pdu.state_key.is_none() { |
|
|
|
|
// TODO: Some auth needs to be done for non state events
|
|
|
|
|
if !db.rooms.is_joined(&pdu.sender, room_id)? { |
|
|
|
|
error!("Sender is not joined {}", pdu.kind); |
|
|
|
|
resolved_map.insert(event_id, Err("Sender not found in room".into())); |
|
|
|
|
continue 'outer; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
append_state(&db, &pdu)?; |
|
|
|
|
resolved_map.insert(event_id, Ok::<(), String>(())); |
|
|
|
|
} else { |
|
|
|
|
let incoming = pdu.convert_for_state_res(); |
|
|
|
|
match state_res::StateResolution::apply_event( |
|
|
|
|
room_id, |
|
|
|
|
&RoomVersionId::Version6, |
|
|
|
|
incoming, |
|
|
|
|
¤t_room_state, |
|
|
|
|
Some(accum_event_map), // TODO: make mut and check on Ok(..) ?
|
|
|
|
|
&db.rooms, |
|
|
|
|
) { |
|
|
|
|
Ok(true) => { |
|
|
|
|
append_state(&db, &pdu)?; |
|
|
|
|
resolved_map.insert(event_id, Ok::<(), String>(())); |
|
|
|
|
} |
|
|
|
|
Ok(false) => { |
|
|
|
|
resolved_map.insert(event_id, Err("Failed event auth".to_string())); |
|
|
|
|
error!("Failed sequential event auth for incoming"); |
|
|
|
|
} |
|
|
|
|
Err(err) => { |
|
|
|
|
resolved_map.insert(event_id, Err(err.to_string())); |
|
|
|
|
error!("{}", err); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
// 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 { pdus: resolved_map }.into()) |
|
|
|
|
Ok(dbg!(send_transaction_message::v1::Response { pdus: resolved_map }).into()) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
fn append_state(db: &Database, pdu: &PduEvent) -> Result<()> { |
|
|
|
|
let count = db.globals.next_count()?; |
|
|
|
|
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, |
|
|
|
|
&utils::to_canonical_object(pdu).expect("Pdu is valid canonical object"), |
|
|
|
|
count, |
|
|
|
|
pdu_id.clone().into(), |
|
|
|
|
&db.globals, |
|
|
|
|
&db.account_data, |
|
|
|
|
&db.admin, |
|
|
|
|
) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
#[cfg_attr(
|
|
|
|
|
|
Devin Ragotzy
5 years ago