feat: basic messaging implementation

Cargo.lock

@@ -73,6 +73,17 @@ version = "1.0.100"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a23eb6b1614318a8071c9b2521f36b424b2c83db5eb3a0fead4a6c0809af6e61"
 
+[[package]]
+name = "async-trait"
+version = "0.1.89"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "autocfg"
 version = "1.5.0"
@@ -186,7 +197,9 @@ name = "irs"
 version = "0.1.0"
 dependencies = [
  "anyhow",
+ "async-trait",
  "clap",
+ "once_cell",
  "tokio",
 ]
 
@@ -247,6 +260,12 @@ dependencies = [
  "memchr",
 ]
 
+[[package]]
+name = "once_cell"
+version = "1.21.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "42f5e15c9953c5e4ccceeb2e7382a716482c34515315f7b03532b8b4e8393d2d"
+
 [[package]]
 name = "once_cell_polyfill"
 version = "1.70.1"

Cargo.toml

@@ -5,5 +5,7 @@ edition = "2024"
 
 [dependencies]
 anyhow = "1.0.100"
+async-trait = "0.1.89"
 clap = { version = "4.5.48", features = ["derive"] }
+once_cell = "1.21.3"
 tokio = { version = "1.47.1", features = ["full"] }

src/commands/cap.rs

@@ -1,18 +1,21 @@
+use async_trait::async_trait;
+
 use crate::{
     commands::{IrcAction, IrcHandler},
     sender::IrcResponse,
+    user::User,
 };
 
 pub struct Cap;
 
