From 6275562f35c0b8b7f057c085dbecdd833fec4316 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Fri, 6 Feb 2026 13:08:06 -0600 Subject: [PATCH] Resolve issues from TUI implementation --- Cargo.lock | 343 +++++++++++++++++++++++++++++++++++++++- Cargo.toml | 10 ++ src/main.rs | 124 ++++++++++++++- src/tui/app.rs | 150 ++++++++++++++++++ src/tui/events.rs | 68 ++++++++ src/tui/mod.rs | 8 + src/tui/state.rs | 166 +++++++++++++++++++ src/tui/ui.rs | 194 +++++++++++++++++++++++ src/websocket_client.rs | 306 ++++++++++++++++++++++++++++++++++- 9 files changed, 1349 insertions(+), 20 deletions(-) create mode 100644 src/tui/app.rs create mode 100644 src/tui/events.rs create mode 100644 src/tui/mod.rs create mode 100644 src/tui/state.rs create mode 100644 src/tui/ui.rs diff --git a/Cargo.lock b/Cargo.lock index 9c40c12..1b48d79 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -52,6 +52,12 @@ dependencies = [ "memchr", ] +[[package]] +name = "allocator-api2" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" + [[package]] name = "android_system_properties" version = "0.1.5" @@ -306,6 +312,21 @@ version = "1.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" +[[package]] +name = "cassowary" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df8670b8c7b9dae1793364eafadf7239c40d669904660c5960d74cfd80b46a53" + +[[package]] +name = "castaway" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dec551ab6e7578819132c713a93c022a05d60159dc86e7a7050223577484c55a" +dependencies = [ + "rustversion", +] + [[package]] name = "cbc" version = "0.1.2" @@ -434,6 +455,20 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b05b61dc5112cbb17e4b6cd61790d9845d13888356391624cbe7e41efeac1e75" +[[package]] +name = "compact_str" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b79c4069c6cad78e2e0cdfcbd26275770669fb39fd308a752dc110e83b9af32" +dependencies = [ + "castaway", + "cfg-if", + "itoa", + "rustversion", + "ryu", + "static_assertions", +] + [[package]] name = "const-oid" version = "0.9.6" @@ -481,6 +516,31 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "crossterm" +version = "0.28.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "829d955a0bb380ef178a640b91779e3987da38c9aea133b20614cfed8cdea9c6" +dependencies = [ + "bitflags", + "crossterm_winapi", + "mio", + "parking_lot", + "rustix 0.38.44", + "signal-hook", + "signal-hook-mio", + "winapi", +] + +[[package]] +name = "crossterm_winapi" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "acdd7c62a3665c7f6830a51635d9ac9b23ed385797f70a83bb8bafe9c572ab2b" +dependencies = [ + "winapi", +] + [[package]] name = "crypto-bigint" version = "0.5.5" @@ -581,6 +641,40 @@ dependencies = [ "syn", ] +[[package]] +name = "darling" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "25ae13da2f202d56bd7f91c25fba009e7717a1e4a1cc98a76d844b65ae912e9d" +dependencies = [ + "darling_core", + "darling_macro", +] + +[[package]] +name = "darling_core" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9865a50f7c335f53564bb694ef660825eb8610e0a53d3e11bf1b0d3df31e03b0" +dependencies = [ + "ident_case", + "proc-macro2", + "quote", + "strsim", + "syn", +] + +[[package]] +name = "darling_macro" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3984ec7bd6cfa798e62b4a642426a5be0e68f9401cfc2a01e3fa9ea2fcdb8d" +dependencies = [ + "darling_core", + "quote", + "syn", +] + [[package]] name = "data-encoding" version = "2.10.0" @@ -988,6 +1082,8 @@ version = "0.15.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" dependencies = [ + "allocator-api2", + "equivalent", "foldhash", ] @@ -1079,6 +1175,17 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "hostname" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "617aaa3557aef3810a6369d0a99fac8a080891b68bd9f9812a1eeda0c0730cbd" +dependencies = [ + "cfg-if", + "libc", + "windows-link", +] + [[package]] name = "http" version = "1.4.0" @@ -1128,6 +1235,12 @@ dependencies = [ "cc", ] +[[package]] +name = "ident_case" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9e0384b61958566e926dc50660321d12159025e767c18e043daf26b70104c39" + [[package]] name = "indexmap" version = "2.13.0" @@ -1138,6 +1251,15 @@ dependencies = [ "hashbrown 0.16.1", ] +[[package]] +name = "indoc" +version = "2.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "79cf5c93f93228cf8efb3ba362535fb11199ac548a09ce117c9b1adc3030d706" +dependencies = [ + "rustversion", +] + [[package]] name = "inout" version = "0.1.4" @@ -1148,6 +1270,19 @@ dependencies = [ "generic-array 0.14.7", ] +[[package]] +name = "instability" +version = "0.3.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "357b7205c6cd18dd2c86ed312d1e70add149aea98e7ef72b9fdf0270e555c11d" +dependencies = [ + "darling", + "indoc", + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "internal-russh-forked-ssh-key" version = "0.6.16+upstream-0.6.7" @@ -1185,6 +1320,15 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" +[[package]] +name = "itertools" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "413ee7dfc52ee1a4949ceeb7dbc8a33f2d6c088194d9f922fb8318faf1f01186" +dependencies = [ + "either", +] + [[package]] name = "itertools" version = "0.14.0" @@ -1307,18 +1451,42 @@ version = "0.2.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" +[[package]] +name = "linux-raw-sys" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d26c52dbd32dccf2d10cac7725f8eae5296885fb5703b261f7d0a0739ec807ab" + [[package]] name = "linux-raw-sys" version = "0.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "df1d3c3b53da64cf5760482273a98e575c651a67eec7f77df96b5b642de8f039" +[[package]] +name = "lock_api" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" +dependencies = [ + "scopeguard", +] + [[package]] name = "log" version = "0.4.29" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" +[[package]] +name = "lru" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "234cf4f4a04dc1f57e24b96cc0cd600cf2af460d4161ac5ecdd0af8e1f3b2a38" +dependencies = [ + "hashbrown 0.15.5", +] + [[package]] name = "matchers" version = "0.2.0" @@ -1363,6 +1531,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a69bcab0ad47271a0234d9422b131806bf3968021e5dc9328caf2d4cd58557fc" dependencies = [ "libc", + "log", "wasi", "windows-sys 0.61.2", ] @@ -1573,6 +1742,29 @@ dependencies = [ "windows-strings", ] +[[package]] +name = "parking_lot" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" +dependencies = [ + "lock_api", + "parking_lot_core", +] + +[[package]] +name = "parking_lot_core" +version = "0.9.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" +dependencies = [ + "cfg-if", + "libc", + "redox_syscall", + "smallvec", + "windows-link", +] + [[package]] name = "password-hash" version = "0.5.0" @@ -1584,6 +1776,12 @@ dependencies = [ "subtle", ] +[[package]] +name = "paste" +version = "1.0.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a" + [[package]] name = "pastey" version = "0.1.1" @@ -1793,7 +1991,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "343d3bd7056eda839b03204e68deff7d1b13aba7af2b2fd16890697274262ee7" dependencies = [ "heck", - "itertools", + "itertools 0.14.0", "log", "multimap", "petgraph", @@ -1812,7 +2010,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b" dependencies = [ "anyhow", - "itertools", + "itertools 0.14.0", "proc-macro2", "quote", "syn", @@ -1907,6 +2105,36 @@ version = "0.10.0-rc-3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f66ee92bc15280519ef199a274fe0cafff4245d31bc39aaa31c011ad56cb1f05" +[[package]] +name = "ratatui" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eabd94c2f37801c20583fc49dd5cd6b0ba68c716787c2dd6ed18571e1e63117b" +dependencies = [ + "bitflags", + "cassowary", + "compact_str", + "crossterm", + "indoc", + "instability", + "itertools 0.13.0", + "lru", + "paste", + "strum", + "unicode-segmentation", + "unicode-truncate", + "unicode-width 0.2.0", +] + +[[package]] +name = "redox_syscall" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" +dependencies = [ + "bitflags", +] + [[package]] name = "regex" version = "1.12.2" @@ -2084,6 +2312,19 @@ dependencies = [ "nom", ] +[[package]] +name = "rustix" +version = "0.38.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fdb5bc1ae2baa591800df16c9ca78619bf65c0488b41b96ccec5d11220d8c154" +dependencies = [ + "bitflags", + "errno", + "libc", + "linux-raw-sys 0.4.15", + "windows-sys 0.52.0", +] + [[package]] name = "rustix" version = "1.1.3" @@ -2093,7 +2334,7 @@ dependencies = [ "bitflags", "errno", "libc", - "linux-raw-sys", + "linux-raw-sys 0.11.0", "windows-sys 0.61.2", ] @@ -2139,6 +2380,12 @@ version = "1.0.22" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" +[[package]] +name = "ryu" +version = "1.0.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a50f4cf475b65d88e057964e0e9bb1f0aa9bbb2036dc65c64596b42932536984" + [[package]] name = "salsa20" version = "0.10.2" @@ -2148,6 +2395,12 @@ dependencies = [ "cipher", ] +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + [[package]] name = "scrypt" version = "0.11.0" @@ -2291,6 +2544,27 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" +[[package]] +name = "signal-hook" +version = "0.3.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d881a16cf4426aa584979d30bd82cb33429027e42122b169753d6ef1085ed6e2" +dependencies = [ + "libc", + "signal-hook-registry", +] + +[[package]] +name = "signal-hook-mio" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b75a19a7a740b25bc7944bdee6172368f988763b744e3d4dfe753f6b4ece40cc" +dependencies = [ + "libc", + "mio", + "signal-hook", +] + [[package]] name = "signal-hook-registry" version = "1.4.8" @@ -2415,12 +2689,40 @@ dependencies = [ "sha2 0.10.9", ] +[[package]] +name = "static_assertions" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" + [[package]] name = "strsim" version = "0.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" +[[package]] +name = "strum" +version = "0.26.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fec0f0aef304996cf250b31b5a10dee7980c85da9d759361292b8bca5a18f06" +dependencies = [ + "strum_macros", +] + +[[package]] +name = "strum_macros" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4c6bee85a5a24955dc440386795aa378cd9cf82acd5f764469152d2270e581be" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "rustversion", + "syn", +] + [[package]] name = "subtle" version = "2.6.1" @@ -2458,7 +2760,7 @@ dependencies = [ "fastrand", "getrandom 0.3.4", "once_cell", - "rustix", + "rustix 1.1.3", "windows-sys 0.61.2", ] @@ -2589,13 +2891,17 @@ dependencies = [ name = "towerops-agent" version = "0.1.0" dependencies = [ + "anyhow", "async-trait", "clap", + "crossterm", "futures", "home", + "hostname", "prost", "prost-build", "prost-types", + "ratatui", "russh", "serde", "serde_json", @@ -2701,6 +3007,35 @@ version = "1.0.22" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9312f7c4f6ff9069b165498234ce8be658059c6728633667c526e27dc2cf1df5" +[[package]] +name = "unicode-segmentation" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f6ccf251212114b54433ec949fd6a7841275f9ada20dddd2f29e9ceea4501493" + +[[package]] +name = "unicode-truncate" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3644627a5af5fa321c95b9b235a72fd24cd29c648c2c379431e6628655627bf" +dependencies = [ + "itertools 0.13.0", + "unicode-segmentation", + "unicode-width 0.1.14", +] + +[[package]] +name = "unicode-width" +version = "0.1.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7dd6e30e90baa6f72411720665d41d89b9a3d039dc45b8faea1ddd07f617f6af" + +[[package]] +name = "unicode-width" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fc81956842c57dac11422a97c3b8195a1ff727f06e85c84ed2e8aa277c9a0fd" + [[package]] name = "universal-hash" version = "0.5.1" diff --git a/Cargo.toml b/Cargo.toml index f56c8b4..459debd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -21,6 +21,16 @@ russh = "0.57" async-trait = "0.1" home = "=0.5.12" zeroize = { version = "1", features = ["derive"] } +hostname = "0.4" +anyhow = "1.0" + +# TUI dependencies (optional feature) +ratatui = { version = "0.29", optional = true } +crossterm = { version = "0.28", optional = true } + +[features] +default = ["tui"] +tui = ["ratatui", "crossterm"] [build-dependencies] prost-build = "0.14" diff --git a/src/main.rs b/src/main.rs index 21d98c8..5386d94 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3,6 +3,8 @@ mod proto; pub mod secret; mod snmp; mod ssh; +#[cfg(feature = "tui")] +mod tui; mod version; mod websocket_client; @@ -16,16 +18,45 @@ use tokio::time::sleep; use tracing_subscriber::EnvFilter; use websocket_client::AgentClient; -fn init_logger() { +fn init_logger(suppress_stdout: bool) { // Use LOG_LEVEL env var (fall back to RUST_LOG for backwards compatibility) let filter = env::var("LOG_LEVEL") .or_else(|_| env::var("RUST_LOG")) .unwrap_or_else(|_| "info".to_string()); - tracing_subscriber::fmt() + let builder = tracing_subscriber::fmt() .with_env_filter(EnvFilter::new(&filter)) - .with_target(false) - .init(); + .with_target(false); + + if suppress_stdout { + // Route to sink when TUI is active (using closure to create MakeWriter) + builder.with_writer(std::io::sink).init(); + } else { + builder.init(); + } +} + +/// Determine if TUI should be enabled based on CLI flags and TTY detection +#[cfg(feature = "tui")] +fn should_enable_tui(args: &Args) -> bool { + // If explicitly disabled, never enable + if args.no_tui { + return false; + } + + // If explicitly enabled, always enable + if let Some(explicit) = args.tui { + return explicit; + } + + // Auto-detect: check if stdout is a TTY + use std::io::IsTerminal; + std::io::stdout().is_terminal() +} + +#[cfg(not(feature = "tui"))] +fn should_enable_tui(_args: &Args) -> bool { + false } /// Convert HTTP(S) URL to WebSocket URL @@ -113,12 +144,17 @@ struct Args { /// SNMPv3 priv password (for --snmpv3-test) #[arg(long, default_value = "")] snmpv3_priv_pass: String, + + /// Enable TUI mode (auto-detects TTY by default) + #[arg(long, env = "TOWEROPS_TUI")] + tui: Option, + + /// Force disable TUI even if TTY is available + #[arg(long, env = "TOWEROPS_NO_TUI", default_value_t = false)] + no_tui: bool, } fn main() { - // Initialize logging - init_logger(); - // Build Tokio runtime with larger stack size for SNMPv3 crypto operations let runtime = tokio::runtime::Builder::new_multi_thread() .enable_all() @@ -132,6 +168,12 @@ fn main() { async fn async_main() { let args = Args::parse(); + // Determine if TUI should be enabled (before initializing logger) + let tui_enabled = should_enable_tui(&args); + + // Initialize logging (suppress stdout if TUI is enabled) + init_logger(tui_enabled); + // Handle MikroTik test mode if args.mikrotik_test { run_mikrotik_test(&args).await; @@ -181,12 +223,58 @@ async fn async_main() { let (shutdown_tx, shutdown_rx) = watch::channel(false); // Spawn signal handler for graceful shutdown + let shutdown_tx_clone = shutdown_tx.clone(); tokio::spawn(async move { wait_for_shutdown_signal().await; tracing::info!("Shutdown signal received, initiating graceful shutdown..."); - let _ = shutdown_tx.send(true); + let _ = shutdown_tx_clone.send(true); }); + // Setup TUI if enabled + #[cfg(feature = "tui")] + let (event_bus, _tui_handle) = if tui_enabled { + use std::sync::Mutex; + + // Get hostname for TUI state + let hostname = hostname::get() + .ok() + .and_then(|h| h.into_string().ok()) + .unwrap_or_else(|| "unknown".to_string()); + + // Create event bus and shared state + let event_bus = tui::EventBus::new(100); + let agent_state = Arc::new(Mutex::new(tui::AgentState::new( + hostname, + env!("CARGO_PKG_VERSION").to_string(), + ))); + + // Spawn TUI task + match tui::TuiApp::new(agent_state.clone(), event_bus.subscribe()) { + Ok(mut tui_app) => { + let shutdown_rx_clone = shutdown_rx.clone(); + let tui_handle = tokio::spawn(async move { + if let Err(e) = tui_app.run(shutdown_rx_clone).await { + eprintln!("TUI task failed: {}", e); + } + // When TUI exits (user presses 'q'), trigger shutdown + let _ = shutdown_tx.send(true); + }); + + (Some(event_bus), Some(tui_handle)) + } + Err(e) => { + eprintln!("Failed to initialize TUI: {}", e); + eprintln!("Falling back to log mode"); + (None, None) + } + } + } else { + (None, None) + }; + + #[cfg(not(feature = "tui"))] + let event_bus: Option<()> = None; + // Retry loop with exponential backoff let mut retry_delay = Duration::from_secs(1); let max_retry_delay = Duration::from_secs(60); @@ -216,6 +304,26 @@ async fn async_main() { } // Connect to Towerops server via WebSocket + #[cfg(feature = "tui")] + let mut client = match AgentClient::connect(&ws_url, &token, event_bus.clone()).await { + Ok(client) => { + tracing::info!("Successfully connected to server"); + // Mark as connected for health check + connected.store(true, Ordering::Relaxed); + // Reset retry delay on successful connection + retry_delay = Duration::from_secs(1); + attempt = 0; + client + } + Err(e) => { + tracing::error!("Failed to connect to server: {}", e); + // Mark as disconnected for health check + connected.store(false, Ordering::Relaxed); + continue; + } + }; + + #[cfg(not(feature = "tui"))] let mut client = match AgentClient::connect(&ws_url, &token).await { Ok(client) => { tracing::info!("Successfully connected to server"); diff --git a/src/tui/app.rs b/src/tui/app.rs new file mode 100644 index 0000000..b9eeeb6 --- /dev/null +++ b/src/tui/app.rs @@ -0,0 +1,150 @@ +use std::io::{self, Stdout}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use crossterm::{ + event::{self, Event, KeyCode, KeyEvent}, + execute, + terminal::{disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen}, +}; +use ratatui::{backend::CrosstermBackend, Terminal}; +use tokio::sync::{broadcast, watch}; +use tokio::time::interval; + +use super::events::AgentEvent; +use super::state::AgentState; +use super::ui; + +pub struct TuiApp { + state: Arc>, + event_rx: broadcast::Receiver, + terminal: Terminal>, + scroll_offset: usize, +} + +impl TuiApp { + pub fn new( + state: Arc>, + event_rx: broadcast::Receiver, + ) -> anyhow::Result { + // Setup terminal + enable_raw_mode()?; + let mut stdout = io::stdout(); + execute!(stdout, EnterAlternateScreen)?; + let backend = CrosstermBackend::new(stdout); + let terminal = Terminal::new(backend)?; + + Ok(Self { + state, + event_rx, + terminal, + scroll_offset: 0, + }) + } + + pub async fn run(&mut self, mut shutdown_rx: watch::Receiver) -> anyhow::Result<()> { + let mut tick_interval = interval(Duration::from_millis(100)); // 10 FPS + + loop { + tokio::select! { + // Check for shutdown signal + _ = shutdown_rx.changed() => { + break; + } + + // Handle agent events + result = self.event_rx.recv() => { + match result { + Ok(event) => { + let mut state = self.state.lock().unwrap(); + state.apply_event(&event); + } + Err(broadcast::error::RecvError::Lagged(skipped)) => { + tracing::warn!("TUI lagged behind, skipped {} events", skipped); + } + Err(broadcast::error::RecvError::Closed) => { + break; + } + } + } + + // Render tick + _ = tick_interval.tick() => { + self.render()?; + } + + // Handle keyboard input (non-blocking) + _ = tokio::time::sleep(Duration::from_millis(50)) => { + if event::poll(Duration::from_millis(0))? { + if let Event::Key(key) = event::read()? { + if self.handle_key(key) { + break; // User pressed 'q' to quit + } + } + } + } + } + } + + self.cleanup()?; + Ok(()) + } + + fn render(&mut self) -> anyhow::Result<()> { + let state = self.state.lock().unwrap().clone(); + let scroll_offset = self.scroll_offset; + + self.terminal.draw(|f| { + ui::render(f, &state, scroll_offset); + })?; + + Ok(()) + } + + fn handle_key(&mut self, key: KeyEvent) -> bool { + match key.code { + KeyCode::Char('q') => { + return true; // Signal to quit + } + KeyCode::Up => { + self.scroll_offset = self.scroll_offset.saturating_add(1); + } + KeyCode::Down => { + self.scroll_offset = self.scroll_offset.saturating_sub(1); + } + KeyCode::PageUp => { + self.scroll_offset = self.scroll_offset.saturating_add(10); + } + KeyCode::PageDown => { + self.scroll_offset = self.scroll_offset.saturating_sub(10); + } + KeyCode::Home => { + // Go to end (most recent events) + self.scroll_offset = 0; + } + KeyCode::End => { + // Go to beginning (oldest events) + let state = self.state.lock().unwrap(); + self.scroll_offset = state.recent_events.len().saturating_sub(1); + } + _ => {} + } + false + } + + fn cleanup(&mut self) -> anyhow::Result<()> { + disable_raw_mode()?; + execute!(self.terminal.backend_mut(), LeaveAlternateScreen,)?; + self.terminal.show_cursor()?; + Ok(()) + } +} + +impl Drop for TuiApp { + fn drop(&mut self) { + // Ensure terminal is restored even on panic + let _ = disable_raw_mode(); + let _ = execute!(self.terminal.backend_mut(), LeaveAlternateScreen,); + let _ = self.terminal.show_cursor(); + } +} diff --git a/src/tui/events.rs b/src/tui/events.rs new file mode 100644 index 0000000..9edb5ee --- /dev/null +++ b/src/tui/events.rs @@ -0,0 +1,68 @@ +use tokio::sync::broadcast; + +#[derive(Debug, Clone)] +pub enum AgentEvent { + Connected { + agent_id: String, + }, + Disconnected, + JobReceived { + job_id: String, + device_id: String, + job_type: String, + }, + JobCompleted { + job_id: String, + device_id: String, + duration_ms: u64, + }, + JobFailed { + job_id: String, + device_id: String, + error: String, + }, + SnmpResultSent { + device_id: String, + oid_count: usize, + }, + MikrotikResultSent { + device_id: String, + sentence_count: usize, + }, + PollerCreated { + device_id: String, + total_count: usize, + }, + PollerRemoved { + device_id: String, + total_count: usize, + }, + HeartbeatSent, + PhxHeartbeatSent, + Error { + message: String, + }, +} + +#[derive(Clone)] +pub struct EventBus { + tx: broadcast::Sender, +} + +impl EventBus { + pub fn new(capacity: usize) -> Self { + let (tx, _) = broadcast::channel(capacity); + Self { tx } + } + + pub fn subscribe(&self) -> broadcast::Receiver { + self.tx.subscribe() + } + + pub fn send( + &self, + event: AgentEvent, + ) -> Result> { + self.tx.send(event) + } +} diff --git a/src/tui/mod.rs b/src/tui/mod.rs new file mode 100644 index 0000000..62510c7 --- /dev/null +++ b/src/tui/mod.rs @@ -0,0 +1,8 @@ +mod app; +mod events; +mod state; +mod ui; + +pub use app::TuiApp; +pub use events::{AgentEvent, EventBus}; +pub use state::AgentState; diff --git a/src/tui/state.rs b/src/tui/state.rs new file mode 100644 index 0000000..4e35378 --- /dev/null +++ b/src/tui/state.rs @@ -0,0 +1,166 @@ +use std::collections::VecDeque; +use std::time::Instant; + +use super::events::AgentEvent; + +#[derive(Clone)] +pub struct AgentState { + // Connection + pub connected: bool, + pub agent_id: String, + + // System info + pub hostname: String, + pub version: String, + pub started_at: Instant, + + // Active jobs + pub active_pollers: usize, + pub active_devices: Vec, + + // Event log (ring buffer, last 100 events) + pub recent_events: VecDeque<(Instant, String)>, + + // Statistics + pub stats: Stats, +} + +#[derive(Clone, Default)] +pub struct Stats { + pub jobs_received: u64, + pub snmp_results_sent: u64, + pub mikrotik_results_sent: u64, + pub heartbeats_sent: u64, + pub errors: u64, + pub last_heartbeat_at: Option, +} + +impl AgentState { + pub fn new(hostname: String, version: String) -> Self { + Self { + connected: false, + agent_id: String::new(), + hostname, + version, + started_at: Instant::now(), + active_pollers: 0, + active_devices: Vec::new(), + recent_events: VecDeque::with_capacity(100), + stats: Stats::default(), + } + } + + pub fn apply_event(&mut self, event: &AgentEvent) { + let now = Instant::now(); + let event_msg = match event { + AgentEvent::Connected { agent_id } => { + self.connected = true; + self.agent_id = agent_id.clone(); + format!("Connected to server (agent_id: {})", agent_id) + } + AgentEvent::Disconnected => { + self.connected = false; + "Disconnected from server".to_string() + } + AgentEvent::JobReceived { + job_id, + device_id, + job_type, + } => { + self.stats.jobs_received += 1; + format!("Job received: {} ({} - {})", job_id, device_id, job_type) + } + AgentEvent::JobCompleted { + job_id, + device_id, + duration_ms, + } => { + format!( + "Job completed: {} ({}) in {}ms", + job_id, device_id, duration_ms + ) + } + AgentEvent::JobFailed { + job_id, + device_id, + error, + } => { + self.stats.errors += 1; + format!("Job failed: {} ({}) - {}", job_id, device_id, error) + } + AgentEvent::SnmpResultSent { + device_id, + oid_count, + } => { + self.stats.snmp_results_sent += 1; + format!("SNMP result sent for {} ({} OIDs)", device_id, oid_count) + } + AgentEvent::MikrotikResultSent { + device_id, + sentence_count, + } => { + self.stats.mikrotik_results_sent += 1; + format!( + "MikroTik result sent for {} ({} sentences)", + device_id, sentence_count + ) + } + AgentEvent::PollerCreated { + device_id, + total_count, + } => { + self.active_pollers = *total_count; + if !self.active_devices.contains(device_id) { + self.active_devices.push(device_id.clone()); + } + format!( + "Poller created for {} ({} total pollers)", + device_id, total_count + ) + } + AgentEvent::PollerRemoved { + device_id, + total_count, + } => { + self.active_pollers = *total_count; + self.active_devices.retain(|d| d != device_id); + format!( + "Poller removed for {} ({} total pollers)", + device_id, total_count + ) + } + AgentEvent::HeartbeatSent => { + self.stats.heartbeats_sent += 1; + self.stats.last_heartbeat_at = Some(now); + "Heartbeat sent".to_string() + } + AgentEvent::PhxHeartbeatSent => "Phoenix heartbeat sent".to_string(), + AgentEvent::Error { message } => { + self.stats.errors += 1; + format!("Error: {}", message) + } + }; + + // Add event to ring buffer (keep last 100) + if self.recent_events.len() >= 100 { + self.recent_events.pop_front(); + } + self.recent_events.push_back((now, event_msg)); + } + + pub fn uptime_seconds(&self) -> u64 { + self.started_at.elapsed().as_secs() + } + + pub fn format_uptime(&self) -> String { + let seconds = self.uptime_seconds(); + let hours = seconds / 3600; + let minutes = (seconds % 3600) / 60; + let secs = seconds % 60; + format!("{}h {}m {}s", hours, minutes, secs) + } + + pub fn last_heartbeat_ago(&self) -> Option { + self.stats.last_heartbeat_at.map(|t| t.elapsed().as_secs()) + } +} diff --git a/src/tui/ui.rs b/src/tui/ui.rs new file mode 100644 index 0000000..091b475 --- /dev/null +++ b/src/tui/ui.rs @@ -0,0 +1,194 @@ +use ratatui::{ + layout::{Constraint, Direction, Layout, Rect}, + style::{Color, Modifier, Style}, + text::{Line, Span}, + widgets::{Block, Borders, List, ListItem, Paragraph}, + Frame, +}; + +use super::state::AgentState; + +pub fn render(f: &mut Frame, state: &AgentState, scroll_offset: usize) { + let chunks = Layout::default() + .direction(Direction::Vertical) + .constraints([ + Constraint::Length(3), // Header + Constraint::Length(3), // Stats + Constraint::Length(8), // Devices + Constraint::Min(0), // Events log + ]) + .split(f.area()); + + render_header(f, chunks[0], state); + render_stats(f, chunks[1], state); + render_devices(f, chunks[2], state); + render_events(f, chunks[3], state, scroll_offset); +} + +fn render_header(f: &mut Frame, area: Rect, state: &AgentState) { + let status_color = if state.connected { + Color::Green + } else { + Color::Red + }; + + let status_text = if state.connected { + "Connected" + } else { + "Disconnected" + }; + + let version_str = format!(" | v{}", state.version); + let uptime_str = format!(" | Uptime: {}", state.format_uptime()); + + let header_lines = vec![ + Line::from(vec![ + Span::raw("Status: "), + Span::styled( + status_text, + Style::default() + .fg(status_color) + .add_modifier(Modifier::BOLD), + ), + Span::raw(" | Agent ID: "), + Span::styled(&state.agent_id, Style::default().fg(Color::Cyan)), + Span::raw(version_str), + ]), + Line::from(vec![ + Span::raw("Hostname: "), + Span::styled(&state.hostname, Style::default().fg(Color::Yellow)), + Span::raw(uptime_str), + ]), + ]; + + let header = Paragraph::new(header_lines).block( + Block::default() + .title("Towerops Agent") + .borders(Borders::ALL), + ); + + f.render_widget(header, area); +} + +fn render_stats(f: &mut Frame, area: Rect, state: &AgentState) { + let heartbeat_ago = state + .last_heartbeat_ago() + .map(|s| format!("{}s ago", s)) + .unwrap_or_else(|| "never".to_string()); + + let heartbeat_str = format!(" | Last Heartbeat: {}", heartbeat_ago); + + let stats_lines = vec![ + Line::from(vec![ + Span::raw("Jobs: "), + Span::styled( + format!("{}", state.stats.jobs_received), + Style::default().fg(Color::Cyan), + ), + Span::raw(" | SNMP: "), + Span::styled( + format!("{}", state.stats.snmp_results_sent), + Style::default().fg(Color::Green), + ), + Span::raw(" | MikroTik: "), + Span::styled( + format!("{}", state.stats.mikrotik_results_sent), + Style::default().fg(Color::Green), + ), + Span::raw(" | Errors: "), + Span::styled( + format!("{}", state.stats.errors), + Style::default().fg(if state.stats.errors > 0 { + Color::Red + } else { + Color::White + }), + ), + ]), + Line::from(vec![ + Span::raw("Active Pollers: "), + Span::styled( + format!("{}", state.active_pollers), + Style::default().fg(Color::Yellow), + ), + Span::raw(heartbeat_str), + ]), + ]; + + let stats = Paragraph::new(stats_lines) + .block(Block::default().title("Statistics").borders(Borders::ALL)); + + f.render_widget(stats, area); +} + +fn render_devices(f: &mut Frame, area: Rect, state: &AgentState) { + let device_items: Vec = state + .active_devices + .iter() + .take(10) + .map(|device| { + ListItem::new(Line::from(vec![ + Span::raw("• "), + Span::styled(device, Style::default().fg(Color::Cyan)), + ])) + }) + .collect(); + + let more_count = state.active_devices.len().saturating_sub(10); + let title = if more_count > 0 { + format!("Active Devices ({} more)", more_count) + } else { + "Active Devices".to_string() + }; + + let devices = + List::new(device_items).block(Block::default().title(title).borders(Borders::ALL)); + + f.render_widget(devices, area); +} + +fn render_events(f: &mut Frame, area: Rect, state: &AgentState, scroll_offset: usize) { + let event_items: Vec = state + .recent_events + .iter() + .rev() + .skip(scroll_offset) + .take(area.height.saturating_sub(2) as usize) + .map(|(timestamp, message)| { + let elapsed = timestamp.elapsed().as_secs(); + let time_str = if elapsed < 60 { + format!("{}s ago", elapsed) + } else if elapsed < 3600 { + format!("{}m ago", elapsed / 60) + } else { + format!("{}h ago", elapsed / 3600) + }; + + let color = if message.contains("Error") || message.contains("failed") { + Color::Red + } else if message.contains("sent") || message.contains("completed") { + Color::Green + } else if message.contains("received") { + Color::Yellow + } else { + Color::White + }; + + ListItem::new(Line::from(vec![ + Span::styled( + format!("[{}] ", time_str), + Style::default().fg(Color::DarkGray), + ), + Span::styled(message, Style::default().fg(color)), + ])) + }) + .collect(); + + let events = List::new(event_items).block( + Block::default() + .title("Recent Events (Press 'q' to quit, ↑/↓ to scroll)") + .borders(Borders::ALL), + ); + + f.render_widget(events, area); +} diff --git a/src/websocket_client.rs b/src/websocket_client.rs index 78efba9..7b81cc8 100644 --- a/src/websocket_client.rs +++ b/src/websocket_client.rs @@ -67,6 +67,9 @@ pub struct AgentClient { phx_heartbeat_ref: u64, /// Cached hostname (computed once at startup, avoids blocking /proc reads) cached_hostname: String, + /// Optional event bus for TUI updates + #[cfg(feature = "tui")] + event_bus: Option, } impl AgentClient { @@ -75,12 +78,121 @@ impl AgentClient { /// # Arguments /// * `url` - Server URL (e.g., "wss://towerops.net") /// * `token` - Agent authentication token + /// * `event_bus` - Optional event bus for TUI updates /// /// # Example /// ```no_run - /// let client = AgentClient::connect("wss://towerops.net", "token123").await?; + /// let client = AgentClient::connect("wss://towerops.net", "token123", None).await?; /// ``` + #[cfg(feature = "tui")] + pub async fn connect( + url: &str, + token: &SecretString, + event_bus: Option, + ) -> Result { + Self::connect_internal(url, token, event_bus).await + } + + #[cfg(not(feature = "tui"))] pub async fn connect(url: &str, token: &SecretString) -> Result { + Self::connect_internal(url, token).await + } + + #[cfg(feature = "tui")] + async fn connect_internal( + url: &str, + token: &SecretString, + event_bus: Option, + ) -> Result { + // Strip trailing slash from base URL to avoid double slashes + let base_url = url.trim_end_matches('/'); + let ws_url = format!("{}/socket/agent/websocket", base_url); + tracing::info!( + "Connecting to WebSocket: {} (timeout: {}s)", + ws_url, + CONNECTION_TIMEOUT.as_secs() + ); + + // Wrap connection in timeout to avoid hanging indefinitely on bad network + let (ws_stream, _) = match timeout(CONNECTION_TIMEOUT, connect_async(&ws_url)).await { + Ok(Ok(result)) => result, + Ok(Err(e)) => { + tracing::error!("WebSocket connection failed: {}", e); + return Err(format!("Failed to connect to WebSocket: {}", e).into()); + } + Err(_) => { + tracing::error!( + "WebSocket connection timed out after {}s", + CONNECTION_TIMEOUT.as_secs() + ); + return Err(format!( + "Connection timed out after {}s", + CONNECTION_TIMEOUT.as_secs() + ) + .into()); + } + }; + + tracing::info!("Connected to Towerops server at {}", url); + + let agent_id = generate_agent_id(); + let (result_tx, result_rx) = mpsc::channel(RESULT_CHANNEL_CAPACITY); + let (mikrotik_result_tx, mikrotik_result_rx) = mpsc::channel(RESULT_CHANNEL_CAPACITY); + let (credential_test_tx, credential_test_rx) = mpsc::channel(RESULT_CHANNEL_CAPACITY); + + // Split the WebSocket stream so reads and writes can proceed concurrently. + // The write half is owned by a dedicated writer task. + let (ws_write, ws_read) = ws_stream.split(); + let (ws_write_tx, ws_write_rx) = mpsc::channel::(WS_WRITE_CHANNEL_CAPACITY); + + tokio::spawn(ws_writer_task(ws_write, ws_write_rx)); + + // Join Phoenix channel with token in payload + let join_msg = PhoenixMessage { + topic: format!("agent:{}", agent_id), + event: "phx_join".to_string(), + payload: serde_json::json!({"token": token.expose()}), + reference: Some("1".to_string()), + }; + + let join_text = serde_json::to_string(&join_msg)?; + ws_write_tx + .send(WsMessage::Text(join_text.into())) + .await + .map_err(|e| format!("Failed to send join message: {}", e))?; + tracing::info!( + "Sent channel join request with token for agent:{}", + agent_id + ); + + let client = Self { + ws_read, + ws_write_tx, + agent_id: agent_id.clone(), + result_tx, + result_rx, + mikrotik_result_tx, + mikrotik_result_rx, + credential_test_tx, + credential_test_rx, + poller_registry: PollerRegistry::new(), + phx_heartbeat_ref: 0, + cached_hostname: get_hostname(), + #[cfg(feature = "tui")] + event_bus: event_bus.clone(), + }; + + // Publish connected event + #[cfg(feature = "tui")] + client.publish_event(crate::tui::AgentEvent::Connected { + agent_id: agent_id.clone(), + }); + + Ok(client) + } + + #[cfg(not(feature = "tui"))] + async fn connect_internal(url: &str, token: &SecretString) -> Result { // Strip trailing slash from base URL to avoid double slashes let base_url = url.trim_end_matches('/'); let ws_url = format!("{}/socket/agent/websocket", base_url); @@ -158,6 +270,19 @@ impl AgentClient { }) } + /// Publish an event to the TUI event bus (if enabled). + #[cfg(feature = "tui")] + fn publish_event(&self, event: crate::tui::AgentEvent) { + if let Some(bus) = &self.event_bus { + let _ = bus.send(event); // Fire-and-forget + } + } + + #[cfg(not(feature = "tui"))] + fn publish_event(&self, _event: ()) { + // No-op when TUI is disabled + } + /// Main event loop for agent operation. /// /// Handles: @@ -197,16 +322,22 @@ impl AgentClient { } Some(Ok(WsMessage::Close(_))) => { tracing::info!("Server closed connection"); + #[cfg(feature = "tui")] + self.publish_event(crate::tui::AgentEvent::Disconnected); self.poller_registry.shutdown_all(); break Ok(()); } Some(Err(e)) => { tracing::error!("WebSocket error: {}", e); + #[cfg(feature = "tui")] + self.publish_event(crate::tui::AgentEvent::Disconnected); self.poller_registry.shutdown_all(); break Err(e.into()); } None => { tracing::info!("Connection closed"); + #[cfg(feature = "tui")] + self.publish_event(crate::tui::AgentEvent::Disconnected); self.poller_registry.shutdown_all(); break Ok(()); } @@ -216,15 +347,29 @@ impl AgentClient { // Receive SNMP results from job tasks Some(snmp_result) = self.result_rx.recv() => { + #[cfg(feature = "tui")] + let device_id = snmp_result.device_id.clone(); + if let Err(e) = self.send_snmp_result(snmp_result).await { tracing::error!("Error sending SNMP result: {}", e); + #[cfg(feature = "tui")] + self.publish_event(crate::tui::AgentEvent::Error { + message: format!("SNMP result send failed for {}: {}", device_id, e), + }); } } // Receive MikroTik results from job tasks Some(mikrotik_result) = self.mikrotik_result_rx.recv() => { + #[cfg(feature = "tui")] + let device_id = mikrotik_result.device_id.clone(); + if let Err(e) = self.send_mikrotik_result(mikrotik_result).await { tracing::error!("Error sending MikroTik result: {}", e); + #[cfg(feature = "tui")] + self.publish_event(crate::tui::AgentEvent::Error { + message: format!("MikroTik result send failed for {}: {}", device_id, e), + }); } } @@ -239,6 +384,13 @@ impl AgentClient { _ = heartbeat_interval.tick() => { if let Err(e) = self.send_heartbeat().await { tracing::error!("Error sending heartbeat: {}", e); + #[cfg(feature = "tui")] + self.publish_event(crate::tui::AgentEvent::Error { + message: format!("Heartbeat failed: {}", e), + }); + } else { + #[cfg(feature = "tui")] + self.publish_event(crate::tui::AgentEvent::HeartbeatSent); } // Log active poller count let count = self.poller_registry.count(); @@ -251,6 +403,9 @@ impl AgentClient { _ = phx_heartbeat_interval.tick() => { if let Err(e) = self.send_phx_heartbeat().await { tracing::error!("Error sending Phoenix heartbeat: {}", e); + } else { + #[cfg(feature = "tui")] + self.publish_event(crate::tui::AgentEvent::PhxHeartbeatSent); } } } @@ -317,6 +472,14 @@ impl AgentClient { device_id ); self.poller_registry.remove(&device_id); + + #[cfg(feature = "tui")] + if let Some(ref bus) = self.event_bus { + let _ = bus.send(crate::tui::AgentEvent::PollerRemoved { + device_id: device_id.clone(), + total_count: self.poller_registry.count(), + }); + } } } @@ -324,22 +487,85 @@ impl AgentClient { let job_type = JobType::try_from(job.job_type).unwrap_or(JobType::Poll); tracing::info!("Executing job: {} (type: {:?})", job.job_id, job_type); + #[cfg(feature = "tui")] + { + if let Some(ref bus) = self.event_bus { + let _ = bus.send(crate::tui::AgentEvent::JobReceived { + job_id: job.job_id.clone(), + device_id: job.device_id.clone(), + job_type: format!("{:?}", job_type), + }); + } + } + match job_type { JobType::Mikrotik => { // Execute MikroTik API job let mikrotik_result_tx = self.mikrotik_result_tx.clone(); + #[cfg(feature = "tui")] + let event_bus = self.event_bus.clone(); + let job_id = job.job_id.clone(); + let device_id = job.device_id.clone(); + tokio::spawn(async move { - if let Err(e) = execute_mikrotik_job(job, mikrotik_result_tx).await { - tracing::error!("MikroTik job execution failed: {}", e); + let start_time = std::time::Instant::now(); + match execute_mikrotik_job(job, mikrotik_result_tx).await { + Ok(_) => { + #[cfg(feature = "tui")] + if let Some(ref bus) = event_bus { + let _ = bus.send(crate::tui::AgentEvent::JobCompleted { + job_id, + device_id, + duration_ms: start_time.elapsed().as_millis() as u64, + }); + } + } + Err(e) => { + tracing::error!("MikroTik job execution failed: {}", e); + #[cfg(feature = "tui")] + if let Some(ref bus) = event_bus { + let _ = bus.send(crate::tui::AgentEvent::JobFailed { + job_id, + device_id, + error: e.to_string(), + }); + } + } } }); } JobType::TestCredentials => { // Execute credential test let credential_test_tx = self.credential_test_tx.clone(); + #[cfg(feature = "tui")] + let event_bus = self.event_bus.clone(); + let job_id = job.job_id.clone(); + let device_id = job.device_id.clone(); + tokio::spawn(async move { - if let Err(e) = execute_credential_test(job, credential_test_tx).await { - tracing::error!("Credential test execution failed: {}", e); + let start_time = std::time::Instant::now(); + match execute_credential_test(job, credential_test_tx).await { + Ok(_) => { + #[cfg(feature = "tui")] + if let Some(ref bus) = event_bus { + let _ = bus.send(crate::tui::AgentEvent::JobCompleted { + job_id, + device_id, + duration_ms: start_time.elapsed().as_millis() as u64, + }); + } + } + Err(e) => { + tracing::error!("Credential test execution failed: {}", e); + #[cfg(feature = "tui")] + if let Some(ref bus) = event_bus { + let _ = bus.send(crate::tui::AgentEvent::JobFailed { + job_id, + device_id, + error: e.to_string(), + }); + } + } } }); } @@ -347,9 +573,42 @@ impl AgentClient { // Execute SNMP job (discovery or polling) let result_tx = self.result_tx.clone(); let poller_registry = self.poller_registry.clone(); + #[cfg(feature = "tui")] + let event_bus = self.event_bus.clone(); + #[cfg(feature = "tui")] + let event_bus_for_task = event_bus.clone(); + let job_id = job.job_id.clone(); + let device_id = job.device_id.clone(); + tokio::spawn(async move { - if let Err(e) = execute_snmp_job(job, result_tx, poller_registry).await { - tracing::error!("SNMP job execution failed: {}", e); + let start_time = std::time::Instant::now(); + #[cfg(feature = "tui")] + let result = execute_snmp_job(job, result_tx, poller_registry, event_bus_for_task).await; + #[cfg(not(feature = "tui"))] + let result = execute_snmp_job(job, result_tx, poller_registry).await; + + match result { + Ok(_) => { + #[cfg(feature = "tui")] + if let Some(ref bus) = event_bus { + let _ = bus.send(crate::tui::AgentEvent::JobCompleted { + job_id, + device_id, + duration_ms: start_time.elapsed().as_millis() as u64, + }); + } + } + Err(e) => { + tracing::error!("SNMP job execution failed: {}", e); + #[cfg(feature = "tui")] + if let Some(ref bus) = event_bus { + let _ = bus.send(crate::tui::AgentEvent::JobFailed { + job_id, + device_id, + error: e.to_string(), + }); + } + } } }); } @@ -365,7 +624,7 @@ impl AgentClient { version: env!("CARGO_PKG_VERSION").to_string(), hostname: self.cached_hostname.clone(), uptime_seconds: get_uptime_seconds(), - ip_address: get_local_ip().unwrap_or_else(|| "".to_string()), + ip_address: get_local_ip().unwrap_or_default(), }; let binary = heartbeat.encode_to_vec(); @@ -433,6 +692,13 @@ impl AgentClient { .map_err(|e| format!("Writer task closed: {}", e))?; tracing::debug!("Sent SNMP result for device {}", result.device_id); + + #[cfg(feature = "tui")] + self.publish_event(crate::tui::AgentEvent::SnmpResultSent { + device_id: result.device_id.clone(), + oid_count: result.oid_values.len(), + }); + Ok(()) } @@ -458,6 +724,13 @@ impl AgentClient { result.device_id, result.job_id ); + + #[cfg(feature = "tui")] + self.publish_event(crate::tui::AgentEvent::MikrotikResultSent { + device_id: result.device_id.clone(), + sentence_count: result.sentences.len(), + }); + Ok(()) } @@ -522,6 +795,7 @@ async fn execute_snmp_job( job: AgentJob, result_tx: mpsc::Sender, poller_registry: PollerRegistry, + #[cfg(feature = "tui")] event_bus: Option, ) -> Result<()> { let mut snmp_device = job.snmp_device.ok_or("Job missing SNMP device info")?; @@ -588,8 +862,24 @@ async fn execute_snmp_job( snmp_device.v3_auth_password.zeroize(); snmp_device.v3_priv_password.zeroize(); + #[cfg(feature = "tui")] + let count_before = poller_registry.count(); + let poller = poller_registry.get_or_create(job.device_id.clone(), device_config); + #[cfg(feature = "tui")] + { + let count_after = poller_registry.count(); + if count_after > count_before { + if let Some(ref bus) = event_bus { + let _ = bus.send(crate::tui::AgentEvent::PollerCreated { + device_id: job.device_id.clone(), + total_count: count_after, + }); + } + } + } + let mut oid_values: HashMap = HashMap::new(); for query in job.queries {