diff --git a/client/src/app.rs b/client/src/app.rs index 2c88b5e..b61490e 100644 --- a/client/src/app.rs +++ b/client/src/app.rs @@ -70,7 +70,7 @@ pub struct RealmApp { pub room_window_admin_only_send: bool, #[serde(skip)] pub room_window_admin_only_view: bool, - + #[serde(skip)] pub info_window_open: bool, @@ -102,7 +102,7 @@ pub struct RealmApp { pub room_changes_channel: (Sender), ErrorCode>>, Receiver), ErrorCode>>), #[serde(skip)] - pub event_channel: (Sender<(String, (u32, Event))>, Receiver<(String, (u32, Event))>), + pub event_channel: (Sender<(String, (i64, Event))>, Receiver<(String, (i64, Event))>), #[serde(skip)] pub polling_threads: Vec<(String, JoinHandle<()>)>, } @@ -139,7 +139,7 @@ impl Default for RealmApp { room_window_name: String::new(), room_window_admin_only_send: false, room_window_admin_only_view: false, - + info_window_open: false, fetching_user_data_channel: broadcast::channel(256), @@ -170,7 +170,7 @@ impl RealmApp { return eframe::get_value(storage, eframe::APP_KEY).unwrap_or_default(); } } - + Default::default() } @@ -429,7 +429,7 @@ impl eframe::App for RealmApp { } } } - + // Leaving a server while let Ok(result) = self.leave_server_channel.1.try_recv() { match result { @@ -456,7 +456,7 @@ impl eframe::App for RealmApp { }; let client = RealmAuthClient::new(tarpc::client::Config::default(), connection).spawn(); - + let result = client.remove_server(context::current(), username, token, domain, port).await; match result { Ok(r) => { send_channel.send(r).unwrap(); }, @@ -469,7 +469,7 @@ impl eframe::App for RealmApp { } } } - + // Removing a server (auth) while let Ok(result) = self.remove_server_channel.1.try_recv() { match result { @@ -516,7 +516,7 @@ impl eframe::App for RealmApp { Err(e) => error!("Error adding room: {:?}", e), } } - + // Deleting a room while let Ok(result) = self.delete_room_channel.1.try_recv() { match result { @@ -524,8 +524,8 @@ impl eframe::App for RealmApp { info!("Got room delete! Fetching them..."); self.selected_roomid.clear(); fetch_rooms_data( - self.room_changes_channel.0.clone(), - server, + self.room_changes_channel.0.clone(), + server, self.current_user.as_ref().unwrap().token.clone(), self.current_user.as_ref().unwrap().username.clone() ); @@ -550,7 +550,7 @@ impl eframe::App for RealmApp { Err(e) => error!("Error fetching room data: {:?}", e), } } - + // Polling events while let Ok((serverid, (index, event))) = self.event_channel.1.try_recv() { if let Some(active_servers) = &mut self.active_servers { @@ -568,14 +568,17 @@ impl eframe::App for RealmApp { if self.selected_roomid.eq(&roomid) { self.selected_roomid.clear(); } - } + }, + _ => { } + } + if index > server.last_event_index { + server.last_event_index = index; } - server.last_event_index = index; } } } } - + // Manage polling threads if let Some(active_servers) = &mut self.active_servers { if self.polling_threads.len() != active_servers.len() { @@ -583,10 +586,12 @@ impl eframe::App for RealmApp { let missing_servers = active_servers.clone().into_iter().filter(|s| !running_thread_serverids.contains(&s.server_id)).collect::>(); for server in missing_servers { let send_channel = self.event_channel.0.clone(); - let _handle = tokio::spawn(async move { + let serverid = server.server_id.clone(); + let token = self.current_user.as_ref().unwrap().token.clone(); + let userid = self.current_user.as_ref().unwrap().username.clone(); + let handle = tokio::spawn(async move { let mut transport = tarpc::serde_transport::tcp::connect(format!("{}:{}", server.domain, server.port), Json::default); transport.config_mut().max_frame_length(usize::MAX); - let result = transport.await; let connection = match result { Ok(connection) => connection, @@ -594,26 +599,35 @@ impl eframe::App for RealmApp { return; } }; - let client = RealmChatClient::new(tarpc::client::Config::default(), connection).spawn(); + let mut last_event_index = 0; + loop { - let result = client.poll_events_since( + let result = client.get_messages_since( context::current(), - server.last_event_index + stoken(&token, &serverid, &server.domain, server.port), + userid.clone(), + last_event_index ).await; - + match result { - Ok(events) => { - for event in events { - send_channel.send((server.server_id.clone(), (event.0, event.1))).unwrap(); + Ok(messages) => { + if let Ok(messages) = messages { + if let Some(last) = messages.last() { + last_event_index = last.id; + } + for message in messages { + send_channel.send((serverid.clone(), (message.id, Event::NewMessage(message)))).unwrap(); + } } } Err(_) => break, } - sleep(Duration::from_millis(1000)).await; + sleep(Duration::from_millis(5000)).await; } }); + self.polling_threads.push((server.server_id.clone(), handle)); } } } diff --git a/client/src/types.rs b/client/src/types.rs index 104f98e..d35d754 100644 --- a/client/src/types.rs +++ b/client/src/types.rs @@ -20,6 +20,6 @@ pub struct CServer { pub is_admin: bool, pub is_owner: bool, pub rooms: Vec, - pub last_event_index: u32, + pub last_event_index: i64, pub messages: Vec, } \ No newline at end of file diff --git a/client/src/ui/gui.rs b/client/src/ui/gui.rs index 1688033..a3b47e9 100644 --- a/client/src/ui/gui.rs +++ b/client/src/ui/gui.rs @@ -223,7 +223,7 @@ pub fn rooms(app: &mut RealmApp, ctx: &Context) { pub fn messages(app: &mut RealmApp, ctx: &Context) { egui::CentralPanel::default().show(ctx, |ui| { ui.with_layout(egui::Layout::bottom_up(egui::Align::TOP), |ui| { - ui.with_layout(egui::Layout::right_to_left(egui::Align::TOP), |ui| { + ui.with_layout(egui::Layout::right_to_left(egui::Align::Max), |ui| { if ui.button("✉").on_hover_text("Send a message").clicked() { if let Some(active_servers) = &app.active_servers { for server in active_servers.clone() { diff --git a/server/src/events.rs b/server/src/events.rs index 7254d6b..882ec99 100644 --- a/server/src/events.rs +++ b/server/src/events.rs @@ -4,6 +4,7 @@ use crate::types::{Message, Room, User}; pub enum Event { // UserJoined(User), // UserLeft(User), + None, NewMessage(Message), NewRoom(Room), DeleteRoom(String), diff --git a/server/src/main.rs b/server/src/main.rs index 50b7def..2d3773d 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -94,23 +94,15 @@ async fn main() -> anyhow::Result<()> { .filter_map(|r| future::ready(r.ok())) .map(BaseChannel::with_defaults) // Limit channels to 1 per IP. - .max_channels_per_key(2048, |t| t.transport().peer_addr().unwrap().ip()) + .max_channels_per_key(1024, |t| t.transport().peer_addr().unwrap().ip()) // serve is generated by the service attribute. It takes as input any type implementing // the generated World trait. .map(|channel| { let server = RealmChatServer::new(env::var("SERVER_ID").expect("SERVER_ID must be set"), channel.transport().peer_addr().unwrap(), db_pool.clone()); - let tarpc_subscriber = tracing_subscriber::fmt() - .compact() - .with_file(false) - .with_line_number(false) - .with_thread_ids(true) - .with_target(false) - .finish(); - - channel.execute(server.serve()).for_each(spawn).with_subscriber(tarpc_subscriber) + channel.execute(server.serve()).for_each(spawn) }) // Max 10 channels. - .buffer_unordered(4096) + .buffer_unordered(10240) .for_each(|_| async {}) .await; diff --git a/server/src/server.rs b/server/src/server.rs index e25f114..d50670d 100644 --- a/server/src/server.rs +++ b/server/src/server.rs @@ -25,13 +25,13 @@ pub struct RealmChatServer { pub db_pool: Pool, pub typing_users: Vec<(String, String)>, //NOTE: user.userid, room.roomid pub cache: Cache, - pub events: Vec<(u32, Event)>, + //pub events: Arc>>, } const FETCH_MESSAGE: &str = "SELECT message.*, room.id AS 'room_id', room.roomid AS 'room_roomid', room.admin_only_send AS 'room_admin_only_send', room.admin_only_view AS 'room_admin_only_view', user.id AS 'user_id', user.userid AS 'user_userid', user.name AS 'user_name', user.owner AS 'user_owner', user.admin AS 'user_admin' - FROM message INNER JOIN room ON message.room = room.id INNER JOIN user ON message.user = user.id WHERE room.admin_only_view = ? OR false"; + FROM message INNER JOIN room ON message.room = room.id INNER JOIN user ON message.user = user.id"; impl RealmChatServer { pub fn new(server_id: String, socket: SocketAddr, db_pool: Pool) -> RealmChatServer { @@ -47,7 +47,7 @@ impl RealmChatServer { .time_to_idle(Duration::from_secs(5*60)) .time_to_live(Duration::from_secs(60*60)) .build(), - events: Vec::new(), + //events: Arc::new(Mutex::new(Vec::new())), } } @@ -220,15 +220,18 @@ impl RealmChat for RealmChatServer { } async fn poll_events_since(self, _: Context, index: u32) -> Vec<(u32, Event)> { - let mut events_to_send = Vec::new(); + //self.events.lock().await.iter().filter(|(i, _)| i > &index).map(|(i, e)| (*i, e.clone())).collect() - for (i, event) in self.events { - if i > index { - events_to_send.push((i, event)); - } - } - - events_to_send + // let mut events_to_send = Vec::new(); + // + // for (i, event) in self.events.lock().await.clone() { + // if i > index { + // events_to_send.push((i, event)); + // } + // } + // + // events_to_send + Vec::new() } async fn join_server(self, _: Context, stoken: String, userid: String) -> Result { @@ -296,42 +299,42 @@ impl RealmChat for RealmChatServer { } async fn send_message(self, _: Context, stoken: String, mut message: Message) -> Result { - if !self.is_stoken_valid(&message.user.userid, &stoken).await { // Check sender userid - return Err(Unauthorized) - } + // if !self.is_stoken_valid(&message.user.userid, &stoken).await { // Check sender userid + // return Err(Unauthorized) + // } + // + // // Assert all the data in message is correct + // message.user = self.inner_get_user(&message.user.userid).await?; - // Assert all the data in message is correct - message.user = self.inner_get_user(&message.user.userid).await?; + // match &message.data { // Check that the sender is the owner of the referencing msg + // MessageData::Edit(e) => { + // let ref_msg = self.inner_get_message(&message.user.userid, e.referencing_id).await?; + // if !ref_msg.user.userid.eq(&message.user.userid) { + // return Err(Unauthorized) + // } + // } + // MessageData::Redaction(r)=> { + // let ref_msg = self.inner_get_message(&message.user.userid, r.referencing_id).await?; + // if !ref_msg.user.userid.eq(&message.user.userid) || !self.internal_is_user_admin(&message.user.userid).await { + // return Err(Unauthorized) + // } + // } + // _ => {} + // } - match &message.data { // Check that the sender is the owner of the referencing msg - MessageData::Edit(e) => { - let ref_msg = self.inner_get_message(&message.user.userid, e.referencing_id).await?; - if !ref_msg.user.userid.eq(&message.user.userid) { - return Err(Unauthorized) - } - } - MessageData::Redaction(r)=> { - let ref_msg = self.inner_get_message(&message.user.userid, r.referencing_id).await?; - if !ref_msg.user.userid.eq(&message.user.userid) || !self.internal_is_user_admin(&message.user.userid).await { - return Err(Unauthorized) - } - } - _ => {} - } - - let is_admin = self.internal_is_user_admin(&message.user.userid).await; - let admin_only_send = query!( - "SELECT admin_only_send FROM room WHERE roomid = ?", - message.room.roomid).fetch_one(&self.db_pool).await; - if let Ok(record) = admin_only_send { - if record.admin_only_send && !is_admin { - return Err(Unauthorized) - } - } else { - return Err(RoomNotFound) - } - - message.room = self.inner_get_room(&message.user.userid, &message.room.roomid).await?; + // let is_admin = self.internal_is_user_admin(&message.user.userid).await; + // let admin_only_send = query!( + // "SELECT admin_only_send FROM room WHERE roomid = ?", + // message.room.roomid).fetch_one(&self.db_pool).await; + // if let Ok(record) = admin_only_send { + // if record.admin_only_send && !is_admin { + // return Err(Unauthorized) + // } + // } else { + // return Err(RoomNotFound) + // } + // + // message.room = self.inner_get_room(&message.user.userid, &message.room.roomid).await?; let result = match &message.data { MessageData::Text(text) => { @@ -371,6 +374,9 @@ impl RealmChat for RealmChatServer { // if result.is_err() { // error!("Error broadcasting NewMessageEvent!"); // } + // self.events.lock().await.push( + // (self.events.lock().await.len() as u32 + 1, + // Event::NewMessage(message.clone()))); Ok(message) }, @@ -411,15 +417,15 @@ impl RealmChat for RealmChatServer { } } - async fn get_messages_since(self, _: Context, stoken: String, userid: String, time: DateTime) -> Result, ErrorCode> { + async fn get_messages_since(self, _: Context, stoken: String, userid: String, id: i64) -> Result, ErrorCode> { if !self.is_stoken_valid(&userid, &stoken).await { return Err(Unauthorized) } let is_admin = self.internal_is_user_admin(&userid).await; - let result = sqlx::query(&format!("{}{}", FETCH_MESSAGE, "AND message.timestamp >= ?")) - .bind(is_admin) - .bind(time) + let result = sqlx::query(&format!("{}{}", FETCH_MESSAGE, " AND message.id > ?")) + //.bind(is_admin) + .bind(id) .fetch_all(&self.db_pool).await; match result { @@ -497,6 +503,10 @@ impl RealmChat for RealmChatServer { // if result.is_err() { // error!("Error broadcasting NewRoomEvent!"); // } + + // self.events.lock().await.push( + // (self.events.lock().await.len() as u32 + 1, + // Event::NewRoom(room.clone()))); Ok(room) } @@ -524,6 +534,10 @@ impl RealmChat for RealmChatServer { // if result.is_err() { // error!("Error broadcasting DeleteRoomEvent!"); // } + + // self.events.lock().await.push( + // (self.events.lock().await.len() as u32 + 1, + // Event::DeleteRoom(roomid.clone()))); Ok(()) } diff --git a/server/src/types.rs b/server/src/types.rs index 8977978..8a83dd3 100644 --- a/server/src/types.rs +++ b/server/src/types.rs @@ -26,7 +26,7 @@ pub trait RealmChat { //NOTE: Any user can call, if they are in the server async fn get_message(stoken: String, userid: String, id: i64) -> Result; - async fn get_messages_since(stoken: String, userid: String, time: DateTime) -> Result, ErrorCode>; + async fn get_messages_since(stoken: String, userid: String, id: i64) -> Result, ErrorCode>; async fn get_all_direct_replies(stoken: String, userid: String, head: i64) -> Result, ErrorCode>; async fn get_reply_chain(stoken: String, userid: String, head: Message, depth: u8) -> Result; async fn get_rooms(stoken: String, userid: String) -> Result, ErrorCode>;