+#[async_trait]
 impl IrcHandler for Cap {
-    fn handle(&self, _arguments: Vec<String>) -> super::IrcAction {
+    async fn handle(
-        // TODO: parse the args, etc
+        &self,
-        IrcAction::SendText(IrcResponse {
+        arguments: Vec<String>,
-            command: "CAP".into(),
+        _authenticated: bool,
-            receiver: "*".into(),
+        _user_state: &mut User,
-            arguments: Some("LS".into()),
+    ) -> super::IrcAction {
-            message: "TODO".into(),
+        IrcAction::DoNothing
-        })
     }
 }

src/commands/mod.rs

@@ -1,29 +1,54 @@
 #![allow(dead_code)]
-use std::{collections::HashMap, io::BufWriter, net::TcpStream};
+use std::collections::HashMap;
 
 use anyhow::{Result, anyhow};
+use async_trait::async_trait;
+use tokio::{io::BufWriter, net::TcpStream};
 
-use crate::{commands::cap::Cap, sender::IrcResponse};
+use crate::{
+    commands::{cap::Cap, nick::Nick, ping::Ping, privmsg::PrivMsg, user::User as UserHandler},
+    sender::IrcResponse,
+    user::User,
+};
 
 mod cap;
+mod nick;
+mod ping;
+mod privmsg;
+mod user;
 
+#[derive(Debug)]
 pub struct IrcCommand {
     command: String,
     arguments: Vec<String>,
 }
 
+pub struct IrcMessage {
+    pub sender: String, // TODO: replace with hostmask
+    pub message: String,
+}
+
 pub enum IrcAction {
     MultipleActions(Vec<Self>),
-    ModifyDatabase(DatabaseAction),
     SendText(IrcResponse),
+    ErrorAuthenticateFirst,
+    DoNothing,
 }
 
 pub enum DatabaseAction {}
 
-pub trait IrcHandler {
+#[async_trait]
-    fn handle(&self, command: Vec<String>) -> IrcAction;
+pub trait IrcHandler: Send + Sync {
+    async fn handle(
+        &self,
+        command: Vec<String>,
+        authenticated: bool,
+        user_state: &mut User,
+    ) -> IrcAction;
 }
 
+pub struct SendMessage(Option<String>);
+
 impl IrcCommand {
     pub fn new(command_with_arguments: String) -> Self {
         let split_command: Vec<&str> = command_with_arguments
@@ -32,10 +57,24 @@ impl IrcCommand {
             .collect();
         let command = split_command[0].to_owned();
         let mut arguments = Vec::new();
+        let mut buffer: Option<String> = None;
 
         split_command[1..]
             .iter()
-            .for_each(|e| arguments.push(e.to_string()));
+            .for_each(|e| match (buffer.as_mut(), e.starts_with(":")) {
+                (None, false) => arguments.push(e.to_string()),
+                (None, true) => {
+                    buffer = Some(e[1..].to_string());
+                }
+                (Some(buf), starts_with_colon) => {
+                    buf.push(' ');
+                    buf.push_str(if starts_with_colon { &e[1..] } else { &e });
+                }
+            });
+
+        if let Some(buf) = buffer {
+            arguments.push(buf.to_string());
+        }
 
         Self {
             command: command,
@@ -43,35 +82,56 @@ impl IrcCommand {
         }
     }
 
-    pub fn execute(&self, writer: &mut BufWriter<&TcpStream>, hostname: &str) -> Result<()> {
+    pub async fn execute(
+        &self,
+        writer: &mut BufWriter<TcpStream>,
+        hostname: &str,
+        user_state: &mut User,
+    ) -> Result<()> {
         let mut command_map: HashMap<String, &dyn IrcHandler> = HashMap::new();
 
         // Command map is defined here
         command_map.insert("CAP".to_owned(), &Cap);
+        command_map.insert("NICK".to_owned(), &Nick);
+        command_map.insert("USER".to_owned(), &UserHandler);
+        command_map.insert("PRIVMSG".to_owned(), &PrivMsg);
+        command_map.insert("PING".to_owned(), &Ping);
+
+        println!("{self:#?}");
 
         let command_to_execute = command_map
             .get(&self.command.to_uppercase())
             .map(|v| *v)
             .ok_or(anyhow!("unknown command!"))?;
 
-        let action = command_to_execute.handle(self.arguments.clone());
+        let action = command_to_execute
-        action.execute(writer, hostname);
+            .handle(
+                self.arguments.clone(),
+                user_state.is_populated(),
+                user_state,
+            )
+            .await;
+        action.execute(writer, hostname, user_state).await;
 
         Ok(())
     }
 }
 
 impl IrcAction {
-    pub fn execute(&self, writer: &mut BufWriter<&TcpStream>, hostname: &str) {
+    pub async fn execute(
+        &self,
+        writer: &mut BufWriter<TcpStream>,
+        hostname: &str,
+        user_state: &mut User,
+    ) {
         match self {
-            IrcAction::MultipleActions(actions) => {
+            /*IrcAction::MultipleActions(actions) => {
                 for action in actions {
-                    action.execute(writer, hostname);
+                    action.execute(writer, hostname, user_state);
                 }
-            }
+            }*/
-
             IrcAction::SendText(msg) => {
-                msg.send(hostname, writer).unwrap();
+                msg.send(hostname, writer, false).await.unwrap();
             }
 
             _ => {}

src/commands/nick.rs

@@ -0,0 +1,22 @@
+use async_trait::async_trait;
+
+use crate::{
+    commands::{IrcAction, IrcHandler},
+    user::User,
+};
+
+pub struct Nick;
+
+#[async_trait]
+impl IrcHandler for Nick {
+    async fn handle(
+        &self,
+        command: Vec<String>,
+        _authenticated: bool,
+        user_state: &mut User,
+    ) -> IrcAction {
+        user_state.nickname = Some(command[0].clone());
+
+        IrcAction::DoNothing
+    }
+}

src/commands/ping.rs

@@ -0,0 +1,30 @@
+use async_trait::async_trait;
+
+use crate::{
+    commands::{IrcAction, IrcHandler},
+    sender::IrcResponse,
+    user::User,
+};
+
+pub struct Ping;
+
+#[async_trait]
+impl IrcHandler for Ping {
+    async fn handle(
+        &self,
+        command: Vec<String>,
+        authenticated: bool,
+        user_state: &mut User,
+    ) -> IrcAction {
+        if authenticated {
+            IrcAction::SendText(IrcResponse {
+                sender: None,
+                command: "PONG".into(),
+                receiver: user_state.nickname.clone().unwrap(),
+                message: command[0].clone(),
+            })
+        } else {
+            IrcAction::DoNothing
+        }
+    }
+}

src/commands/privmsg.rs

@@ -0,0 +1,40 @@
+use async_trait::async_trait;
+use tokio::sync::broadcast::Sender;
+
+use crate::{
+    CONNECTED_USERS, SENDER,
+    commands::{IrcAction, IrcHandler},
+    messages::Message,
+    user::User,
+};
+
+pub struct PrivMsg;
+
+#[async_trait]
+impl IrcHandler for PrivMsg {
+    async fn handle(
+        &self,
+        command: Vec<String>,
+        authenticated: bool,
+        user_state: &mut User,
+    ) -> IrcAction {
+        if !authenticated {
+            return IrcAction::ErrorAuthenticateFirst;
+        }
+        let connected_users = CONNECTED_USERS.lock().await;
+        let sender = SENDER.lock().await.clone().unwrap();
+
+        println!("{connected_users:#?}");
+        drop(connected_users);
+
+        let message = Message {
+            sender: user_state.clone().unwrap_all(),
+            receiver: command[0].clone(),
+            text: command[1].clone(),
+        };
+        println!("SENDING: {message:#?}");
+        sender.send(message).unwrap();
+
+        IrcAction::DoNothing
+    }
+}

src/commands/user.rs

@@ -0,0 +1,23 @@
+use async_trait::async_trait;
+
+use crate::{
+    commands::{IrcAction, IrcHandler},
+    user::User as UserState,
+};
+
+pub struct User;
+
+#[async_trait]
+impl IrcHandler for User {
+    async fn handle(
+        &self,
+        command: Vec<String>,
+        _authenticated: bool,
+        user_state: &mut UserState,
+    ) -> IrcAction {
+        user_state.username = Some(command[0].clone());
+        user_state.realname = Some(command[3].clone());
+
+        IrcAction::DoNothing
+    }
+}

src/login.rs

@@ -0,0 +1,53 @@
+use anyhow::Result;
+use tokio::{io::BufWriter, net::TcpStream};
+
+use crate::{ServerInfo, sender::IrcResponseCodes, user::User};
+
+pub async fn send_motd(
+    server_info: ServerInfo,
+    user_info: User,
+    writer: &mut BufWriter<TcpStream>,
+) -> Result<()> {
+    let user_info = user_info.unwrap_all();
+    let server_version = &format!("IRS-v{}", env!("CARGO_PKG_VERSION")) as &str;
+
+    let welcome_text = format!(
+        "Welcome to the {} Internet Relay Chat Network {}",
+        server_info.network_name, user_info.nickname
+    );
+    let yourhost_text = format!(
+        "Your host is {}, running version {}",
+        server_info.server_hostname, server_version
+    );
+    let myinfo_text = format!("{} {} i b", server_info.server_hostname, server_version);
+    let isupport_text = format!(
+        "CHANTYPES=# NETWORK={} :are supported by this server",
+        server_info.network_name
+    );
+
+    IrcResponseCodes::Welcome
+        .into_irc_response(user_info.username.clone(), welcome_text)
+        .send(&server_info.server_hostname, writer, true)
+        .await?;
+    IrcResponseCodes::YourHost
+        .into_irc_response(user_info.username.clone(), yourhost_text)
+        .send(&server_info.server_hostname, writer, true)
+        .await?;
+    IrcResponseCodes::MyInfo
+        .into_irc_response(user_info.username.clone(), myinfo_text)
+        .send(&server_info.server_hostname, writer, false)
+        .await?;
+    IrcResponseCodes::ISupport
+        .into_irc_response(user_info.username.clone(), isupport_text)
+        .send(&server_info.server_hostname, writer, false)
+        .await?;
+    IrcResponseCodes::NoMotd
+        .into_irc_response(
+            user_info.username.clone(),
+            "MOTD not implemented yet".into(),
+        )
+        .send(&server_info.server_hostname, writer, true)
+        .await?;
+
+    Ok(())
+}

src/main.rs

@@ -1,61 +1,201 @@
 use std::{
+    collections::{HashMap, HashSet},
     io::{BufRead, BufReader, BufWriter},
     net::{SocketAddr, TcpListener, TcpStream},
     str::FromStr,
+    sync::mpsc,
+    time::Duration,
 };
 
-use anyhow::Result;
+use anyhow::{Result, bail};
-use tokio::spawn;
+use once_cell::sync::Lazy;
+use tokio::{
+    io::{AsyncBufReadExt, BufReader as TokioBufReader, BufWriter as TokioBufWriter},
+    net::TcpStream as TokioTcpStream,
+    spawn,
+    sync::{
+        Mutex,
+        broadcast::{self, Receiver, Sender},
+    },
+    time::sleep,
+};
 
-use crate::sender::IrcResponseCodes;
+use crate::{
+    login::send_motd,
+    messages::Message,
+    sender::{IrcResponse, IrcResponseCodes},
+    user::{User, UserUnwrapped},
+};
 
 mod commands;
+mod login;
+mod messages;
 mod sender;
+mod user;
+
+pub static CONNECTED_USERS: Lazy<Mutex<HashSet<UserUnwrapped>>> =
+    Lazy::new(|| Mutex::new(HashSet::new()));
+pub static SENDER: Lazy<Mutex<Option<Sender<Message>>>> = Lazy::new(|| Mutex::new(None));
+
+#[allow(dead_code)]
+#[derive(Clone)]
+struct ServerInfo {
+    ip: String,
+    port: String,
+    server_hostname: String,
+    network_name: String,
+    operators: Vec<String>,
+}
 
 #[tokio::main]
 async fn main() -> Result<()> {
-    let ip = "0.0.0.0";
+    let info = ServerInfo {
-    let port = "6667";
+        ip: "0.0.0.0".into(),
-    let server_hostname = "irc.blah.blah";
+        port: "6667".into(),
+        server_hostname: "irc.blah.blah".into(),
+        network_name: "TeamDunno".into(),
+        operators: Vec::new(),
+    };
     // TODO: ^ pull these from a config file
 
-    let listener = TcpListener::bind(SocketAddr::from_str(&format!("{}:{}", ip, port))?)?;
+    let listener = TcpListener::bind(SocketAddr::from_str(&format!("{}:{}", info.ip, info.port))?)?;
+    let (tx, mut _rx) = broadcast::channel::<Message>(32);
+    let mut sender_mut = SENDER.lock().await;
+    *sender_mut = Some(tx.clone());
+    drop(sender_mut);
 
     for stream in listener.incoming() {
         let stream = stream?;
+        stream.set_nonblocking(true)?;
+        let tx_thread = tx.clone();
+        let info = info.clone();
 
-        spawn(async move { handle_connection(stream, server_hostname).await.unwrap() });
+        spawn(async move {
+            handle_connection(stream, info, /*&mut rx_thread,*/ tx_thread)
+                .await
+                .unwrap()
+        });
     }
 
     Ok(())
 }
 
-async fn handle_connection(stream: TcpStream, hostname: &str) -> Result<()> {
+async fn handle_connection(stream: TcpStream, info: ServerInfo, tx: Sender<Message>) -> Result<()> {
-    let reader_stream = stream.try_clone()?;
+    let stream_tcp = stream.try_clone()?;
-    let mut reader = BufReader::new(&reader_stream);
+    let mut message_receiver = tx.clone().subscribe();
-    let mut writer = BufWriter::new(&stream);
+    let mut tcp_reader = TokioBufReader::new(TokioTcpStream::from_std(stream.try_clone()?)?);
-    let mut buffer = String::new();
+    let mut tcp_writer = TokioBufWriter::new(TokioTcpStream::from_std(stream)?);
+    let mut state = User::default();
 
     loop {
-        buffer.clear();
+        tokio::select! {
-        if reader.read_line(&mut buffer).unwrap() == 0 {
+            result = tcp_listener(&stream_tcp, state.clone(), &info, &mut tcp_reader) => {
-            break;
+                match result {
-        }
+                    Ok(modified_user) => {
+                        state = modified_user;
+                    }
 
-        let command = commands::IrcCommand::new(buffer.clone());
+                    Err(_) => {
-        match command.execute(&mut writer, hostname) {
+                        break;
-            Ok(_) => {}
+                    }
-            Err(error) => {
+                }
-                let error_string = format!("error processing your command: {error:#?}\n");
+            },
-                let error = IrcResponseCodes::UnknownCommand;
+            result = message_listener(&state, &mut message_receiver, &mut tcp_writer) => {
-
+                match result {
-                error
+                    Ok(_) => {},
-                    .into_irc_response("*".into(), error_string.into())
+                    Err(_) => {
-                    .send(hostname, &mut writer)
+                        // break;
-                    .unwrap();
+                    }
-            }
+                }
+            },
+            _ = sleep(Duration::from_millis(200)) => {},
         }
     }
 
+    stream_tcp.shutdown(std::net::Shutdown::Both)?;
+
+    Ok(())
+}
+
+async fn tcp_listener(
+    stream: &TcpStream,
+    mut state: User,
+    info: &ServerInfo,
+    reader: &mut TokioBufReader<TokioTcpStream>,
+) -> Result<User> {
+    let mut buffer = String::new();
+
+    let mut writer = TokioBufWriter::new(TokioTcpStream::from_std(stream.try_clone()?)?);
+
+    buffer.clear();
+    match reader.read_line(&mut buffer).await {
+        Ok(0) => bail!("invalid response"),
+        Ok(_) => {}
+
+        Err(_) => {
+            let mut conneted_users = CONNECTED_USERS.lock().await;
+            let _ = conneted_users.remove(&state.clone().unwrap_all());
+
+            bail!("client disconnected")
+        }
+    }
+
+    let command = commands::IrcCommand::new(buffer.clone());
+    match command
+        .execute(&mut writer, &info.server_hostname, &mut state)
+        .await
+    {
+        Ok(_) => {}
+        Err(error) => {
+            let error_string = format!("error processing your command: {error:#?}\n");
+            let error = IrcResponseCodes::UnknownCommand;
+
+            error
+                .into_irc_response("*".into(), error_string.into())
+                .send(&info.server_hostname, &mut writer, true)
+                .await
+                .unwrap();
+        }
+    }
+
+    if !state.identified && state.is_populated() {
+        send_motd(info.clone(), state.clone(), &mut writer).await?;
+
+        state.identified = true;
+        CONNECTED_USERS
+            .lock()
+            .await
+            .insert(state.clone().unwrap_all());
+    }
+
+    Ok(state)
+}
+
+async fn message_listener(
+    user_wrapped: &User,
+    receiver: &mut Receiver<Message>,
+    writer: &mut TokioBufWriter<TokioTcpStream>,
+) -> Result<()> {
+    if !user_wrapped.is_populated() {
+        bail!("user has not registered yet, returning...");
+    }
+
+    let user = user_wrapped.unwrap_all();
+
+    let message: Message = receiver.recv().await.unwrap();
+    println!("{message:#?}");
+
+    if user.nickname.clone().to_ascii_lowercase() == message.receiver.to_ascii_lowercase() {
+        IrcResponse {
+            sender: Some(message.sender.hostmask()),
+            command: "PRIVMSG".into(),
+            message: message.text,
+            receiver: user.username.clone(),
+        }
+        .send("", writer, true)
+        .await
+        .unwrap();
+    }
+
     Ok(())
 }

src/messages.rs

@@ -0,0 +1,9 @@
+use crate::user::UserUnwrapped;
+
+#[allow(dead_code)]
+#[derive(Debug, Clone)]
+pub struct Message {
+    pub sender: UserUnwrapped,
+    pub receiver: String,
+    pub text: String,
+}

src/sender.rs

@@ -1,59 +1,81 @@
-use std::{
+use std::io::Write;
-    io::{BufWriter, Write},
+
+use anyhow::Result;
+use tokio::{
+    io::{AsyncWriteExt, BufWriter},
     net::TcpStream,
 };
 
-use anyhow::Result;
+#[derive(Clone)]
-
 pub struct IrcResponse {
+    pub sender: Option<String>,
     pub command: String,
     pub receiver: String,
-    pub arguments: Option<String>,
     pub message: String,
 }
 
 #[derive(Clone, Copy)]
 pub enum IrcResponseCodes {
     UnknownCommand,
+    Welcome,
+    YourHost,
+    MyInfo,
+    ISupport,
+    NoMotd,
 }
 
 impl IrcResponse {
-    pub fn send(&self, hostname: &str, writer: &mut BufWriter<&TcpStream>) -> Result<()> {
+    pub async fn send(
-        let mut response = format!(":{} {} {} ", hostname, self.command, self.receiver);
+        &self,
+        hostname: &str,
+        writer: &mut BufWriter<TcpStream>,
+        prepend_column: bool,
+    ) -> Result<()> {
+        let mut response = format!(
+            ":{} {} {} ",
+            self.sender.clone().unwrap_or(hostname.to_string()),
+            self.command,
+            self.receiver
+        );
 
-        if let Some(arguments) = &self.arguments {
+        if prepend_column {
-            response.push_str(&format!("{} ", arguments));
+            response.push_str(&format!(":{}\r\n", self.message.trim_end()));
-        };
+        } else {
+            response.push_str(&format!("{}\r\n", self.message.trim_end()));
+        }
 
-        response.push_str(&format!(":{}\n", self.message.trim_end()));
+        writer.write_all(response.as_bytes()).await?;
-
+        writer.flush().await?;
-        writer.write_all(response.as_bytes())?;
-        writer.flush()?;
 
         Ok(())
     }
 }
 
-impl From<IrcResponseCodes> for u32 {
+impl From<IrcResponseCodes> for &str {
     fn from(value: IrcResponseCodes) -> Self {
         match value {
-            IrcResponseCodes::UnknownCommand => 421,
+            IrcResponseCodes::UnknownCommand => "421",
+            IrcResponseCodes::Welcome => "001",
+            IrcResponseCodes::YourHost => "002",
+            IrcResponseCodes::MyInfo => "004",
+            IrcResponseCodes::ISupport => "005",
+            IrcResponseCodes::NoMotd => "422",
         }
     }
 }
 
 impl From<IrcResponseCodes> for String {
     fn from(value: IrcResponseCodes) -> Self {
-        Into::<u32>::into(value).to_string()
+        Into::<&str>::into(value).to_string()
     }
 }
 
 impl IrcResponseCodes {
     pub fn into_irc_response(&self, receiver: String, message: String) -> IrcResponse {
         IrcResponse {
+            sender: None,
             command: (*self).into(),
             receiver,
-            arguments: None,
             message,
         }
     }

src/user.rs

@@ -0,0 +1,72 @@
+#![allow(dead_code)]
+
+use std::borrow::Borrow;
+
+#[derive(Clone, Debug, Hash, Eq, PartialEq)]
+pub struct User {
+    pub nickname: Option<String>,
+    pub username: Option<String>,
+    pub realname: Option<String>,
+    pub identified: bool,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq, Hash)]
+pub struct UserUnwrapped {
+    pub nickname: String,
+    pub username: String,
+    pub realname: String,
+    pub identified: bool,
+}
+
+impl User {
+    pub fn is_populated(&self) -> bool {
+        self.realname.is_some() && self.username.is_some() && self.nickname.is_some()
+    }
+
+    pub fn unwrap_all(&self) -> UserUnwrapped {
+        UserUnwrapped {
+            nickname: self.nickname.clone().unwrap(),
+            username: self.username.clone().unwrap(),
+            realname: self.realname.clone().unwrap(),
+            identified: self.identified,
+        }
+    }
+
+    pub fn default() -> Self {
+        Self {
+            nickname: None,
+            username: None,
+            realname: None,
+            identified: false,
+        }
+    }
+}
+
+impl UserUnwrapped {
+    pub fn hostmask(&self) -> String {
+        format!(
+            "{}!~{}@{}",
+            self.nickname.clone(),
+            self.realname.clone(),
+            "unimplement.ed"
+        )
+    }
+}
+
+impl PartialEq<String> for UserUnwrapped {
+    fn eq(&self, other: &String) -> bool {
+        self.username == other.clone()
+    }
+}
+
+impl PartialEq<UserUnwrapped> for String {
+    fn eq(&self, other: &UserUnwrapped) -> bool {
+        self == &other.username.clone()
+    }
+}
+
+impl Borrow<String> for UserUnwrapped {
+    fn borrow(&self) -> &String {
+        &self.username
+    }
+}