|
|
|
@ -21,7 +21,7 @@ use ruma::{ |
|
|
|
OutgoingRequest, |
|
|
|
OutgoingRequest, |
|
|
|
}, |
|
|
|
}, |
|
|
|
directory::{IncomingFilter, IncomingRoomNetwork}, |
|
|
|
directory::{IncomingFilter, IncomingRoomNetwork}, |
|
|
|
EventId, RoomId, ServerName, ServerSigningKeyId, UserId, |
|
|
|
EventId, RoomId, RoomVersionId, ServerName, ServerSigningKeyId, UserId, |
|
|
|
}; |
|
|
|
}; |
|
|
|
use std::{ |
|
|
|
use std::{ |
|
|
|
collections::BTreeMap, |
|
|
|
collections::BTreeMap, |
|
|
|
@ -445,7 +445,14 @@ pub async fn send_transaction_message_route<'a>( |
|
|
|
// discard the event whereas the Client Server API's /send/{eventType} endpoint
|
|
|
|
// discard the event whereas the Client Server API's /send/{eventType} endpoint
|
|
|
|
// would return a M_BAD_JSON error.
|
|
|
|
// would return a M_BAD_JSON error.
|
|
|
|
let mut resolved_map = BTreeMap::new(); |
|
|
|
let mut resolved_map = BTreeMap::new(); |
|
|
|
for pdu in &body.pdus { |
|
|
|
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.
|
|
|
|
// 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.
|
|
|
|
// state-res checks signatures - 2. Passes signature checks, otherwise event is dropped.
|
|
|
|
@ -467,12 +474,13 @@ pub async fn send_transaction_message_route<'a>( |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
// The events that must be resolved to catch up to the incoming event
|
|
|
|
// The events that must be resolved to catch up to the incoming event
|
|
|
|
let mut missing = BTreeMap::new(); |
|
|
|
let mut missing = vec![]; |
|
|
|
let mut seen = BTreeMap::new(); |
|
|
|
let mut seen = BTreeMap::new(); |
|
|
|
|
|
|
|
|
|
|
|
let mut prev_ids = pdu.prev_events.to_vec(); |
|
|
|
let mut prev_ids = pdu.prev_events.to_vec(); |
|
|
|
// Don't kill our server with state-res
|
|
|
|
// Don't kill our server with state-res
|
|
|
|
if prev_ids.len() > 20 { |
|
|
|
// TODO: set this at a reasonable level this is for debug/wip purposes
|
|
|
|
|
|
|
|
if prev_ids.len() > 5 { |
|
|
|
resolved_map.insert( |
|
|
|
resolved_map.insert( |
|
|
|
event_id, |
|
|
|
event_id, |
|
|
|
Err("Event has abnormally large prev_events count".into()), |
|
|
|
Err("Event has abnormally large prev_events count".into()), |
|
|
|
@ -480,39 +488,48 @@ pub async fn send_transaction_message_route<'a>( |
|
|
|
continue; |
|
|
|
continue; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
while let Some(id) = prev_ids.pop() { |
|
|
|
// This is `while let Some(event_id) = prev_ids.pop()` but with a fancy continue
|
|
|
|
|
|
|
|
// in the case of a failed request to the server that sent the event
|
|
|
|
|
|
|
|
'inner: loop { |
|
|
|
|
|
|
|
let id = if let Some(id) = prev_ids.pop() { |
|
|
|
|
|
|
|
id |
|
|
|
|
|
|
|
} else { |
|
|
|
|
|
|
|
break 'inner; |
|
|
|
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
|
|
match db |
|
|
|
match db |
|
|
|
.rooms |
|
|
|
.rooms |
|
|
|
.pdu_state_hash(&db.rooms.get_pdu_id(&id)?.unwrap_or_default())? |
|
|
|
.pdu_state_hash(&db.rooms.get_pdu_id(&id)?.unwrap_or_default())? |
|
|
|
{ |
|
|
|
{ |
|
|
|
|
|
|
|
// We know the state snapshot for this events parents so we can simply auth the
|
|
|
|
|
|
|
|
// incoming event and append to DB and append to state if it passes
|
|
|
|
Some(state_hash) => { |
|
|
|
Some(state_hash) => { |
|
|
|
seen = db.rooms.state_full(&state_hash)?; |
|
|
|
seen = db.rooms.state_full(&state_hash)?; |
|
|
|
break; |
|
|
|
break 'inner; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
// We need to fill in information about this event's `prev_events` (parents)
|
|
|
|
None => { |
|
|
|
None => { |
|
|
|
// TODO: as soon as https://github.com/ruma/ruma/pull/364 is accepted
|
|
|
|
|
|
|
|
// use `::get_event::v1::Request...`
|
|
|
|
|
|
|
|
let last_event = vec![id.clone()]; |
|
|
|
|
|
|
|
let mut req = |
|
|
|
|
|
|
|
ruma::api::federation::event::get_missing_events::v1::Request::new( |
|
|
|
|
|
|
|
room_id, |
|
|
|
|
|
|
|
&[], |
|
|
|
|
|
|
|
&last_event, |
|
|
|
|
|
|
|
); |
|
|
|
|
|
|
|
req.limit = ruma::uint!(1); |
|
|
|
|
|
|
|
// We have a state event so we need info for state-res
|
|
|
|
// We have a state event so we need info for state-res
|
|
|
|
let get_event = match send_request(&db.globals, body.body.origin.clone(), req) |
|
|
|
match send_request( |
|
|
|
.await |
|
|
|
&db.globals, |
|
|
|
|
|
|
|
body.body.origin.clone(), |
|
|
|
|
|
|
|
ruma::api::federation::event::get_event::v1::Request::new(&id), |
|
|
|
|
|
|
|
) |
|
|
|
|
|
|
|
.await |
|
|
|
{ |
|
|
|
{ |
|
|
|
Ok(res) => { |
|
|
|
Ok(res) => { |
|
|
|
for ev in &res.events { |
|
|
|
let (_, val) = crate::pdu::process_incoming_pdu(&res.pdu); |
|
|
|
let (id, val) = crate::pdu::process_incoming_pdu(ev); |
|
|
|
let prev_pdu = serde_json::from_value::<PduEvent>( |
|
|
|
let pdu = state_res::StateEvent::from_id_canon_obj(id.clone(), val) |
|
|
|
serde_json::to_value(&val) |
|
|
|
.expect("Pdu is a valid StateEvent"); |
|
|
|
.expect("CanonicalJsonObj is a valid JsonValue"), |
|
|
|
|
|
|
|
) |
|
|
|
prev_ids.extend(pdu.prev_event_ids()); |
|
|
|
.expect("all ruma pdus are conduit pdus"); |
|
|
|
missing.insert(id, pdu); |
|
|
|
|
|
|
|
} |
|
|
|
// TODO: do we need this
|
|
|
|
|
|
|
|
assert_eq!(room_id, &prev_pdu.room_id); |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
prev_ids.extend(prev_pdu.prev_events.to_vec()); |
|
|
|
|
|
|
|
missing.push(prev_pdu); |
|
|
|
} |
|
|
|
} |
|
|
|
// We can't hard fail because there are some valid errors, just
|
|
|
|
// We can't hard fail because there are some valid errors, just
|
|
|
|
// keep checking PDU's
|
|
|
|
// keep checking PDU's
|
|
|
|
@ -521,18 +538,142 @@ pub async fn send_transaction_message_route<'a>( |
|
|
|
// {"errcode":"M_FORBIDDEN","error":"Host not in room."}
|
|
|
|
// {"errcode":"M_FORBIDDEN","error":"Host not in room."}
|
|
|
|
Err(err) => { |
|
|
|
Err(err) => { |
|
|
|
resolved_map.insert(event_id, Err(err.to_string())); |
|
|
|
resolved_map.insert(event_id, Err(err.to_string())); |
|
|
|
continue; |
|
|
|
// We have to give up on this PDU
|
|
|
|
|
|
|
|
continue 'outer; |
|
|
|
} |
|
|
|
} |
|
|
|
}; |
|
|
|
}; |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
// If it is not a state event, we can skip state-res... maybe
|
|
|
|
// Now build up
|
|
|
|
|
|
|
|
let mut state_snapshot = seen |
|
|
|
|
|
|
|
.iter() |
|
|
|
|
|
|
|
.map(|(k, v)| (k.clone(), v.event_id.clone())) |
|
|
|
|
|
|
|
.collect(); |
|
|
|
|
|
|
|
let mut accum_event_map = seen |
|
|
|
|
|
|
|
.iter() |
|
|
|
|
|
|
|
.map(|(_, v)| (v.event_id.clone(), v.convert_for_state_res())) |
|
|
|
|
|
|
|
.collect::<BTreeMap<_, Arc<state_res::StateEvent>>>(); |
|
|
|
|
|
|
|
// TODO: this only accounts for sequentially missing events no holes will be filled
|
|
|
|
|
|
|
|
// and I'm still not sure what happens when fork introduces multiple `prev_events`
|
|
|
|
|
|
|
|
//
|
|
|
|
|
|
|
|
// We need to go from oldest (furthest ancestor of the incoming event) to the
|
|
|
|
|
|
|
|
// prev_event of the incoming event so we reverse the order oldest -> most recent
|
|
|
|
|
|
|
|
for missing_pdu in missing.into_iter().rev() { |
|
|
|
|
|
|
|
// 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()); |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
let count = db.globals.next_count()?; |
|
|
|
|
|
|
|
let mut pdu_id = missing_pdu.room_id().as_bytes().to_vec(); |
|
|
|
|
|
|
|
pdu_id.push(0xff); |
|
|
|
|
|
|
|
pdu_id.extend_from_slice(&count.to_be_bytes()); |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// Since we know the state from event to event we can do the
|
|
|
|
|
|
|
|
// slightly more efficient force_state
|
|
|
|
|
|
|
|
db.rooms.force_state( |
|
|
|
|
|
|
|
room_id, |
|
|
|
|
|
|
|
state_snapshot |
|
|
|
|
|
|
|
.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(), |
|
|
|
|
|
|
|
)?; |
|
|
|
|
|
|
|
// Now append the new missing event to DB it will be part of the
|
|
|
|
|
|
|
|
// next events state_snapshot
|
|
|
|
|
|
|
|
db.rooms.append_pdu( |
|
|
|
|
|
|
|
&PduEvent::from(&*missing_pdu), |
|
|
|
|
|
|
|
&utils::to_canonical_object(&*missing_pdu) |
|
|
|
|
|
|
|
.expect("Pdu is valid canonical object"), |
|
|
|
|
|
|
|
count, |
|
|
|
|
|
|
|
pdu_id.clone().into(), |
|
|
|
|
|
|
|
&db.globals, |
|
|
|
|
|
|
|
&db.account_data, |
|
|
|
|
|
|
|
&db.admin, |
|
|
|
|
|
|
|
)?; |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// 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(), |
|
|
|
|
|
|
|
); |
|
|
|
|
|
|
|
accum_event_map.insert(missing_pdu.event_id(), missing_pdu); |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
Ok(false) => { |
|
|
|
|
|
|
|
error!("{:?}", missing_pdu); |
|
|
|
|
|
|
|
continue; |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
Err(e) => { |
|
|
|
|
|
|
|
error!("{}", e); |
|
|
|
|
|
|
|
// I think we need to keep going until we reach
|
|
|
|
|
|
|
|
// the incoming pdu so do not error here
|
|
|
|
|
|
|
|
continue; // The continue is not needed but to remind us that this is not an error
|
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
// All between the event we know about and the incoming event need to be accounted
|
|
|
|
|
|
|
|
// for
|
|
|
|
|
|
|
|
} 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!("Sender is not joined {}", missing_pdu.kind); |
|
|
|
|
|
|
|
// I think we need to keep going until we reach
|
|
|
|
|
|
|
|
// the incoming pdu so do not error here
|
|
|
|
|
|
|
|
continue; |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
let count = db.globals.next_count()?; |
|
|
|
|
|
|
|
let mut pdu_id = missing_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, &missing_pdu)?; |
|
|
|
|
|
|
|
db.rooms.append_pdu( |
|
|
|
|
|
|
|
&missing_pdu, |
|
|
|
|
|
|
|
&value, |
|
|
|
|
|
|
|
count, |
|
|
|
|
|
|
|
pdu_id.into(), |
|
|
|
|
|
|
|
&db.globals, |
|
|
|
|
|
|
|
&db.account_data, |
|
|
|
|
|
|
|
&db.admin, |
|
|
|
|
|
|
|
)?; |
|
|
|
|
|
|
|
// Continue this inner for loop adding all the missing events
|
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// Back to the original incoming event
|
|
|
|
|
|
|
|
// If it is a non state event we still must add it and create a statehash for it
|
|
|
|
if value.get("state_key").is_none() { |
|
|
|
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)? { |
|
|
|
if !db.rooms.is_joined(&pdu.sender, room_id)? { |
|
|
|
warn!("Sender is not joined {}", pdu.kind); |
|
|
|
error!("Sender is not joined {}", pdu.kind); |
|
|
|
resolved_map.insert(event_id, Err("User is not in this room".into())); |
|
|
|
// I think we need to keep going until we reach
|
|
|
|
|
|
|
|
// the incoming pdu so do not error here
|
|
|
|
continue; |
|
|
|
continue; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
@ -551,183 +692,49 @@ pub async fn send_transaction_message_route<'a>( |
|
|
|
&db.account_data, |
|
|
|
&db.account_data, |
|
|
|
&db.admin, |
|
|
|
&db.admin, |
|
|
|
)?; |
|
|
|
)?; |
|
|
|
|
|
|
|
|
|
|
|
resolved_map.insert(event_id, Ok::<(), String>(())); |
|
|
|
resolved_map.insert(event_id, Ok::<(), String>(())); |
|
|
|
continue; |
|
|
|
continue; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
// We have a state event so we need info for state-res
|
|
|
|
// If we have iterated through the incoming missing events sequentially we know that
|
|
|
|
let get_state_response = match send_request( |
|
|
|
// the original incoming event is the youngest child and so can be simply authed and append
|
|
|
|
&db.globals, |
|
|
|
// to the state
|
|
|
|
body.body.origin.clone(), |
|
|
|
// If we have holes or a fork I am less sure what can be guaranteed about our state?
|
|
|
|
ruma::api::federation::event::get_room_state::v1::Request { |
|
|
|
match state_res::StateResolution::apply_event( |
|
|
|
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"), |
|
|
|
|
|
|
|
), |
|
|
|
|
|
|
|
) |
|
|
|
|
|
|
|
}) |
|
|
|
|
|
|
|
// Add the incoming event to their state this will ensure it is within the
|
|
|
|
|
|
|
|
// resolved state if indeed it passes state-res
|
|
|
|
|
|
|
|
.chain(Some(( |
|
|
|
|
|
|
|
event_id.clone(), |
|
|
|
|
|
|
|
Arc::new( |
|
|
|
|
|
|
|
state_res::StateEvent::from_id_canon_obj(event_id.clone(), value.clone()) |
|
|
|
|
|
|
|
.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, |
|
|
|
room_id, |
|
|
|
&ruma::RoomVersionId::Version6, |
|
|
|
&RoomVersionId::Version6, |
|
|
|
&[ |
|
|
|
pdu.convert_for_state_res(), // We know this a state event
|
|
|
|
our_current_state |
|
|
|
&state_snapshot, |
|
|
|
.iter() |
|
|
|
Some(accum_event_map.clone()), // TODO: make mut and check on Ok(..) ?
|
|
|
|
.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())) |
|
|
|
|
|
|
|
// We must ensure that our incoming event is part of state-res and not
|
|
|
|
|
|
|
|
// accidentally removed from the BTree because of being sorted first by
|
|
|
|
|
|
|
|
// event_id
|
|
|
|
|
|
|
|
.chain(Some(( |
|
|
|
|
|
|
|
( |
|
|
|
|
|
|
|
pdu.kind.clone(), |
|
|
|
|
|
|
|
pdu.state_key |
|
|
|
|
|
|
|
.clone() |
|
|
|
|
|
|
|
.expect("Found state event without state_key"), |
|
|
|
|
|
|
|
), |
|
|
|
|
|
|
|
pdu.event_id.clone(), |
|
|
|
|
|
|
|
))) |
|
|
|
|
|
|
|
.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, |
|
|
|
&db.rooms, |
|
|
|
) { |
|
|
|
) { |
|
|
|
Ok(resolved) if resolved.values().any(|id| &event_id == id) => { |
|
|
|
Ok(true) => { |
|
|
|
let res_state = resolved.iter().map(|(k, v)| { |
|
|
|
let count = db.globals.next_count()?; |
|
|
|
Ok::<_, Error>(( |
|
|
|
let mut pdu_id = room_id.as_bytes().to_vec(); |
|
|
|
k.clone(), |
|
|
|
pdu_id.push(0xff); |
|
|
|
db.rooms |
|
|
|
pdu_id.extend_from_slice(&count.to_be_bytes()); |
|
|
|
.get_pdu(v)? |
|
|
|
|
|
|
|
.ok_or_else(|| Error::bad_database("Pdu with eventId not found"))?, |
|
|
|
db.rooms.append_to_state(&pdu_id, &pdu)?; |
|
|
|
)) |
|
|
|
db.rooms.append_pdu( |
|
|
|
}); |
|
|
|
&pdu, |
|
|
|
// If the event is older than the last event in pduid_pdu Tree then find the
|
|
|
|
&value, |
|
|
|
// closest ancestor we know of and insert after the known ancestor by
|
|
|
|
count, |
|
|
|
// altering the known events pduid to = same roomID + same count bytes + 0x1
|
|
|
|
pdu_id.into(), |
|
|
|
// pushing a single byte every time a simple append cannot be done.
|
|
|
|
&db.globals, |
|
|
|
match db.rooms.get_latest_pduid_before( |
|
|
|
&db.account_data, |
|
|
|
room_id, |
|
|
|
&db.admin, |
|
|
|
&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_to_state(&pdu_id, &pdu)?; |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
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_to_state(&pdu_id, &pdu)?; |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
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>(())); |
|
|
|
resolved_map.insert(event_id, Ok::<(), String>(())); |
|
|
|
} |
|
|
|
} |
|
|
|
// If the eventId is not found in the resolved state auth has failed
|
|
|
|
Ok(false) => { |
|
|
|
Ok(res) => { |
|
|
|
resolved_map.insert(event_id, Err("Failed event auth".into())); |
|
|
|
dbg!(res); |
|
|
|
error!("{:?}", pdu); |
|
|
|
resolved_map.insert( |
|
|
|
|
|
|
|
event_id, |
|
|
|
|
|
|
|
Err("This event failed authentication, not found in resolved set".into()), |
|
|
|
|
|
|
|
); |
|
|
|
|
|
|
|
} |
|
|
|
} |
|
|
|
Err(e) => { |
|
|
|
Err(err) => { |
|
|
|
resolved_map.insert(event_id, Err(e.to_string())); |
|
|
|
resolved_map.insert(event_id, Err(err.to_string())); |
|
|
|
|
|
|
|
error!("{}", err); |
|
|
|
} |
|
|
|
} |
|
|
|
}; |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
Ok(dbg!(send_transaction_message::v1::Response { pdus: resolved_map }).into()) |
|
|
|
Ok(dbg!(send_transaction_message::v1::Response { pdus: resolved_map }).into()) |
|
|
|
|
Devin Ragotzy
5 years ago