joshuafhiggins/realm-chat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

my suffering is over

Browse Source
This commit is contained in:
Joshua Higgins
2024-10-16 00:06:48 -04:00
Unverified
parent 54a0b7b00b
commit 3cfad53e06
7 changed files with 108 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
client/src/app.rs
View File

@@ -102,7 +102,7 @@ pub struct RealmApp {
pub room_changes_channel: (Sender<Result<(CServer, Vec<Room>), ErrorCode>>, Receiver<Result<(CServer, Vec<Room>), ErrorCode>>), pub room_changes_channel: (Sender<Result<(CServer, Vec<Room>), ErrorCode>>, Receiver<Result<(CServer, Vec<Room>), ErrorCode>>),
#[serde(skip)] #[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)] #[serde(skip)]
pub polling_threads: Vec<(String, JoinHandle<()>)>, pub polling_threads: Vec<(String, JoinHandle<()>)>,
} }
@@ -568,13 +568,16 @@ impl eframe::App for RealmApp {
if self.selected_roomid.eq(&roomid) { if self.selected_roomid.eq(&roomid) {
self.selected_roomid.clear(); self.selected_roomid.clear();
} }
},
_ => { }
} }
} if index > server.last_event_index {
server.last_event_index = index; server.last_event_index = index;
} }
} }
} }
} }
}
// Manage polling threads // Manage polling threads
if let Some(active_servers) = &mut self.active_servers { if let Some(active_servers) = &mut self.active_servers {
@@ -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::<Vec<CServer>>(); let missing_servers = active_servers.clone().into_iter().filter(|s| !running_thread_serverids.contains(&s.server_id)).collect::<Vec<CServer>>();
for server in missing_servers { for server in missing_servers {
let send_channel = self.event_channel.0.clone(); 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); let mut transport = tarpc::serde_transport::tcp::connect(format!("{}:{}", server.domain, server.port), Json::default);
transport.config_mut().max_frame_length(usize::MAX); transport.config_mut().max_frame_length(usize::MAX);
let result = transport.await; let result = transport.await;
let connection = match result { let connection = match result {
Ok(connection) => connection, Ok(connection) => connection,
@@ -594,26 +599,35 @@ impl eframe::App for RealmApp {
return; return;
} }
}; };
let client = RealmChatClient::new(tarpc::client::Config::default(), connection).spawn(); let client = RealmChatClient::new(tarpc::client::Config::default(), connection).spawn();
let mut last_event_index = 0;
loop { loop {
let result = client.poll_events_since( let result = client.get_messages_since(
context::current(), context::current(),
server.last_event_index stoken(&token, &serverid, &server.domain, server.port),
userid.clone(),
last_event_index
).await; ).await;
match result { match result {
Ok(events) => { Ok(messages) => {
for event in events { if let Ok(messages) = messages {
send_channel.send((server.server_id.clone(), (event.0, event.1))).unwrap(); 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, Err(_) => break,
} }
sleep(Duration::from_millis(1000)).await; sleep(Duration::from_millis(5000)).await;
} }
}); });
self.polling_threads.push((server.server_id.clone(), handle));
} }
} }
} }

2
client/src/types.rs
View File

@@ -20,6 +20,6 @@ pub struct CServer {
pub is_admin: bool, pub is_admin: bool,
pub is_owner: bool, pub is_owner: bool,
pub rooms: Vec<Room>, pub rooms: Vec<Room>,
pub last_event_index: u32, pub last_event_index: i64,
pub messages: Vec<Message>, pub messages: Vec<Message>,
} }

2
client/src/ui/gui.rs
View File

