Merge pull request #12 from selimaj-dev/logging-and-config

Logging and config

Cargo.lock

@@ -3887,8 +3887,11 @@ dependencies = [
  "hypersdk",
  "pulse-ui",
  "pulse-wire",
+ "rand 0.8.7",
+ "serde",
  "serde_json",
  "tokio",
+ "toml",
 ]
 
 [[package]]
@@ -4728,6 +4731,15 @@ dependencies = [
  "zmij",
 ]
 
+[[package]]
+name = "serde_spanned"
+version = "1.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6662b5879511e06e8999a8a235d848113e942c9124f211511b16466ee2995f26"
+dependencies = [
+ "serde_core",
+]
+
 [[package]]
 name = "serde_urlencoded"
 version = "0.7.1"
@@ -5239,6 +5251,21 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "toml"
+version = "1.1.3+spec-1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "53c96ecdfa941c8fc4fcaed14f99ada8ebed502eef533015095a07e3301d4c3c"
+dependencies = [
+ "indexmap 2.14.0",
+ "serde_core",
+ "serde_spanned",
+ "toml_datetime",
+ "toml_parser",
+ "toml_writer",
+ "winnow",
+]
+
 [[package]]
 name = "toml_datetime"
 version = "1.1.1+spec-1.1.0"
@@ -5269,6 +5296,12 @@ dependencies = [
  "winnow",
 ]
 
+[[package]]
+name = "toml_writer"
+version = "1.1.2+spec-1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7d56353a2a665ad0f41a421187180aab746c8c325620617ad883a99a1cbe66d2"
+
 [[package]]
 name = "tonic"
 version = "0.14.6"

Cargo.toml

@@ -8,10 +8,20 @@ pulse-ui = { workspace = true }
 pulse-wire = { workspace = true }
 
 chrono = "0.4.45"
-tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "fs", "io-util", "time"] }
+tokio = { workspace = true, features = [
+    "rt-multi-thread",
+    "macros",
+    "net",
+    "fs",
+    "io-util",
+    "time",
+] }
 crossterm = { workspace = true }
 hypersdk = "0.2.14"
 serde_json = "1"
+rand = "0.8.7"
+toml = "1.1.3"
+serde = { version = "1.0.229", features = ["serde_derive"] }
 
 [workspace]
 members = ["pulse-macros", "pulse-ui", "pulse-wire"]

src/engine/config.rs

@@ -0,0 +1,54 @@
+use std::path::PathBuf;
+
+pub fn home_dir() -> tokio::io::Result<PathBuf> {
+    std::env::home_dir().ok_or_else(|| {
+        tokio::io::Error::new(std::io::ErrorKind::NotFound, "Unable to get home directory")
+    })
+}
+
+pub fn pulse_directory() -> tokio::io::Result<PathBuf> {
+    Ok(home_dir()?.join(".config/pulse-trader"))
+}
+
+pub fn pulse_config_directory() -> tokio::io::Result<PathBuf> {
+    Ok(home_dir()?.join(".config/pulse-trader/config.toml"))
+}
+
+#[derive(Debug, Default, serde::Serialize, serde::Deserialize)]
+pub struct WatchList {
+    pub symbols: Vec<String>,
+}
+
+#[derive(Debug, Default, serde::Serialize, serde::Deserialize)]
+pub struct Config {
+    pub watchlist: WatchList,
+}
+
+impl Config {
+    pub async fn new() -> tokio::io::Result<Self> {
+        let path = pulse_config_directory()?;
+
+        if !path.exists() {
+            let default = Self::default();
+
+            tokio::fs::create_dir_all(path.parent().unwrap()).await?;
+            tokio::fs::write(&path, default.to_string()?).await?;
+
+            return Ok(default);
+        }
+
+        let output = tokio::fs::read_to_string(path).await?;
+
+        Self::from_str(&output)
+    }
+
+    pub fn from_str(s: &str) -> tokio::io::Result<Self> {
+        toml::from_str(s)
+            .map_err(|v| tokio::io::Error::new(std::io::ErrorKind::InvalidInput, v.to_string()))
+    }
+
+    pub fn to_string(&self) -> tokio::io::Result<String> {
+        toml::to_string(self)
+            .map_err(|v| tokio::io::Error::new(std::io::ErrorKind::Other, v.to_string()))
+    }
+}

src/engine/engine.rs

@@ -1,60 +1,102 @@
-use crate::terminal::TerminalServer;
+use std::sync::Arc;
 
