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

feat: implement appservices

Browse source 
this also reverts some stateres changes
This commit is contained in:
Timo Kösters 2020-12-08 10:33:44 +01:00
parent d62f17a91a
commit 6e5b35ea92
No known key found for this signature in database
GPG key ID: 24DA7517711A2BA4
26 changed files with 696 additions and 584 deletions
Download patch file Download diff file Expand all files Collapse all files

61
src/database/admin.rs
View file

@ -10,7 +10,9 @@ use ruma::{
use tokio::select;
pub enum AdminCommand {
SendTextMessage(message::TextMessageEventContent),
RegisterAppservice(serde_yaml::Value),
ListAppservices,
SendMessage(message::MessageEventContent),
}
#[derive(Clone)]
@ -44,28 +46,49 @@ impl Admin {
warn!("Conduit instance does not have an #admins room. Logging to that room will not work.");
}
let send_message = |message: message::MessageEventContent| {
if let Some(conduit_room) = &conduit_room {
db.rooms
.build_and_append_pdu(
PduBuilder {
event_type: EventType::RoomMessage,
content: serde_json::to_value(message)
.expect("event is valid, we just created it"),
unsigned: None,
state_key: None,
redacts: None,
},
&conduit_user,
&conduit_room,
&db.globals,
&db.sending,
&db.admin,
&db.account_data,
&db.appservice,
)
.unwrap();
}
};
loop {
select! {
Some(event) = receiver.next() => {
match event {
AdminCommand::SendTextMessage(message) => {
if let Some(conduit_room) = &conduit_room {
db.rooms.build_and_append_pdu(
PduBuilder {
event_type: EventType::RoomMessage,
content: serde_json::to_value(message).expect("event is valid, we just created it"),
unsigned: None,
state_key: None,
redacts: None,
},
&conduit_user,
&conduit_room,
&db.globals,
&db.sending,
&db.admin,
&db.account_data,
).unwrap();
}
AdminCommand::RegisterAppservice(yaml) => {
db.appservice.register_appservice(yaml).unwrap(); // TODO handle error
}
AdminCommand::ListAppservices => {
let appservices = db.appservice.iter_ids().collect::<Vec<_>>();
let count = appservices.len();
let output = format!(
"Appservices ({}): {}",
count,
appservices.into_iter().filter_map(|r| r.ok()).collect::<Vec<_>>().join(", ")
);
send_message(message::MessageEventContent::text_plain(output));
}
AdminCommand::SendMessage(message) => {
send_message(message);
}
}
}

67
src/database/appservice.rs Normal file
View file

@ -0,0 +1,67 @@
use crate::{utils, Error, Result};
use std::collections::HashMap;
use std::sync::{Arc, RwLock};
#[derive(Clone)]
pub struct Appservice {
pub(super) cached_registrations: Arc<RwLock<HashMap<String, serde_yaml::Value>>>,
pub(super) id_appserviceregistrations: sled::Tree,
}
impl Appservice {
pub fn register_appservice(&self, yaml: serde_yaml::Value) -> Result<()> {
// TODO: Rumaify
let id = yaml.get("id").unwrap().as_str().unwrap();
self.id_appserviceregistrations
.insert(id, serde_yaml::to_string(&yaml).unwrap().as_bytes())?;
self.cached_registrations
.write()
.unwrap()
.insert(id.to_owned(), yaml);
Ok(())
}
pub fn get_registration(&self, id: &str) -> Result<Option<serde_yaml::Value>> {
self.cached_registrations
.read()
.unwrap()
.get(id)
.map_or_else(
|| {
Ok(self
.id_appserviceregistrations
.get(id)?
.map(|bytes| {
Ok::<_, Error>(serde_yaml::from_slice(&bytes).map_err(|_| {
Error::bad_database(
"Invalid registration bytes in id_appserviceregistrations.",
)
})?)
})
.transpose()?)
},
|r| Ok(Some(r.clone())),
)
}
pub fn iter_ids(&self) -> impl Iterator<Item = Result<String>> {
self.id_appserviceregistrations.iter().keys().map(|id| {
Ok(utils::string_from_bytes(&id?).map_err(|_| {
Error::bad_database("Invalid id bytes in id_appserviceregistrations.")
})?)
})
}
pub fn iter_all<'a>(
&'a self,
) -> impl Iterator<Item = Result<(String, serde_yaml::Value)>> + 'a {
self.iter_ids().filter_map(|id| id.ok()).map(move |id| {
Ok((
id.clone(),
self.get_registration(&id)?
.expect("iter_ids only returns appservices that exist"),
))
})
}
}

12
src/database/globals.rs
View file

@ -1,10 +1,10 @@
use crate::{database::Config, utils, Error, Result};
use trust_dns_resolver::TokioAsyncResolver;
use std::collections::HashMap;
use log::error;
use ruma::ServerName;
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::RwLock;
use trust_dns_resolver::TokioAsyncResolver;
pub const COUNTER: &str = "c";
@ -59,9 +59,11 @@ impl Globals {
config,
keypair: Arc::new(keypair),
reqwest_client: reqwest::Client::new(),
dns_resolver: TokioAsyncResolver::tokio_from_system_conf().await.map_err(|_| {
Error::bad_config("Failed to set up trust dns resolver with system config.")
})?,
dns_resolver: TokioAsyncResolver::tokio_from_system_conf()
.await
.map_err(|_| {
Error::bad_config("Failed to set up trust dns resolver with system config.")
})?,
actual_destination_cache: Arc::new(RwLock::new(HashMap::new())),
})
}