@@ -223,7 +223,7 @@ pub fn rooms(app: &mut RealmApp, ctx: &Context) {
pub fn messages(app: &mut RealmApp, ctx: &Context) { pub fn messages(app: &mut RealmApp, ctx: &Context) {
egui::CentralPanel::default().show(ctx, |ui| { egui::CentralPanel::default().show(ctx, |ui| {
ui.with_layout(egui::Layout::bottom_up(egui::Align::TOP), |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 ui.button("✉").on_hover_text("Send a message").clicked() {
if let Some(active_servers) = &app.active_servers { if let Some(active_servers) = &app.active_servers {
for server in active_servers.clone() { for server in active_servers.clone() {

1
server/src/events.rs
View File

@@ -4,6 +4,7 @@ use crate::types::{Message, Room, User};
pub enum Event { pub enum Event {
// UserJoined(User), // UserJoined(User),
// UserLeft(User), // UserLeft(User),
None,
NewMessage(Message), NewMessage(Message),
NewRoom(Room), NewRoom(Room),
DeleteRoom(String), DeleteRoom(String),

14
server/src/main.rs
View File

@@ -94,23 +94,15 @@ async fn main() -> anyhow::Result<()> {
.filter_map(|r| future::ready(r.ok())) .filter_map(|r| future::ready(r.ok()))
.map(BaseChannel::with_defaults) .map(BaseChannel::with_defaults)
// Limit channels to 1 per IP. // 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 // serve is generated by the service attribute. It takes as input any type implementing
// the generated World trait. // the generated World trait.
.map(|channel| { .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 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() channel.execute(server.serve()).for_each(spawn)
.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)
}) })
// Max 10 channels. // Max 10 channels.
.buffer_unordered(4096) .buffer_unordered(10240)
.for_each(|_| async {}) .for_each(|_| async {})
.await; .await;

112
server/src/server.rs
View File

@@ -25,13 +25,13 @@ pub struct RealmChatServer {
pub db_pool: Pool<Sqlite>, pub db_pool: Pool<Sqlite>,
pub typing_users: Vec<(String, String)>, //NOTE: user.userid, room.roomid pub typing_users: Vec<(String, String)>, //NOTE: user.userid, room.roomid
pub cache: Cache<String, String>, pub cache: Cache<String, String>,
pub events: Vec<(u32, Event)>, //pub events: Arc<Mutex<Vec<(u32, Event)>>>,
} }
const FETCH_MESSAGE: &str = "SELECT message.*, 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', 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' 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 { impl RealmChatServer {
pub fn new(server_id: String, socket: SocketAddr, db_pool: Pool<Sqlite>) -> RealmChatServer { pub fn new(server_id: String, socket: SocketAddr, db_pool: Pool<Sqlite>) -> RealmChatServer {
@@ -47,7 +47,7 @@ impl RealmChatServer {
.time_to_idle(Duration::from_secs(5*60)) .time_to_idle(Duration::from_secs(5*60))
.time_to_live(Duration::from_secs(60*60)) .time_to_live(Duration::from_secs(60*60))
.build(), .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)> { 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 { // let mut events_to_send = Vec::new();
if i > index { //
events_to_send.push((i, event)); // for (i, event) in self.events.lock().await.clone() {
} // if i > index {
} // events_to_send.push((i, event));
// }
events_to_send // }
//
// events_to_send
Vec::new()
} }
async fn join_server(self, _: Context, stoken: String, userid: String) -> Result<User, ErrorCode> { async fn join_server(self, _: Context, stoken: String, userid: String) -> Result<User, ErrorCode> {
@@ -296,42 +299,42 @@ impl RealmChat for RealmChatServer {
} }
async fn send_message(self, _: Context, stoken: String, mut message: Message) -> Result<Message, ErrorCode> { async fn send_message(self, _: Context, stoken: String, mut message: Message) -> Result<Message, ErrorCode> {
if !self.is_stoken_valid(&message.user.userid, &stoken).await { // Check sender userid // if !self.is_stoken_valid(&message.user.userid, &stoken).await { // Check sender userid
return Err(Unauthorized) // 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 // match &message.data { // Check that the sender is the owner of the referencing msg
message.user = self.inner_get_user(&message.user.userid).await?; // 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 // let is_admin = self.internal_is_user_admin(&message.user.userid).await;
MessageData::Edit(e) => { // let admin_only_send = query!(
let ref_msg = self.inner_get_message(&message.user.userid, e.referencing_id).await?; // "SELECT admin_only_send FROM room WHERE roomid = ?",
if !ref_msg.user.userid.eq(&message.user.userid) { // message.room.roomid).fetch_one(&self.db_pool).await;
return Err(Unauthorized) // if let Ok(record) = admin_only_send {
} // if record.admin_only_send && !is_admin {
} // return Err(Unauthorized)
MessageData::Redaction(r)=> { // }
let ref_msg = self.inner_get_message(&message.user.userid, r.referencing_id).await?; // } else {
if !ref_msg.user.userid.eq(&message.user.userid) || !self.internal_is_user_admin(&message.user.userid).await { // return Err(RoomNotFound)
return Err(Unauthorized) // }
} //
} // 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 { let result = match &message.data {
MessageData::Text(text) => { MessageData::Text(text) => {
@@ -371,6 +374,9 @@ impl RealmChat for RealmChatServer {
// if result.is_err() { // if result.is_err() {
// error!("Error broadcasting NewMessageEvent!"); // error!("Error broadcasting NewMessageEvent!");
// } // }
// self.events.lock().await.push(
// (self.events.lock().await.len() as u32 + 1,
// Event::NewMessage(message.clone())));
Ok(message) Ok(message)
}, },
@@ -411,15 +417,15 @@ impl RealmChat for RealmChatServer {
} }
} }
async fn get_messages_since(self, _: Context, stoken: String, userid: String, time: DateTime<Utc>) -> Result<Vec<Message>, ErrorCode> { async fn get_messages_since(self, _: Context, stoken: String, userid: String, id: i64) -> Result<Vec<Message>, ErrorCode> {
if !self.is_stoken_valid(&userid, &stoken).await { if !self.is_stoken_valid(&userid, &stoken).await {
return Err(Unauthorized) return Err(Unauthorized)
} }
let is_admin = self.internal_is_user_admin(&userid).await; let is_admin = self.internal_is_user_admin(&userid).await;
let result = sqlx::query(&format!("{}{}", FETCH_MESSAGE, "AND message.timestamp >= ?")) let result = sqlx::query(&format!("{}{}", FETCH_MESSAGE, " AND message.id > ?"))
.bind(is_admin) //.bind(is_admin)
.bind(time) .bind(id)
.fetch_all(&self.db_pool).await; .fetch_all(&self.db_pool).await;
match result { match result {
@@ -498,6 +504,10 @@ impl RealmChat for RealmChatServer {
// error!("Error broadcasting NewRoomEvent!"); // error!("Error broadcasting NewRoomEvent!");
// } // }
// self.events.lock().await.push(
// (self.events.lock().await.len() as u32 + 1,
// Event::NewRoom(room.clone())));
Ok(room) Ok(room)
} }
Err(_) => Err(MalformedDBResponse) Err(_) => Err(MalformedDBResponse)
@@ -525,6 +535,10 @@ impl RealmChat for RealmChatServer {
// error!("Error broadcasting DeleteRoomEvent!"); // error!("Error broadcasting DeleteRoomEvent!");
// } // }
// self.events.lock().await.push(
// (self.events.lock().await.len() as u32 + 1,
// Event::DeleteRoom(roomid.clone())));
Ok(()) Ok(())
} }
Err(_) => Err(MalformedDBResponse) Err(_) => Err(MalformedDBResponse)

2
server/src/types.rs
View File

@@ -26,7 +26,7 @@ pub trait RealmChat {
//NOTE: Any user can call, if they are in the server //NOTE: Any user can call, if they are in the server
async fn get_message(stoken: String, userid: String, id: i64) -> Result<Message, ErrorCode>; async fn get_message(stoken: String, userid: String, id: i64) -> Result<Message, ErrorCode>;
async fn get_messages_since(stoken: String, userid: String, time: DateTime<Utc>) -> Result<Vec<Message>, ErrorCode>; async fn get_messages_since(stoken: String, userid: String, id: i64) -> Result<Vec<Message>, ErrorCode>;
async fn get_all_direct_replies(stoken: String, userid: String, head: i64) -> Result<Vec<Message>, ErrorCode>; async fn get_all_direct_replies(stoken: String, userid: String, head: i64) -> Result<Vec<Message>, ErrorCode>;
async fn get_reply_chain(stoken: String, userid: String, head: Message, depth: u8) -> Result<ReplyChain, ErrorCode>; async fn get_reply_chain(stoken: String, userid: String, head: Message, depth: u8) -> Result<ReplyChain, ErrorCode>;
async fn get_rooms(stoken: String, userid: String) -> Result<Vec<Room>, ErrorCode>; async fn get_rooms(stoken: String, userid: String) -> Result<Vec<Room>, ErrorCode>;