char
/
conduit
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

feat: room joining, room based /sync responses

next
timokoesters 2020-04-06 13:46:46 +02:00
parent 4d4cff7120
commit 10bb96fcf7
No known key found for this signature in database
GPG Key ID: 356E705610F626D5
5 changed files with 134 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
Cargo.lock generated
View File

@ -1050,9 +1050,8 @@ dependencies = [
[[package]] [[package]]
name = "ruma-client-api" name = "ruma-client-api"
version = "0.7.1" version = "0.7.2"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "git+https://github.com/ruma/ruma-client-api.git#fe92c2940a2db80509e9a9f162c0f68f3ec3d0a4"
checksum = "b390a86d36e87cc56111802bfd281eed1095f5097a89677101d0271d8e6b1306"
dependencies = [ dependencies = [
"http", "http",
"js_int", "js_int",
@ -1193,9 +1192,9 @@ dependencies = [
[[package]] [[package]]
name = "serde_json" name = "serde_json"
version = "1.0.50" version = "1.0.51"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "78a7a12c167809363ec3bd7329fc0a3369056996de43c4b37ef3cd54a6ce4867" checksum = "da07b57ee2623368351e9a0488bb0b261322a15a6e0ae53e243cbdc0f4208da9"
dependencies = [ dependencies = [
"itoa", "itoa",
"ryu", "ryu",

2
Cargo.toml
View File

@ -9,7 +9,7 @@ edition = "2018"
[dependencies] [dependencies]
rocket = { git = "https://github.com/SergioBenitez/Rocket.git", branch = "async", features = ["tls"] } rocket = { git = "https://github.com/SergioBenitez/Rocket.git", branch = "async", features = ["tls"] }
http = "0.2.1" http = "0.2.1"
ruma-client-api = "0.7.1" ruma-client-api = { git = "https://github.com/ruma/ruma-client-api.git" }
pretty_env_logger = "0.4.0" pretty_env_logger = "0.4.0"
log = "0.4.8" log = "0.4.8"
sled = "0.31.0" sled = "0.31.0"

75
src/data.rs
View File

@ -107,6 +107,29 @@ impl Data {
.unwrap(); .unwrap();
} }
pub fn room_join(&self, room_id: &RoomId, user_id: &UserId) {
self.db.userid_roomids.add(
user_id.to_string().as_bytes(),
room_id.to_string().as_bytes().into(),
);
self.db.roomid_userids.add(
room_id.to_string().as_bytes(),
user_id.to_string().as_bytes().into(),
);
}
pub fn rooms_joined(&self, user_id: &UserId) -> Vec<RoomId> {
self.db
.userid_roomids
.get_iter(user_id.to_string().as_bytes())
.values()
.map(|room_id| {
RoomId::try_from(&*utils::string_from_bytes(&room_id.unwrap()))
.expect("user joined valid room ids")
})
.collect()
}
pub fn pdu_get(&self, event_id: &EventId) -> Option<RoomV3Pdu> { pub fn pdu_get(&self, event_id: &EventId) -> Option<RoomV3Pdu> {
self.db self.db
.eventid_pduid .eventid_pduid
@ -157,7 +180,7 @@ impl Data {
room_id: RoomId, room_id: RoomId,
sender: UserId, sender: UserId,
event_type: EventType, event_type: EventType,
content: MessageEventContent, content: serde_json::Value,
) -> EventId { ) -> EventId {
// prev_events are the leaves of the current graph. This method removes all leaves from the // prev_events are the leaves of the current graph. This method removes all leaves from the
// room and replaces them with our event // room and replaces them with our event
@ -184,7 +207,7 @@ impl Data {
origin: self.hostname.clone(), origin: self.hostname.clone(),
origin_server_ts: utils::millis_since_unix_epoch(), origin_server_ts: utils::millis_since_unix_epoch(),
kind: event_type, kind: event_type,
content: serde_json::to_value(content).expect("message content is valid json"), content,
state_key: None, state_key: None,
prev_events, prev_events,
depth: depth.try_into().unwrap(), depth: depth.try_into().unwrap(),
@ -207,9 +230,10 @@ impl Data {
self.pdu_leaves_replace(&room_id, &pdu.event_id); self.pdu_leaves_replace(&room_id, &pdu.event_id);
// The new value will need a new index. We store the last used index in 'n' + id // The new value will need a new index. We store the last used index in 'n'
let mut count_key: Vec<u8> = vec![b'n']; // The count will go up regardless of the room_id
count_key.extend_from_slice(&room_id.to_string().as_bytes()); // This is also the next_batch/since value
let count_key: Vec<u8> = vec![b'n'];
// Increment the last index and use that // Increment the last index and use that
let index = utils::u64_from_bytes( let index = utils::u64_from_bytes(
@ -225,7 +249,7 @@ impl Data {
pdu_id.extend_from_slice(room_id.to_string().as_bytes()); pdu_id.extend_from_slice(room_id.to_string().as_bytes());
pdu_id.push(b'#'); // Add delimiter so we don't find rooms starting with the same id pdu_id.push(b'#'); // Add delimiter so we don't find rooms starting with the same id
pdu_id.extend_from_slice(index.to_string().as_bytes()); pdu_id.extend_from_slice(&index.to_be_bytes());
self.db self.db
.pduid_pdus .pduid_pdus
@ -240,37 +264,30 @@ impl Data {
pdu.event_id pdu.event_id
} }
/// Returns a vector of all PDUs. /// Returns a vector of all PDUs in a room.
pub fn pdus_all(&self) -> Vec<PduEvent> { pub fn pdus_all(&self, room_id: &RoomId) -> Vec<PduEvent> {
self.pdus_since( self.pdus_since(room_id, "".to_owned())
self.db
.eventid_pduid
.iter()
.values()
.next()
.unwrap()
.map(|key| utils::string_from_bytes(&key))
.expect("there should be at least one pdu"),
)
} }
/// Returns a vector of all events that happened after the event with id `since`. /// Returns a vector of all events in a room that happened after the event with id `since`.
pub fn pdus_since(&self, since: String) -> Vec<PduEvent> { pub fn pdus_since(&self, room_id: &RoomId, since: String) -> Vec<PduEvent> {
let mut pdus = Vec::new(); let mut pdus = Vec::new();
if let Some(room_id) = since.rsplitn(2, '#').nth(1) { // Create the first part of the full pdu id
let mut current = since.clone(); let mut pdu_id = vec![b'd'];
pdu_id.extend_from_slice(room_id.to_string().as_bytes());
pdu_id.push(b'#'); // Add delimiter so we don't find rooms starting with the same id
while let Some((key, value)) = self.db.pduid_pdus.get_gt(current).unwrap() { let mut current = pdu_id.clone();
if key.starts_with(&room_id.to_string().as_bytes()) { current.extend_from_slice(since.as_bytes());
current = utils::string_from_bytes(&key);
while let Some((key, value)) = self.db.pduid_pdus.get_gt(&current).unwrap() {
if key.starts_with(&pdu_id) {
current = key.to_vec();
pdus.push(serde_json::from_slice(&value).expect("pdu in db is valid"));
} else { } else {
break; break;
} }
pdus.push(serde_json::from_slice(&value).expect("pdu is valid"));
}
} else {
debug!("event at `since` not found");
} }
pdus pdus
} }

38
src/database.rs
View File

@ -55,8 +55,10 @@ pub struct Database {
pub deviceid_token: sled::Tree, pub deviceid_token: sled::Tree,
pub token_userid: sled::Tree, pub token_userid: sled::Tree,
pub pduid_pdus: sled::Tree, pub pduid_pdus: sled::Tree,
pub roomid_pduleaves: MultiValue,
pub eventid_pduid: sled::Tree, pub eventid_pduid: sled::Tree,
pub roomid_pduleaves: MultiValue,
pub roomid_userids: MultiValue,
pub userid_roomids: MultiValue,
_db: sled::Db, _db: sled::Db,
} }
@ -76,8 +78,10 @@ impl Database {
deviceid_token: db.open_tree("deviceid_token").unwrap(), deviceid_token: db.open_tree("deviceid_token").unwrap(),
token_userid: db.open_tree("token_userid").unwrap(), token_userid: db.open_tree("token_userid").unwrap(),
pduid_pdus: db.open_tree("pduid_pdus").unwrap(), pduid_pdus: db.open_tree("pduid_pdus").unwrap(),
roomid_pduleaves: MultiValue(db.open_tree("roomid_pduleaves").unwrap()),
eventid_pduid: db.open_tree("eventid_pduid").unwrap(), eventid_pduid: db.open_tree("eventid_pduid").unwrap(),
roomid_pduleaves: MultiValue(db.open_tree("roomid_pduleaves").unwrap()),
roomid_userids: MultiValue(db.open_tree("roomid_userids").unwrap()),
userid_roomids: MultiValue(db.open_tree("userid_roomids").unwrap()),
_db: db, _db: db,
} }
} }
@ -86,7 +90,7 @@ impl Database {
println!("# UserId -> Password:"); println!("# UserId -> Password:");
for (k, v) in self.userid_password.iter().map(|r| r.unwrap()) { for (k, v) in self.userid_password.iter().map(|r| r.unwrap()) {
println!( println!(
"{} -> {}", "{:?} -> {:?}",
String::from_utf8_lossy(&k), String::from_utf8_lossy(&k),
String::from_utf8_lossy(&v), String::from_utf8_lossy(&v),
); );
@ -94,7 +98,7 @@ impl Database {
println!("\n# UserId -> DeviceIds:"); println!("\n# UserId -> DeviceIds:");
for (k, v) in self.userid_deviceids.iter_all().map(|r| r.unwrap()) { for (k, v) in self.userid_deviceids.iter_all().map(|r| r.unwrap()) {
println!( println!(
"{} -> {}", "{:?} -> {:?}",
String::from_utf8_lossy(&k), String::from_utf8_lossy(&k),
String::from_utf8_lossy(&v), String::from_utf8_lossy(&v),
); );
@ -102,7 +106,7 @@ impl Database {
println!("\n# DeviceId -> Token:"); println!("\n# DeviceId -> Token:");
for (k, v) in self.deviceid_token.iter().map(|r| r.unwrap()) { for (k, v) in self.deviceid_token.iter().map(|r| r.unwrap()) {
println!( println!(
"{} -> {}", "{:?} -> {:?}",
String::from_utf8_lossy(&k), String::from_utf8_lossy(&k),
String::from_utf8_lossy(&v), String::from_utf8_lossy(&v),
); );
@ -110,7 +114,7 @@ impl Database {
println!("\n# Token -> UserId:"); println!("\n# Token -> UserId:");
for (k, v) in self.token_userid.iter().map(|r| r.unwrap()) { for (k, v) in self.token_userid.iter().map(|r| r.unwrap()) {
println!( println!(
"{} -> {}", "{:?} -> {:?}",
String::from_utf8_lossy(&k), String::from_utf8_lossy(&k),
String::from_utf8_lossy(&v), String::from_utf8_lossy(&v),
); );
@ -118,7 +122,23 @@ impl Database {
println!("\n# RoomId -> PDU leaves:"); println!("\n# RoomId -> PDU leaves:");
for (k, v) in self.roomid_pduleaves.iter_all().map(|r| r.unwrap()) { for (k, v) in self.roomid_pduleaves.iter_all().map(|r| r.unwrap()) {
println!( println!(
"{} -> {}", "{:?} -> {:?}",
String::from_utf8_lossy(&k),
String::from_utf8_lossy(&v),
);
}
println!("\n# RoomId -> UserIds:");
for (k, v) in self.roomid_userids.iter_all().map(|r| r.unwrap()) {
println!(
"{:?} -> {:?}",
String::from_utf8_lossy(&k),
String::from_utf8_lossy(&v),
);
}
println!("\n# UserId -> RoomIds:");
for (k, v) in self.userid_roomids.iter_all().map(|r| r.unwrap()) {
println!(
"{:?} -> {:?}",
String::from_utf8_lossy(&k), String::from_utf8_lossy(&k),
String::from_utf8_lossy(&v), String::from_utf8_lossy(&v),
); );
@ -126,7 +146,7 @@ impl Database {
println!("\n# PDU Id -> PDUs:"); println!("\n# PDU Id -> PDUs:");
for (k, v) in self.pduid_pdus.iter().map(|r| r.unwrap()) { for (k, v) in self.pduid_pdus.iter().map(|r| r.unwrap()) {
println!( println!(
"{} -> {}", "{:?} -> {:?}",
String::from_utf8_lossy(&k), String::from_utf8_lossy(&k),
String::from_utf8_lossy(&v), String::from_utf8_lossy(&v),
); );
@ -134,7 +154,7 @@ impl Database {
println!("\n# EventId -> PDU Id:"); println!("\n# EventId -> PDU Id:");
for (k, v) in self.eventid_pduid.iter().map(|r| r.unwrap()) { for (k, v) in self.eventid_pduid.iter().map(|r| r.unwrap()) {
println!( println!(
"{} -> {}", "{:?} -> {:?}",
String::from_utf8_lossy(&k), String::from_utf8_lossy(&k),
String::from_utf8_lossy(&v), String::from_utf8_lossy(&v),
); );

71
src/main.rs
View File

@ -15,14 +15,16 @@ use ruma_client_api::{
error::{Error, ErrorKind}, error::{Error, ErrorKind},
r0::{ r0::{
account::register, alias::get_alias, membership::join_room_by_id, account::register, alias::get_alias, membership::join_room_by_id,
message::create_message_event, session::login, sync::sync_events, message::create_message_event, room::create_room, session::login, sync::sync_events,
}, },
unversioned::get_supported_versions, unversioned::get_supported_versions,
}; };
use ruma_events::{collections::all::RoomEvent, room::message::MessageEvent, EventResult}; use ruma_events::{
use ruma_identifiers::{EventId, UserId}; collections::all::RoomEvent, room::message::MessageEvent, EventResult, EventType,
};
use ruma_identifiers::{EventId, RoomId, UserId};
use ruma_wrapper::{MatrixResult, Ruma}; use ruma_wrapper::{MatrixResult, Ruma};
use serde_json::map::Map; use serde_json::{json, map::Map};
use std::{ use std::{
collections::HashMap, collections::HashMap,
convert::{TryFrom, TryInto}, convert::{TryFrom, TryInto},
@ -168,6 +170,29 @@ fn login_route(data: State<Data>, body: Ruma<login::Request>) -> MatrixResult<lo
})); }));
} }
#[post("/_matrix/client/r0/createRoom", data = "<body>")]
fn create_room_route(
data: State<Data>,
body: Ruma<create_room::Request>,
) -> MatrixResult<create_room::Response> {
// TODO: check if room is unique
let room_id = RoomId::new(data.hostname()).expect("host is valid");
data.room_join(
&room_id,
body.user_id.as_ref().expect("user is authenticated"),
);
data.pdu_append(
room_id.clone(),
body.user_id.clone().expect("user is authenticated"),
EventType::RoomMessage,
json!({"msgtype": "m.text", "body": "Hello"}),
);
MatrixResult(Ok(create_room::Response { room_id }))
}
#[get("/_matrix/client/r0/directory/room/<room_alias>")] #[get("/_matrix/client/r0/directory/room/<room_alias>")]
fn get_alias_route(room_alias: String) -> MatrixResult<get_alias::Response> { fn get_alias_route(room_alias: String) -> MatrixResult<get_alias::Response> {
// TODO // TODO
@ -193,10 +218,14 @@ fn get_alias_route(room_alias: String) -> MatrixResult<get_alias::Response> {
#[post("/_matrix/client/r0/rooms/<_room_id>/join", data = "<body>")] #[post("/_matrix/client/r0/rooms/<_room_id>/join", data = "<body>")]
fn join_room_by_id_route( fn join_room_by_id_route(
_room_id: String, data: State<Data>,
body: Ruma<join_room_by_id::Request>, body: Ruma<join_room_by_id::Request>,
_room_id: String,
) -> MatrixResult<join_room_by_id::Response> { ) -> MatrixResult<join_room_by_id::Response> {
// TODO data.room_join(
&body.room_id,
body.user_id.as_ref().expect("user is authenticated"),
);
MatrixResult(Ok(join_room_by_id::Response { MatrixResult(Ok(join_room_by_id::Response {
room_id: body.room_id.clone(), room_id: body.room_id.clone(),
})) }))
@ -213,37 +242,28 @@ fn create_message_event_route(
_txn_id: String, _txn_id: String,
body: Ruma<create_message_event::Request>, body: Ruma<create_message_event::Request>,
) -> MatrixResult<create_message_event::Response> { ) -> MatrixResult<create_message_event::Response> {
if let Ok(content) = body.data.clone().into_result() {
let event_id = data.pdu_append( let event_id = data.pdu_append(
body.room_id.clone(), body.room_id.clone(),
body.user_id.clone().expect("user is authenticated"), body.user_id.clone().expect("user is authenticated"),
body.event_type.clone(), body.event_type.clone(),
content, body.json_body,
); );
MatrixResult(Ok(create_message_event::Response { event_id })) MatrixResult(Ok(create_message_event::Response { event_id }))
} else {
error!("No data found");
MatrixResult(Err(Error {
kind: ErrorKind::NotFound,
message: "Room not found.".to_owned(),
status_code: http::StatusCode::NOT_FOUND,
}))
}
} }
#[get("/_matrix/client/r0/sync", data = "<_body>")] #[get("/_matrix/client/r0/sync", data = "<body>")]
fn sync_route( fn sync_route(
data: State<Data>, data: State<Data>,
_body: Ruma<sync_events::Request>, body: Ruma<sync_events::Request>,
) -> MatrixResult<sync_events::Response> { ) -> MatrixResult<sync_events::Response> {
let mut joined_rooms = HashMap::new(); let mut joined_rooms = HashMap::new();
{ let joined_roomids = data.rooms_joined(body.user_id.as_ref().expect("user is authenticated"));
let pdus = data.pdus_all(); for room_id in joined_roomids {
let mut room_events = Vec::new(); let room_events = data
.pdus_all(&room_id)
for pdu in pdus { .into_iter()
room_events.push(pdu.to_room_event()); .map(|pdu| pdu.to_room_event())
} .collect();
joined_rooms.insert( joined_rooms.insert(
"!roomid:localhost".try_into().unwrap(), "!roomid:localhost".try_into().unwrap(),
@ -309,6 +329,7 @@ fn main() {
get_supported_versions_route, get_supported_versions_route,
register_route, register_route,
login_route, login_route,
create_room_route,
get_alias_route, get_alias_route,
join_room_by_id_route, join_room_by_id_route,
create_message_event_route, create_message_event_route,