diff --git a/Cargo.lock b/Cargo.lock index 04277be..5329cde 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2085,6 +2085,14 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "iota" +version = "0.1.0" +dependencies = [ + "iota-cli", + "tokio", +] + [[package]] name = "iota-auth" version = "0.1.0" @@ -2151,6 +2159,7 @@ dependencies = [ "hkdf 0.12.4", "hyper", "hyper-util", + "iota-ipc", "iota-logger", "iota-state", "iota-storage", @@ -2178,7 +2187,6 @@ dependencies = [ "tokio", "tokio-tungstenite", "tungstenite", - "uuid", "walkdir", "warp", "x448", @@ -2195,27 +2203,41 @@ dependencies = [ ] [[package]] -name = "iota-core" +name = "iota-daemon" version = "0.1.0" dependencies = [ - "dashmap", - "iota-cli", + "iota-daemon-lib", + "iota-ipc", "iota-logger", "iota-state", "iota-storage", - "iota-terms", - "iota-updater", - "iota-util", - "json", - "mtp", "omikron-connector", - "once_cell", - "pnet", - "ratatui", - "reqwest", "tokio", "web-server", - "web-ui", +] + +[[package]] +name = "iota-daemon-lib" +version = "0.1.0" +dependencies = [ + "iota-ipc", + "iota-logger", + "iota-state", + "iota-storage", + "iota-util", + "mtp", + "omikron-connector", + "sysinfo", + "tokio", +] + +[[package]] +name = "iota-ipc" +version = "0.1.0" +dependencies = [ + "serde", + "serde_json", + "tokio", ] [[package]] @@ -2228,6 +2250,7 @@ dependencies = [ "mtp", "once_cell", "ratatui", + "tokio", ] [[package]] @@ -2238,6 +2261,7 @@ dependencies = [ "json", "mtp", "once_cell", + "serde", "sysinfo", "tokio", ] @@ -4805,9 +4829,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.53.0" +version = "1.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d988bcd52dbe076d3d46903332f58c912b87a2c49b1428419a5845154762ffee" +checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" dependencies = [ "bytes", "libc", diff --git a/Cargo.toml b/Cargo.toml index ff17473..ecd2457 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,7 +9,10 @@ members = [ "iota-terms", "iota-state", "iota-cli", - "iota-core", + "iota", + "iota-daemon", + "iota-daemon-lib", + "iota-ipc", "omikron-connector", "web-server", "web-ui", diff --git a/flake.lock b/flake.lock index 0d30539..3042264 100644 --- a/flake.lock +++ b/flake.lock @@ -5,11 +5,11 @@ "nixpkgs-lib": "nixpkgs-lib" }, "locked": { - "lastModified": 1778716662, - "narHash": "sha256-m1Yf0wZ8j1OHjTc2UwHwyQRSnNeSgLJOd7q5Y45hzi4=", + "lastModified": 1782949081, + "narHash": "sha256-vp6Y/Grm98ESt6ceOkWiHWyZRDV3J1RID4w+6NWK9yA=", "owner": "hercules-ci", "repo": "flake-parts", - "rev": "f7c1a2d347e4c52d5fb8d10cb4d94b5884e546fb", + "rev": "17c9d6cdfc60c64f4ee8d306f9bc0b4ccb51481e", "type": "github" }, "original": { @@ -20,11 +20,11 @@ }, "nixpkgs": { "locked": { - "lastModified": 1780243769, - "narHash": "sha256-x5UQuRsH3MqI0U9afaXSNqzTPSeZlRLvFAav2Ux1pNw=", + "lastModified": 1784497964, + "narHash": "sha256-vlHUuqAcbcH2RKmHbPiuQzbv1pnzzavXnI62RD0bqCU=", "owner": "nixos", "repo": "nixpkgs", - "rev": "331800de5053fcebacf6813adb5db9c9dca22a0c", + "rev": "241313f4e8e508cb9b13278c2b0fa25b9ca27163", "type": "github" }, "original": { @@ -36,11 +36,11 @@ }, "nixpkgs-lib": { "locked": { - "lastModified": 1777168982, - "narHash": "sha256-GOkGPcboWE9BmGCRMLX3worL4EMnsnG8MyKmXNeYuhQ=", + "lastModified": 1782614948, + "narHash": "sha256-ePjCwr1sNm9NYUqywL7QfK3JnlS015msC+eBu2zKlp8=", "owner": "nix-community", "repo": "nixpkgs.lib", - "rev": "f5901329dade4a6ea039af1433fb087bd9c1fe14", + "rev": "db3f255737b94216eb71cce308e2912cf6bc2d7c", "type": "github" }, "original": { @@ -53,8 +53,7 @@ "inputs": { "flake-parts": "flake-parts", "nixpkgs": "nixpkgs", - "rust-overlay": "rust-overlay", - "ttp": "ttp" + "rust-overlay": "rust-overlay" } }, "rust-overlay": { @@ -64,11 +63,11 @@ ] }, "locked": { - "lastModified": 1780543271, - "narHash": "sha256-oPJ7eJN1sM37v92Rp/eyQL7/rUm0BOvXEBAoq/zN0cM=", + "lastModified": 1784526465, + "narHash": "sha256-L37teKC6oINWG4PGZLIqbphMWvSQ0PEz+aWxAk+rIDw=", "owner": "oxalica", "repo": "rust-overlay", - "rev": "c30ca201c5093540cf792f6982f81ba1aa0f3514", + "rev": "58c6334db52d51fc5dd8877c90b01f00cf8a696b", "type": "github" }, "original": { @@ -76,22 +75,6 @@ "repo": "rust-overlay", "type": "github" } - }, - "ttp": { - "flake": false, - "locked": { - "lastModified": 1780494955, - "narHash": "sha256-i2VRRF6yNips3c4JHgfvmvMxb0HTkTCn69lmsKZLHRw=", - "ref": "refs/heads/main", - "rev": "23438fa8f884e6ad0d32ca1004c0dedcce0cc8d2", - "revCount": 125, - "type": "git", - "url": "https://git.methanium.net/tensamin/ttp.git" - }, - "original": { - "type": "git", - "url": "https://git.methanium.net/tensamin/ttp.git" - } } }, "root": "root", diff --git a/flake.nix b/flake.nix index 8cf53d7..607fb77 100644 --- a/flake.nix +++ b/flake.nix @@ -8,10 +8,6 @@ url = "github:oxalica/rust-overlay"; inputs.nixpkgs.follows = "nixpkgs"; }; - ttp = { - url = "git+https://git.methanium.net/tensamin/ttp.git"; - flake = false; - }; }; outputs = inputs @ { @@ -19,7 +15,6 @@ nixpkgs, flake-parts, rust-overlay, - ttp, ... }: flake-parts.lib.mkFlake {inherit inputs;} { @@ -47,9 +42,10 @@ packages = { default = self'.packages.iota; iota = pkgs.rustPlatform.buildRustPackage { - pname = "iota"; + pname = "iota-daemon"; version = "0.1.0"; src = ./.; + cargoBuildFlags = ["-p" "iota-daemon"]; cargoLock = { lockFile = ./Cargo.lock; allowBuiltinFetchGit = true; @@ -57,15 +53,9 @@ nativeBuildInputs = with pkgs; [cmake perl pkg-config]; buildInputs = with pkgs; [openssl sqlite]; dontUseCmakeConfigure = true; - preConfigure = '' - if [ -d ../cargo-vendor-dir/ttp-core-0.1.0 ]; then - cp ${ttp}/ttp-codec.json ../cargo-vendor-dir/ttp-codec.json - fi - ''; postInstall = '' - mv $out/bin/iota-core $out/bin/iota for f in $out/bin/*; do - if [ "$(basename "$f")" != "iota" ]; then + if [ "$(basename "$f")" != "iota-daemon" ]; then rm "$f" fi done @@ -95,8 +85,7 @@ then cfg.settingsFile else pkgs.writeText "iota-config.json" (builtins.toJSON cfg.settings); - descriptionText = "Tensamin Iota"; - #+ lib.optionalString cfg.useTmux " (attach TUI: tmux -S ${cfg.dataDir}/tmux.sock attach -t iota)"; + descriptionText = "Tensamin Iota daemon"; in { options.services.iota = { enable = lib.mkEnableOption "Enable the Iota service."; @@ -132,12 +121,6 @@ description = "Whether to open the firewall for ports used by Iota."; }; - ttpBind = lib.mkOption { - type = lib.types.str; - default = "0.0.0.0"; - description = "IP address to bind the TTP/QUIC server to."; - }; - bindAddress = lib.mkOption { type = lib.types.str; default = "0.0.0.0"; @@ -150,12 +133,6 @@ description = "The Iota package to use."; }; - useTmux = lib.mkOption { - type = lib.types.bool; - default = true; - description = "Whether to run Iota inside a tmux session for shared TUI access."; - }; - settings = lib.mkOption { type = lib.types.attrs; default = {}; @@ -181,19 +158,7 @@ users.groups.iota = {}; - systemd.services.iota = let - iotaTmuxCmd = pkgs.writeShellScript "iota-tmux-cmd" '' - mkdir -p ${cfg.dataDir} - echo "[$(date)] Running Iota..." - ${cfg.package}/bin/iota - status=$? - echo "" - echo "[$(date)] Iota exited with status: $status" - echo "Press any key to exit..." - read -r -n 1 - exit $status - ''; - in { + systemd.services.iota = { description = descriptionText; wantedBy = ["multi-user.target"]; after = ["network.target"]; @@ -205,29 +170,7 @@ Group = "iota"; WorkingDirectory = cfg.dataDir; - ExecStart = - if cfg.useTmux - then - pkgs.writeShellScript "iota-start" '' - set -e - export TMUX_TMPDIR=${cfg.dataDir} - ${pkgs.coreutils}/bin/mkdir -p ${cfg.dataDir} - ${pkgs.coreutils}/bin/chown iota:iota ${cfg.dataDir} - - echo "[iota-start] Creating tmux session..." - if ! ${pkgs.tmux}/bin/tmux -S ${cfg.dataDir}/tmux.sock new-session -d -s iota "${iotaTmuxCmd}"; then - echo "[iota-start] ERROR: tmux new-session failed" - exit 1 - fi - echo "[iota-start] tmux session created, waiting..." - echo "[iota-start] Run 'tmux -S ${cfg.dataDir}/tmux.sock attach -t iota' to attach to the tmux session." - - while ${pkgs.tmux}/bin/tmux -S ${cfg.dataDir}/tmux.sock has-session -t iota 2>/dev/null; do - sleep 2 - done - echo "[iota-start] tmux session ended" - '' - else "${cfg.package}/bin/iota"; + ExecStart = "${cfg.package}/bin/iota-daemon"; ExecStartPre = [ ("+" @@ -245,6 +188,8 @@ Restart = "always"; RestartSec = "5s"; + RuntimeDirectory = "iota"; + RuntimeDirectoryMode = "0750"; AmbientCapabilities = ["CAP_NET_BIND_SERVICE"]; CapabilityBoundingSet = ["CAP_NET_BIND_SERVICE"]; @@ -262,8 +207,8 @@ LockPersonality = true; MemoryDenyWriteExecute = true; Environment = [ - "TTP_BIND=${cfg.ttpBind}" "BIND_ADDRESS=${cfg.bindAddress}" + "IOTA_SOCKET=/run/iota/iota.sock" ]; } // lib.optionalAttrs (cfg.environmentFiles != []) { diff --git a/iota-cli/Cargo.toml b/iota-cli/Cargo.toml index 9fa889f..2c5f868 100644 --- a/iota-cli/Cargo.toml +++ b/iota-cli/Cargo.toml @@ -3,16 +3,26 @@ name = "iota-cli" version = "0.1.0" edition = "2024" +[features] +legacy-commands = [ + "dep:iota-logger", + "dep:iota-storage", + "dep:iota-util", + "dep:mtp", + "dep:omikron-connector", +] + [dependencies] -iota-logger = { path = "../iota-logger" } +iota-logger = { path = "../iota-logger", optional = true } iota-state = { path = "../iota-state" } -iota-storage = { path = "../iota-storage" } +iota-storage = { path = "../iota-storage", optional = true } iota-terms = { path = "../iota-terms" } -iota-util = { path = "../iota-util" } -omikron-connector = { path = "../omikron-connector" } + iota-util = { path = "../iota-util", optional = true } +iota-ipc = { path = "../iota-ipc" } +omikron-connector = { path = "../omikron-connector", optional = true } +mtp = { git = "https://git.methanium.net/Methanium/mtp.git", optional = true } -mtp = { git = "https://git.methanium.net/Methanium/mtp.git" } actix-web = { version = "4", features = ["rustls-0_23"] } actix-web-actors = "4" @@ -56,7 +66,6 @@ sysinfo = "0.38.3" tokio = { version = "1.50.0", features = ["full"] } tokio-tungstenite = { version = "*", features = ["native-tls"] } tungstenite = "*" -uuid = { version = "*", features = ["v4"] } walkdir = "2.5.0" warp = "*" x448 = { version = "*" } diff --git a/iota-cli/src/elements/console_card.rs b/iota-cli/src/elements/console_card.rs index 820ae1e..5ece1d9 100644 --- a/iota-cli/src/elements/console_card.rs +++ b/iota-cli/src/elements/console_card.rs @@ -1,10 +1,17 @@ use crossterm::event::{KeyCode, KeyEvent}; -use iota_logger::{log, log_command, log_cv}; +#[cfg(feature = "legacy-commands")] +use iota_logger::{log, log_cv}; +#[cfg(feature = "legacy-commands")] use iota_state::{ACTIVE_TASKS, RELOAD, SHUTDOWN}; +#[cfg(feature = "legacy-commands")] use iota_storage::users::{user_manager, user_profile::UserProfile}; +#[cfg(feature = "legacy-commands")] use iota_storage::util::config_util::modify_config; +#[cfg(feature = "legacy-commands")] use iota_util::file_util; +#[cfg(feature = "legacy-commands")] use mtp::codec::{CommunicationType, CommunicationValue}; +#[cfg(feature = "legacy-commands")] use omikron_connector::omikron_connection::OMIKRON_CONNECTION; use ratatui::{ Frame, @@ -13,7 +20,6 @@ use ratatui::{ text::{Line, Span}, widgets::{Block, Borders, Paragraph}, }; -use uuid::Uuid; use std::{ any::Any, @@ -25,11 +31,12 @@ use tokio::time::Instant; use crate::{ elements::elements::{Element, InteractableElement, JoinableElement}, interaction_result::InteractionResult, - ui::FPS, + ipc_client::IpcClient, util::borders::draw_block_joins, }; pub struct ConsoleCard { + ipc: Arc, focused: bool, pub title: String, pub content: String, @@ -44,8 +51,9 @@ pub struct ConsoleCard { } impl ConsoleCard { - pub fn new(title: &str, content: &str) -> Self { + pub fn new(title: &str, content: &str, ipc: Arc) -> Self { ConsoleCard { + ipc, focused: false, title: title.to_string(), content: content.to_string(), @@ -290,22 +298,17 @@ impl InteractableElement for ConsoleCard { match key.code { KeyCode::Enter => { if self.content.is_empty() { - log!(""); return InteractionResult::Handled; } let command = self.content.clone(); - let id = Uuid::new_v4(); - let id = id.to_string(); - let id = id.split_at(8).0; - let task_id = format!("command_{}_{}", command, id); - ACTIVE_TASKS.insert(task_id.clone()); - - log_command!("{}", command); - + let ipc = self.ipc.clone(); + let seq = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as u64; tokio::spawn(async move { - run_command(&command).await; - ACTIVE_TASKS.remove(&task_id); + let _ = ipc.send_command(seq, command).await; }); self.content.clear(); @@ -379,6 +382,7 @@ impl InteractableElement for ConsoleCard { } } +#[cfg(feature = "legacy-commands")] pub async fn run_command(command: &str) { let parts = command.split(" ").collect::>(); @@ -503,6 +507,7 @@ pub async fn run_command(command: &str) { } } +#[cfg(feature = "legacy-commands")] pub async fn ping(time: u64) { let conn = OMIKRON_CONNECTION.clone(); diff --git a/iota-cli/src/elements/graph_card.rs b/iota-cli/src/elements/graph_card.rs index c7ec2d6..178ffaa 100644 --- a/iota-cli/src/elements/graph_card.rs +++ b/iota-cli/src/elements/graph_card.rs @@ -1,7 +1,7 @@ use std::{any::Any, sync::Arc}; use crossterm::event::KeyEvent; -use iota_state::APP_STATE; +use iota_state::ClientState; use ratatui::{ Frame, layout::Rect, @@ -34,11 +34,29 @@ impl GRAPHS { } } - pub fn get_graph(&self) -> Vec<(f64, f64)> { + pub fn get_graph(&self, state: &ClientState) -> Vec<(f64, f64)> { match self { - GRAPHS::Ram => APP_STATE.lock().unwrap().with_width(28).ram.clone(), - GRAPHS::Cpu => APP_STATE.lock().unwrap().with_width(28).cpu.clone(), - GRAPHS::Ping => APP_STATE.lock().unwrap().with_width(28).ping.clone(), + GRAPHS::Ram => state + .app + .lock() + .unwrap_or_else(|error| error.into_inner()) + .with_width(28) + .ram + .clone(), + GRAPHS::Cpu => state + .app + .lock() + .unwrap_or_else(|error| error.into_inner()) + .with_width(28) + .cpu + .clone(), + GRAPHS::Ping => state + .app + .lock() + .unwrap_or_else(|error| error.into_inner()) + .with_width(28) + .ping + .clone(), } } @@ -54,6 +72,7 @@ impl GRAPHS { #[allow(unused)] pub struct GraphCard { ui: Arc, + state: ClientState, graph_type: GRAPHS, focused: bool, @@ -66,9 +85,10 @@ pub struct GraphCard { } impl GraphCard { - pub fn new(ui: Arc, graph_type: GRAPHS, title: String) -> Self { + pub fn new(ui: Arc, state: ClientState, graph_type: GRAPHS, title: String) -> Self { Self { ui, + state, graph_type, focused: false, title, @@ -93,7 +113,7 @@ impl Element for GraphCard { fn render(&self, f: &mut Frame, r: Rect) { if self.open { - let graph = self.graph_type.get_graph(); + let graph = self.graph_type.get_graph(&self.state); let unit = self.graph_type.get_unit(); let min_x = graph.first().map(|(x, _)| *x).unwrap_or(0.0); let max_x = graph.last().map(|(x, _)| *x).unwrap_or(100.0); diff --git a/iota-cli/src/elements/log_card.rs b/iota-cli/src/elements/log_card.rs index a0ce8fd..c415ba9 100755 --- a/iota-cli/src/elements/log_card.rs +++ b/iota-cli/src/elements/log_card.rs @@ -1,9 +1,8 @@ -use crate::app_state::APP_STATE; use crate::elements::elements::{Element, InteractableElement, JoinableElement}; use crate::interaction_result::InteractionResult; use crate::util::borders::draw_block_joins; use crossterm::event::{KeyCode, KeyEvent}; -use iota_logger::PrintType; +use iota_state::{ClientState, UiLogEntry}; use ratatui::{ Frame, layout::Rect, @@ -12,60 +11,9 @@ use ratatui::{ widgets::{Block, Borders, Paragraph}, }; use std::any::Any; -use std::time::{SystemTime, UNIX_EPOCH}; - -#[derive(Clone, Debug)] -pub struct UiLogEntry { - pub timestamp_ms: u128, - pub sender: PrintType, - pub message: String, - pub is_error: bool, -} - -impl UiLogEntry { - pub fn format_timestamp(&self) -> String { - let secs = (self.timestamp_ms / 1000) as i64; - let hours = (secs / 3600) % 24; - let minutes = (secs / 60) % 60; - let seconds = secs % 60; - format!("{:02}:{:02}:{:02}", hours, minutes, seconds) - } -} - -impl From for UiLogEntry { - fn from(entry: LogEntry) -> Self { - Self { - timestamp_ms: entry.timestamp_ms, - sender: entry.sender, - message: entry.message, - is_error: entry.is_error, - } - } -} - -#[derive(Clone, Debug)] -pub struct LogEntry { - pub timestamp_ms: u128, - pub sender: PrintType, - pub message: String, - pub is_error: bool, -} - -impl LogEntry { - pub fn new(sender: PrintType, message: String, is_error: bool) -> Self { - Self { - timestamp_ms: SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap() - .as_millis(), - sender, - message, - is_error, - } - } -} pub struct LogCard { + state: ClientState, focused: bool, selected: bool, scroll_offset: usize, @@ -76,8 +24,9 @@ pub struct LogCard { } impl LogCard { - pub fn new() -> Self { + pub fn new(state: ClientState) -> Self { Self { + state, focused: false, selected: false, scroll_offset: 0, @@ -89,21 +38,17 @@ impl LogCard { } fn get_logs(&self) -> Vec { - let state = APP_STATE.lock().unwrap(); + let state = self + .state + .app + .lock() + .unwrap_or_else(|error| error.into_inner()); state .get_logs() .iter() .map(|e| UiLogEntry { timestamp_ms: e.timestamp_ms, - sender: match e.sender.as_str() { - "Call" => PrintType::Call, - "Client" => PrintType::Client, - "Iota" => PrintType::Iota, - "Omikron" => PrintType::Omikron, - "Omega" => PrintType::Omega, - "Command" => PrintType::Command, - _ => PrintType::General, - }, + sender: e.sender.clone(), message: e.message.clone(), is_error: e.is_error, }) @@ -196,12 +141,24 @@ impl LogCard { line.push_str(×tamp); } - result.push((line, entry.sender.prefix_color(), entry.is_error)); + result.push((line, Self::sender_color(&entry.sender), entry.is_error)); } result } + fn sender_color(sender: &str) -> Color { + match sender { + "Call" => Color::Magenta, + "Client" => Color::Green, + "Iota" => Color::Yellow, + "Omikron" => Color::Blue, + "Omega" => Color::Cyan, + "Command" => Color::LightGreen, + _ => Color::LightCyan, + } + } + fn build_all_lines( &self, entries: Vec, diff --git a/iota-cli/src/input_handler.rs b/iota-cli/src/input_handler.rs index 524ce2e..5cdf4fa 100644 --- a/iota-cli/src/input_handler.rs +++ b/iota-cli/src/input_handler.rs @@ -1,14 +1,12 @@ use crate::ui::UI; use crossterm::event::{Event, KeyEvent, KeyEventKind, KeyModifiers, poll, read}; -use iota_state::{RELOAD, SHUTDOWN, UNIQUE}; use std::sync::Arc; -use std::sync::atomic::Ordering; use std::time::Duration; pub fn setup_input_handler(ui: Arc) { tokio::spawn(async move { loop { - if *SHUTDOWN.read().await { + if ui.is_shutdown() { break; } @@ -27,7 +25,6 @@ pub fn setup_input_handler(ui: Arc) { match event_result { Ok(Some(key_event)) => { handle_input(key_event, ui.clone()).await; - UNIQUE.store(true, Ordering::Relaxed); } Ok(_) => {} Err(e) => { @@ -43,11 +40,11 @@ pub async fn handle_input(key: KeyEvent, ui: Arc) { match (key.code, key.modifiers) { (crossterm::event::KeyCode::Char('q'), KeyModifiers::CONTROL) | (crossterm::event::KeyCode::Char('c'), KeyModifiers::CONTROL) => { - *SHUTDOWN.write().await = true; + ui.request_shutdown(); } (crossterm::event::KeyCode::Char('r'), KeyModifiers::CONTROL) => { - *RELOAD.write().await = true; - *SHUTDOWN.write().await = true; + let _ = ui.send_restart().await; + ui.request_shutdown(); } _ => { ui.handle_input(key).await; diff --git a/iota-cli/src/ipc_client.rs b/iota-cli/src/ipc_client.rs new file mode 100644 index 0000000..c594f7e --- /dev/null +++ b/iota-cli/src/ipc_client.rs @@ -0,0 +1,83 @@ +use iota_ipc::{ClientMessage, DaemonMessage, read_msg, write_msg}; +use iota_state::{ClientState, UiLogEntry}; +use std::io::Result; +use std::path::Path; +use std::sync::Arc; +use tokio::net::UnixStream; +use tokio::net::unix::OwnedWriteHalf; +use tokio::sync::Mutex; + +/* The TUI owns this cache. IPC updates replace daemon snapshots and append + * logs, so rendering never reaches into daemon-owned storage or connections. */ +pub struct IpcClient { + state: ClientState, + writer: Mutex, +} + +impl IpcClient { + pub async fn connect(path: impl AsRef) -> Result> { + let stream = UnixStream::connect(path).await?; + let (mut reader, writer) = stream.into_split(); + let client = Arc::new(Self { + state: ClientState::new(), + writer: Mutex::new(writer), + }); + let reader_client = client.clone(); + tokio::spawn(async move { + while let Ok(message) = read_msg::<_, DaemonMessage>(&mut reader).await { + reader_client.apply(message).await; + } + }); + client.send(ClientMessage::Subscribe).await?; + Ok(client) + } + + pub fn state(&self) -> ClientState { + self.state.clone() + } + + pub async fn send_command(&self, seq: u64, line: String) -> Result<()> { + self.send(ClientMessage::Command { seq, line }).await + } + + async fn send(&self, message: ClientMessage) -> Result<()> { + let mut writer = self.writer.lock().await; + write_msg(&mut *writer, &message).await + } + + async fn apply(&self, message: DaemonMessage) { + let mut state = self + .state + .app + .lock() + .unwrap_or_else(|error| error.into_inner()); + match message { + DaemonMessage::LogEntry(entry) => state.push_log(UiLogEntry { + timestamp_ms: entry.timestamp_ms, + sender: entry.sender, + message: entry.message, + is_error: entry.is_error, + }), + DaemonMessage::StateUpdate(snapshot) => { + state.cpu = snapshot.cpu; + state.ram = snapshot.ram; + state.ping = snapshot.ping; + state.net_up = snapshot.net_up; + state.net_down = snapshot.net_down; + state.sys_info = snapshot.sys_info; + } + DaemonMessage::CommandResult { + success, message, .. + } => state.push_log(UiLogEntry { + timestamp_ms: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis(), + sender: "Command".into(), + message, + is_error: !success, + }), + DaemonMessage::Pong { .. } => {} + } + } +} diff --git a/iota-cli/src/lib.rs b/iota-cli/src/lib.rs index aaa23ef..6986d72 100644 --- a/iota-cli/src/lib.rs +++ b/iota-cli/src/lib.rs @@ -18,5 +18,6 @@ pub mod util { } pub mod app_state; pub mod input_handler; +pub mod ipc_client; pub mod interaction_result; pub mod ui; diff --git a/iota-cli/src/screens/main_screen.rs b/iota-cli/src/screens/main_screen.rs index e10947d..576afeb 100644 --- a/iota-cli/src/screens/main_screen.rs +++ b/iota-cli/src/screens/main_screen.rs @@ -36,22 +36,23 @@ impl MainScreen { vec![Some(1), Some(4)], ]; - let mut log_card = LogCard::new(); + let state = ui.client_state(); + let mut log_card = LogCard::new(state.clone()); log_card.set_borders(Borders::TOP.union(Borders::RIGHT).union(Borders::LEFT)); - let mut console_card = ConsoleCard::new("Console", ""); + let mut console_card = ConsoleCard::new("Console", "", ui.ipc()); console_card.set_joins(Borders::TOP); elements.push(Box::new(log_card)); elements.push(Box::new(console_card)); - let mut ram_graph = GraphCard::new(ui.clone(), GRAPHS::Ram, "RAM".into()); + let mut ram_graph = GraphCard::new(ui.clone(), state.clone(), GRAPHS::Ram, "RAM".into()); ram_graph.set_borders(Borders::TOP.union(Borders::LEFT).union(Borders::RIGHT)); elements.push(Box::new(ram_graph)); - let mut cpu_graph = GraphCard::new(ui.clone(), GRAPHS::Cpu, "CPU".into()); + let mut cpu_graph = GraphCard::new(ui.clone(), state.clone(), GRAPHS::Cpu, "CPU".into()); cpu_graph.set_borders(Borders::TOP.union(Borders::LEFT).union(Borders::RIGHT)); cpu_graph.set_joins(Borders::TOP); elements.push(Box::new(cpu_graph)); - let mut ping_graph = GraphCard::new(ui.clone(), GRAPHS::Ping, "Ping".into()); + let mut ping_graph = GraphCard::new(ui.clone(), state, GRAPHS::Ping, "Ping".into()); ping_graph.set_joins(Borders::TOP); elements.push(Box::new(ping_graph)); diff --git a/iota-cli/src/ui.rs b/iota-cli/src/ui.rs index a44ee74..f298a92 100644 --- a/iota-cli/src/ui.rs +++ b/iota-cli/src/ui.rs @@ -1,15 +1,17 @@ use crate::{ input_handler::setup_input_handler, interaction_result::InteractionResult, - screens::screens::Screen, + ipc_client::IpcClient, screens::screens::Screen, }; use crossterm::event::KeyEvent; -use iota_state::{ACTIVE_TASKS, SHUTDOWN, UNIQUE}; use once_cell::sync::Lazy; use ratatui::{Terminal, backend::CrosstermBackend, init}; use std::{ collections::VecDeque, io::Stdout, - sync::{Arc, Mutex, atomic::Ordering}, + sync::{ + Arc, Mutex, + atomic::{AtomicBool, Ordering}, + }, time::Duration, }; use tokio::{sync::RwLock, time::Instant}; @@ -19,14 +21,15 @@ use tokio::{sync::RwLock, time::Instant}; pub static FPS: Lazy> = Lazy::new(|| RwLock::new((0.0, 0.0))); pub struct UI { + ipc: Arc, + shutdown: AtomicBool, pub terminal: Arc>>>, screen_stack: Arc>>>, } -pub fn start_tui() -> Arc { - let ui = Arc::new(UI::new()); +pub fn start_tui(ipc: Arc) -> Arc { + let ui = Arc::new(UI::new(ipc)); let uic = ui.clone(); - ACTIVE_TASKS.insert("UI Renderer".to_string()); tokio::spawn(async move { let mut last_render = Instant::now(); @@ -39,11 +42,11 @@ pub fn start_tui() -> Arc { let mut skipped = 0; loop { - if *SHUTDOWN.read().await { + if uic.is_shutdown() { break; } - if skipped > 5 || UNIQUE.load(Ordering::Relaxed) { + if skipped > 5 { uic.render().await; skip_samples.push_back(skipped); @@ -88,27 +91,47 @@ pub fn start_tui() -> Arc { *FPS.write().await = (avg_fps, avg_skips_percentage); last_render = Instant::now(); - UNIQUE.store(false, Ordering::Relaxed); } else { skipped += 1; } tokio::time::sleep(Duration::from_millis(16)).await; } - ACTIVE_TASKS.remove("UI Renderer"); ratatui::restore(); }); setup_input_handler(ui.clone()); ui } impl UI { - pub fn new() -> Self { + pub fn new(ipc: Arc) -> Self { let terminal = init(); Self { + ipc, + shutdown: AtomicBool::new(false), terminal: Arc::new(Mutex::new(terminal)), screen_stack: Arc::new(RwLock::new(Vec::new())), } } + pub fn ipc(&self) -> Arc { + self.ipc.clone() + } + + pub fn client_state(&self) -> iota_state::ClientState { + self.ipc.state() + } + + pub fn is_shutdown(&self) -> bool { + self.shutdown.load(Ordering::Relaxed) + } + + pub fn request_shutdown(&self) { + self.shutdown.store(true, Ordering::Relaxed); + } + + pub async fn send_restart(&self) -> std::io::Result<()> { + self.ipc.send_command(0, "restart".into()).await + } + pub async fn set_screen(&self, screen: Box) { self.screen_stack.write().await.push(screen); } @@ -140,7 +163,7 @@ impl UI { stack.pop(); if stack.is_empty() { - *SHUTDOWN.write().await = true; + self.request_shutdown(); } } InteractionResult::Handled => {} diff --git a/iota-core/src/main.rs b/iota-core/src/main.rs index fd794e6..b0a3bb6 100644 --- a/iota-core/src/main.rs +++ b/iota-core/src/main.rs @@ -6,7 +6,7 @@ use tokio::time::{Duration, sleep}; use iota_state::{ACTIVE_TASKS, APP_STATE, AppState, RELOAD, SHUTDOWN}; use iota_cli::screens::main_screen::MainScreen; -use iota_cli::ui::start_tui; +use iota_cli::{ipc_client::IpcClient, ui::start_tui}; use iota_logger::{self as logger, language_creator}; use iota_logger::{log, log_t}; use iota_storage::users::user_manager; @@ -22,7 +22,10 @@ async fn main() { *RELOAD.write().await = false; *SHUTDOWN.write().await = false; - let ui = start_tui(); + let ipc = IpcClient::connect("/run/iota/iota.sock") + .await + .expect("iota-daemon must be running before starting iota-core"); + let ui = start_tui(ipc); let (eula, tos_pp) = match consent_state::check(ui.clone()).await { Ok(v) => v, diff --git a/iota-daemon-lib/Cargo.toml b/iota-daemon-lib/Cargo.toml new file mode 100644 index 0000000..abb2245 --- /dev/null +++ b/iota-daemon-lib/Cargo.toml @@ -0,0 +1,15 @@ +[package] +name = "iota-daemon-lib" +version = "0.1.0" +edition = "2024" + +[dependencies] +iota-ipc = { path = "../iota-ipc" } +iota-logger = { path = "../iota-logger" } +iota-state = { path = "../iota-state" } +iota-storage = { path = "../iota-storage" } +iota-util = { path = "../iota-util" } +omikron-connector = { path = "../omikron-connector" } +mtp = { git = "https://git.methanium.net/Methanium/mtp.git" } +sysinfo = "0.38.3" +tokio = { version = "1.50.0", features = ["full"] } diff --git a/iota-daemon-lib/src/command_router.rs b/iota-daemon-lib/src/command_router.rs new file mode 100644 index 0000000..11e2ddd --- /dev/null +++ b/iota-daemon-lib/src/command_router.rs @@ -0,0 +1,114 @@ +use crate::DaemonRuntime; +use iota_ipc::DaemonMessage; +use iota_logger::{log, log_command}; +use iota_storage::users::user_manager; +use iota_storage::util::config_util::modify_config; +use mtp::codec::{CommunicationType, CommunicationValue}; +use omikron_connector::omikron_connection::OMIKRON_CONNECTION; +use std::sync::Arc; +use std::time::Duration; + +#[derive(Clone)] +pub struct CommandRouter { + runtime: Arc, +} + +impl CommandRouter { + pub fn new(runtime: Arc) -> Self { + Self { runtime } + } + + pub async fn route(&self, seq: u64, line: String) -> DaemonMessage { + log_command!("{}", line); + let result = self.execute(&line).await; + DaemonMessage::CommandResult { + seq, + success: result.is_ok(), + message: result.unwrap_or_else(|error| error), + } + } + + async fn execute(&self, line: &str) -> Result { + let parts = line + .trim_start_matches('/') + .split_whitespace() + .collect::>(); + match parts.as_slice() { + ["tasks"] => Ok(self + .runtime + .state + .active_tasks + .iter() + .map(|task| task.to_string()) + .collect::>() + .join(", ")), + ["help"] => Ok( + "Available commands: tasks, ping, user, reconnect, regenerate, reload, shutdown" + .into(), + ), + ["ping"] => self.ping(20).await, + ["ping", seconds] => self.ping(seconds.parse::().unwrap_or(20)).await, + ["user", "add", username] => { + let (user, _) = omikron_connector::user_ops::create_user(username).await; + user.map(|user| format!("Created user {}", user.user_id)) + .ok_or_else(|| "User creation failed".into()) + } + ["user", "remove", username] => { + let user = user_manager::get_user_by_username(username) + .ok_or_else(|| "Username does not exist".to_string())?; + let message = CommunicationValue::new(CommunicationType::DeleteUser) + .with_sender(user.user_id as u64); + OMIKRON_CONNECTION + .send_message(&message) + .await + .map_err(|error| error.to_string())?; + user_manager::remove_user(user.user_id); + Ok(format!("Removed user {}", user.user_id)) + } + ["user", "list"] => Ok(user_manager::get_users() + .into_iter() + .map(|user| format!("{} ({})", user.username, user.user_id)) + .collect::>() + .join("\n")), + ["reconnect"] => { + OMIKRON_CONNECTION.reconnect().await; + Ok("Reconnected to Omikron server".into()) + } + ["regenerate", "keys"] => { + modify_config(|config| { + config.public_key = None; + config.private_key = None; + config.iota_id = None; + }); + OMIKRON_CONNECTION.reconnect().await; + Ok("Key pair regenerated and Omikron reconnection requested".into()) + } + ["reload"] | ["restart"] => { + *self.runtime.state.reload.write().await = true; + *self.runtime.state.shutdown.write().await = true; + Ok("Daemon restart requested".into()) + } + ["shutdown"] | ["stop"] => { + *self.runtime.state.shutdown.write().await = true; + Ok("Daemon shutdown requested".into()) + } + _ => Err("Unknown command".into()), + } + } + + async fn ping(&self, seconds: u64) -> Result { + let response = OMIKRON_CONNECTION + .await_response( + &CommunicationValue::new(CommunicationType::Ping), + Some(Duration::from_secs(seconds)), + ) + .await; + match response { + Ok(value) => { + log!("{}", iota_logger::format_cv(&value)); + Ok("Ping response received".into()) + } + Err(error) => Err(format!("Ping error: {error:?}")), + } + } +} diff --git a/iota-daemon-lib/src/daemon_state.rs b/iota-daemon-lib/src/daemon_state.rs new file mode 100644 index 0000000..ef26e63 --- /dev/null +++ b/iota-daemon-lib/src/daemon_state.rs @@ -0,0 +1,72 @@ +use iota_ipc::StateSnapshot; +use iota_state::DaemonState; +use std::sync::Arc; +use std::time::Duration; +use sysinfo::{RefreshKind, System}; + +/* This wrapper exposes daemon state as IPC-safe snapshots while preserving a + * single owned state instance for all daemon subsystems. */ +#[derive(Clone, Default)] +pub struct DaemonRuntime { + pub state: Arc, +} + +impl DaemonRuntime { + pub fn new() -> Self { + Self { + state: Arc::new(DaemonState::new()), + } + } + + pub fn snapshot(&self) -> StateSnapshot { + let state = self + .state + .app + .lock() + .unwrap_or_else(|error| error.into_inner()); + StateSnapshot { + cpu: state.cpu.clone(), + ram: state.ram.clone(), + ping: state.ping.clone(), + net_up: state.net_up.clone(), + net_down: state.net_down.clone(), + sys_info: state.sys_info.clone(), + } + } + + pub fn spawn_system_monitor(&self) { + let runtime = self.clone(); + tokio::spawn(async move { + runtime.state.active_tasks.insert("System monitor".into()); + let mut system = System::new_with_specifics(RefreshKind::everything()); + let mut counter = 0.0; + loop { + if *runtime.state.shutdown.read().await { + break; + } + system.refresh_cpu_all(); + system.refresh_memory(); + let cpu = system.global_cpu_usage() as f64; + let total_memory = system.total_memory(); + let ram = if total_memory == 0 { + 0.0 + } else { + system.used_memory() as f64 / total_memory as f64 * 100.0 + }; + { + let mut state = runtime + .state + .app + .lock() + .unwrap_or_else(|error| error.into_inner()); + state.push_cpu((counter, cpu)); + state.push_ram((counter, ram)); + state.sys_info = format!("CPU: {cpu:.1}% RAM: {ram:.1}%"); + } + counter += 1.0; + tokio::time::sleep(Duration::from_millis(500)).await; + } + runtime.state.active_tasks.remove("System monitor"); + }); + } +} diff --git a/iota-daemon-lib/src/ipc_server.rs b/iota-daemon-lib/src/ipc_server.rs new file mode 100644 index 0000000..11f62f4 --- /dev/null +++ b/iota-daemon-lib/src/ipc_server.rs @@ -0,0 +1,113 @@ +use crate::{CommandRouter, DaemonRuntime}; +use iota_ipc::{ClientMessage, DaemonMessage, read_msg, write_msg}; +use std::io::Result; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::{env, os::fd::FromRawFd}; +use tokio::net::{UnixListener, UnixStream}; +use tokio::sync::broadcast; + +pub struct IpcServer { + path: PathBuf, + runtime: Arc, + messages: broadcast::Sender, +} + +impl IpcServer { + pub fn new( + path: impl Into, + runtime: Arc, + messages: broadcast::Sender, + ) -> Self { + Self { + path: path.into(), + runtime, + messages, + } + } + + pub async fn run(self) -> Result<()> { + if let Some(parent) = self.path.parent() { + tokio::fs::create_dir_all(parent).await?; + } + let listener = match activated_listener()? { + Some(listener) => listener, + None => { + remove_stale_socket(&self.path).await?; + UnixListener::bind(&self.path)? + } + }; + loop { + let (stream, _) = listener.accept().await?; + let runtime = self.runtime.clone(); + let messages = self.messages.clone(); + tokio::spawn(async move { + let _ = handle_client(stream, runtime, messages).await; + }); + } + } +} + +/* systemd hands the first socket-activated file descriptor to the service as + * descriptor 3. Manual launches continue to bind the configured socket path. */ +fn activated_listener() -> Result> { + let listen_fds = env::var("LISTEN_FDS") + .ok() + .and_then(|value| value.parse::().ok()); + let listen_pid = env::var("LISTEN_PID") + .ok() + .and_then(|value| value.parse::().ok()); + if listen_fds != Some(1) || listen_pid != Some(std::process::id()) { + return Ok(None); + } + let listener = unsafe { std::os::unix::net::UnixListener::from_raw_fd(3) }; + UnixListener::from_std(listener).map(Some) +} + +async fn remove_stale_socket(path: &Path) -> Result<()> { + match tokio::fs::symlink_metadata(path).await { + Ok(_) => tokio::fs::remove_file(path).await, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(error), + } +} + +async fn handle_client( + stream: UnixStream, + runtime: Arc, + messages: broadcast::Sender, +) -> Result<()> { + let (mut reader, mut writer) = stream.into_split(); + let mut outgoing = messages.subscribe(); + let initial = DaemonMessage::StateUpdate(runtime.snapshot()); + write_msg(&mut writer, &initial).await?; + let writer_task = tokio::spawn(async move { + while let Ok(message) = outgoing.recv().await { + if write_msg(&mut writer, &message).await.is_err() { + break; + } + } + }); + let router = CommandRouter::new(runtime.clone()); + loop { + match read_msg::<_, ClientMessage>(&mut reader).await { + Ok(ClientMessage::Command { seq, line }) => { + let result = router.route(seq, line).await; + let _ = messages.send(result); + } + Ok(ClientMessage::Subscribe) => { + let _ = messages.send(DaemonMessage::StateUpdate(runtime.snapshot())); + } + Ok(ClientMessage::Ping { seq }) => { + let _ = messages.send(DaemonMessage::Pong { seq }); + } + Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => break, + Err(error) => { + writer_task.abort(); + return Err(error); + } + } + } + writer_task.abort(); + Ok(()) +} diff --git a/iota-daemon-lib/src/lib.rs b/iota-daemon-lib/src/lib.rs new file mode 100644 index 0000000..7f10773 --- /dev/null +++ b/iota-daemon-lib/src/lib.rs @@ -0,0 +1,8 @@ +pub mod command_router; +pub mod daemon_state; +pub mod ipc_server; +pub mod log_broadcaster; + +pub use command_router::CommandRouter; +pub use daemon_state::DaemonRuntime; +pub use ipc_server::IpcServer; diff --git a/iota-daemon-lib/src/log_broadcaster.rs b/iota-daemon-lib/src/log_broadcaster.rs new file mode 100644 index 0000000..1ce3c88 --- /dev/null +++ b/iota-daemon-lib/src/log_broadcaster.rs @@ -0,0 +1,21 @@ +use iota_ipc::{DaemonMessage, LogEntry}; +use iota_logger::subscribe; +use tokio::sync::broadcast; + +/* The daemon adapts logger output to the wire protocol so the logger stays + * independent from both the socket implementation and TUI state. */ +pub fn spawn(message_tx: broadcast::Sender) { + let Some(mut logs) = subscribe() else { + return; + }; + tokio::spawn(async move { + while let Ok(entry) = logs.recv().await { + let _ = message_tx.send(DaemonMessage::LogEntry(LogEntry { + timestamp_ms: entry.timestamp_ms, + sender: entry.sender, + message: entry.message, + is_error: entry.is_error, + })); + } + }); +} diff --git a/iota-daemon/Cargo.toml b/iota-daemon/Cargo.toml new file mode 100644 index 0000000..e5d5864 --- /dev/null +++ b/iota-daemon/Cargo.toml @@ -0,0 +1,14 @@ +[package] +name = "iota-daemon" +version = "0.1.0" +edition = "2024" + +[dependencies] +iota-daemon-lib = { path = "../iota-daemon-lib" } +iota-ipc = { path = "../iota-ipc" } +iota-logger = { path = "../iota-logger" } +iota-state = { path = "../iota-state" } +iota-storage = { path = "../iota-storage" } +omikron-connector = { path = "../omikron-connector" } +web-server = { path = "../web-server" } +tokio = { version = "1.50.0", features = ["full"] } diff --git a/iota-daemon/src/main.rs b/iota-daemon/src/main.rs new file mode 100644 index 0000000..2cdb8ff --- /dev/null +++ b/iota-daemon/src/main.rs @@ -0,0 +1,58 @@ +use iota_daemon_lib::{DaemonRuntime, IpcServer, log_broadcaster}; +use iota_logger::{self as logger, log, log_t}; +use iota_storage::users::user_manager; +use iota_storage::util::config_util::CONFIG; +use std::path::PathBuf; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::broadcast; + +fn socket_path() -> PathBuf { + std::env::var_os("IOTA_SOCKET") + .map(PathBuf::from) + .unwrap_or_else(|| PathBuf::from("/run/iota/iota.sock")) +} + +#[tokio::main(flavor = "multi_thread")] +async fn main() { + logger::startup(); + iota_storage::util::config_util::load_config(); + if user_manager::load_users().await.is_err() { + log_t!("user_load_failed"); + } + let runtime = Arc::new(DaemonRuntime::new()); + runtime.spawn_system_monitor(); + let (messages, _) = broadcast::channel(512); + log_broadcaster::spawn(messages.clone()); + let state_updates = runtime.clone(); + let state_messages = messages.clone(); + tokio::spawn(async move { + loop { + if *state_updates.state.shutdown.read().await { + break; + } + let _ = state_messages.send(iota_ipc::DaemonMessage::StateUpdate( + state_updates.snapshot(), + )); + tokio::time::sleep(Duration::from_millis(500)).await; + } + }); + + let port = CONFIG.load().port; + if !web_server::start(port).await { + log!("Failed to start the MTP web server on port {}", port); + } + let _ = omikron_connector::omikron_connection::get_omikron_connection().await; + + let server = IpcServer::new(socket_path(), runtime.clone(), messages); + tokio::spawn(async move { + if let Err(error) = server.run().await { + eprintln!("iota-daemon IPC server failed: {error}"); + } + }); + log!("iota-daemon started"); + while !*runtime.state.shutdown.read().await { + tokio::time::sleep(Duration::from_millis(250)).await; + } + log!("iota-daemon stopping"); +} diff --git a/iota-ipc/Cargo.toml b/iota-ipc/Cargo.toml new file mode 100644 index 0000000..8fb9e01 --- /dev/null +++ b/iota-ipc/Cargo.toml @@ -0,0 +1,9 @@ +[package] +name = "iota-ipc" +version = "0.1.0" +edition = "2024" + +[dependencies] +serde = { version = "1", features = ["derive"] } +serde_json = "1" +tokio = { version = "1.53.1", features = ["io-util", "macros", "rt"] } diff --git a/iota-ipc/src/lib.rs b/iota-ipc/src/lib.rs new file mode 100644 index 0000000..8cbd6fa --- /dev/null +++ b/iota-ipc/src/lib.rs @@ -0,0 +1,5 @@ +pub mod protocol; +pub mod transport; + +pub use protocol::{ClientMessage, DaemonMessage, LogEntry, StateSnapshot}; +pub use transport::{read_msg, write_msg}; diff --git a/iota-ipc/src/protocol.rs b/iota-ipc/src/protocol.rs new file mode 100644 index 0000000..09a723b --- /dev/null +++ b/iota-ipc/src/protocol.rs @@ -0,0 +1,42 @@ +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(tag = "type", content = "data", rename_all = "snake_case")] +pub enum ClientMessage { + Command { seq: u64, line: String }, + Subscribe, + Ping { seq: u64 }, +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(tag = "type", content = "data", rename_all = "snake_case")] +pub enum DaemonMessage { + LogEntry(LogEntry), + StateUpdate(StateSnapshot), + CommandResult { + seq: u64, + success: bool, + message: String, + }, + Pong { + seq: u64, + }, +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct LogEntry { + pub timestamp_ms: u128, + pub sender: String, + pub message: String, + pub is_error: bool, +} + +#[derive(Clone, Debug, Default, Deserialize, Serialize)] +pub struct StateSnapshot { + pub cpu: Vec<(f64, f64)>, + pub ram: Vec<(f64, f64)>, + pub ping: Vec<(f64, f64)>, + pub net_up: Vec<(f64, f64)>, + pub net_down: Vec<(f64, f64)>, + pub sys_info: String, +} diff --git a/iota-ipc/src/transport.rs b/iota-ipc/src/transport.rs new file mode 100644 index 0000000..811b0ee --- /dev/null +++ b/iota-ipc/src/transport.rs @@ -0,0 +1,59 @@ +use serde::Serialize; +use serde::de::DeserializeOwned; +use std::io::{Error, ErrorKind, Result}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; + +const MAX_MESSAGE_SIZE: usize = 1024 * 1024; + +/* Length-prefixing preserves message boundaries on a byte stream and bounds + * allocations before JSON is deserialized. */ +pub async fn write_msg(writer: &mut W, message: &T) -> Result<()> +where + W: AsyncWrite + Unpin, + T: Serialize, +{ + let payload = + serde_json::to_vec(message).map_err(|error| Error::new(ErrorKind::InvalidData, error))?; + let len = u32::try_from(payload.len()) + .map_err(|_| Error::new(ErrorKind::InvalidData, "IPC message is too large"))?; + writer.write_u32(len).await?; + writer.write_all(&payload).await?; + writer.flush().await +} + +pub async fn read_msg(reader: &mut R) -> Result +where + R: AsyncRead + Unpin, + T: DeserializeOwned, +{ + let len = reader.read_u32().await? as usize; + if len > MAX_MESSAGE_SIZE { + return Err(Error::new( + ErrorKind::InvalidData, + "IPC message exceeds limit", + )); + } + let mut payload = vec![0; len]; + reader.read_exact(&mut payload).await?; + serde_json::from_slice(&payload).map_err(|error| Error::new(ErrorKind::InvalidData, error)) +} + +#[cfg(test)] +mod tests { + use super::{read_msg, write_msg}; + use crate::protocol::ClientMessage; + + #[tokio::test] + async fn round_trips_framed_messages() { + let (mut writer, mut reader) = tokio::io::duplex(1024); + let message = ClientMessage::Command { + seq: 4, + line: "help".into(), + }; + write_msg(&mut writer, &message) + .await + .expect("write succeeds"); + let received: ClientMessage = read_msg(&mut reader).await.expect("read succeeds"); + assert!(matches!(received, ClientMessage::Command { seq: 4, line } if line == "help")); + } +} diff --git a/iota-logger/Cargo.toml b/iota-logger/Cargo.toml index 23f4a66..530f2f6 100644 --- a/iota-logger/Cargo.toml +++ b/iota-logger/Cargo.toml @@ -12,3 +12,4 @@ mtp = { git = "https://git.methanium.net/Methanium/mtp.git" } ratatui = "0.30.0" json = "0.12.4" once_cell = "1.21.4" +tokio = { version = "1.50.0", features = ["sync"] } diff --git a/iota-logger/src/lib.rs b/iota-logger/src/lib.rs index 01ebbff..486a04c 100644 --- a/iota-logger/src/lib.rs +++ b/iota-logger/src/lib.rs @@ -10,11 +10,13 @@ use std::{ use mtp::codec::{CommunicationValue, DataTypeId, DataValue, Version}; use ratatui::style::Color; -use iota_state::{APP_STATE, UNIQUE, UiLogEntry}; +use iota_state::{UNIQUE, UiLogEntry}; +use tokio::sync::broadcast; pub mod language_creator; pub mod language_manager; static LOGGER: OnceLock> = OnceLock::new(); +static LOG_BROADCASTER: OnceLock> = OnceLock::new(); #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] #[allow(unused)] @@ -51,9 +53,15 @@ struct LogMessage { message: Option, } +/* The logger owns file persistence while consumers receive rendered entries + * through a process-local broadcast subscription. */ pub fn startup() { let (tx, rx) = mpsc::channel::(); - LOGGER.set(tx).expect("Logger already initialized"); + if LOGGER.set(tx).is_err() { + return; + } + let (broadcast_tx, _) = broadcast::channel(512); + let _ = LOG_BROADCASTER.set(broadcast_tx.clone()); thread::spawn(move || { let working_dir = iota_util::file_util::get_directory(); @@ -106,12 +114,15 @@ pub fn startup() { is_error: msg.is_error, }; - let mut state = APP_STATE.lock().unwrap(); - state.push_log(entry); + let _ = broadcast_tx.send(entry); } }); } +pub fn subscribe() -> Option> { + LOG_BROADCASTER.get().map(broadcast::Sender::subscribe) +} + fn format_timestamp_inline(timestamp_ms: u128) -> String { let secs = (timestamp_ms / 1000) as i64; let hours = (secs / 3600) % 24; diff --git a/iota-state/Cargo.toml b/iota-state/Cargo.toml index e246e12..4275556 100644 --- a/iota-state/Cargo.toml +++ b/iota-state/Cargo.toml @@ -3,6 +3,10 @@ name = "iota-state" version = "0.1.0" edition = "2024" +[features] +default = ["legacy-globals"] +legacy-globals = [] + [dependencies] dashmap = "6.1.0" once_cell = "1.21.3" @@ -10,3 +14,4 @@ tokio = { version = "1.50.0", features = ["full"] } json = "*" sysinfo = "0.38.3" mtp = { git = "https://git.methanium.net/Methanium/mtp.git" } +serde = { version = "1", features = ["derive"] } diff --git a/iota-state/src/lib.rs b/iota-state/src/lib.rs index 1e5d1c2..3b29d08 100644 --- a/iota-state/src/lib.rs +++ b/iota-state/src/lib.rs @@ -1,13 +1,67 @@ use dashmap::DashSet; use json::{JsonValue, object}; +#[cfg(feature = "legacy-globals")] use once_cell::sync::Lazy; use std::collections::VecDeque; -use std::sync::{Arc, LazyLock, Mutex, atomic::AtomicBool}; +#[cfg(feature = "legacy-globals")] +use std::sync::LazyLock; +use std::sync::{Arc, Mutex, atomic::AtomicBool}; +#[cfg(feature = "legacy-globals")] use std::thread; +#[cfg(feature = "legacy-globals")] use std::time::Duration; +#[cfg(feature = "legacy-globals")] use sysinfo::{RefreshKind, System}; use tokio::sync::RwLock; +/* Process-owned daemon state and TUI-local state must be separate because IPC, + * rather than shared memory, is the boundary between the two binaries. */ +#[derive(Clone)] +pub struct DaemonState { + pub app: Arc>, + pub shutdown: Arc>, + pub reload: Arc>, + pub active_tasks: Arc>, +} + +impl DaemonState { + pub fn new() -> Self { + Self { + app: Arc::new(Mutex::new(AppState::new())), + shutdown: Arc::new(RwLock::new(false)), + reload: Arc::new(RwLock::new(false)), + active_tasks: Arc::new(DashSet::new()), + } + } +} + +impl Default for DaemonState { + fn default() -> Self { + Self::new() + } +} + +/* The TUI keeps only the daemon data it renders. This state is never shared + * with the daemon and is populated from daemon IPC messages. */ +#[derive(Clone)] +pub struct ClientState { + pub app: Arc>, +} + +impl ClientState { + pub fn new() -> Self { + Self { + app: Arc::new(Mutex::new(AppState::new())), + } + } +} + +impl Default for ClientState { + fn default() -> Self { + Self::new() + } +} + pub const MAX_POINTS: usize = 1000; pub const MAX_LOGS: usize = 100; @@ -148,13 +202,22 @@ impl AppState { } } +#[cfg(feature = "legacy-globals")] +#[deprecated(note = "use DaemonState or ClientState")] pub static APP_STATE: LazyLock>> = LazyLock::new(|| Arc::new(Mutex::new(AppState::new()))); +#[cfg(feature = "legacy-globals")] +#[deprecated(note = "use DaemonState")] pub static SHUTDOWN: Lazy> = Lazy::new(|| RwLock::new(false)); +#[cfg(feature = "legacy-globals")] +#[deprecated(note = "use DaemonState")] pub static RELOAD: Lazy> = Lazy::new(|| RwLock::new(true)); +#[cfg(feature = "legacy-globals")] +#[deprecated(note = "use DaemonState")] pub static ACTIVE_TASKS: Lazy> = Lazy::new(|| DashSet::new()); +#[cfg(feature = "legacy-globals")] pub fn setup() { ACTIVE_TASKS.insert("System info loader".to_string()); tokio::spawn(async move { diff --git a/iota-util/src/crypto_util.rs b/iota-util/src/crypto_util.rs index 010ceae..64056d9 100644 --- a/iota-util/src/crypto_util.rs +++ b/iota-util/src/crypto_util.rs @@ -42,10 +42,12 @@ pub fn decrypt_challenge(encrypted: &str, keyring: &Keyring) -> Result String { +pub fn export(data: &[u8], format: DataFormat) -> Result { match format { - DataFormat::Raw => String::from_utf8_lossy(data).to_string(), - DataFormat::Base64 => STANDARD.encode(data), - DataFormat::Hex => hex::encode(data), + DataFormat::Raw => { + String::from_utf8(data.to_vec()).map_err(|error| format!("utf8 decode error: {error}")) + } + DataFormat::Base64 => Ok(STANDARD.encode(data)), + DataFormat::Hex => Ok(hex::encode(data)), } } diff --git a/iota/Cargo.toml b/iota/Cargo.toml new file mode 100644 index 0000000..32525e2 --- /dev/null +++ b/iota/Cargo.toml @@ -0,0 +1,8 @@ +[package] +name = "iota" +version = "0.1.0" +edition = "2024" + +[dependencies] +iota-cli = { path = "../iota-cli" } +tokio = { version = "1.50.0", features = ["full"] } diff --git a/iota/src/main.rs b/iota/src/main.rs new file mode 100644 index 0000000..845cb83 --- /dev/null +++ b/iota/src/main.rs @@ -0,0 +1,29 @@ +use iota_cli::{ipc_client::IpcClient, screens::main_screen::MainScreen, ui::start_tui}; +use std::path::PathBuf; + +fn socket_path() -> PathBuf { + std::env::var_os("IOTA_SOCKET") + .map(PathBuf::from) + .unwrap_or_else(|| PathBuf::from("/run/iota/iota.sock")) +} + +#[tokio::main(flavor = "multi_thread")] +async fn main() { + let path = socket_path(); + let ipc = match IpcClient::connect(&path).await { + Ok(client) => client, + Err(error) => { + eprintln!( + "Cannot connect to iota-daemon at {}: {error}", + path.display() + ); + std::process::exit(1); + } + }; + let ui = start_tui(ipc); + ui.set_screen(Box::new(MainScreen::new(ui.clone()).await)) + .await; + while !ui.is_shutdown() { + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } +} diff --git a/systemd/iota-daemon.service b/systemd/iota-daemon.service new file mode 100644 index 0000000..3f3b819 --- /dev/null +++ b/systemd/iota-daemon.service @@ -0,0 +1,16 @@ +[Unit] +Description=Tensamin Iota daemon +After=network-online.target +Wants=network-online.target +Requires=iota-daemon.socket + +[Service] +Type=simple +ExecStart=/usr/bin/iota-daemon +Restart=on-failure +RuntimeDirectory=iota +RuntimeDirectoryMode=0750 +Environment=IOTA_SOCKET=/run/iota/iota.sock + +[Install] +WantedBy=multi-user.target diff --git a/systemd/iota-daemon.socket b/systemd/iota-daemon.socket new file mode 100644 index 0000000..22ab691 --- /dev/null +++ b/systemd/iota-daemon.socket @@ -0,0 +1,12 @@ +[Unit] +Description=Tensamin Iota daemon IPC socket + +[Socket] +ListenStream=/run/iota/iota.sock +SocketMode=0660 +SocketUser=iota +SocketGroup=iota +RemoveOnStop=true + +[Install] +WantedBy=sockets.target