diff --git a/src/main.rs b/src/main.rs index c5d1570..7e62770 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,5 +1,6 @@ mod cli; mod macros; +mod node_requests; mod plugin; mod requests; mod server; @@ -22,6 +23,17 @@ fn main() -> Result<()> { } else { vfs::read_config(&root.join("config.json"))? }; + + if let Ok(a) = std::env::var("AXIOM_NODE") { + if a == "true" { + config + .build_req(&root, crate::server::Server::call_node_request) + .run()?; + + return Ok(()); + } + } + config.build(&root).run()?; Ok(()) } diff --git a/src/node_requests/chunk.rs b/src/node_requests/chunk.rs new file mode 100644 index 0000000..374c37c --- /dev/null +++ b/src/node_requests/chunk.rs @@ -0,0 +1,18 @@ +use crate::{server::Server, types::message::ServerMessage, utils::client::Client}; +use std::sync::Arc; + +crate::logger!(LOGGER "Chunk Loader"); + +pub fn load_chunk( + server: &Arc, + client: &Client, + channel_id: &str, + chunk_id: usize, +) -> crate::Result<()> { + let mut chunk = server + .db + .get_chunk_node(&client.get_uuid()?, channel_id, chunk_id)?; + chunk.reverse(); + client.send(ServerMessage::Chunk(chunk))?; + Ok(()) +} diff --git a/src/node_requests/indicator.rs b/src/node_requests/indicator.rs new file mode 100644 index 0000000..665bd1c --- /dev/null +++ b/src/node_requests/indicator.rs @@ -0,0 +1,6 @@ +use crate::{server::Server, utils::client::Client}; +use std::sync::Arc; + +pub fn start_typing(server: &Arc, client: &Client, channel_id: &str) -> crate::Result<()> { + Ok(()) +} diff --git a/src/node_requests/message.rs b/src/node_requests/message.rs new file mode 100644 index 0000000..78600aa --- /dev/null +++ b/src/node_requests/message.rs @@ -0,0 +1,83 @@ +use std::sync::Arc; + +use anyhow::anyhow; + +use crate::{server::Server, types, utils::client::Client}; + +crate::logger!(LOGGER "Message Manager"); + +pub fn send( + server: &Arc, + client: &Client, + channel_id: &str, + contents: &str, +) -> crate::Result<()> { + LOGGER.info(format!("SendMessage to {channel_id}: {contents}")); + + if contents.is_empty() { + client.send(types::message::ResponseError::InvalidRequest(format!( + "Invalid message: empty message" + )))?; + + return Ok(()); + } + + let msg = server.db.insert_message( + &channel_id, + &client.get_uuid()?, + &contents, + chrono::Utc::now().timestamp(), + )?; + + server.broadcast_to( + &[&msg.channel_id, &msg.from], + types::message::ServerMessage::MessageCreate(msg.clone()), + )?; + + Ok(()) +} + +pub fn edit( + server: &Arc, + client: &Client, + message_id: i64, + new_contents: &str, +) -> crate::Result<()> { + LOGGER.info(format!("EditMessage {message_id}: {new_contents}")); + let Some(msg) = server.db.get_message_by_id(message_id)? else { + return Err(anyhow!("Message does not exist")); + }; + + if msg.from != client.get_uuid()? { + return Err(anyhow!("You are not the author of this message")); + } + + server.db.edit_message(message_id, new_contents)?; + + server.broadcast_to( + &[&msg.channel_id, &msg.from], + types::message::ServerMessage::MessageCreate(msg.clone()), + )?; + + Ok(()) +} + +pub fn delete(server: &Arc, client: &Client, message_id: i64) -> crate::Result<()> { + LOGGER.info(format!("DeleteMessage {message_id}")); + let Some(msg) = server.db.get_message_by_id(message_id)? else { + return Err(anyhow!("Message does not exist")); + }; + + if msg.from != client.get_uuid()? { + return Err(anyhow!("You are not the author of this message")); + } + + server.db.delete_message(message_id)?; + + server.broadcast_to( + &[&msg.channel_id, &msg.from], + types::message::ServerMessage::MessageDelete { message_id }, + )?; + + Ok(()) +} diff --git a/src/node_requests/mod.rs b/src/node_requests/mod.rs new file mode 100644 index 0000000..cd4a322 --- /dev/null +++ b/src/node_requests/mod.rs @@ -0,0 +1,86 @@ +pub mod chunk; +pub mod indicator; +pub mod message; +pub mod voice; + +use std::sync::Arc; + +use crate::{ + server::Server, + types::message::{ClientMessage, ServerMessage, WsMessage}, + utils::client::Client, +}; + +impl Server { + pub fn call_node_request( + self: &Arc, + req: &WsMessage, + client: &Client, + ) -> crate::Result<()> { + match req { + WsMessage::Message(req) => match req { + ClientMessage::SendMessage { + channel_id, + contents, + } => { + message::send(self, client, channel_id, contents)?; + } + + ClientMessage::EditMessage { + message_id, + new_contents, + } => message::edit(self, client, *message_id, new_contents)?, + + ClientMessage::DeleteMessage { message_id } => { + message::delete(self, client, *message_id)? + } + + ClientMessage::LoadChunk { + chunk_id, + channel_id, + } => chunk::load_chunk(self, client, channel_id, *chunk_id)?, + + ClientMessage::Typing { channel_id } => { + indicator::start_typing(self, client, channel_id)? + } + + ClientMessage::JoinVoice { channel_id } => voice::join(self, client, channel_id)?, + ClientMessage::LeaveVoice { channel_id } => voice::leave(self, client, channel_id)?, + }, + + WsMessage::Binary(data) => { + // Self::LOGGER.info(format!("Binary message: {data:?}")); + voice::voice(self, client, data)?; + } + + WsMessage::String(s) => { + Self::LOGGER.info(format!("String message: {s}")); + } + } + + Ok(()) + } + + pub fn broadcast_to( + self: &Arc, + targets: &[&String], + message: ServerMessage, + ) -> crate::Result<()> { + for c in self.clients.lock().unwrap().iter() { + if !targets.contains(&&c.get_uuid()?) { + continue; + } + + let c = c.clone(); + let server = self.clone(); + let message = message.clone(); + std::thread::spawn(move || { + server + .wrap_err(&c, c.send(&message)) + .expect("Failed to broadcast"); + }); + } + + Ok(()) + } +} diff --git a/src/node_requests/voice.rs b/src/node_requests/voice.rs new file mode 100644 index 0000000..924a34d --- /dev/null +++ b/src/node_requests/voice.rs @@ -0,0 +1,61 @@ +use crate::{server::Server, utils::client::Client}; +use std::sync::Arc; + +crate::logger!(LOGGER "Voice chat"); + +pub fn join(server: &Arc, client: &Client, channel_id: &str) -> crate::Result<()> { + let user_id = client.get_uuid()?; + + let voice_id = server + .voice + .lock() + .unwrap() + .set(user_id.clone(), channel_id.to_string()); + + server.broadcast(crate::types::message::ServerMessage::VoiceJoin { + user_id, + channel_id: channel_id.to_string(), + voice_id: voice_id, + }); + + Ok(()) +} + +pub fn leave(server: &Arc, client: &Client, channel_id: &str) -> crate::Result<()> { + let user_id = client.get_uuid()?; + + let Some(voice_id) = server.voice.lock().unwrap().remove(channel_id, &user_id) else { + return Ok(()); + }; + + server.broadcast(crate::types::message::ServerMessage::VoiceLeave { + user_id, + channel_id: channel_id.to_string(), + voice_id, + }); + + Ok(()) +} + +pub fn voice(server: &Arc, client: &Client, data: &[u8]) -> crate::Result<()> { + let v = server.voice.lock().unwrap(); + let user_id = client.get_uuid()?; + let Some((channel_id, voice_id)) = v.find_user(&user_id) else { + return Ok(()); + }; + + let mut targets = v.get(&channel_id); + + if let Some(pos) = targets.iter().position(|x| *x == &user_id) { + targets.remove(pos); + } + + let prefix = voice_id.to_le_bytes(); + let mut payload = Vec::with_capacity(prefix.len() + data.len()); + payload.extend_from_slice(&prefix); + payload.extend_from_slice(data); + + server.broadcast_bin_to(&targets, payload)?; + + Ok(()) +} diff --git a/src/requests/mod.rs b/src/requests/mod.rs index c7844f5..969b11d 100644 --- a/src/requests/mod.rs +++ b/src/requests/mod.rs @@ -12,7 +12,7 @@ use crate::{ }; impl Server { - pub fn call_request( + pub fn call_server_request( self: &Arc, req: &WsMessage, client: &Client, diff --git a/src/server.rs b/src/server.rs index c34397d..005427a 100644 --- a/src/server.rs +++ b/src/server.rs @@ -25,7 +25,10 @@ use crate::{ cli, logger, plugin::{Plugin, loader::PluginLoader, types::LoaderMessage}, requests::voice, - types::{self, message::WsMessage}, + types::{ + self, + message::{ClientMessage, WsMessage}, + }, utils::{self, auth, client::Client, voice::Voice}, }; @@ -47,6 +50,7 @@ pub struct Server { pub shutting_down: AtomicBool, pub indicators: Mutex>, pub voice: Mutex, + pub call_request: fn(&Arc, &WsMessage, &Client) -> crate::Result<()>, } impl Default for ServerConfig { @@ -66,6 +70,14 @@ impl ServerConfig { Server::new_config(root, self) } + pub fn build_req( + self, + root: &Path, + req: fn(&Arc, &WsMessage, &Client) -> crate::Result<()>, + ) -> Arc { + Server::new_req_config(root, req, self) + } + pub fn from_str(s: &str) -> std::result::Result { serde_json::from_str(s) } @@ -74,6 +86,24 @@ impl ServerConfig { impl Server { logger!(LOGGER "Server"); + pub fn new_req_config( + root: &Path, + call_request: fn(&Arc, &WsMessage, &Client) -> crate::Result<()>, + config: ServerConfig, + ) -> Arc { + Arc::new(Self { + db: utils::database::Database::new(&config).unwrap(), + root: root.to_path_buf(), + config, + clients: Mutex::new(HashSet::new()), + plugins: Mutex::new(Vec::new()), + shutting_down: AtomicBool::new(false), + indicators: Mutex::new(Vec::new()), + voice: Mutex::new(Voice::new()), + call_request: call_request, + }) + } + pub fn new_config(root: &Path, config: ServerConfig) -> Arc { Arc::new(Self { db: utils::database::Database::new(&config).unwrap(), @@ -84,6 +114,7 @@ impl Server { shutting_down: AtomicBool::new(false), indicators: Mutex::new(Vec::new()), voice: Mutex::new(Voice::new()), + call_request: Self::call_server_request, }) } @@ -214,7 +245,7 @@ impl Server { } } - self.wrap_err(&client, self.call_request(r, &client))?; + self.wrap_err(&client, (self.call_request)(self, r, &client))?; } } Ok(()) diff --git a/src/utils/database.rs b/src/utils/database.rs index ac78f1d..9cfb52e 100644 --- a/src/utils/database.rs +++ b/src/utils/database.rs @@ -129,6 +129,38 @@ impl Database { Ok(messages) } + + pub fn get_chunk_node( + &self, + author: &str, + channel_id: &str, + chunk_id: usize, + ) -> Result> { + let mut stmt = self.0.prepare( + "SELECT id, channel_id, user_id, contents, timestamp + FROM chat + WHERE channel_id = ?1 AND user_id = ?3 + ORDER BY id DESC + LIMIT 16 OFFSET (?2 * 16)", + )?; + + let rows = stmt.query_map(params![channel_id, chunk_id, author], |row| { + Ok(Message { + id: row.get::<_, i64>(0)?, + channel_id: row.get::<_, String>(1)?, + from: row.get::<_, String>(2)?, + contents: row.get::<_, String>(3)?, + timestamp: row.get::<_, i64>(4)?, + }) + })?; + + let mut messages = Vec::new(); + for row in rows { + messages.push(row?); + } + + Ok(messages) + } } unsafe impl Send for Database {}