-const WATCH_LIST_SYMBOLS: &[&str] = &["BTC", "ETH", "SOL", "XRP"];
+use tokio::{sync::Mutex, task::JoinHandle};
 
+use crate::{config::Config, terminal::TerminalServer};
+
+#[derive(Debug, Clone)]
 pub struct Engine {
-    pub terminal_server: TerminalServer,
+    pub terminal_server: Arc<TerminalServer>,
+    pub config: Arc<Mutex<Config>>,
 }
 
 impl Engine {
-    pub fn new() -> Self {
+    pub async fn new() -> tokio::io::Result<Arc<Self>> {
-        Self {
+        let config = Arc::new(Mutex::new(Config::new().await?));
-            terminal_server: TerminalServer::new(),
+
+        Ok(Arc::new_cyclic(|engine| Self {
+            terminal_server: TerminalServer::new(engine.clone()),
+            config,
+        }))
+    }
+
+    pub async fn spawn_terminal_server(&self) -> JoinHandle<tokio::io::Result<()>> {
+        let terminal_server = self.terminal_server.clone();
+
+        tokio::spawn(async move { terminal_server.run().await })
+    }
+
+    pub async fn spawn_broadcaster(&self) -> JoinHandle<tokio::io::Result<()>> {
+        let s = self.clone();
+
+        tokio::spawn(async move { s.run_broadcaster().await })
+    }
+
+    pub async fn run_broadcaster(&self) -> tokio::io::Result<()> {
+        let mut refresh = tokio::time::interval(tokio::time::Duration::from_secs(5));
+
+        loop {
+            refresh.tick().await;
+
+            let watch_list = &self.config.lock().await.watchlist.symbols;
+
+            match crate::fetch::fetch_watch_list(watch_list).await {
+                Ok(watch_list) => {
+                    if let Err(error) = self
+                        .terminal_server
+                        .broadcast(
+                            pulse_wire::terminal::TerminalServerMessage::WatchListUpdated(
+                                watch_list,
+                            ),
+                        )
+                        .await
+                    {
+                        eprintln!("Failed to broadcast Hyperliquid watch list: {error}");
+                    }
+                }
+                Err(error) => eprintln!("Failed to refresh Hyperliquid watch list: {error}"),
+            }
         }
     }
 
-    pub fn spawn_terminal_server(&self) {
+    pub async fn run_engine(&self) -> tokio::io::Result<()> {
-        let terminal_server = self.terminal_server.clone();
-
-        tokio::spawn(async move {
-            terminal_server
-                .run()
-                .await
-                .expect("Failed to run terminal server");
-        });
-    }
-
-    pub fn spawn_broadcaster(&mut self) {
-        let terminal_server = self.terminal_server.clone();
-
-        tokio::spawn(async move {
-            let mut refresh = tokio::time::interval(tokio::time::Duration::from_secs(5));
-
-            loop {
-                refresh.tick().await;
-
-                match crate::fetch::fetch_watch_list(WATCH_LIST_SYMBOLS).await {
-                    Ok(watch_list) => {
-                        if let Err(error) = terminal_server
-                            .broadcast(
-                                pulse_wire::terminal::TerminalServerMessage::WatchListUpdated(
-                                    watch_list,
-                                ),
-                            )
-                            .await
-                        {
-                            eprintln!("Failed to broadcast Hyperliquid watch list: {error}");
-                        }
-                    }
-                    Err(error) => eprintln!("Failed to refresh Hyperliquid watch list: {error}"),
-                }
-            }
-        });
-    }
-
-    pub async fn run_engine(&mut self) -> tokio::io::Result<()> {
         loop {
             tokio::time::sleep(tokio::time::Duration::from_millis(5000)).await;
         }
     }
+
+    pub async fn execute_command(&self, command: &str, args: Vec<&str>) -> tokio::io::Result<()> {
+        match command {
+            "config" => {
+                const MESSAGE: &str = "Invalid command arguments, usage: config <reload>";
+
+                if args.len() != 1 {
+                    self.terminal_server.error("config", MESSAGE).await?;
+
+                    return Ok(());
+                }
+
+                match args[0] {
+                    "reload" => {
+                        *self.config.lock().await = Config::new().await?;
+                    }
+
+                    _ => {
+                        self.terminal_server.error("config", MESSAGE).await?;
+                    }
+                }
+            }
+
+            _ => {
+                self.terminal_server
+                    .error(
+                        "Command executor",
+                        &format!("Command '{}' not found", command),
+                    )
+                    .await?;
+            }
+        }
+
+        Ok(())
+    }
 }

src/engine/fetch.rs

@@ -12,7 +12,7 @@ fn number(value: &Value, field: &str) -> Result<f64, String> {
         .map_err(|error| format!("could not parse asset context field {field} ({raw}): {error}"))
 }
 
-pub async fn fetch_watch_list(symbols: &[&str]) -> Result<Vec<WatchListItem>, String> {
+pub async fn fetch_watch_list(symbols: &[String]) -> Result<Vec<WatchListItem>, String> {
     let response = hypersdk::hypercore::mainnet()
         .meta_and_asset_ctxs(None)
         .await
@@ -73,7 +73,7 @@ pub async fn fetch_watch_list(symbols: &[&str]) -> Result<Vec<WatchListItem>, St
         .iter()
         .map(|symbol| {
             by_symbol
-                .remove(*symbol)
+                .remove(symbol.as_str())
                 .ok_or_else(|| format!("{symbol} is not in the Hyperliquid perpetual universe"))
         })
         .collect()

src/engine/main.rs

@@ -1,16 +1,22 @@
+pub mod config;
 pub mod engine;
 pub mod fetch;
 pub mod terminal;
 
 #[tokio::main]
 async fn main() -> tokio::io::Result<()> {
-    let mut engine = engine::Engine::new();
+    let engine = engine::Engine::new().await?;
 
-    engine.spawn_terminal_server();
+    let terminal_server = engine.spawn_terminal_server().await;
 
-    engine.spawn_broadcaster();
+    let broadcaster = engine.spawn_broadcaster().await;
 
     engine.run_engine().await?;
 
+    let (terminal_server, broadcaster) = tokio::join!(terminal_server, broadcaster);
+
+    terminal_server??;
+    broadcaster??;
+
     Ok(())
 }

src/engine/terminal.rs

@@ -1,6 +1,12 @@
-use std::sync::Arc;
+use std::{
+    collections::HashMap,
+    sync::{Arc, Weak},
+};
 
-use pulse_wire::{PulseWire, terminal::TerminalClientMessage};
+use pulse_wire::{
+    PulseWire,
+    terminal::{EventLog, LogKind, TerminalClientMessage},
+};
 use tokio::{
     io::{AsyncReadExt, AsyncWriteExt},
     net::{
@@ -10,19 +16,25 @@ use tokio::{
     sync::Mutex,
 };
 
-#[derive(Debug, Clone)]
+use crate::engine::Engine;
+
+#[derive(Debug)]
 pub struct TerminalServer {
-    clients: Arc<Mutex<Vec<OwnedWriteHalf>>>,
+    clients: Mutex<HashMap<usize, OwnedWriteHalf>>,
+    logs: Mutex<Vec<EventLog>>,
+    engine: Weak<Engine>,
 }
 
 impl TerminalServer {
-    pub fn new() -> Self {
+    pub fn new(engine: Weak<Engine>) -> Arc<Self> {
-        Self {
+        Arc::new(Self {
-            clients: Arc::new(Mutex::new(Vec::new())),
+            clients: Mutex::new(HashMap::new()),
-        }
+            logs: Mutex::new(Vec::new()),
+            engine,
+        })
     }
 
-    pub async fn run(&self) -> tokio::io::Result<()> {
+    pub async fn run(self: &Arc<Self>) -> tokio::io::Result<()> {
         let path = pulse_wire::server_path();
 
         if path.exists() {
@@ -38,19 +50,31 @@ impl TerminalServer {
 
             let (reader, writer) = stream.into_split();
 
-            self.clients.lock().await.push(writer);
+            let id = rand::random();
+
+            self.clients.lock().await.insert(id, writer);
 
             let s = self.clone();
 
             tokio::spawn(async move {
-                if let Err(err) = s.handle_client(reader).await {
+                if let Err(err) = s.handle_client(&id, reader).await {
                     eprintln!("Terminal connection error: {err}");
                 }
             });
         }
     }
 
-    async fn handle_client(&self, mut reader: OwnedReadHalf) -> tokio::io::Result<()> {
+    async fn handle_client(
+        self: &Arc<Self>,
+        id: &usize,
+        mut reader: OwnedReadHalf,
+    ) -> tokio::io::Result<()> {
+        self.send_to(
+            id,
+            pulse_wire::terminal::TerminalServerMessage::SetLogs(self.logs.lock().await.clone()),
+        )
+        .await?;
+
         loop {
             let mut len_buf = [0u8; size_of::<usize>()];
             let size = reader.read_exact(&mut len_buf).await?;
@@ -69,24 +93,13 @@ impl TerminalServer {
                 TerminalClientMessage::ExecuteCommand(command) => {
                     let command = command.as_str();
 
-                    let (command, _args) = if let Some((command, args)) = command.split_once(" ") {
+                    let (command, args) = if let Some((command, args)) = command.split_once(" ") {
                         (command, args.split(" ").collect())
                     } else {
                         (command, Vec::new())
                     };
 
-                    match command {
+                    self.get_engine().execute_command(command, args).await?;
-                        _ => {
-                            self.broadcast(pulse_wire::terminal::TerminalServerMessage::AddLog(
-                                pulse_wire::terminal::EventLog {
-                                    kind: pulse_wire::terminal::LogKind::Err,
-                                    name: "Command executor".to_string(),
-                                    message: format!("Command '{}' not found", command),
-                                },
-                            ))
-                            .await?;
-                        }
-                    }
                 }
             }
         }
@@ -95,32 +108,91 @@ impl TerminalServer {
     }
 
     pub async fn broadcast(
-        &self,
+        self: &Arc<Self>,
         message: pulse_wire::terminal::TerminalServerMessage,
     ) -> tokio::io::Result<()> {
         let msg = message.to_com();
 
         let mut clients = self.clients.lock().await;
 
-        for i in (0..clients.len()).rev() {
+        let mut remove_clients = Vec::new();
-            if let Err(e) = Self::send_to_client(&mut clients, i, &msg).await {
+
-                clients.remove(i);
+        for (id, client) in clients.iter_mut() {
+            if let Err(e) = Self::send_to_client(client, &msg).await {
+                remove_clients.push(*id);
                 println!("{e:?}");
             }
         }
 
+        for id in remove_clients {
+            clients.remove(&id);
+        }
+
         Ok(())
     }
 
-    pub async fn send_to_client(
+    pub async fn send_to(
-        clients: &mut tokio::sync::MutexGuard<'_, Vec<OwnedWriteHalf>>,
+        self: &Arc<Self>,
-        i: usize,
+        id: &usize,
-        msg: &[u8],
+        message: pulse_wire::terminal::TerminalServerMessage,
     ) -> tokio::io::Result<()> {
-        clients[i].write(&msg.len().to_le_bytes()).await?;
+        Self::send_to_client(
-        clients[i].write(msg).await?;
+            self.clients.lock().await.get_mut(id).ok_or_else(|| {
-        clients[i].flush().await?;
+                tokio::io::Error::new(
+                    std::io::ErrorKind::Other,
+                    format!("Client({id}) does not exist"),
+                )
+            })?,
+            &message.to_com(),
+        )
+        .await
+    }
+
+    pub async fn send_to_client(client: &mut OwnedWriteHalf, msg: &[u8]) -> tokio::io::Result<()> {
+        client.write(&msg.len().to_le_bytes()).await?;
+        client.write(msg).await?;
+        client.flush().await?;
 
         Ok(())
     }
+
+    pub async fn log(
+        self: &Arc<Self>,
+        kind: LogKind,
+        name: &str,
+        message: &str,
+    ) -> tokio::io::Result<()> {
+        let log = EventLog {
+            kind,
+            name: name.to_string(),
+            message: message.to_string(),
+        };
+
+        self.logs.lock().await.push(log.clone());
+
+        self.broadcast(pulse_wire::terminal::TerminalServerMessage::AddLog(log))
+            .await
+    }
+
+    pub async fn info(self: &Arc<Self>, name: &str, message: &str) -> tokio::io::Result<()> {
+        self.log(LogKind::Info, name, message).await
+    }
+
+    pub async fn warn(self: &Arc<Self>, name: &str, message: &str) -> tokio::io::Result<()> {
+        self.log(LogKind::Warn, name, message).await
+    }
+
+    pub async fn error(self: &Arc<Self>, name: &str, message: &str) -> tokio::io::Result<()> {
+        self.log(LogKind::Err, name, message).await
+    }
+
+    pub async fn debug(self: &Arc<Self>, name: &str, message: &str) -> tokio::io::Result<()> {
+        self.log(LogKind::Debug, name, message).await
+    }
+
+    pub fn get_engine(&self) -> Arc<Engine> {
+        self.engine
+            .upgrade()
+            .expect("Failed to upgrade engine(Weak) to Arc")
+    }
 }