Node mode

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(())
 }

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<Server>,
+    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(())
+}

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<Server>, client: &Client, channel_id: &str) -> crate::Result<()> {
+    Ok(())
+}

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<Server>,
+    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<Server>,
+    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<Server>, 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(())
+}

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<Self>,
+        req: &WsMessage<ClientMessage>,
+        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<Self>,
+        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(())
+    }
+}

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<Server>, 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<Server>, 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<Server>, 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(())
+}

src/requests/mod.rs

@@ -12,7 +12,7 @@ use crate::{
 };
 
 impl Server {
-    pub fn call_request(
+    pub fn call_server_request(
         self: &Arc<Self>,
         req: &WsMessage<ClientMessage>,
         client: &Client,

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<Vec<crate::requests::indicator::IndicatorContext>>,
     pub voice: Mutex<crate::utils::voice::Voice>,
+    pub call_request: fn(&Arc<Self>, &WsMessage<ClientMessage>, &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<Server>, &WsMessage<ClientMessage>, &Client) -> crate::Result<()>,
+    ) -> Arc<Server> {
+        Server::new_req_config(root, req, self)
+    }
+
     pub fn from_str(s: &str) -> std::result::Result<Self, serde_json::Error> {
         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<Self>, &WsMessage<ClientMessage>, &Client) -> crate::Result<()>,
+        config: ServerConfig,
+    ) -> Arc<Self> {
+        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<Self> {
         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(())

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<Vec<Message>> {
+        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 {}