diff --git a/Cargo.toml b/Cargo.toml index ab224ae..9f5ad03 100644 --- a/Cargo.toml +++ b/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" diff --git a/src/api/auth.rs b/src/api/auth.rs index 7da093d..85cda65 100644 --- a/src/api/auth.rs +++ b/src/api/auth.rs @@ -24,8 +24,8 @@ impl Auth { .map_err(|e| e.into()) } - pub async fn validate_token(&self, token: &str) -> Result { - let client = ApiClient::new(token.to_string()); + pub async fn validate_token(&self, token: &str, base_url: Option) -> Result { + let client = ApiClient::new(token.to_string(), base_url); client .get::(Endpoint::CurrentUser) diff --git a/src/api/client.rs b/src/api/client.rs index 761787f..4754340 100644 --- a/src/api/client.rs +++ b/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) -> 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(&self, endpoint: Endpoint) -> Result { - 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::(Endpoint::Config).await; assert!(result.is_ok(), "Failed to get config: {:?}", result.err()); diff --git a/src/api/events.rs b/src/api/events.rs new file mode 100644 index 0000000..d468bef --- /dev/null +++ b/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, + }, +} + +#[derive(Debug, Clone, Deserialize)] +#[serde(tag = "type")] +#[allow(unused)] +pub enum ServerEvent { + Error { + error: String, + }, + Authenticated, + Logout, + Bulk { + v: Vec, + }, + Pong { + data: Value, + }, + Ready { + users: Option>, + servers: Option>, + channels: Option>, + members: Option>, + emojis: Option>, + user_settings: Option>, + channel_unreads: Option>, + policy_changes: Option>, + }, + 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>, + }, + 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>, + }, + ServerDelete { + id: String, + }, + ServerMemberUpdate { + id: ServerMemberId, + data: Value, + clear: Option>, + }, + ServerMemberJoin { + id: String, + user: String, + member: Value, + }, + ServerMemberLeave { + id: String, + user: String, + }, + ServerRoleUpdate { + id: String, + role_id: String, + data: Value, + clear: Option>, + }, + ServerRoleDelete { + id: String, + role_id: String, + }, + UserUpdate { + id: String, + data: Value, + clear: Option>, + }, + 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), +} diff --git a/src/api/mod.rs b/src/api/mod.rs index 9ee5abf..1c9346f 100644 --- a/src/api/mod.rs +++ b/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"; diff --git a/src/api/ws.rs b/src/api/ws.rs new file mode 100644 index 0000000..a6064ad --- /dev/null +++ b/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, +} + +impl WsClient { + pub async fn connect(base_url: Option) -> Result<(Self, mpsc::Receiver)> { + 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::(OUTGOING_BUFFER_SIZE); + let (tx_incoming, rx_incoming) = mpsc::channel::(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::(&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 { + self.tx_outgoing.clone() + } + + pub async fn dispatch_event(event: ServerEvent, tx: &mpsc::Sender) { + 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(); + } + } +} diff --git a/src/app.rs b/src/app.rs index 59857dd..8d6e28a 100644 --- a/src/app.rs +++ b/src/app.rs @@ -1,10 +1,23 @@ +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use log::{debug, error, info, warn}; use ratatui::crossterm::event::{KeyCode, KeyEvent}; +use tokio::sync::mpsc::Receiver; +use tokio::time; -use crate::action::Action; -use crate::api::auth::Auth; -use crate::api::client::ApiClient; -use crate::input::InputState; -use crate::{Result, cache::CacheStore}; +use crate::{ + Result, + action::Action, + api::{ + API_BASE_URL, + auth::Auth, + client::ApiClient, + events::{ClientEvent, ServerEvent}, + ws::WsClient, + }, + cache::CacheStore, + input::InputState, +}; pub enum AppState { InputToken, @@ -19,21 +32,24 @@ pub struct App { pub auth: Auth, pub should_quit: bool, pub input_state: InputState, - pub client: ApiClient, + pub api_base_url: String, + pub api_client: ApiClient, + pub ws_client: WsClient, + pub ws_rx: Receiver, #[allow(unused)] pub cache: CacheStore, } impl App { - pub async fn new() -> Result { + pub async fn new(api_base_url: Option, ws_base_url: Option) -> Result { let auth = Auth::new().map_err(|e| anyhow::anyhow!(e))?; - let mut client = ApiClient::new(String::new()); + let mut api_client = ApiClient::new(String::new(), api_base_url.clone()); let state = if let Ok(token) = auth.token_entry.get_secret().await { - match auth.validate_token(&token).await { + match auth.validate_token(&token, api_base_url.clone()).await { Ok(authenticated_client) => { - client = authenticated_client; + api_client = authenticated_client; AppState::LoggedIn } Err(e) => AppState::Error(e), @@ -42,6 +58,8 @@ impl App { AppState::InputToken }; + let (ws_client, ws_rx) = WsClient::connect(ws_base_url).await?; + let cache = CacheStore::new()?; Ok(Self { @@ -49,7 +67,10 @@ impl App { input_text: String::new(), auth, should_quit: false, - client, + api_base_url: api_base_url.unwrap_or(API_BASE_URL.to_string()), + api_client, + ws_client, + ws_rx, cache, input_state: InputState::default(), }) @@ -61,10 +82,14 @@ impl App { KeyCode::Enter => { if !self.input_text.is_empty() { self.state = AppState::ValidatingToken; - match self.auth.validate_token(&self.input_text).await { + match self + .auth + .validate_token(&self.input_text, Some(self.api_base_url.clone())) + .await + { Ok(client) => match self.auth.store_token(&self.input_text).await { Ok(_) => { - self.client = client; + self.api_client = client; self.state = AppState::LoggedIn; } Err(detailed_err) => { @@ -103,4 +128,58 @@ impl App { } Ok(()) } + + pub async fn authenticate_ws(&mut self, token: &str) -> Result<()> { + self.ws_client + .send_event(ClientEvent::Authenticate { + token: token.into(), + }) + .await?; + + let mut is_authenticated = false; + while let Some(event) = self.ws_rx.recv().await { + match event { + ServerEvent::Authenticated => { + info!("Successfully authenticated!"); + is_authenticated = true; + break; + } + ServerEvent::Error { error } => { + error!("Error authenticating: {error}"); + return Ok(()); + } + _ => {} + } + } + + if is_authenticated { + let tx_ping = self.ws_client.clone_sender(); + + tokio::spawn(async move { + let mut interval = time::interval(Duration::from_secs(20)); + + loop { + interval.tick().await; + + let timestamp = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_millis() as u64; + + if tx_ping + .send(ClientEvent::Ping { data: timestamp }) + .await + .is_err() + { + warn!("Stopped pinging: channel closed."); + break; + } + } + }); + + debug!("Started pinging every 20s."); + } + + Ok(()) + } } diff --git a/src/main.rs b/src/main.rs index af3f991..108cc07 100644 --- a/src/main.rs +++ b/src/main.rs @@ -9,6 +9,7 @@ mod ui; use std::{fs, path::PathBuf}; use app::App; +use log::debug; use ratatui::crossterm::event::{self, Event}; pub const LOG_FILE: &str = "logs"; @@ -46,7 +47,12 @@ async fn main() -> anyhow::Result<(), Box> { let mut terminal = ratatui::init(); - let mut app = App::new().await?; + let api_base_url = std::env::var("API_BASE_URL").ok(); + let ws_base_url = std::env::var("WS_BASE_URL").ok(); + + let mut app = App::new(api_base_url.clone(), ws_base_url.clone()).await?; + + app.authenticate_ws(&app.api_client.clone_token()).await?; loop { terminal.draw(|f| ui::render(f, &app))?; @@ -62,6 +68,10 @@ async fn main() -> anyhow::Result<(), Box> { break; } } + + if let Some(event) = app.ws_rx.recv().await { + debug!("Received WebSocket event: {event:?}"); + } } ratatui::restore();