7
src/database/media.rs
View file

@ -290,7 +290,12 @@ impl Media {
file: thumbnail_bytes.to_vec(),
}))
} else {
Ok(None)
// Couldn't parse file to generate thumbnail, send original
Ok(Some(FileMeta {
filename,
content_type,
file: file.to_vec(),
}))
}
} else {
Ok(None)

122
src/database/rooms.rs
View file

@ -36,16 +36,6 @@ use super::admin::AdminCommand;
/// hashing the entire state.
pub type StateHashId = IVec;
/// An enum that represents the two valid states when searching
/// for an events "parent".
///
/// An events parent is any event we are aware of that is part of
/// the events `prev_events` array.
pub(crate) enum ClosestParent {
Append,
Insert(u64),
}
#[derive(Clone)]
pub struct Rooms {
pub edus: edus::RoomEdus,
@ -411,54 +401,6 @@ impl Rooms {
}
}
/// Recursively search for a PDU from our DB that is also in the
/// `prev_events` field of the incoming PDU.
///
/// First we check if the last PDU inserted to the given room is a parent
/// if not we recursively check older `prev_events` to insert the incoming
/// event after.
pub(crate) fn get_latest_pduid_before(
&self,
room: &RoomId,
incoming_prev_ids: &[EventId],
their_state: &BTreeMap<EventId, Arc<StateEvent>>,
) -> Result<Option<ClosestParent>> {
match self.pduid_pdu.scan_prefix(room.as_bytes()).last() {
Some(Ok(val))
if incoming_prev_ids.contains(
&serde_json::from_slice::<PduEvent>(&val.1)
.map_err(|_| {
Error::bad_database("last DB entry contains invalid PDU bytes")
})?
.event_id,
) =>
{
Ok(Some(ClosestParent::Append))
}
_ => {
let mut prev_ids = incoming_prev_ids.to_vec();
while let Some(id) = prev_ids.pop() {
match self.get_pdu_id(&id)? {
Some(pdu_id) => {
return Ok(Some(ClosestParent::Insert(self.pdu_count(&pdu_id)?)));
}
None => {
prev_ids.extend(their_state.get(&id).map_or(
Err(Error::BadServerResponse(
"Failed to find previous event for PDU in state",
)),
// `prev_event_ids` will return an empty Vec instead of failing
// so it works perfect for our use here
|pdu| Ok(pdu.prev_event_ids()),
)?);
}
}
}
Ok(None)
}
}
}
/// Returns the leaf pdus of a room.
pub fn get_pdu_leaves(&self, room_id: &RoomId) -> Result<Vec<EventId>> {
let mut prefix = room_id.as_bytes().to_vec();
@ -583,18 +525,59 @@ impl Rooms {
.as_ref()
== Some(&pdu.room_id)
{
let mut parts = body.split_whitespace().skip(1);
let mut lines = body.lines();
let command_line = lines.next().expect("each string has at least one line");
let body = lines.collect::<Vec<_>>();
let mut parts = command_line.split_whitespace().skip(1);
if let Some(command) = parts.next() {
let args = parts.collect::<Vec<_>>();
admin.send(AdminCommand::SendTextMessage(
message::TextMessageEventContent {
body: format!("Command: {}, Args: {:?}", command, args),
formatted: None,
relates_to: None,
new_content: None,
},
));
match command {
"register_appservice" => {
if body.len() > 2
&& body[0].trim() == "```"
&& body.last().unwrap().trim() == "```"
{
let appservice_config = body[1..body.len() - 1].join("\n");
let parsed_config = serde_yaml::from_str::<serde_yaml::Value>(
&appservice_config,
);
match parsed_config {
Ok(yaml) => {
admin.send(AdminCommand::RegisterAppservice(yaml));
}
Err(e) => {
admin.send(AdminCommand::SendMessage(
message::MessageEventContent::text_plain(
format!(
"Could not parse appservice config: {}",
e
),
),
));
}
}
} else {
admin.send(AdminCommand::SendMessage(
message::MessageEventContent::text_plain(
"Expected code block in command body.",
),
));
}
}
"list_appservices" => {
admin.send(AdminCommand::ListAppservices);
}
_ => {
admin.send(AdminCommand::SendMessage(
message::MessageEventContent::text_plain(format!(
"Command: {}, Args: {:?}",
command, args
)),
));
}
}
}
}
}
@ -675,6 +658,7 @@ impl Rooms {
sending: &super::sending::Sending,
admin: &super::admin::Admin,
account_data: &super::account_data::AccountData,
appservice: &super::appservice::Appservice,
) -> Result<EventId> {
let PduBuilder {
event_type,
@ -923,6 +907,10 @@ impl Rooms {
sending.send_pdu(&server, &pdu_id)?;
}
for appservice in appservice.iter_all().filter_map(|r| r.ok()) {
sending.send_pdu_appservice(&appservice.0, &pdu_id)?;
}
Ok(pdu.event_id)
}

240
src/database/sending.rs
View file

@ -1,26 +1,35 @@
use std::{collections::HashMap, convert::TryFrom, time::SystemTime};
use crate::{server_server, utils, Error, PduEvent, Result};
use crate::{appservice_server, server_server, utils, Error, PduEvent, Result};
use federation::transactions::send_transaction_message;
use log::{debug, warn};
use log::warn;
use rocket::futures::stream::{FuturesUnordered, StreamExt};
use ruma::{api::federation, ServerName};
use ruma::{
api::{appservice, federation},
ServerName,
};
use sled::IVec;
use tokio::select;
#[derive(Clone)]
pub struct Sending {
/// The state for a given state hash.
pub(super) servernamepduids: sled::Tree, // ServernamePduId = ServerName + PduId
pub(super) servercurrentpdus: sled::Tree, // ServerCurrentPdus = ServerName + PduId (pduid can be empty for reservation)
pub(super) servernamepduids: sled::Tree, // ServernamePduId = (+)ServerName + PduId
pub(super) servercurrentpdus: sled::Tree, // ServerCurrentPdus = (+)ServerName + PduId (pduid can be empty for reservation)
}
impl Sending {
pub fn start_handler(&self, globals: &super::globals::Globals, rooms: &super::rooms::Rooms) {
pub fn start_handler(
&self,
globals: &super::globals::Globals,
rooms: &super::rooms::Rooms,
appservice: &super::appservice::Appservice,
) {
let servernamepduids = self.servernamepduids.clone();
let servercurrentpdus = self.servercurrentpdus.clone();
let rooms = rooms.clone();
let globals = globals.clone();
let appservice = appservice.clone();
tokio::spawn(async move {
let mut futures = FuturesUnordered::new();
@ -28,7 +37,7 @@ impl Sending {
// Retry requests we could not finish yet
let mut current_transactions = HashMap::new();
for (server, pdu) in servercurrentpdus
for (server, pdu, is_appservice) in servercurrentpdus
.iter()
.filter_map(|r| r.ok())
.map(|(key, _)| {
@ -38,45 +47,61 @@ impl Sending {
Error::bad_database("Invalid bytes in servercurrentpdus.")
})?;
let server = utils::string_from_bytes(&server).map_err(|_| {
Error::bad_database("Invalid server bytes in server_currenttransaction")
})?;
// Appservices start with a plus
let (server, is_appservice) = if server.starts_with("+") {
(&server[1..], true)
} else {
(&*server, false)
};
Ok::<_, Error>((
Box::<ServerName>::try_from(utils::string_from_bytes(&server).map_err(
|_| {
Error::bad_database(
"Invalid server bytes in server_currenttransaction",
)
},
)?)
.map_err(|_| {
Box::<ServerName>::try_from(server).map_err(|_| {
Error::bad_database(
"Invalid server string in server_currenttransaction",
)
})?,
IVec::from(pdu),
is_appservice,
))
})
.filter_map(|r| r.ok())
.filter(|(_, pdu)| !pdu.is_empty()) // Skip reservation key
.filter(|(_, pdu, _)| !pdu.is_empty()) // Skip reservation key
.take(50)
// This should not contain more than 50 anyway
{
current_transactions
.entry(server)
.entry((server, is_appservice))
.or_insert_with(Vec::new)
.push(pdu);
}
for (server, pdus) in current_transactions {
futures.push(Self::handle_event(server, pdus, &globals, &rooms));
for ((server, is_appservice), pdus) in current_transactions {
futures.push(Self::handle_event(
server,
is_appservice,
pdus,
&globals,
&rooms,
&appservice,
));
}
let mut subscriber = servernamepduids.watch_prefix(b"");
loop {
select! {
Some(server) = futures.next() => {
debug!("sending response: {:?}", &server);
match server {
Ok((server, _response)) => {
let mut prefix = server.as_bytes().to_vec();
Some(response) = futures.next() => {
match response {
Ok((server, is_appservice)) => {
let mut prefix = if is_appservice {
"+".as_bytes().to_vec()
} else {
Vec::new()
};
prefix.extend_from_slice(server.as_bytes());
prefix.push(0xff);
for key in servercurrentpdus
@ -109,13 +134,13 @@ impl Sending {
servernamepduids.remove(&current_key).unwrap();
}
futures.push(Self::handle_event(server, new_pdus, &globals, &rooms));
futures.push(Self::handle_event(server, is_appservice, new_pdus, &globals, &rooms, &appservice));
} else {
servercurrentpdus.remove(&prefix).unwrap();
// servercurrentpdus with the prefix should be empty now
}
}
Err((server, e)) => {
Err((server, _is_appservice, e)) => {
warn!("Couldn't send transaction to {}: {}", server, e)
// TODO: exponential backoff
}
@ -126,24 +151,37 @@ impl Sending {
let servernamepduid = key.clone();
let mut parts = servernamepduid.splitn(2, |&b| b == 0xff);
if let Some((server, pdu_id)) = utils::string_from_bytes(
if let Some((server, is_appservice, pdu_id)) = utils::string_from_bytes(
parts
.next()
.expect("splitn will always return 1 or more elements"),
)
.map_err(|_| Error::bad_database("ServerName in servernamepduid bytes are invalid."))
.and_then(|server_str| Box::<ServerName>::try_from(server_str)
.map_err(|_| Error::bad_database("ServerName in servernamepduid is invalid.")))
.map(|server_str| {
// Appservices start with a plus
if server_str.starts_with("+") {
(server_str[1..].to_owned(), true)
} else {
(server_str, false)
}
})
.and_then(|(server_str, is_appservice)| Box::<ServerName>::try_from(server_str)
.map_err(|_| Error::bad_database("ServerName in servernamepduid is invalid.")).map(|s| (s, is_appservice)))
.ok()
.and_then(|server| parts
.and_then(|(server, is_appservice)| parts
.next()
.ok_or_else(|| Error::bad_database("Invalid servernamepduid in db."))
.ok()
.map(|pdu_id| (server, pdu_id))
.map(|pdu_id| (server, is_appservice, pdu_id))
)
// TODO: exponential backoff
.filter(|(server, _)| {
let mut prefix = server.to_string().as_bytes().to_vec();
.filter(|(server, is_appservice, _)| {
let mut prefix = if *is_appservice {
"+".as_bytes().to_vec()
} else {
Vec::new()
};
prefix.extend_from_slice(server.as_bytes());
prefix.push(0xff);
servercurrentpdus
@ -154,7 +192,7 @@ impl Sending {
servercurrentpdus.insert(&key, &[]).unwrap();
servernamepduids.remove(&key).unwrap();
futures.push(Self::handle_event(server, vec![pdu_id.into()], &globals, &rooms));
futures.push(Self::handle_event(server, is_appservice, vec![pdu_id.into()], &globals, &rooms, &appservice));
}
}
}
@ -172,56 +210,102 @@ impl Sending {
Ok(())
}
pub fn send_pdu_appservice(&self, appservice_id: &str, pdu_id: &[u8]) -> Result<()> {
let mut key = "+".as_bytes().to_vec();
key.extend_from_slice(appservice_id.as_bytes());
key.push(0xff);
key.extend_from_slice(pdu_id);
self.servernamepduids.insert(key, b"")?;
Ok(())
}
async fn handle_event(
server: Box<ServerName>,
is_appservice: bool,
pdu_ids: Vec<IVec>,
globals: &super::globals::Globals,
rooms: &super::rooms::Rooms,
) -> std::result::Result<
(Box<ServerName>, send_transaction_message::v1::Response),
(Box<ServerName>, Error),
> {
let pdu_jsons = pdu_ids
.iter()
.map(|pdu_id| {
Ok::<_, (Box<ServerName>, Error)>(
// TODO: check room version and remove event_id if needed
serde_json::from_str(
PduEvent::convert_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 servernamepduids not found in db.",
),
)
})?,
)
.json()
.get(),
appservice: &super::appservice::Appservice,
) -> std::result::Result<(Box<ServerName>, bool), (Box<ServerName>, bool, Error)> {
if is_appservice {
let pdu_jsons = pdu_ids
.iter()
.map(|pdu_id| {
Ok::<_, (Box<ServerName>, Error)>(
rooms
.get_pdu_from_id(pdu_id)
.map_err(|e| (server.clone(), e))?
.ok_or_else(|| {
(
server.clone(),
Error::bad_database(
"Event in servernamepduids not found in db.",
),
)
})?
.to_any_event(),
)
.expect("Raw<..> is always valid"),
)
})
.filter_map(|r| r.ok())
.collect::<Vec<_>>();
})
.filter_map(|r| r.ok())
.collect::<Vec<_>>();
appservice_server::send_request(
&globals,
appservice
.get_registration(server.as_str())
.unwrap()
.unwrap(), // TODO: handle error
appservice::event::push_events::v1::Request {
events: &pdu_jsons,
txn_id: &utils::random_string(16),
},
)
.await
.map(|_response| (server.clone(), is_appservice))
.map_err(|e| (server, is_appservice, e))
} else {
let pdu_jsons = pdu_ids
.iter()
.map(|pdu_id| {
Ok::<_, (Box<ServerName>, Error)>(
// TODO: check room version and remove event_id if needed
serde_json::from_str(
PduEvent::convert_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 servernamepduids not found in db.",
),
)
})?,
)
.json()
.get(),
)
.expect("Raw<..> is always valid"),
)
})
.filter_map(|r| r.ok())
.collect::<Vec<_>>();
server_server::send_request(
&globals,
server.clone(),
send_transaction_message::v1::Request {
origin: globals.server_name(),
pdus: &pdu_jsons,
edus: &[],
origin_server_ts: SystemTime::now(),
transaction_id: &utils::random_string(16),
},
)
.await
.map(|response| (server.clone(), response))
.map_err(|e| (server, e))
server_server::send_request(
&globals,
server.clone(),
send_transaction_message::v1::Request {
origin: globals.server_name(),
pdus: &pdu_jsons,
edus: &[],
origin_server_ts: SystemTime::now(),
transaction_id: &utils::random_string(16),
},
)
.await
.map(|_response| (server.clone(), is_appservice))
.map_err(|e| (server, is_appservice, e))
}
}
}

8
src/database/transaction_ids.rs
View file

@ -11,13 +11,13 @@ impl TransactionIds {
pub fn add_txnid(
&self,
user_id: &UserId,
device_id: &DeviceId,
device_id: Option<&DeviceId>,
txn_id: &str,
data: &[u8],
) -> Result<()> {
let mut key = user_id.as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(device_id.as_bytes());
key.extend_from_slice(device_id.map(|d| d.as_bytes()).unwrap_or_default());
key.push(0xff);
key.extend_from_slice(txn_id.as_bytes());
@ -29,12 +29,12 @@ impl TransactionIds {
pub fn existing_txnid(
&self,
user_id: &UserId,
device_id: &DeviceId,
device_id: Option<&DeviceId>,
txn_id: &str,
) -> Result<Option<IVec>> {
let mut key = user_id.as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(device_id.as_bytes());
key.extend_from_slice(device_id.map(|d| d.as_bytes()).unwrap_or_default());
key.push(0xff);
key.extend_from_slice(txn_id.as_bytes());