Implemented WebSocket connection

Cargo.toml

@@ -18,9 +18,16 @@ pickledb = { version = "0.5.1", features = ["bincode"] }
 # Directories
 dirs = "6.0.0"
 
+# WebSockets
+tokio-tungstenite = { version = "0.30.0", features = ["native-tls"] }
+futures-util = "0.3.31"
+
 keyring-lib = { version = "1.0.3", features = ["tokio"] }
 ratatui = "0.30.2"
-reqwest = { version = "0.13.4", default-features = false, features = ["json", "rustls"] }
+reqwest = { version = "0.13.4", default-features = false, features = [
+  "json",
+  "rustls",
+] }
 serde = { version = "1", features = ["derive"] }
 tokio = { version = "1.52.3", features = ["full"] }
 serde_json = "1.0.150"

src/api/auth.rs

@@ -24,8 +24,8 @@ impl Auth {
             .map_err(|e| e.into())
     }
 
-    pub async fn validate_token(&self, token: &str) -> Result<ApiClient> {
-        let client = ApiClient::new(token.to_string());
+    pub async fn validate_token(&self, token: &str, base_url: Option<String>) -> Result<ApiClient> {
+        let client = ApiClient::new(token.to_string(), base_url);
 
         client
             .get::<Value>(Endpoint::CurrentUser)

src/api/client.rs

@@ -2,7 +2,7 @@ use anyhow::{Result, anyhow};
 use reqwest::Client;
 use serde::de::DeserializeOwned;
 
-const BASE_URL: &str = "https://api.stoat.chat";
+use crate::api::API_BASE_URL;
 
 #[derive(Debug)]
 #[allow(unused)]
@@ -35,20 +35,22 @@ impl Endpoint {
 pub struct ApiClient {
     client: Client,
     token: String,
+    base_url: String,
 }
 
 impl ApiClient {
-    pub fn new(token: String) -> Self {
+    pub fn new(token: String, base_url: Option<String>) -> Self {
         Self {
             client: Client::new(),
             token,
+            base_url: base_url.unwrap_or(API_BASE_URL.to_string()),
         }
     }
 
     /// Makes a GET request to the specified endpoint and deserializes the JSON response into `T`.
     /// The `endpoint` should start with a slash, e.g., `/users/@me`.
     pub async fn get<T: DeserializeOwned>(&self, endpoint: Endpoint) -> Result<T> {
-        let url = format!("{}{}", BASE_URL, endpoint.path());
+        let url = format!("{}{}", self.base_url, endpoint.path());
 
         let response = self
             .client
@@ -71,6 +73,10 @@ impl ApiClient {
             ))
         }
     }
+
+    pub fn clone_token(&self) -> String {
+        self.token.clone()
+    }
 }
 
 #[cfg(test)]
@@ -81,7 +87,7 @@ mod tests {
     #[tokio::test]
     async fn test_get_config() {
         // Token doesn't matter for the root config endpoint, but we provide a dummy one
-        let client = ApiClient::new("dummy_token".to_string());
+        let client = ApiClient::new("dummy_token".to_string(), None);
 
         let result = client.get::<Value>(Endpoint::Config).await;
         assert!(result.is_ok(), "Failed to get config: {:?}", result.err());

src/api/events.rs

@@ -0,0 +1,179 @@
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+
+#[derive(Debug, Clone, Serialize)]
+#[serde(tag = "type")]
+#[allow(unused)]
+pub enum ClientEvent {
+    Authenticate { token: String },
+    BeginTyping { channel: String },
+    EndTyping { channel: String },
+    Ping { data: u64 },
+    Subscribe { server_id: String },
+}
+
+#[derive(Debug, Clone, Deserialize)]
+#[allow(unused)]
+pub struct ServerMemberId {
+    pub server: String,
+    pub user: String,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+#[serde(tag = "event_type")]
+#[allow(unused)]
+pub enum AuthEvent {
+    DeleteSession {
+        user_id: String,
+        session_id: String,
+    },
+    DeleteAllSessions {
+        user_id: String,
+        exclude_session_id: Option<String>,
+    },
+}
+
+#[derive(Debug, Clone, Deserialize)]
+#[serde(tag = "type")]
+#[allow(unused)]
+pub enum ServerEvent {
+    Error {
+        error: String,
+    },
+    Authenticated,
+    Logout,
+    Bulk {
+        v: Vec<ServerEvent>,
+    },
+    Pong {
+        data: Value,
+    },
+    Ready {
+        users: Option<Vec<Value>>,
+        servers: Option<Vec<Value>>,
+        channels: Option<Vec<Value>>,
+        members: Option<Vec<Value>>,
+        emojis: Option<Vec<Value>>,
+        user_settings: Option<Vec<Value>>,
+        channel_unreads: Option<Vec<Value>>,
+        policy_changes: Option<Vec<Value>>,
+    },
+    Message(Value),
+    MessageUpdate {
+        id: String,
+        channel: String,
+        data: Value,
+    },
+    MessageAppend {
+        id: String,
+        channel: String,
+        append: Value,
+    },
+    MessageDelete {
+        id: String,
+        channel: String,
+    },
+    MessageReact {
+        id: String,
+        channel_id: String,
+        user_id: String,
+        emoji_id: String,
+    },
+    MessageUnreact {
+        id: String,
+        channel_id: String,
+        user_id: String,
+        emoji_id: String,
+    },
+    MessageRemoveReaction {
+        id: String,
+        channel_id: String,
+        emoji_id: String,
+    },
+    ChannelCreate(Value),
+    ChannelUpdate {
+        id: String,
+        data: Value,
+        clear: Option<Vec<String>>,
+    },
+    ChannelDelete {
+        id: String,
+    },
+    ChannelGroupJoin {
+        id: String,
+        user: String,
+    },
+    ChannelGroupLeave {
+        id: String,
+        user: String,
+    },
+    ChannelStartTyping {
+        id: String,
+        user: String,
+    },
+    ChannelStopTyping {
+        id: String,
+        user: String,
+    },
+    ChannelAck {
+        id: String,
+        user: String,
+        message_id: String,
+    },
+    ServerCreate(Value),
+    ServerUpdate {
+        id: String,
+        data: Value,
+        clear: Option<Vec<String>>,
+    },
+    ServerDelete {
+        id: String,
+    },
+    ServerMemberUpdate {
+        id: ServerMemberId,
+        data: Value,
+        clear: Option<Vec<String>>,
+    },
+    ServerMemberJoin {
+        id: String,
+        user: String,
+        member: Value,
+    },
+    ServerMemberLeave {
+        id: String,
+        user: String,
+    },
+    ServerRoleUpdate {
+        id: String,
+        role_id: String,
+        data: Value,
+        clear: Option<Vec<String>>,
+    },
+    ServerRoleDelete {
+        id: String,
+        role_id: String,
+    },
+    UserUpdate {
+        id: String,
+        data: Value,
+        clear: Option<Vec<String>>,
+    },
+    UserRelationship {
+        id: String,
+        user: Value,
+        status: String,
+    },
+    UserPlatformWipe {
+        user_id: String,
+        flags: Value,
+    },
+    EmojiCreate(Value),
+    EmojiUpdate {
+        id: String,
+        data: Value,
+    },
+    EmojiDelete {
+        id: String,
+    },
+    Auth(AuthEvent),
+}

src/api/mod.rs

@@ -1,2 +1,7 @@
 pub mod auth;
 pub mod client;
+pub mod events;
+pub mod ws;
+
+pub const API_BASE_URL: &str = "https://api.stoat.chat";
+pub const WS_BASE_URL: &str = "wss://events.stoat.chat";

src/api/ws.rs

@@ -0,0 +1,85 @@
+use crate::{
+    Result,
+    api::{
+        WS_BASE_URL,
+        events::{ClientEvent, ServerEvent},
+    },
+};
+use futures_util::{SinkExt, StreamExt};
+use log::{error, info};
+use tokio::sync::mpsc;
+use tokio_tungstenite::{connect_async, tungstenite::protocol::Message as WsMessage};
+
+const OUTGOING_BUFFER_SIZE: usize = 32;
+const INCOMING_BUFFER_SIZE: usize = 100;
+
+pub struct WsClient {
+    tx_outgoing: mpsc::Sender<ClientEvent>,
+}
+
+impl WsClient {
+    pub async fn connect(base_url: Option<String>) -> Result<(Self, mpsc::Receiver<ServerEvent>)> {
+        let (ws_stream, _) =
+            connect_async(base_url.unwrap_or(WS_BASE_URL.to_string()).as_str()).await?;
+        let (mut write, mut read) = ws_stream.split();
+
+        let (tx_outgoing, mut rx_outgoing) = mpsc::channel::<ClientEvent>(OUTGOING_BUFFER_SIZE);
+        let (tx_incoming, rx_incoming) = mpsc::channel::<ServerEvent>(INCOMING_BUFFER_SIZE);
+
+        tokio::spawn(async move {
+            while let Some(msg) = read.next().await {
+                match msg {
+                    Ok(WsMessage::Text(text)) => match serde_json::from_str::<ServerEvent>(&text) {
+                        Ok(event) => {
+                            Self::dispatch_event(event, &tx_incoming).await;
+                        }
+                        Err(e) => {
+                            error!("Error deserializing ServerEvent: {e}\nBrut data: {text}");
+                            break;
+                        }
+                    },
+                    Ok(WsMessage::Close(_)) => {
+                        info!("WS Connexion closed by server.");
+                        break;
+                    }
+                    Err(e) => {
+                        error!("WS Error: {e}");
+                        break;
+                    }
+                    _ => {}
+                }
+            }
+        });
+
+        tokio::spawn(async move {
+            while let Some(event) = rx_outgoing.recv().await {
+                if let Ok(json) = serde_json::to_string(&event)
+                    && let Err(e) = write.send(WsMessage::Text(json.into())).await
+                {
+                    error!("Error sending WsMessage: {e}");
+                    break;
+                }
+            }
+        });
+
+        Ok((Self { tx_outgoing }, rx_incoming))
+    }
+
+    pub async fn send_event(&self, event: ClientEvent) -> Result<()> {
+        self.tx_outgoing.send(event).await.map_err(|e| e.into())
+    }
+
+    pub fn clone_sender(&self) -> mpsc::Sender<ClientEvent> {
+        self.tx_outgoing.clone()
+    }
+
+    pub async fn dispatch_event(event: ServerEvent, tx: &mpsc::Sender<ServerEvent>) {
+        if let ServerEvent::Bulk { v } = event {
+            for sub_event in v {
+                Box::pin(Self::dispatch_event(sub_event, tx)).await;
+            }
+        } else {
+            tx.send(event).await.ok();
+        }
+    }
+}