magmaus3/continuwuity
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

improvement: get_missing_events route and cleanup

Browse source
This commit is contained in:
Timo Kösters 2020-09-25 12:26:29 +02:00
parent ab332363ce
commit 0d6159c2da
No known key found for this signature in database
GPG key ID: 24DA7517711A2BA4
5 changed files with 95 additions and 80 deletions
Download patch file Download diff file Expand all files Collapse all files

52
src/database/sending.rs
View file

@ -1,10 +1,10 @@
use std::{collections::HashSet, convert::TryFrom, time::SystemTime};
use crate::{server_server, utils, Error, Result};
use crate::{server_server, utils, Error, PduEvent, Result};
use federation::transactions::send_transaction_message;
use log::warn;
use rocket::futures::stream::{FuturesUnordered, StreamExt};
use ruma::{api::federation, Raw, ServerName};
use ruma::{api::federation, ServerName};
use sled::IVec;
use tokio::select;
@ -83,49 +83,27 @@ impl Sending {
(Box<ServerName>, send_transaction_message::v1::Response),
(Box<ServerName>, Error),
> {
let mut pdu_json = rooms
.get_pdu_json_from_id(&pdu_id)
.map_err(|e| (server.clone(), e))?
.ok_or_else(|| {
(
server.clone(),
Error::bad_database("Event in serverpduids not found in db."),
)
})?;
if let Some(unsigned) = pdu_json
.as_object_mut()
.expect("json is object")
.get_mut("unsigned")
{
unsigned
.as_object_mut()
.expect("unsigned is object")
.remove("transaction_id");
}
pdu_json
.as_object_mut()
.expect("json is object")
.remove("event_id");
let raw_json =
serde_json::from_value::<Raw<_>>(pdu_json).expect("Raw::from_value always works");
let globals = &globals;
let pdus = vec![raw_json];
let transaction_id = utils::random_string(16);
let pdu_json = PduEvent::to_outgoing_federation_event(
rooms
.get_pdu_json_from_id(&pdu_id)
.map_err(|e| (server.clone(), e))?
.ok_or_else(|| {
(
server.clone(),
Error::bad_database("Event in serverpduids not found in db."),
)
})?,
);
server_server::send_request(
&globals,
server.clone(),
send_transaction_message::v1::Request {
origin: globals.server_name(),
pdus: &pdus,
pdus: &[pdu_json],
edus: &[],
origin_server_ts: SystemTime::now(),
transaction_id: &transaction_id,
transaction_id: &utils::random_string(16),
},
)
.await

1
src/main.rs
View file

@ -125,6 +125,7 @@ fn setup_rocket() -> rocket::Rocket {
server_server::get_public_rooms_route,
server_server::get_public_rooms_filtered_route,
server_server::send_transaction_message_route,
server_server::get_missing_events_route,
],
)
.attach(AdHoc::on_attach("Config", |mut rocket| async {

43
src/pdu.rs
View file

@ -1,7 +1,6 @@
use crate::Error;
use js_int::UInt;
use ruma::{
events::pdu::PduStub,
events::{
pdu::EventHash, room::member::MemberEventContent, AnyEvent, AnyRoomEvent, AnyStateEvent,
AnyStrippedStateEvent, AnySyncRoomEvent, AnySyncStateEvent, EventType, StateEvent,
@ -200,32 +199,26 @@ impl PduEvent {
serde_json::from_value(json).expect("Raw::from_value always works")
}
pub fn to_outgoing_federation_event(&self) -> Raw<PduStub> {
let mut unsigned = self.unsigned.clone();
unsigned.remove("transaction_id");
let mut json = json!({
"room_id": self.room_id,
"sender": self.sender,
"origin_server_ts": self.origin_server_ts,
"type": self.kind,
"content": self.content,
"prev_events": self.prev_events,
"depth": self.depth,
"auth_events": self.auth_events,
"unsigned": unsigned,
"hashes": self.hashes,
"signatures": self.signatures,
});
if let Some(state_key) = &self.state_key {
json["state_key"] = json!(state_key);
}
if let Some(redacts) = &self.redacts {
json["redacts"] = json!(redacts);
pub fn to_outgoing_federation_event(
mut pdu_json: serde_json::Value,
) -> Raw<ruma::events::pdu::PduStub> {
if let Some(unsigned) = pdu_json
.as_object_mut()
.expect("json is object")
.get_mut("unsigned")
{
unsigned
.as_object_mut()
.expect("unsigned is object")
.remove("transaction_id");
}
serde_json::from_value(json).expect("Raw::from_value always works")
pdu_json
.as_object_mut()
.expect("json is object")
.remove("event_id");
serde_json::from_value::<Raw<_>>(pdu_json).expect("Raw::from_value always works")
}
}

47
src/server_server.rs
View file

@ -3,13 +3,13 @@ use http::header::{HeaderValue, AUTHORIZATION, HOST};
use log::warn;
use rocket::{get, post, put, response::content::Json, State};
use ruma::{
api::federation::directory::get_public_rooms_filtered,
api::{
federation::{
directory::get_public_rooms,
directory::{get_public_rooms, get_public_rooms_filtered},
discovery::{
get_server_keys, get_server_version::v1 as get_server_version, ServerKey, VerifyKey,
},
event::get_missing_events,
transactions::send_transaction_message,
},
OutgoingRequest,
@ -373,3 +373,46 @@ pub fn send_transaction_message_route<'a>(
}
.into())
}
#[cfg_attr(
feature = "conduit_bin",
post("/_matrix/federation/v1/get_missing_events/<_>", data = "<body>")
)]
pub fn get_missing_events_route<'a>(
db: State<'a, Database>,
body: Ruma<get_missing_events::v1::Request<'_>>,
) -> ConduitResult<get_missing_events::v1::Response> {
let mut queued_events = body.latest_events.clone();
let mut events = Vec::new();
let mut i = 0;
while i < queued_events.len() && events.len() < u64::from(body.limit) as usize {
if let Some(pdu) = db.rooms.get_pdu_json(&queued_events[i])? {
if body.earliest_events.contains(
&serde_json::from_value(
pdu.get("event_id")
.cloned()
.ok_or_else(|| Error::bad_database("Event in db has no event_id field."))?,
)
.map_err(|_| Error::bad_database("Invalid event_id field in pdu in db."))?,
) {
i += 1;
continue;
}
queued_events.extend_from_slice(
&serde_json::from_value::<Vec<EventId>>(
pdu.get("prev_events").cloned().ok_or_else(|| {
Error::bad_database("Invalid prev_events field of pdu in db.")
})?,
)
.map_err(|_| Error::bad_database("Invalid prev_events content in pdu in db."))?,
);
events.push(PduEvent::to_outgoing_federation_event(pdu));
}
i += 1;
}
dbg!(&events);
Ok(get_missing_events::v1::Response { events }.into())
}