From 4508f317be424fcf1b30f3ba3e1fc5fe0363df69 Mon Sep 17 00:00:00 2001 From: Stephen Waits Date: Sun, 5 Apr 2026 12:21:27 -0600 Subject: [PATCH] feat: replace direct P2P networking with iroh-gossip swarm and DHT peer discovery Replace manually configured peer connections with a passphrase-based gossip swarm. Peers sharing the same passphrase automatically discover each other via mainline DHT and join an iroh-gossip topic. Key changes: - Add gossip.rs: iroh-gossip swarm with DHT bootstrap, heartbeat, and graceful shutdown that publishes neighbor list - Add setup.rs: interactive config wizard replacing generate-config - Add rate_limit.rs: token-bucket TX rate limiter derived from LoRa air time - Remove network.rs: old direct P2P connection management - Config: remove [[peers]] section, add passphrase to [bridge], update radio defaults to MeshCore US/Canada recommended values - Packet: rename Envelope to GossipFrame, simplify wire format (drop version byte and seq number) - Router: unify dedup into single content-hash cache, add rate limiting - Identity: use ephemeral keypair per launch instead of persistent secret key - Radio: improve config negotiation (GetConfig-first strategy), add device info tracking - TUI: complete redesign with network/swarm panel, sparkline activity charts, color-coded RSSI/SNR, and shutdown overlay --- Cargo.lock | 196 +++++++++++- Cargo.toml | 4 + src/config.rs | 151 ++-------- src/gossip.rs | 612 +++++++++++++++++++++++++++++++++++++ src/main.rs | 138 ++++----- src/network.rs | 411 ------------------------- src/packet.rs | 114 ++++--- src/radio.rs | 181 +++++++---- src/rate_limit.rs | 174 +++++++++++ src/router.rs | 341 +++++++++------------ src/setup.rs | 177 +++++++++++ src/tui.rs | 755 ++++++++++++++++++++++++++++------------------ 12 files changed, 2025 insertions(+), 1229 deletions(-) create mode 100644 src/gossip.rs delete mode 100644 src/network.rs create mode 100644 src/rate_limit.rs create mode 100644 src/setup.rs diff --git a/Cargo.lock b/Cargo.lock index 995986d..135e7c9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -239,6 +239,15 @@ dependencies = [ "cpufeatures 0.3.0", ] +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + [[package]] name = "block-buffer" version = "0.11.0" @@ -274,6 +283,9 @@ name = "bytes" version = "1.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" +dependencies = [ + "serde", +] [[package]] name = "cassowary" @@ -468,6 +480,21 @@ dependencies = [ "libc", ] +[[package]] +name = "crc" +version = "3.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5eb8a2a1cd12ab0d987a5d5e825195d372001a4094a0376319d5a0ad71c1ba0d" +dependencies = [ + "crc-catalog", +] + +[[package]] +name = "crc-catalog" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "19d374276b40fb8bbdee95aef7c7fa6b5316ec764510eb64b8dd0e2ed0d7e7f5" + [[package]] name = "critical-section" version = "1.2.0" @@ -560,7 +587,7 @@ dependencies = [ "cfg-if", "cpufeatures 0.2.17", "curve25519-dalek-derive", - "digest", + "digest 0.11.0-rc.10", "fiat-crypto", "rand_core", "rustc_version", @@ -735,13 +762,23 @@ version = "0.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ab03c107fafeb3ee9f5925686dbb7a73bc76e3932abb0d2b365cb64b169cf04c" +[[package]] +name = "digest" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer 0.10.4", + "crypto-common 0.1.7", +] + [[package]] name = "digest" version = "0.11.0-rc.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "afa94b64bfc6549e6e4b5a3216f22593224174083da7a90db47e951c4fb31725" dependencies = [ - "block-buffer", + "block-buffer 0.11.0", "const-oid", "crypto-common 0.2.1", ] @@ -821,11 +858,15 @@ dependencies = [ "crossterm", "dirs", "donglora-client", + "futures-lite", "hex", "iroh", + "iroh-gossip", + "mainline", "rand", "ratatui", "serde", + "sha2 0.10.9", "tokio", "tokio-util", "toml", @@ -844,6 +885,12 @@ dependencies = [ "ucobs", ] +[[package]] +name = "dyn-clone" +version = "1.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" + [[package]] name = "ed25519" version = "3.0.0-rc.4" @@ -865,7 +912,7 @@ dependencies = [ "ed25519", "rand_core", "serde", - "sha2", + "sha2 0.11.0-rc.2", "signature", "subtle", "zeroize", @@ -958,6 +1005,23 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +[[package]] +name = "fixedbitset" +version = "0.5.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99" + +[[package]] +name = "flume" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" +dependencies = [ + "futures-core", + "futures-sink", + "spin 0.9.8", +] + [[package]] name = "fnv" version = "1.0.7" @@ -1023,6 +1087,19 @@ dependencies = [ "futures-sink", ] +[[package]] +name = "futures-concurrency" +version = "7.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "175cd8cca9e1d45b87f18ffa75088f2099e3c4fe5e2f83e42de112560bea8ea6" +dependencies = [ + "fixedbitset", + "futures-core", + "futures-lite", + "pin-project", + "smallvec", +] + [[package]] name = "futures-core" version = "0.3.32" @@ -1743,17 +1820,47 @@ dependencies = [ "curve25519-dalek", "data-encoding", "derive_more", - "digest", + "digest 0.11.0-rc.10", "ed25519-dalek", "n0-error", "rand_core", "serde", - "sha2", + "sha2 0.11.0-rc.2", "url", "zeroize", "zeroize_derive", ] +[[package]] +name = "iroh-gossip" +version = "0.97.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4db5b64f3cb0a0c8b68b57888acd4cefcd2f0774f1a132d2a498cbb2a92fbc55" +dependencies = [ + "blake3", + "bytes", + "data-encoding", + "derive_more", + "ed25519-dalek", + "futures-concurrency", + "futures-lite", + "futures-util", + "hex", + "indexmap", + "iroh", + "iroh-base", + "iroh-metrics", + "irpc", + "n0-error", + "n0-future", + "postcard", + "rand", + "serde", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "iroh-metrics" version = "0.38.3" @@ -1829,6 +1936,33 @@ dependencies = [ "z32", ] +[[package]] +name = "irpc" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4f47b7c52662d673df377b5ac40c121c7ff56eb764e520fae6543686132f7957" +dependencies = [ + "futures-util", + "irpc-derive", + "n0-error", + "n0-future", + "serde", + "tokio", + "tokio-util", + "tracing", +] + +[[package]] +name = "irpc-derive" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "83c1a4b460634aeed6dc01236a0047867de70e30562d91a0ad031dcb3ac33fb4" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "is_terminal_polyfill" version = "1.70.2" @@ -2000,6 +2134,28 @@ dependencies = [ "libc", ] +[[package]] +name = "mainline" +version = "6.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ff27d378ca495eaf3be8616d5d7319c1c18e93fd60e13698fcdc7e19448f1a4" +dependencies = [ + "crc", + "document-features", + "dyn-clone", + "ed25519-dalek", + "flume", + "futures-lite", + "getrandom 0.3.4", + "lru 0.16.3", + "serde", + "serde_bencode", + "serde_bytes", + "sha1_smol", + "thiserror", + "tracing", +] + [[package]] name = "matchers" version = "0.2.0" @@ -2050,6 +2206,7 @@ version = "0.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "af4782b4baf92d686d161c15460c83d16ebcfd215918763903e9619842665cae" dependencies = [ + "anyhow", "n0-error-macros", "spez", ] @@ -3057,6 +3214,16 @@ dependencies = [ "serde_derive", ] +[[package]] +name = "serde_bencode" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a70dfc7b7438b99896e7f8992363ab8e2c4ba26aa5ec675d32d1c3c2c33d413e" +dependencies = [ + "serde", + "serde_bytes", +] + [[package]] name = "serde_bytes" version = "0.11.19" @@ -3140,6 +3307,23 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "sha1_smol" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbfa15b3dddfee50a0fff136974b3e1bde555604ba463834a7eb7deb6417705d" + +[[package]] +name = "sha2" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "digest 0.10.7", +] + [[package]] name = "sha2" version = "0.11.0-rc.2" @@ -3148,7 +3332,7 @@ checksum = "d1e3878ab0f98e35b2df35fe53201d088299b41a6bb63e3e34dada2ac4abd924" dependencies = [ "cfg-if", "cpufeatures 0.2.17", - "digest", + "digest 0.11.0-rc.10", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index a6c8b2a..82c45ee 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,6 +9,8 @@ description = "Peer-to-peer LoRa bridge using iroh and DongLoRa" [dependencies] donglora-client = { path = "../donglora/clients/rust" } iroh = "0.97" +iroh-gossip = "0.97" +mainline = { version = "6", features = ["async"] } tokio = { version = "1", features = ["full"] } clap = { version = "4", features = ["derive"] } serde = { version = "1", features = ["derive"] } @@ -22,8 +24,10 @@ tracing-appender = "0.2" anyhow = "1" dirs = "6" bytes = "1" +futures-lite = "2" rand = "0.9" hex = "0.4.3" +sha2 = "0.10" tokio-util = { version = "0.7.18", features = ["rt"] } [lints.clippy] diff --git a/src/config.rs b/src/config.rs index c7c486a..fdb1934 100644 --- a/src/config.rs +++ b/src/config.rs @@ -2,7 +2,6 @@ use std::path::{Path, PathBuf}; use anyhow::{Context, Result, bail}; use donglora_client::{Bandwidth, RadioConfig, TX_POWER_MAX}; -use iroh::{PublicKey, SecretKey}; use serde::{Deserialize, Serialize}; /// Top-level bridge configuration. @@ -12,11 +11,9 @@ pub struct Config { pub radio: RadioSection, #[serde(default)] pub bridge: BridgeSection, - #[serde(default)] - pub peers: Vec, } -#[derive(Debug, Deserialize, Serialize)] +#[derive(Debug, Clone, Deserialize, Serialize)] pub struct RadioSection { /// Serial port path (auto-detect if omitted). pub port: Option, @@ -32,36 +29,34 @@ pub struct RadioSection { pub sync_word: u16, #[serde(default = "default_tx_power")] pub tx_power: String, - #[serde(default)] + #[serde(default = "default_preamble")] pub preamble: u16, #[serde(default = "default_true")] pub cad: bool, } -#[derive(Debug, Deserialize, Serialize)] +#[derive(Debug, Clone, Deserialize, Serialize)] pub struct BridgeSection { + /// Shared passphrase. All bridges with the same passphrase join the same swarm. + #[serde(default)] + pub passphrase: String, #[serde(default = "default_dedup_window")] pub dedup_window_secs: u64, #[serde(default = "default_tx_queue_size")] pub tx_queue_size: usize, + /// Optional manual override for the radio TX rate limit (packets per second). + /// If omitted, a rate is calculated from the radio config. + pub rate_limit_pps: Option, pub log_file: Option, } -#[derive(Debug, Clone, Deserialize, Serialize)] -pub struct PeerEntry { - pub node_id: String, - pub name: Option, - #[serde(default)] - pub addresses: Vec, -} - -// ── Defaults ────────────────────────────────────────────────────── +// ── Defaults (MeshCore US/Canada Recommended) ──────────────────── fn default_frequency() -> u32 { - 915_000_000 + 910_525_000 } fn default_bandwidth() -> String { - "125kHz".into() + "62.5kHz".into() } fn default_sf() -> u8 { 7 @@ -70,16 +65,19 @@ fn default_cr() -> u8 { 5 } fn default_sync_word() -> u16 { - 0x1424 + 0x3444 } fn default_tx_power() -> String { - "max".into() + "22".into() +} +fn default_preamble() -> u16 { + 16 } fn default_true() -> bool { true } fn default_dedup_window() -> u64 { - 30 + 300 } fn default_tx_queue_size() -> usize { 32 @@ -95,7 +93,7 @@ impl Default for RadioSection { coding_rate: default_cr(), sync_word: default_sync_word(), tx_power: default_tx_power(), - preamble: 0, + preamble: default_preamble(), cad: true, } } @@ -104,8 +102,10 @@ impl Default for RadioSection { impl Default for BridgeSection { fn default() -> Self { Self { + passphrase: String::new(), dedup_window_secs: default_dedup_window(), tx_queue_size: default_tx_queue_size(), + rate_limit_pps: None, log_file: None, } } @@ -153,31 +153,6 @@ impl RadioSection { } } -// ── Peer parsing ────────────────────────────────────────────────── - -impl PeerEntry { - /// Parse the node_id string into an iroh PublicKey. - pub fn public_key(&self) -> Result { - self.node_id - .parse::() - .map_err(|e| anyhow::anyhow!("invalid node_id '{}': {e}", self.node_id)) - } - - /// Display name: the configured name, or a truncated NodeId. - pub fn display_name(&self) -> String { - if let Some(name) = &self.name { - name.clone() - } else { - let id = &self.node_id; - if id.len() > 12 { - format!("{}...", &id[..12]) - } else { - id.clone() - } - } - } -} - // ── File I/O ────────────────────────────────────────────────────── /// Default config directory: `$XDG_CONFIG_HOME/donglora-bridge/` or `~/.config/donglora-bridge/`. @@ -193,12 +168,6 @@ pub fn config_path() -> Result { Ok(config_dir()?.join("config.toml")) } -/// Default secret key file path (alongside default config). -#[allow(dead_code)] -pub fn secret_key_path() -> Result { - Ok(config_dir()?.join("config.key")) -} - /// Default log file path: `$XDG_STATE_HOME/donglora-bridge/bridge.log`. pub fn default_log_path() -> Result { let dir = dirs::state_dir() @@ -208,79 +177,15 @@ pub fn default_log_path() -> Result { Ok(dir.join("bridge.log")) } -/// Load config from a TOML file. +/// Load config from a TOML file. Returns None-passphrase error if not set. pub fn load_config(path: &Path) -> Result { let text = std::fs::read_to_string(path) .with_context(|| format!("reading config: {}", path.display()))?; - toml::from_str(&text).context("parsing config TOML") -} - -/// Load or generate the persistent secret key. -pub fn load_or_create_secret_key(path: &Path) -> Result { - if path.exists() { - let hex = std::fs::read_to_string(path) - .with_context(|| format!("reading secret key: {}", path.display()))?; - let hex = hex.trim(); - let bytes = hex::decode(hex).context("decoding secret key hex")?; - let arr: [u8; 32] = bytes - .try_into() - .map_err(|_| anyhow::anyhow!("secret key must be 32 bytes"))?; - Ok(SecretKey::from_bytes(&arr)) - } else { - let key = SecretKey::generate(&mut rand::rng()); - if let Some(parent) = path.parent() { - std::fs::create_dir_all(parent) - .with_context(|| format!("creating directory: {}", parent.display()))?; - } - let hex = hex::encode(key.to_bytes()); - std::fs::write(path, &hex) - .with_context(|| format!("writing secret key: {}", path.display()))?; - Ok(key) + let cfg: Config = toml::from_str(&text).context("parsing config TOML")?; + if cfg.bridge.passphrase.is_empty() || cfg.bridge.passphrase == "change-me" { + bail!( + "passphrase is required — run `donglora-bridge config` to set it" + ); } -} - -/// Generate a default config file with the given NodeId embedded as a comment. -pub fn generate_default_config(node_id: &str) -> String { - format!( - r#"# donglora-bridge configuration -# Auto-generated — add peers to get started! -# Your NodeId: {node_id} - -[radio] -# port = "/dev/ttyACM0" # auto-detect if omitted -frequency = 915_000_000 -bandwidth = "125kHz" -spreading_factor = 7 -coding_rate = 5 -sync_word = 0x1424 -tx_power = "max" -preamble = 0 # 0 = firmware default (16 symbols) -cad = true - -[bridge] -dedup_window_secs = 30 -tx_queue_size = 32 -# log_file = "/tmp/donglora-bridge.log" - -# Add peers below. Each peer needs the remote bridge's NodeId. -# Share your NodeId (above) with peers so they can add you too. -# -# [[peers]] -# node_id = "paste-node-id-here" -# name = "My Friend's Bridge" -# addresses = ["1.2.3.4:1234"] # optional direct address hints -"# - ) -} - -/// Write the default config to disk (first-run flow). -pub fn write_default_config(path: &Path, node_id: &str) -> Result<()> { - if let Some(parent) = path.parent() { - std::fs::create_dir_all(parent) - .with_context(|| format!("creating directory: {}", parent.display()))?; - } - let content = generate_default_config(node_id); - std::fs::write(path, content) - .with_context(|| format!("writing config: {}", path.display()))?; - Ok(()) + Ok(cfg) } diff --git a/src/gossip.rs b/src/gossip.rs new file mode 100644 index 0000000..2beb654 --- /dev/null +++ b/src/gossip.rs @@ -0,0 +1,612 @@ +use std::collections::HashSet; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use anyhow::{Context, Result}; +use bytes::Bytes; +use iroh::endpoint::Incoming; +use iroh::{Endpoint, PublicKey, SecretKey}; +use futures_lite::StreamExt; +use iroh_gossip::api::Event; +use iroh_gossip::net::{Gossip as GossipActor, GOSSIP_ALPN}; +use iroh_gossip::TopicId; +use mainline::async_dht::AsyncDht; +use mainline::{Dht, MutableItem, SigningKey}; +use sha2::{Digest, Sha256}; +use tokio::sync::mpsc; +use tokio_util::sync::CancellationToken; +use tracing::{debug, info, warn}; + +use crate::packet::GossipFrame; + +/// Maximum number of peers to store in a DHT record. +/// 28 × 32 bytes = 896 bytes raw. Bencoded: 28 × ~36 = ~1008. +/// Conservative cap to stay under BEP44's 1000-byte value limit. +const MAX_DHT_PEERS: usize = 27; + +/// DHT heartbeat interval (base, before jitter). +const DHT_HEARTBEAT_SECS: u64 = 300; + +/// Maximum random jitter added to heartbeat interval. +const DHT_HEARTBEAT_JITTER_SECS: u64 = 60; + + +// ── Key derivation ─────────────────────────────────────────────── + +/// Cryptographic keys derived from a shared passphrase. +pub struct PassphraseKeys { + pub topic_id: TopicId, + pub dht_signing_key: SigningKey, + pub dht_salt: [u8; 20], +} + +impl PassphraseKeys { + /// Derive all keys from a passphrase. + pub fn derive(passphrase: &str) -> Self { + // Application-specific prefix to avoid DHT collisions with other apps. + const APP_PREFIX: &str = "donglora-bridge:"; + + // Topic ID: blake3 hash of prefixed passphrase. + let topic_bytes = *blake3::hash(format!("{APP_PREFIX}{passphrase}").as_bytes()).as_bytes(); + let topic_id = TopicId::from_bytes(topic_bytes); + + // DHT signing key: ed25519 from SHA-256 of prefixed passphrase + "sign". + let sign_seed = Sha256::digest(format!("{APP_PREFIX}sign:{passphrase}").as_bytes()); + let seed_bytes: [u8; 32] = sign_seed.into(); + let dht_signing_key = SigningKey::from_bytes(&seed_bytes); + + // DHT salt: first 20 bytes of SHA-256 of prefixed passphrase + "salt". + let salt_hash = Sha256::digest(format!("{APP_PREFIX}salt:{passphrase}").as_bytes()); + let mut dht_salt = [0u8; 20]; + dht_salt.copy_from_slice(&salt_hash[..20]); + + Self { + topic_id, + dht_signing_key, + dht_salt, + } + } +} + +// ── Swarm state (for TUI) ─────────────────────────────────────── + +/// Observable state of the gossip swarm. +#[derive(Debug, Clone)] +pub struct SwarmState { + pub topic_hash: String, + pub neighbor_count: usize, + pub dht_status: DhtStatus, + pub last_dht_publish: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum DhtStatus { + Bootstrapping, + Ready, + PublishFailed, +} + +impl std::fmt::Display for DhtStatus { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Bootstrapping => write!(f, "Bootstrapping"), + Self::Ready => write!(f, "Ready"), + Self::PublishFailed => write!(f, "Failed"), + } + } +} + +// ── Events to router ──────────────────────────────────────────── + +/// Events from the gossip swarm to the router. +pub enum GossipEvent { + /// A gossip frame received from the swarm. + Frame(GossipFrame), + /// Neighbor count changed. + NeighborChanged(usize), +} + +// ── Gossip handle ─────────────────────────────────────────────── + +/// Handle to the gossip swarm. Manages iroh endpoint, gossip subscription, and DHT. +pub struct Gossip { + endpoint: Endpoint, + gossip: GossipActor, + state: Arc>, + neighbors: Arc>>, + dht_signing_key: SigningKey, + dht_salt: [u8; 20], + cancel: CancellationToken, +} + +impl Gossip { + /// Create and start the gossip network. + /// + /// Spawns background tasks for connection acceptance, gossip event forwarding, + /// gossip broadcasting, and DHT heartbeat. + /// + /// Returns `(handle, event_rx, frame_tx)`: + /// - `event_rx`: channel of events from the swarm to the router + /// - `frame_tx`: channel for the router to send frames to broadcast + pub async fn new( + secret_key: SecretKey, + keys: &PassphraseKeys, + ) -> Result<(Self, mpsc::Receiver, mpsc::Sender)> { + let cancel = CancellationToken::new(); + + // Build iroh endpoint with n0 discovery. + let endpoint = Endpoint::builder(iroh::endpoint::presets::N0) + .secret_key(secret_key) + .alpns(vec![GOSSIP_ALPN.to_vec()]) + .bind() + .await + .context("binding iroh endpoint")?; + + let my_id = endpoint.id(); + info!("iroh endpoint bound, id: {}", my_id.fmt_short()); + + // Build gossip actor. + let gossip = GossipActor::builder().spawn(endpoint.clone()); + + // Bootstrap from DHT. + let bootstrap_peers = dht_get_peers(keys).await; + if bootstrap_peers.is_empty() { + info!("no peers found in DHT — we are the first node"); + } else { + info!("found {} peer(s) in DHT", bootstrap_peers.len()); + } + + // Subscribe to gossip topic. + let topic = gossip + .subscribe(keys.topic_id, bootstrap_peers) + .await + .context("subscribing to gossip topic")?; + + let (sender, receiver) = topic.split(); + + // Initial swarm state. + let topic_hash = hex::encode(&keys.topic_id.as_bytes()[..8]); + let state = Arc::new(Mutex::new(SwarmState { + topic_hash, + neighbor_count: 0, + dht_status: DhtStatus::Bootstrapping, + last_dht_publish: None, + })); + let neighbor_count = Arc::new(AtomicUsize::new(0)); + let neighbors: Arc>> = Arc::new(Mutex::new(HashSet::new())); + + // Channels. + let (event_tx, event_rx) = mpsc::channel::(256); + let (frame_tx, frame_rx) = mpsc::channel::(256); + + let handle = Self { + endpoint: endpoint.clone(), + gossip: gossip.clone(), + state: state.clone(), + neighbors: neighbors.clone(), + dht_signing_key: keys.dht_signing_key.clone(), + dht_salt: keys.dht_salt, + cancel: cancel.clone(), + }; + + // Spawn accept loop. + spawn_accept_loop(endpoint.clone(), gossip.clone(), cancel.clone()); + + // Spawn gossip receiver task. + spawn_receiver(receiver, event_tx, state.clone(), neighbor_count.clone(), neighbors.clone(), cancel.clone()); + + // Clone sender for DHT heartbeat to feed discovered peers via join_peers. + let dht_sender = sender.clone(); + + // Spawn gossip broadcaster task. + spawn_broadcaster(sender, frame_rx, cancel.clone()); + + // Spawn DHT heartbeat. + spawn_dht_heartbeat( + my_id, + keys.dht_signing_key.clone(), + keys.dht_salt, + dht_sender, + neighbors, + state, + neighbor_count, + cancel, + ); + + // Try to join (non-blocking — fire and forget with timeout). + // The receiver task handles events regardless. + + Ok((handle, event_rx, frame_tx)) + } + + /// Snapshot of swarm state for TUI. + pub fn swarm_state(&self) -> SwarmState { + self.state.lock().unwrap_or_else(|e| e.into_inner()).clone() + } + + /// Graceful shutdown. Publishes neighbors (without ourselves) to DHT + /// so other nodes can still bootstrap from live peers. + pub async fn shutdown(&self) { + // Publish neighbors-only to DHT before shutting down. + let neighbor_ids: Vec<[u8; 32]> = self + .neighbors + .lock() + .unwrap_or_else(|e| e.into_inner()) + .iter() + .map(|id| *id.as_bytes()) + .collect(); + + if let Ok(dht) = Dht::client() { + let dht = dht.as_async(); + let value = encode_peer_list(&neighbor_ids); + let seq = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or(0); + let item = MutableItem::new( + self.dht_signing_key.clone(), + &value, + seq, + Some(self.dht_salt.as_ref()), + ); + match dht.put_mutable(item, None).await { + Ok(_) => info!("shutdown: published {} neighbor(s) to DHT (removed self)", neighbor_ids.len()), + Err(e) => debug!("shutdown: DHT publish failed: {e:#}"), + } + } + + self.cancel.cancel(); + let _ = self.gossip.shutdown().await; + self.endpoint.close().await; + } +} + +// ── Background tasks ──────────────────────────────────────────── + +/// Accept incoming iroh connections and forward gossip ALPN to the gossip actor. +fn spawn_accept_loop(endpoint: Endpoint, gossip: GossipActor, cancel: CancellationToken) { + tokio::spawn(async move { + loop { + let incoming = tokio::select! { + incoming = endpoint.accept() => incoming, + () = cancel.cancelled() => break, + }; + let Some(incoming) = incoming else { break }; + let gossip = gossip.clone(); + tokio::spawn(handle_incoming(incoming, gossip)); + } + debug!("accept loop stopped"); + }); +} + +async fn handle_incoming(incoming: Incoming, gossip: GossipActor) { + let Ok(conn) = incoming.await else { return }; + let alpn = conn.alpn(); + if *alpn == *GOSSIP_ALPN { + if let Err(e) = gossip.handle_connection(conn).await { + debug!("gossip handle_connection error: {e:#}"); + } + } else { + debug!("unexpected ALPN: {:?}", String::from_utf8_lossy(alpn)); + } +} + +/// Forward gossip events to the router via channel. +fn spawn_receiver( + mut receiver: iroh_gossip::api::GossipReceiver, + event_tx: mpsc::Sender, + state: Arc>, + neighbor_count: Arc, + neighbors: Arc>>, + cancel: CancellationToken, +) { + tokio::spawn(async move { + loop { + let event = tokio::select! { + event = receiver.next() => event, + () = cancel.cancelled() => break, + }; + let Some(Ok(event)) = event else { break }; + match event { + Event::Received(msg) => { + if let Some(frame) = GossipFrame::decode(&msg.content) { + let _ = event_tx.send(GossipEvent::Frame(frame)).await; + } else { + debug!("received malformed gossip frame ({} bytes)", msg.content.len()); + } + } + Event::NeighborUp(id) => { + let count = neighbor_count.fetch_add(1, Ordering::Relaxed) + 1; + info!("neighbor up: {} (total: {count})", id.fmt_short()); + if let Ok(mut n) = neighbors.lock() { n.insert(id); } + if let Ok(mut s) = state.lock() { s.neighbor_count = count; } + let _ = event_tx.send(GossipEvent::NeighborChanged(count)).await; + } + Event::NeighborDown(id) => { + let count = neighbor_count.fetch_sub(1, Ordering::Relaxed).saturating_sub(1); + info!("neighbor down: {} (total: {count})", id.fmt_short()); + if let Ok(mut n) = neighbors.lock() { n.remove(&id); } + if let Ok(mut s) = state.lock() { s.neighbor_count = count; } + let _ = event_tx.send(GossipEvent::NeighborChanged(count)).await; + } + Event::Lagged => { + warn!("gossip receiver lagged — some messages were dropped"); + } + } + } + debug!("gossip receiver stopped"); + }); +} + +/// Read frames from the router and broadcast them to the swarm. +fn spawn_broadcaster( + sender: iroh_gossip::api::GossipSender, + mut frame_rx: mpsc::Receiver, + cancel: CancellationToken, +) { + tokio::spawn(async move { + loop { + let frame = tokio::select! { + frame = frame_rx.recv() => frame, + () = cancel.cancelled() => break, + }; + let Some(frame) = frame else { break }; + let encoded = Bytes::from(frame.encode()); + if let Err(e) = sender.broadcast(encoded).await { + debug!("gossip broadcast error: {e:#}"); + } + } + debug!("gossip broadcaster stopped"); + }); +} + +/// Fast DHT interval while we have no neighbors. +const DHT_FAST_INTERVAL_SECS: u64 = 5; + +/// Periodically publish our peer list to the DHT and discover new peers. +/// +/// Interval is adaptive: fast (5s) while alone, slow (5 min) once connected. +#[allow(clippy::too_many_arguments)] +fn spawn_dht_heartbeat( + my_id: PublicKey, + signing_key: SigningKey, + salt: [u8; 20], + gossip_sender: iroh_gossip::api::GossipSender, + neighbors: Arc>>, + state: Arc>, + neighbor_count: Arc, + cancel: CancellationToken, +) { + tokio::spawn(async move { + let dht = match Dht::client() { + Ok(d) => d.as_async(), + Err(e) => { + warn!("failed to create DHT client: {e:#}"); + return; + } + }; + + let pub_key_bytes = signing_key.verifying_key().to_bytes(); + + // Initial publish immediately. + dht_publish(&dht, &my_id, &signing_key, &salt, &pub_key_bytes, &neighbors, &state).await; + + loop { + // Fast while alone, slow once we have neighbors. + let alone = neighbor_count.load(Ordering::Relaxed) == 0; + let base_secs = if alone { + DHT_FAST_INTERVAL_SECS + } else { + DHT_HEARTBEAT_SECS + }; + let max_jitter = if alone { 2 } else { DHT_HEARTBEAT_JITTER_SECS }; + let jitter = rand::random::() % max_jitter.max(1); + let sleep_dur = Duration::from_secs(base_secs + jitter); + + tokio::select! { + () = tokio::time::sleep(sleep_dur) => {} + () = cancel.cancelled() => break, + } + + // GET: discover new peers and feed to gossip. + if let Some(record) = dht.get_mutable_most_recent(&pub_key_bytes, Some(&salt)).await { + let peers = decode_peer_list(record.value()); + let new_peers: Vec<_> = peers + .into_iter() + .filter(|id| *id != my_id) + .collect(); + if !new_peers.is_empty() { + info!("DHT discovered {} peer(s), feeding to gossip", new_peers.len()); + if let Err(e) = gossip_sender.join_peers(new_peers).await { + debug!("join_peers error: {e:#}"); + } + } + } + + // PUT: merge-publish our info + neighbors + existing DHT peers. + dht_publish(&dht, &my_id, &signing_key, &salt, &pub_key_bytes, &neighbors, &state).await; + } + debug!("DHT heartbeat stopped"); + }); +} + +async fn dht_publish( + dht: &AsyncDht, + my_id: &PublicKey, + signing_key: &SigningKey, + salt: &[u8; 20], + pub_key_bytes: &[u8; 32], + neighbors: &Arc>>, + state: &Arc>, +) { + // Merge: ourselves + live neighbors + existing DHT peers (preserve others' data). + let mut seen = HashSet::new(); + let mut peer_list: Vec<[u8; 32]> = Vec::with_capacity(MAX_DHT_PEERS); + + // 1. Always include ourselves first. + peer_list.push(*my_id.as_bytes()); + seen.insert(*my_id.as_bytes()); + + // 2. Add our live neighbors. + if let Ok(n) = neighbors.lock() { + for id in n.iter() { + if seen.insert(*id.as_bytes()) && peer_list.len() < MAX_DHT_PEERS { + peer_list.push(*id.as_bytes()); + } + } + } + + // 3. Preserve existing DHT peers we don't already know about. + if peer_list.len() < MAX_DHT_PEERS + && let Some(record) = dht.get_mutable_most_recent(pub_key_bytes, Some(salt)).await + { + for existing in decode_peer_list(record.value()) { + if seen.insert(*existing.as_bytes()) && peer_list.len() < MAX_DHT_PEERS { + peer_list.push(*existing.as_bytes()); + } + } + } + + let value = encode_peer_list(&peer_list); + let seq = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or(0); + + let item = MutableItem::new(signing_key.clone(), &value, seq, Some(salt.as_ref())); + match dht.put_mutable(item, None).await { + Ok(_) => { + debug!("published to DHT (seq={seq}, peers={})", peer_list.len()); + if let Ok(mut s) = state.lock() { + s.dht_status = DhtStatus::Ready; + s.last_dht_publish = Some(Instant::now()); + } + } + Err(e) => { + warn!("DHT publish failed: {e:#}"); + if let Ok(mut s) = state.lock() { + s.dht_status = DhtStatus::PublishFailed; + } + } + } +} + +// ── DHT helpers ───────────────────────────────────────────────── + +/// Query the DHT for existing peers (blocking bootstrap, called once on startup). +async fn dht_get_peers(keys: &PassphraseKeys) -> Vec { + let dht = match Dht::client() { + Ok(d) => d.as_async(), + Err(e) => { + warn!("failed to create DHT client for bootstrap: {e:#}"); + return vec![]; + } + }; + + let pub_key_bytes = keys.dht_signing_key.verifying_key().to_bytes(); + match dht.get_mutable_most_recent(&pub_key_bytes, Some(&keys.dht_salt)).await { + Some(record) => { + let peers = decode_peer_list(record.value()); + debug!("DHT bootstrap: found {} peer(s)", peers.len()); + peers + } + None => { + debug!("DHT bootstrap: no existing record"); + vec![] + } + } +} + +/// Encode a list of 32-byte peer IDs into bytes for DHT storage. +/// Format: [count: u16 LE] [id_0: 32B] [id_1: 32B] ... +fn encode_peer_list(peers: &[[u8; 32]]) -> Vec { + let count = peers.len().min(MAX_DHT_PEERS) as u16; + let mut buf = Vec::with_capacity(2 + (count as usize) * 32); + buf.extend_from_slice(&count.to_le_bytes()); + for id in peers.iter().take(MAX_DHT_PEERS) { + buf.extend_from_slice(id); + } + buf +} + +/// Decode a peer list from DHT record value. +fn decode_peer_list(data: &[u8]) -> Vec { + if data.len() < 2 { + return vec![]; + } + let count = u16::from_le_bytes([data[0], data[1]]) as usize; + let mut peers = Vec::with_capacity(count); + let mut offset = 2; + for _ in 0..count { + if offset + 32 > data.len() { + break; + } + let bytes: [u8; 32] = match data[offset..offset + 32].try_into() { + Ok(b) => b, + Err(_) => break, + }; + if let Ok(pk) = PublicKey::from_bytes(&bytes) { + peers.push(pk); + } + offset += 32; + } + peers +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn passphrase_keys_deterministic() { + let k1 = PassphraseKeys::derive("test-passphrase"); + let k2 = PassphraseKeys::derive("test-passphrase"); + assert_eq!(k1.topic_id, k2.topic_id); + assert_eq!(k1.dht_signing_key.to_bytes(), k2.dht_signing_key.to_bytes()); + assert_eq!(k1.dht_salt, k2.dht_salt); + } + + #[test] + fn passphrase_keys_different() { + let k1 = PassphraseKeys::derive("alpha"); + let k2 = PassphraseKeys::derive("beta"); + assert_ne!(k1.topic_id, k2.topic_id); + assert_ne!(k1.dht_signing_key.to_bytes(), k2.dht_signing_key.to_bytes()); + } + + #[test] + fn peer_list_roundtrip() { + // Generate valid ed25519 public keys. + let keys: Vec<_> = (0..3) + .map(|_| { + let sk = iroh::SecretKey::generate(&mut rand::rng()); + *sk.public().as_bytes() + }) + .collect(); + let encoded = encode_peer_list(&keys); + let decoded = decode_peer_list(&encoded); + assert_eq!(decoded.len(), 3); + for (i, key) in keys.iter().enumerate() { + assert_eq!(*decoded[i].as_bytes(), *key); + } + } + + #[test] + fn peer_list_empty() { + let encoded = encode_peer_list(&[]); + let decoded = decode_peer_list(&encoded); + assert!(decoded.is_empty()); + } + + #[test] + fn peer_list_truncated_data() { + // Only 1 byte — too short for count. + assert!(decode_peer_list(&[0]).is_empty()); + // Count says 2 but only 1 peer present. + let mut data = vec![2, 0]; // count = 2 + data.extend_from_slice(&[0xAA; 32]); // only 1 peer + let decoded = decode_peer_list(&data); + assert_eq!(decoded.len(), 1); + } +} diff --git a/src/main.rs b/src/main.rs index d43067a..88495ab 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,25 +1,28 @@ mod config; -mod network; +mod gossip; mod packet; mod radio; +mod rate_limit; mod router; +mod setup; mod tui; -use std::collections::HashMap; use std::sync::Arc; use std::time::Instant; use anyhow::{Context, Result}; use clap::{Parser, Subcommand}; +use iroh::SecretKey; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; use tracing::info; -use crate::network::{Network, PeerConfig}; +use crate::gossip::PassphraseKeys; +use crate::rate_limit::RateLimiter; use crate::router::Stats; #[derive(Parser)] -#[command(name = "donglora-bridge", about = "Peer-to-peer LoRa bridge using iroh")] +#[command(name = "donglora-bridge", about = "LoRa gossip bridge using iroh")] struct Cli { /// Path to config file (default: ~/.config/donglora-bridge/config.toml) #[arg(long)] @@ -35,10 +38,8 @@ struct Cli { #[derive(Subcommand)] enum Commands { - /// Print this bridge's NodeId and exit - ShowId, - /// Generate an example config and print to stdout - GenerateConfig, + /// Interactive configuration wizard + Config, } #[tokio::main] @@ -51,34 +52,20 @@ async fn main() -> Result<()> { None => config::config_path()?, }; - // Load or create secret key. - // Secret key lives alongside the config file (same dir, .key extension). - let key_path = config_path.with_extension("key"); - let secret_key = config::load_or_create_secret_key(&key_path)?; - let node_id = secret_key.public(); - let node_id_str = node_id.to_string(); - - // Handle subcommands. - match cli.command { - Some(Commands::ShowId) => { - println!("{node_id_str}"); - return Ok(()); - } - Some(Commands::GenerateConfig) => { - print!("{}", config::generate_default_config(&node_id_str)); - return Ok(()); - } - None => {} + // Config subcommand: run wizard with existing config as defaults. + if let Some(Commands::Config) = cli.command { + let existing = if config_path.exists() { + config::load_config(&config_path).ok() + } else { + None + }; + return setup::run_wizard(&config_path, existing.as_ref()); } - // First-run: if config doesn't exist, generate it and exit. + // First-run: no config file → run wizard automatically. if !config_path.exists() { - config::write_default_config(&config_path, &node_id_str)?; - println!("Generated config at {}", config_path.display()); - println!("Your NodeId: {node_id_str}"); - println!(); - println!("Add peers to your config and run again!"); - return Ok(()); + println!(" No config found. Let's set one up!\n"); + return setup::run_wizard(&config_path, None); } // Load config. @@ -94,7 +81,6 @@ async fn main() -> Result<()> { ) .init(); } else { - // Log to file in TUI mode. let log_path = cfg .bridge .log_file @@ -117,38 +103,32 @@ async fn main() -> Result<()> { .init(); } - info!("donglora-bridge starting, NodeId: {node_id_str}"); + // Ephemeral identity — fresh keypair each launch. + let secret_key = SecretKey::generate(&mut rand::rng()); + let node_id = secret_key.public(); + info!("donglora-bridge starting, ephemeral id: {}", node_id.fmt_short()); // Parse radio config. let radio_config = cfg.radio.to_radio_config()?; - let radio_config_str = radio::format_radio_config(&radio_config); - // Parse peer configs. - let mut peer_configs = Vec::new(); - let mut peer_names: HashMap = HashMap::new(); - for entry in &cfg.peers { - let pk = entry.public_key()?; - let name = entry.display_name(); - peer_names.insert(pk, name.clone()); - peer_configs.push(PeerConfig { - public_key: pk, - name, - addresses: entry.addresses.clone(), - }); - } + // Derive passphrase keys. + let keys = PassphraseKeys::derive(&cfg.bridge.passphrase); + info!("topic: {}", hex::encode(&keys.topic_id.as_bytes()[..8])); - if peer_configs.is_empty() && !cli.log_only { - println!("No peers configured. Add peers to {} and run again.", config_path.display()); - println!("Your NodeId: {node_id_str}"); - return Ok(()); - } + // Create rate limiter. + let rate_limiter = RateLimiter::from_radio_config( + radio_config.sf, + radio_config.bw, + radio_config.cr, + radio_config.preamble_len, + cfg.bridge.rate_limit_pps, + ); + info!("rate limiter: {:.1} pps", rate_limiter.rate_pps()); - // Create network. - let network = Arc::new(Network::new(secret_key, peer_configs).await?); - info!("iroh endpoint bound, id: {node_id_str}"); - - // Start network. - let (net_event_rx, net_send_tx) = network.start(); + // Create gossip network. + let (swarm, gossip_event_rx, gossip_frame_tx) = + gossip::Gossip::new(secret_key, &keys).await?; + let swarm = Arc::new(swarm); // Start radio. let (radio_event_rx, radio_tx) = radio::spawn(radio_config, cfg.radio.port.clone()); @@ -158,14 +138,15 @@ async fn main() -> Result<()> { let (log_tx, log_rx) = mpsc::channel(512); let (config_watch_tx, config_watch_rx) = tokio::sync::watch::channel( router::RadioConfigInfo { - config_str: radio_config_str, + active: radio_config, + requested: radio_config, source: radio::ConfigSource::Ours, + device: String::new(), }, ); // Cancellation. let cancel = CancellationToken::new(); - let start_time = Instant::now(); // Spawn router. @@ -174,18 +155,18 @@ async fn main() -> Result<()> { let router_our_id = node_id; let dedup_window = std::time::Duration::from_secs(cfg.bridge.dedup_window_secs); let tx_queue_size = cfg.bridge.tx_queue_size; - let router_peer_names = peer_names; tokio::spawn(async move { router::run( router_our_id, dedup_window, tx_queue_size, - router_peer_names, + radio_config, + rate_limiter, radio_event_rx, radio_tx, - net_event_rx, - net_send_tx, + gossip_event_rx, + gossip_frame_tx, &router_stats, log_tx, config_watch_tx, @@ -197,7 +178,6 @@ async fn main() -> Result<()> { // Run TUI or headless. if cli.log_only { info!("running in headless mode"); - // Just wait for cancellation or ctrl-c. tokio::select! { () = cancel.cancelled() => {}, result = tokio::signal::ctrl_c() => { @@ -208,26 +188,30 @@ async fn main() -> Result<()> { } } } else { - // Run TUI (blocks until q or cancel). - if let Err(e) = tui::run( - node_id, + let terminal = tui::run( config_watch_rx, - network.clone(), + swarm.clone(), stats, log_rx, cancel.clone(), start_time, ) - .await - { - // Restore terminal before printing error. - eprintln!("TUI error: {e:#}"); + .await; + + cancel.cancel(); + swarm.shutdown().await; + info!("donglora-bridge stopped"); + + if let Ok(mut term) = terminal { + tui::restore_terminal(&mut term); } + + return Ok(()); } - // Shutdown. + // Shutdown (headless path). cancel.cancel(); - network.shutdown().await; + swarm.shutdown().await; info!("donglora-bridge stopped"); Ok(()) diff --git a/src/network.rs b/src/network.rs deleted file mode 100644 index 6dc73fb..0000000 --- a/src/network.rs +++ /dev/null @@ -1,411 +0,0 @@ -use std::collections::HashMap; -use std::sync::Arc; -use std::time::Instant; - -use anyhow::{Context, Result}; -use bytes::Bytes; -use iroh::endpoint::Connection; -use iroh::{Endpoint, PublicKey, SecretKey, endpoint::presets}; -use tokio::sync::{Mutex, mpsc}; -use tokio_util::sync::CancellationToken; -use tracing::{debug, info, warn}; - -use crate::packet::Envelope; - -/// ALPN protocol identifier for donglora-bridge. -const ALPN: &[u8] = b"donglora-bridge/0"; - -/// Info about a configured peer. -#[derive(Debug, Clone)] -pub struct PeerConfig { - pub public_key: PublicKey, - pub name: String, - pub addresses: Vec, -} - -/// Observable state of a peer connection. -#[derive(Debug, Clone)] -pub struct PeerState { - pub name: String, - pub public_key: PublicKey, - pub connected: bool, - pub last_seen: Option, - pub packets_in: u64, - pub packets_out: u64, -} - -/// Event from the network to the router. -#[allow(dead_code)] -pub enum NetworkEvent { - /// A packet received from an iroh peer. - Packet { - from: PublicKey, - envelope: Envelope, - }, - /// A peer connected. - PeerConnected(PublicKey), - /// A peer disconnected. - PeerDisconnected(PublicKey), -} - -/// Shared peer connection map. -type PeerConnections = Arc>>; - -/// Shared peer state map. -type PeerStates = Arc>>; - -/// The network layer: manages the iroh endpoint and peer connections. -pub struct Network { - endpoint: Endpoint, - peer_configs: Vec, - connections: PeerConnections, - states: PeerStates, - cancel: CancellationToken, -} - -impl Network { - /// Create a new network layer with the given identity and peer configuration. - pub async fn new(secret_key: SecretKey, peer_configs: Vec) -> Result { - let endpoint = Endpoint::builder(presets::N0) - .secret_key(secret_key) - .alpns(vec![ALPN.to_vec()]) - .bind() - .await - .context("binding iroh endpoint")?; - - // Wait for relay connection before trying to reach peers. - info!("waiting for iroh relay connection..."); - endpoint.online().await; - let addr = endpoint.addr(); - info!("iroh online, addr: {addr:?}"); - - let states = Arc::new(Mutex::new(HashMap::new())); - { - let mut s = states.lock().await; - for pc in &peer_configs { - s.insert( - pc.public_key, - PeerState { - name: pc.name.clone(), - public_key: pc.public_key, - connected: false, - last_seen: None, - packets_in: 0, - packets_out: 0, - }, - ); - } - } - - Ok(Self { - endpoint, - peer_configs, - connections: Arc::new(Mutex::new(HashMap::new())), - states, - cancel: CancellationToken::new(), - }) - } - - /// Get a snapshot of all peer states. - pub async fn peer_states(&self) -> Vec { - self.states.lock().await.values().cloned().collect() - } - - /// Start the network: spawn outbound connection tasks + inbound accept loop. - /// - /// Returns a receiver for network events and a sender for outbound envelopes. - pub fn start( - &self, - ) -> (mpsc::Receiver, mpsc::Sender<(Envelope, Option)>) { - let (event_tx, event_rx) = mpsc::channel::(256); - let (send_tx, send_rx) = mpsc::channel::<(Envelope, Option)>(256); - - // Spawn outbound connection tasks for each configured peer. - for pc in &self.peer_configs { - let endpoint = self.endpoint.clone(); - let pk = pc.public_key; - let addrs = pc.addresses.clone(); - let connections = self.connections.clone(); - let states = self.states.clone(); - let event_tx = event_tx.clone(); - let cancel = self.cancel.clone(); - - tokio::spawn(async move { - outbound_connection_loop(endpoint, pk, addrs, connections, states, event_tx, cancel).await; - }); - } - - // Spawn inbound accept loop. - { - let endpoint = self.endpoint.clone(); - let connections = self.connections.clone(); - let states = self.states.clone(); - let event_tx = event_tx.clone(); - let cancel = self.cancel.clone(); - let known_peers: Vec = - self.peer_configs.iter().map(|pc| pc.public_key).collect(); - - tokio::spawn(async move { - accept_loop(endpoint, known_peers, connections, states, event_tx, cancel).await; - }); - } - - // Spawn broadcast sender task. - { - let connections = self.connections.clone(); - let states = self.states.clone(); - let cancel = self.cancel.clone(); - - tokio::spawn(async move { - broadcast_loop(connections, states, send_rx, cancel).await; - }); - } - - (event_rx, send_tx) - } - - /// Shut down the network layer. - pub async fn shutdown(&self) { - self.cancel.cancel(); - self.endpoint.close().await; - } -} - -/// Maintain a persistent outbound connection to a peer, reconnecting on failure. -async fn outbound_connection_loop( - endpoint: Endpoint, - peer_key: PublicKey, - addresses: Vec, - connections: PeerConnections, - states: PeerStates, - event_tx: mpsc::Sender, - cancel: CancellationToken, -) { - let mut backoff = std::time::Duration::from_secs(1); - let max_backoff = std::time::Duration::from_secs(30); - - loop { - if cancel.is_cancelled() { - return; - } - - // Skip if we already have a connection (e.g., from inbound accept). - { - let conns = connections.lock().await; - if let Some(c) = conns.get(&peer_key) - && c.stable_id() != 0 - { - // Connection exists and looks alive — wait and retry check. - drop(conns); - tokio::select! { - () = cancel.cancelled() => return, - () = tokio::time::sleep(std::time::Duration::from_secs(5)) => continue, - } - } - } - - info!("connecting to peer {}", fmt_key(&peer_key)); - let mut addr = iroh::EndpointAddr::new(peer_key); - // Add our relay URL as a hint — the peer is likely on the same relay network. - for relay_url in endpoint.addr().relay_urls() { - addr = addr.with_relay_url(relay_url.clone()); - } - for a in &addresses { - if let Ok(sock) = a.parse::() { - addr = addr.with_ip_addr(sock); - } - } - debug!("connect addr: {addr:?}"); - match endpoint.connect(addr, ALPN).await { - Ok(conn) => { - backoff = std::time::Duration::from_secs(1); - let remote = conn.remote_id(); - info!("connected to peer {}", fmt_key(&remote)); - - { - let mut conns = connections.lock().await; - conns.insert(remote, conn.clone()); - } - update_peer_connected(&states, &remote, true).await; - let _ = event_tx.send(NetworkEvent::PeerConnected(remote)).await; - - // Read datagrams from this connection. - read_datagrams(conn, remote, connections.clone(), states.clone(), event_tx.clone(), cancel.clone()).await; - - update_peer_connected(&states, &remote, false).await; - let _ = event_tx - .send(NetworkEvent::PeerDisconnected(remote)) - .await; - } - Err(e) => { - warn!("connect to {} failed: {e:#}", fmt_key(&peer_key)); - } - } - - tokio::select! { - () = cancel.cancelled() => return, - () = tokio::time::sleep(backoff) => {}, - } - backoff = (backoff * 2).min(max_backoff); - } -} - -/// Accept inbound connections from known peers. -async fn accept_loop( - endpoint: Endpoint, - known_peers: Vec, - connections: PeerConnections, - states: PeerStates, - event_tx: mpsc::Sender, - cancel: CancellationToken, -) { - loop { - let incoming = tokio::select! { - () = cancel.cancelled() => return, - inc = endpoint.accept() => match inc { - Some(i) => i, - None => return, - }, - }; - - let conn = match incoming.await { - Ok(c) => c, - Err(e) => { - warn!("inbound connection failed: {e:#}"); - continue; - } - }; - - let remote = conn.remote_id(); - if !known_peers.contains(&remote) { - warn!("rejecting unknown peer {}", fmt_key(&remote)); - conn.close(0u32.into(), b"unknown peer"); - continue; - } - - info!("accepted connection from peer {}", fmt_key(&remote)); - { - let mut conns = connections.lock().await; - conns.insert(remote, conn.clone()); - } - update_peer_connected(&states, &remote, true).await; - let _ = event_tx.send(NetworkEvent::PeerConnected(remote)).await; - - // Spawn a reader for this inbound connection. - let conns = connections.clone(); - let sts = states.clone(); - let etx = event_tx.clone(); - let ct = cancel.clone(); - tokio::spawn(async move { - read_datagrams(conn, remote, conns.clone(), sts.clone(), etx.clone(), ct).await; - update_peer_connected(&sts, &remote, false).await; - let _ = etx.send(NetworkEvent::PeerDisconnected(remote)).await; - }); - } -} - -/// Read datagrams from a connection until it closes. -async fn read_datagrams( - conn: Connection, - remote: PublicKey, - connections: PeerConnections, - states: PeerStates, - event_tx: mpsc::Sender, - cancel: CancellationToken, -) { - loop { - let data = tokio::select! { - () = cancel.cancelled() => break, - result = conn.read_datagram() => match result { - Ok(d) => d, - Err(e) => { - debug!("datagram read from {} ended: {e}", fmt_key(&remote)); - break; - } - }, - }; - - if let Some(envelope) = Envelope::decode(&data) { - { - let mut sts = states.lock().await; - if let Some(ps) = sts.get_mut(&remote) { - ps.packets_in += 1; - ps.last_seen = Some(Instant::now()); - } - } - let _ = event_tx - .send(NetworkEvent::Packet { - from: remote, - envelope, - }) - .await; - } else { - warn!("malformed envelope from {}", fmt_key(&remote)); - } - } - - // Clean up connection. - let mut conns = connections.lock().await; - if let Some(existing) = conns.get(&remote) - && existing.stable_id() == conn.stable_id() - { - conns.remove(&remote); - } -} - -/// Broadcast envelopes to all connected peers (except the optional skip key). -async fn broadcast_loop( - connections: PeerConnections, - states: PeerStates, - mut send_rx: mpsc::Receiver<(Envelope, Option)>, - cancel: CancellationToken, -) { - loop { - let (envelope, skip) = tokio::select! { - () = cancel.cancelled() => return, - msg = send_rx.recv() => match msg { - Some(m) => m, - None => return, - }, - }; - - let data = Bytes::from(envelope.encode()); - let conns = connections.lock().await; - - for (pk, conn) in conns.iter() { - if Some(*pk) == skip { - continue; - } - match conn.send_datagram(data.clone()) { - Ok(()) => { - let mut sts = states.lock().await; - if let Some(ps) = sts.get_mut(pk) { - ps.packets_out += 1; - } - } - Err(e) => { - debug!("send to {} failed: {e}", fmt_key(pk)); - } - } - } - } -} - -async fn update_peer_connected(states: &PeerStates, pk: &PublicKey, connected: bool) { - let mut sts = states.lock().await; - if let Some(ps) = sts.get_mut(pk) { - ps.connected = connected; - if connected { - ps.last_seen = Some(Instant::now()); - } - } -} - -fn fmt_key(pk: &PublicKey) -> String { - let s = pk.to_string(); - if s.len() > 12 { - format!("{}...", &s[..12]) - } else { - s - } -} diff --git a/src/packet.rs b/src/packet.rs index 34b8c80..d8134f3 100644 --- a/src/packet.rs +++ b/src/packet.rs @@ -1,10 +1,5 @@ -use iroh::PublicKey; - -/// Wire protocol version. -const VERSION: u8 = 0; - -/// Envelope header size: version(1) + origin(32) + seq(8) + rssi(2) + snr(2) + payload_len(2) = 47 -const HEADER_SIZE: usize = 47; +/// Gossip frame header size: sender(32) + rssi(2) + snr(2) + payload_len(2) = 38 +const FRAME_HEADER_SIZE: usize = 38; /// SNR quality grade for a received LoRa packet. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -38,7 +33,7 @@ impl std::fmt::Display for Grade { } /// Grade a received packet's SNR against the demodulation floor for the given -/// spreading factor. See PROTOCOL.md for the formula. +/// spreading factor. pub fn snr_grade(snr: i16, sf: u8) -> Grade { if !(-32..=32).contains(&snr) { return Grade::Invalid; @@ -54,7 +49,7 @@ pub fn snr_grade(snr: i16, sf: u8) -> Grade { } } -/// A packet received from the radio, before wrapping in an envelope. +/// A packet received from the radio, before wrapping in a gossip frame. #[derive(Debug, Clone)] pub struct RadioPacket { pub rssi: i16, @@ -62,13 +57,13 @@ pub struct RadioPacket { pub payload: Vec, } -/// A bridge envelope — the wire format sent between iroh peers. +/// A gossip frame — the wire format sent through the iroh-gossip swarm. +/// +/// Wire format: `[sender: 32B] [rssi: i16 LE] [snr: i16 LE] [payload_len: u16 LE] [payload: NB]` #[derive(Debug, Clone)] -pub struct Envelope { - /// NodeId of the bridge that first heard this packet on radio. - pub origin: PublicKey, - /// Per-origin sequence number (monotonically increasing). - pub seq: u64, +pub struct GossipFrame { + /// EndpointId (PublicKey bytes) of the bridge that first heard this packet on radio. + pub sender: [u8; 32], /// RSSI at the originating bridge (informational). pub rssi: i16, /// SNR at the originating bridge (informational). @@ -77,13 +72,21 @@ pub struct Envelope { pub payload: Vec, } -impl Envelope { - /// Encode this envelope into bytes for sending as a QUIC datagram. +impl GossipFrame { + /// Create a new gossip frame from a public key and radio packet data. + pub fn new(sender: &iroh::PublicKey, rssi: i16, snr: i16, payload: Vec) -> Self { + Self { + sender: *sender.as_bytes(), + rssi, + snr, + payload, + } + } + + /// Encode this frame into bytes for gossip broadcast. pub fn encode(&self) -> Vec { - let mut buf = Vec::with_capacity(HEADER_SIZE + self.payload.len()); - buf.push(VERSION); - buf.extend_from_slice(self.origin.as_bytes()); - buf.extend_from_slice(&self.seq.to_le_bytes()); + let mut buf = Vec::with_capacity(FRAME_HEADER_SIZE + self.payload.len()); + buf.extend_from_slice(&self.sender); buf.extend_from_slice(&self.rssi.to_le_bytes()); buf.extend_from_slice(&self.snr.to_le_bytes()); let len = self.payload.len() as u16; @@ -92,39 +95,30 @@ impl Envelope { buf } - /// Decode an envelope from bytes received as a QUIC datagram. + /// Decode a gossip frame from bytes received from the swarm. pub fn decode(data: &[u8]) -> Option { - if data.len() < HEADER_SIZE { + if data.len() < FRAME_HEADER_SIZE { return None; } - if data[0] != VERSION { - return None; - } - let origin = PublicKey::from_bytes(data[1..33].try_into().ok()?).ok()?; - let seq = u64::from_le_bytes(data[33..41].try_into().ok()?); - let rssi = i16::from_le_bytes(data[41..43].try_into().ok()?); - let snr = i16::from_le_bytes(data[43..45].try_into().ok()?); - let payload_len = u16::from_le_bytes(data[45..47].try_into().ok()?) as usize; - if data.len() != HEADER_SIZE + payload_len { + let sender: [u8; 32] = data[..32].try_into().ok()?; + let rssi = i16::from_le_bytes(data[32..34].try_into().ok()?); + let snr = i16::from_le_bytes(data[34..36].try_into().ok()?); + let payload_len = u16::from_le_bytes(data[36..38].try_into().ok()?) as usize; + if data.len() != FRAME_HEADER_SIZE + payload_len { return None; } Some(Self { - origin, - seq, + sender, rssi, snr, - payload: data[HEADER_SIZE..].to_vec(), + payload: data[FRAME_HEADER_SIZE..].to_vec(), }) } } -/// Compute a 16-byte content hash for deduplication. -pub fn content_hash(payload: &[u8]) -> [u8; 16] { - let hash = blake3::hash(payload); - let bytes = hash.as_bytes(); - let mut out = [0u8; 16]; - out.copy_from_slice(&bytes[..16]); - out +/// Compute a 32-byte blake3 content hash for deduplication. +pub fn content_hash(payload: &[u8]) -> [u8; 32] { + *blake3::hash(payload).as_bytes() } #[cfg(test)] @@ -146,34 +140,28 @@ mod tests { } #[test] - fn envelope_roundtrip() { + fn gossip_frame_roundtrip() { let key = iroh::SecretKey::generate(&mut rand::rng()); - let env = Envelope { - origin: key.public(), - seq: 42, - rssi: -85, - snr: 8, - payload: vec![1, 2, 3, 4, 5], - }; - let encoded = env.encode(); - let decoded = Envelope::decode(&encoded).unwrap(); - assert_eq!(decoded.origin, env.origin); - assert_eq!(decoded.seq, env.seq); - assert_eq!(decoded.rssi, env.rssi); - assert_eq!(decoded.snr, env.snr); - assert_eq!(decoded.payload, env.payload); + let frame = GossipFrame::new(&key.public(), -85, 8, vec![1, 2, 3, 4, 5]); + let encoded = frame.encode(); + let decoded = GossipFrame::decode(&encoded).expect("decode should succeed"); + assert_eq!(decoded.sender, *key.public().as_bytes()); + assert_eq!(decoded.rssi, -85); + assert_eq!(decoded.snr, 8); + assert_eq!(decoded.payload, vec![1, 2, 3, 4, 5]); } #[test] - fn envelope_reject_short() { - assert!(Envelope::decode(&[0u8; 10]).is_none()); + fn gossip_frame_reject_short() { + assert!(GossipFrame::decode(&[0u8; 10]).is_none()); } #[test] - fn envelope_reject_bad_version() { - let mut data = vec![1u8]; // version 1 - data.extend_from_slice(&[0u8; 46]); - assert!(Envelope::decode(&data).is_none()); + fn gossip_frame_reject_wrong_length() { + // Header says payload_len=5 but only 3 bytes follow. + let mut data = vec![0u8; 38 + 3]; + data[36..38].copy_from_slice(&5u16.to_le_bytes()); + assert!(GossipFrame::decode(&data).is_none()); } #[test] diff --git a/src/radio.rs b/src/radio.rs index 7933f66..dad813d 100644 --- a/src/radio.rs +++ b/src/radio.rs @@ -20,8 +20,8 @@ pub enum ConfigSource { pub enum RadioEvent { /// A valid packet received from the radio (already SNR-filtered). Packet(RadioPacket), - /// Radio connected (or reconnected) with the active config. - Connected(RadioConfig, ConfigSource), + /// Radio connected (or reconnected) with the active config and device info. + Connected(RadioConfig, ConfigSource, String), /// Radio disconnected — attempting reconnect. Disconnected, } @@ -56,7 +56,6 @@ fn radio_loop( loop { match connect_and_run(&config, &port, &event_tx, &mut tx_recv) { Ok(()) => { - // Clean shutdown (channel closed). info!("radio thread exiting"); return; } @@ -77,51 +76,57 @@ fn connect_and_run( event_tx: &mpsc::Sender, tx_recv: &mut mpsc::Receiver>, ) -> Result<()> { - let timeout = Duration::from_secs(2); + // Long initial timeout: the mux serializes commands through the dongle, + // so our GetConfig might have to wait behind other clients' commands. + let timeout = Duration::from_secs(10); info!("connecting to dongle..."); - let mut client = if let Some(port) = &port { - donglora_client::connect(Some(port), timeout)? + let (mut client, device) = if let Some(port) = &port { + let c = donglora_client::connect(Some(port), timeout)?; + (c, shorten_path(port)) } else { - donglora_client::connect_default()? + let c = donglora_client::connect_default()?; + (c, "mux".to_string()) }; + info!("[radio] transport connected: {device}"); - client.ping()?; - info!("dongle connected, configuring radio"); + info!("[radio] sending ping..."); + match client.ping() { + Ok(()) => info!("[radio] ping OK"), + Err(e) => { + info!("[radio] ping FAILED: {e:#}"); + return Err(e); + } + } - // Try to set our config. If the mux rejects it (config already locked by - // another client), fall back to reading whatever config is active. - let (active_config, config_source) = match client.set_config(*config) { - Ok(()) => { - info!("radio configured with our settings"); - (*config, ConfigSource::Ours) + info!("[radio] negotiating config..."); + let (active_config, config_source) = match negotiate_config(&mut client, config) { + Ok(result) => { + info!("[radio] config negotiated: source={:?}, config={}", result.1, format_radio_config(&result.0)); + result } Err(e) => { - warn!("SetConfig failed ({e:#}), reading active config from mux"); - match client.get_config() { - Ok(mux_config) => { - info!( - "using mux config: {}", - format_radio_config(&mux_config) - ); - (mux_config, ConfigSource::Mux) - } - Err(e2) => { - anyhow::bail!("SetConfig failed ({e:#}) and GetConfig also failed ({e2:#})"); - } - } + info!("[radio] config negotiation FAILED: {e:#}"); + return Err(e); } }; - client.start_rx()?; - info!("radio receiving"); + info!("[radio] sending start_rx..."); + match client.start_rx() { + Ok(()) => info!("[radio] start_rx OK"), + Err(e) => { + info!("[radio] start_rx FAILED: {e:#}"); + return Err(e); + } + } - let _ = event_tx.blocking_send(RadioEvent::Connected(active_config, config_source)); + let _ = event_tx.blocking_send(RadioEvent::Connected(active_config, config_source, device)); // Set a short timeout so we can interleave RX and TX. client .transport_mut() .set_timeout(Duration::from_millis(100))?; + info!("[radio] entering main loop"); loop { // Check for TX requests (non-blocking). @@ -143,7 +148,6 @@ fn connect_and_run( ); let pkt = RadioPacket { rssi, snr, payload }; if event_tx.blocking_send(RadioEvent::Packet(pkt)).is_err() { - // Router dropped — shut down. return Ok(()); } } else { @@ -153,33 +157,81 @@ fn connect_and_run( ); } } - Ok(Some(_)) => { - // Non-RxPacket unsolicited response — ignore. - } - Ok(None) => { - // Timeout — no packet, loop back to check TX. - } - Err(e) if is_timeout_error(&e) => { - // Mux sockets return WouldBlock on timeout instead of Ok(0). - // Treat the same as Ok(None) — just a timeout, not a real error. - } + Ok(Some(_)) => {} + Ok(None) => {} + Err(e) if is_timeout_error(&e) => {} Err(e) => return Err(e), } - // If the event channel is closed, the router is gone — exit cleanly. if event_tx.is_closed() { return Ok(()); } } } -/// Check if an anyhow::Error wraps a timeout-like I/O error. +/// Negotiate radio config with the dongle/mux. /// -/// Workaround: donglora-client's `read_frame` handles `TimedOut` but not -/// `WouldBlock`. On Linux, socket read timeouts return EAGAIN (WouldBlock), -/// not TimedOut. Serial ports return Ok(0) so they work fine, but mux -/// sockets surface this as an error. We catch it here instead of letting -/// it trigger the reconnect loop. +/// Strategy: +/// 1. GetConfig first to read what's active (caches the result). +/// 2. If it matches our desired config, use it — no SetConfig needed. +/// 3. If it differs, only try SetConfig if we haven't been rejected before. +/// 4. If SetConfig is rejected, set the flag IMMEDIATELY (before the mux can +/// disconnect us and lose this state), then return the cached GetConfig result. +/// +/// On reconnect, the flag ensures we never retry SetConfig. +fn negotiate_config( + client: &mut donglora_client::Client, + desired: &RadioConfig, +) -> Result<(RadioConfig, ConfigSource)> { + info!("[negotiate] get_config..."); + match client.get_config() { + Ok(cfg) => { + let source = if configs_match(&cfg, desired) { + info!("[negotiate] get_config OK (matches desired): {}", format_radio_config(&cfg)); + ConfigSource::Ours + } else { + info!("[negotiate] get_config OK (differs from desired): {}", format_radio_config(&cfg)); + ConfigSource::Mux + }; + Ok((cfg, source)) + } + Err(e) => { + info!("[negotiate] get_config FAILED: {e:#}"); + // No config on radio — try to set ours. + info!("[negotiate] trying set_config..."); + match client.set_config(*desired) { + Ok(()) => { + info!("[negotiate] set_config OK"); + Ok((*desired, ConfigSource::Ours)) + } + Err(e2) => { + info!("[negotiate] set_config FAILED: {e2:#}"); + anyhow::bail!("GetConfig failed ({e:#}) and SetConfig also failed ({e2:#})"); + } + } + } + } +} + +/// Shorten a device path to just the filename for display. +fn shorten_path(path: &str) -> String { + std::path::Path::new(path) + .file_name() + .map(|f| f.to_string_lossy().into_owned()) + .unwrap_or_else(|| path.to_string()) +} + +fn configs_match(a: &RadioConfig, b: &RadioConfig) -> bool { + a.freq_hz == b.freq_hz + && a.bw == b.bw + && a.sf == b.sf + && a.cr == b.cr + && a.sync_word == b.sync_word + && a.tx_power_dbm == b.tx_power_dbm + && a.preamble_len == b.preamble_len + && a.cad == b.cad +} + fn is_timeout_error(e: &anyhow::Error) -> bool { if let Some(io_err) = e.downcast_ref::() { matches!( @@ -191,25 +243,30 @@ fn is_timeout_error(e: &anyhow::Error) -> bool { } } +/// Format a bandwidth value for display. +pub fn format_bandwidth(bw: donglora_client::Bandwidth) -> &'static str { + match bw { + donglora_client::Bandwidth::Khz7 => "7.8 kHz", + donglora_client::Bandwidth::Khz10 => "10.4 kHz", + donglora_client::Bandwidth::Khz15 => "15.6 kHz", + donglora_client::Bandwidth::Khz20 => "20.8 kHz", + donglora_client::Bandwidth::Khz31 => "31.25 kHz", + donglora_client::Bandwidth::Khz41 => "41.7 kHz", + donglora_client::Bandwidth::Khz62 => "62.5 kHz", + donglora_client::Bandwidth::Khz125 => "125 kHz", + donglora_client::Bandwidth::Khz250 => "250 kHz", + donglora_client::Bandwidth::Khz500 => "500 kHz", + } +} + /// Format a RadioConfig for display. pub fn format_radio_config(config: &RadioConfig) -> String { let freq_mhz = config.freq_hz as f64 / 1_000_000.0; - let bw = match config.bw { - donglora_client::Bandwidth::Khz7 => "7.8kHz", - donglora_client::Bandwidth::Khz10 => "10.4kHz", - donglora_client::Bandwidth::Khz15 => "15.6kHz", - donglora_client::Bandwidth::Khz20 => "20.8kHz", - donglora_client::Bandwidth::Khz31 => "31.25kHz", - donglora_client::Bandwidth::Khz41 => "41.7kHz", - donglora_client::Bandwidth::Khz62 => "62.5kHz", - donglora_client::Bandwidth::Khz125 => "125kHz", - donglora_client::Bandwidth::Khz250 => "250kHz", - donglora_client::Bandwidth::Khz500 => "500kHz", - }; + let bw = format_bandwidth(config.bw); let power = if config.tx_power_dbm == donglora_client::TX_POWER_MAX { "max".to_string() } else { - format!("{}dBm", config.tx_power_dbm) + format!("{} dBm", config.tx_power_dbm) }; let preamble = if config.preamble_len == 0 { 16 diff --git a/src/rate_limit.rs b/src/rate_limit.rs new file mode 100644 index 0000000..6d35ef3 --- /dev/null +++ b/src/rate_limit.rs @@ -0,0 +1,174 @@ +use std::time::{Duration, Instant}; + +use donglora_client::Bandwidth; + +/// Token-bucket rate limiter for radio transmissions. +/// +/// Calculates a safe transmit rate from the LoRa radio configuration (spreading factor, +/// bandwidth, coding rate) and enforces it with a token bucket that allows short bursts. +pub struct RateLimiter { + tokens: f64, + max_tokens: f64, + refill_rate: f64, // tokens per second + last_refill: Instant, +} + +/// Default burst size (max tokens). +const DEFAULT_BURST: f64 = 3.0; + +/// Reference payload size (bytes) used to estimate air time for rate calculation. +const REFERENCE_PAYLOAD_BYTES: usize = 50; + +/// Target duty cycle fraction (0.5 = 50%). +const TARGET_DUTY_CYCLE: f64 = 0.5; + +impl RateLimiter { + /// Create a rate limiter derived from radio config. + /// + /// If `override_pps` is `Some`, use that rate directly instead of calculating. + pub fn from_radio_config( + sf: u8, + bw: Bandwidth, + cr: u8, + preamble: u16, + override_pps: Option, + ) -> Self { + let refill_rate = if let Some(pps) = override_pps { + pps.max(0.01) + } else { + let air_time = lora_air_time(REFERENCE_PAYLOAD_BYTES, sf, bw, cr, preamble); + let secs = air_time.as_secs_f64(); + if secs > 0.0 { + TARGET_DUTY_CYCLE / secs + } else { + 10.0 // fallback + } + }; + Self { + tokens: DEFAULT_BURST, + max_tokens: DEFAULT_BURST, + refill_rate, + last_refill: Instant::now(), + } + } + + /// Try to consume one token. Returns `true` if the transmission is allowed. + pub fn try_acquire(&mut self) -> bool { + self.refill(); + if self.tokens >= 1.0 { + self.tokens -= 1.0; + true + } else { + false + } + } + + /// Current rate in packets per second. + pub fn rate_pps(&self) -> f64 { + self.refill_rate + } + + fn refill(&mut self) { + let now = Instant::now(); + let elapsed = now.duration_since(self.last_refill).as_secs_f64(); + self.tokens = (self.tokens + elapsed * self.refill_rate).min(self.max_tokens); + self.last_refill = now; + } +} + +/// Estimate LoRa packet air time using the Semtech SX1276 formula. +/// +/// Parameters: +/// - `payload_bytes`: payload size in bytes +/// - `sf`: spreading factor (7–12) +/// - `bw`: bandwidth +/// - `cr`: coding rate denominator (5–8, meaning 4/5 to 4/8) +/// - `preamble`: preamble length in symbols (0 = firmware default of 16) +pub fn lora_air_time(payload_bytes: usize, sf: u8, bw: Bandwidth, cr: u8, preamble: u16) -> Duration { + let sf_f = f64::from(sf); + let bw_hz = bandwidth_hz(bw); + let preamble_symbols = if preamble == 0 { 16.0 } else { f64::from(preamble) }; + + // Symbol time in seconds. + let t_sym = 2.0_f64.powf(sf_f) / bw_hz; + + // Preamble time. + let t_preamble = (preamble_symbols + 4.25) * t_sym; + + // Low data rate optimize: enabled when SF >= 11 and BW <= 125 kHz. + let de: f64 = if sf >= 11 && bw_hz <= 125_000.0 { + 1.0 + } else { + 0.0 + }; + + // Payload symbol count (Semtech formula with CRC enabled, explicit header). + let numerator = 8.0 * payload_bytes as f64 - 4.0 * sf_f + 28.0 + 16.0; + let denominator = 4.0 * (sf_f - 2.0 * de); + let n_payload = 8.0 + (numerator / denominator).ceil().max(0.0) * f64::from(cr); + + let t_payload = n_payload * t_sym; + let total = t_preamble + t_payload; + + Duration::from_secs_f64(total) +} + +fn bandwidth_hz(bw: Bandwidth) -> f64 { + match bw { + Bandwidth::Khz7 => 7_800.0, + Bandwidth::Khz10 => 10_400.0, + Bandwidth::Khz15 => 15_600.0, + Bandwidth::Khz20 => 20_800.0, + Bandwidth::Khz31 => 31_250.0, + Bandwidth::Khz41 => 41_700.0, + Bandwidth::Khz62 => 62_500.0, + Bandwidth::Khz125 => 125_000.0, + Bandwidth::Khz250 => 250_000.0, + Bandwidth::Khz500 => 500_000.0, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn air_time_sf7_bw125() { + let t = lora_air_time(50, 7, Bandwidth::Khz125, 5, 16); + // SF7/125kHz/CR4/5/16-preamble/50 bytes: ~100ms range + assert!(t.as_millis() > 50, "air time too short: {t:?}"); + assert!(t.as_millis() < 300, "air time too long: {t:?}"); + } + + #[test] + fn air_time_sf12_bw125() { + let t = lora_air_time(50, 12, Bandwidth::Khz125, 5, 16); + // SF12/125kHz: should be multiple seconds + assert!(t.as_secs() >= 1, "air time too short: {t:?}"); + assert!(t.as_secs() < 10, "air time too long: {t:?}"); + } + + #[test] + fn rate_limiter_allows_burst() { + let mut rl = RateLimiter::from_radio_config(7, Bandwidth::Khz125, 5, 16, None); + // Should allow burst of 3 + assert!(rl.try_acquire()); + assert!(rl.try_acquire()); + assert!(rl.try_acquire()); + // 4th should fail (no time for refill) + assert!(!rl.try_acquire()); + } + + #[test] + fn rate_limiter_override() { + let rl = RateLimiter::from_radio_config(7, Bandwidth::Khz125, 5, 16, Some(42.0)); + assert!((rl.rate_pps() - 42.0).abs() < f64::EPSILON); + } + + #[test] + fn air_time_zero_preamble_uses_default() { + let t_default = lora_air_time(50, 7, Bandwidth::Khz125, 5, 0); + let t_explicit = lora_air_time(50, 7, Bandwidth::Khz125, 5, 16); + assert_eq!(t_default, t_explicit); + } +} diff --git a/src/router.rs b/src/router.rs index c6784ba..a3d9401 100644 --- a/src/router.rs +++ b/src/router.rs @@ -6,15 +6,18 @@ use iroh::PublicKey; use tokio::sync::mpsc; use tracing::{debug, warn}; -use crate::network::NetworkEvent; -use crate::packet::{Envelope, RadioPacket, content_hash}; -use crate::radio::{ConfigSource, RadioEvent, format_radio_config}; +use crate::gossip::GossipEvent; +use crate::packet::{GossipFrame, content_hash}; +use crate::radio::{ConfigSource, RadioEvent}; +use crate::rate_limit::RateLimiter; /// Active radio config info, sent to the TUI via watch channel. #[derive(Debug, Clone)] pub struct RadioConfigInfo { - pub config_str: String, + pub active: donglora_client::RadioConfig, + pub requested: donglora_client::RadioConfig, pub source: ConfigSource, + pub device: String, } /// Packet event for the TUI log. @@ -25,7 +28,6 @@ pub struct PacketLogEntry { pub size: usize, pub snr: Option, pub rssi: Option, - pub peer_name: Option, pub action: PacketAction, } @@ -33,7 +35,7 @@ pub struct PacketLogEntry { pub enum PacketDirection { RadioRx, RadioTx, - NetRx, + GossipRx, } #[derive(Debug, Clone, Copy)] @@ -42,6 +44,7 @@ pub enum PacketAction { Transmitted, DroppedDedup, DroppedQueueFull, + DroppedRateLimit, } impl std::fmt::Display for PacketDirection { @@ -49,7 +52,7 @@ impl std::fmt::Display for PacketDirection { match self { Self::RadioRx => write!(f, "RX "), Self::RadioTx => write!(f, "TX "), - Self::NetRx => write!(f, "NET"), + Self::GossipRx => write!(f, "GSP"), } } } @@ -61,6 +64,7 @@ impl std::fmt::Display for PacketAction { Self::Transmitted => write!(f, "TX"), Self::DroppedDedup => write!(f, "DUP"), Self::DroppedQueueFull => write!(f, "FULL"), + Self::DroppedRateLimit => write!(f, "RATE"), } } } @@ -70,10 +74,12 @@ impl std::fmt::Display for PacketAction { pub struct Stats { pub radio_rx: AtomicU64, pub radio_tx: AtomicU64, - pub forwarded: AtomicU64, - pub dropped_snr: AtomicU64, - pub dropped_dedup: AtomicU64, + pub gossip_rx: AtomicU64, + pub gossip_tx: AtomicU64, + pub dedup_hits: AtomicU64, + pub rate_limit_drops: AtomicU64, pub dropped_queue: AtomicU64, + pub neighbor_count: AtomicU64, pub radio_connected: AtomicU64, // 0 = disconnected, 1 = connected } @@ -82,10 +88,12 @@ impl Stats { StatsSnapshot { radio_rx: self.radio_rx.load(Ordering::Relaxed), radio_tx: self.radio_tx.load(Ordering::Relaxed), - forwarded: self.forwarded.load(Ordering::Relaxed), - dropped_snr: self.dropped_snr.load(Ordering::Relaxed), - dropped_dedup: self.dropped_dedup.load(Ordering::Relaxed), + gossip_rx: self.gossip_rx.load(Ordering::Relaxed), + gossip_tx: self.gossip_tx.load(Ordering::Relaxed), + dedup_hits: self.dedup_hits.load(Ordering::Relaxed), + rate_limit_drops: self.rate_limit_drops.load(Ordering::Relaxed), dropped_queue: self.dropped_queue.load(Ordering::Relaxed), + neighbor_count: self.neighbor_count.load(Ordering::Relaxed), radio_connected: self.radio_connected.load(Ordering::Relaxed) != 0, } } @@ -95,20 +103,22 @@ impl Stats { pub struct StatsSnapshot { pub radio_rx: u64, pub radio_tx: u64, - pub forwarded: u64, - pub dropped_snr: u64, - pub dropped_dedup: u64, + pub gossip_rx: u64, + pub gossip_tx: u64, + pub dedup_hits: u64, + pub rate_limit_drops: u64, pub dropped_queue: u64, + pub neighbor_count: u64, pub radio_connected: bool, } /// Time-bounded dedup cache. -struct DedupCache { - entries: HashMap, +struct DedupCache { + entries: HashMap<[u8; 32], Instant>, ttl: Duration, } -impl DedupCache { +impl DedupCache { fn new(ttl: Duration) -> Self { Self { entries: HashMap::new(), @@ -117,78 +127,97 @@ impl DedupCache { } /// Returns true if the key was already present (i.e., duplicate). - fn check_and_insert(&mut self, key: K) -> bool { + fn check_and_insert(&mut self, key: [u8; 32]) -> bool { let now = Instant::now(); - self.prune(now); - if self.entries.contains_key(&key) { + if let Some(ts) = self.entries.get(&key) + && now.duration_since(*ts) < self.ttl + { return true; } self.entries.insert(key, now); false } - fn insert(&mut self, key: K) { - self.entries.insert(key, Instant::now()); - } - - fn prune(&mut self, now: Instant) { + fn prune(&mut self) { + let now = Instant::now(); self.entries.retain(|_, ts| now.duration_since(*ts) < self.ttl); } } -/// Peer name lookup for log entries. -struct PeerNames { - names: HashMap, -} - -impl PeerNames { - fn get(&self, pk: &PublicKey) -> Option { - self.names.get(pk).cloned() - } -} - /// Run the router — the central packet bus. /// -/// This is an async task that mediates between the radio thread, the iroh -/// network layer, and the TUI. +/// Bridges radio packets to/from the gossip swarm with deduplication and rate limiting. #[allow(clippy::too_many_arguments)] pub async fn run( our_id: PublicKey, dedup_window: Duration, tx_queue_size: usize, - peer_names: HashMap, + requested_radio_config: donglora_client::RadioConfig, + mut rate_limiter: RateLimiter, mut radio_rx: mpsc::Receiver, radio_tx: mpsc::Sender>, - mut net_rx: mpsc::Receiver, - net_tx: mpsc::Sender<(Envelope, Option)>, + mut gossip_rx: mpsc::Receiver, + gossip_tx: mpsc::Sender, stats: &Stats, log_tx: mpsc::Sender, config_tx: tokio::sync::watch::Sender, ) { - let names = PeerNames { names: peer_names }; - let mut seq: u64 = 0; - let mut iroh_dedup: DedupCache<(PublicKey, u64)> = DedupCache::new(dedup_window); - let mut echo_dedup: DedupCache<[u8; 16]> = DedupCache::new(dedup_window); + let mut dedup = DedupCache::new(dedup_window); let mut tx_queue: VecDeque> = VecDeque::with_capacity(tx_queue_size); + let mut last_cleanup = Instant::now(); + let cleanup_interval = Duration::from_secs(30); loop { + // Periodic dedup cleanup. + if last_cleanup.elapsed() >= cleanup_interval { + dedup.prune(); + last_cleanup = Instant::now(); + } + tokio::select! { // Radio events. event = radio_rx.recv() => { let Some(event) = event else { break }; match event { RadioEvent::Packet(pkt) => { - handle_radio_packet( - &our_id, &mut seq, &pkt, - &mut echo_dedup, &net_tx, - stats, &log_tx, &names, - ).await; + stats.radio_rx.fetch_add(1, Ordering::Relaxed); + + let hash = content_hash(&pkt.payload); + if dedup.check_and_insert(hash) { + stats.dedup_hits.fetch_add(1, Ordering::Relaxed); + debug!("radio RX dedup suppressed ({} bytes)", pkt.payload.len()); + let _ = log_tx.send(PacketLogEntry { + timestamp: Instant::now(), + direction: PacketDirection::RadioRx, + size: pkt.payload.len(), + snr: Some(pkt.snr), + rssi: Some(pkt.rssi), + action: PacketAction::DroppedDedup, + }).await; + continue; + } + + // Wrap and broadcast to gossip. + let frame = GossipFrame::new(&our_id, pkt.rssi, pkt.snr, pkt.payload.clone()); + stats.gossip_tx.fetch_add(1, Ordering::Relaxed); + let _ = gossip_tx.send(frame).await; + + let _ = log_tx.send(PacketLogEntry { + timestamp: Instant::now(), + direction: PacketDirection::RadioRx, + size: pkt.payload.len(), + snr: Some(pkt.snr), + rssi: Some(pkt.rssi), + action: PacketAction::Forwarded, + }).await; } - RadioEvent::Connected(active_config, source) => { + RadioEvent::Connected(active_config, source, device) => { stats.radio_connected.store(1, Ordering::Relaxed); let _ = config_tx.send(RadioConfigInfo { - config_str: format_radio_config(&active_config), + active: active_config, + requested: requested_radio_config, source, + device, }); } RadioEvent::Disconnected => { @@ -197,19 +226,70 @@ pub async fn run( } } - // Network events. - event = net_rx.recv() => { + // Gossip events. + event = gossip_rx.recv() => { let Some(event) = event else { break }; match event { - NetworkEvent::Packet { from, envelope } => { - handle_net_packet( - &from, envelope, &mut iroh_dedup, &mut echo_dedup, - &mut tx_queue, tx_queue_size, - stats, &log_tx, &names, - ).await; + GossipEvent::Frame(frame) => { + stats.gossip_rx.fetch_add(1, Ordering::Relaxed); + + let hash = content_hash(&frame.payload); + if dedup.check_and_insert(hash) { + stats.dedup_hits.fetch_add(1, Ordering::Relaxed); + debug!("gossip RX dedup suppressed ({} bytes)", frame.payload.len()); + let _ = log_tx.send(PacketLogEntry { + timestamp: Instant::now(), + direction: PacketDirection::GossipRx, + size: frame.payload.len(), + snr: Some(frame.snr), + rssi: Some(frame.rssi), + action: PacketAction::DroppedDedup, + }).await; + continue; + } + + // Rate limit check. + if !rate_limiter.try_acquire() { + stats.rate_limit_drops.fetch_add(1, Ordering::Relaxed); + debug!("rate limited ({} bytes)", frame.payload.len()); + let _ = log_tx.send(PacketLogEntry { + timestamp: Instant::now(), + direction: PacketDirection::GossipRx, + size: frame.payload.len(), + snr: Some(frame.snr), + rssi: Some(frame.rssi), + action: PacketAction::DroppedRateLimit, + }).await; + continue; + } + + // Enqueue for radio TX. + if tx_queue.len() >= tx_queue_size { + stats.dropped_queue.fetch_add(1, Ordering::Relaxed); + warn!("TX queue full, dropping packet ({} bytes)", frame.payload.len()); + let _ = log_tx.send(PacketLogEntry { + timestamp: Instant::now(), + direction: PacketDirection::GossipRx, + size: frame.payload.len(), + snr: Some(frame.snr), + rssi: Some(frame.rssi), + action: PacketAction::DroppedQueueFull, + }).await; + continue; + } + + tx_queue.push_back(frame.payload.clone()); + let _ = log_tx.send(PacketLogEntry { + timestamp: Instant::now(), + direction: PacketDirection::GossipRx, + size: frame.payload.len(), + snr: Some(frame.snr), + rssi: Some(frame.rssi), + action: PacketAction::Transmitted, + }).await; } - NetworkEvent::PeerConnected(_) | NetworkEvent::PeerDisconnected(_) => { - // State tracked in network layer; TUI reads from there. + GossipEvent::NeighborChanged(count) => { + stats.neighbor_count.store(count as u64, Ordering::Relaxed); } } } @@ -226,7 +306,6 @@ pub async fn run( size, snr: None, rssi: None, - peer_name: None, action: PacketAction::Transmitted, }) .await; @@ -236,135 +315,3 @@ pub async fn run( } } } - -#[allow(clippy::too_many_arguments)] -async fn handle_radio_packet( - our_id: &PublicKey, - seq: &mut u64, - pkt: &RadioPacket, - echo_dedup: &mut DedupCache<[u8; 16]>, - net_tx: &mpsc::Sender<(Envelope, Option)>, - stats: &Stats, - log_tx: &mpsc::Sender, - names: &PeerNames, -) { - stats.radio_rx.fetch_add(1, Ordering::Relaxed); - - // Check echo suppression. - let hash = content_hash(&pkt.payload); - if echo_dedup.check_and_insert(hash) { - stats.dropped_dedup.fetch_add(1, Ordering::Relaxed); - debug!("radio RX echo suppressed ({} bytes)", pkt.payload.len()); - let _ = log_tx - .send(PacketLogEntry { - timestamp: Instant::now(), - direction: PacketDirection::RadioRx, - size: pkt.payload.len(), - snr: Some(pkt.snr), - rssi: Some(pkt.rssi), - peer_name: None, - action: PacketAction::DroppedDedup, - }) - .await; - return; - } - - // Wrap and forward to all peers. - *seq += 1; - let envelope = Envelope { - origin: *our_id, - seq: *seq, - rssi: pkt.rssi, - snr: pkt.snr, - payload: pkt.payload.clone(), - }; - - stats.forwarded.fetch_add(1, Ordering::Relaxed); - let _ = net_tx.send((envelope, None)).await; - let _ = log_tx - .send(PacketLogEntry { - timestamp: Instant::now(), - direction: PacketDirection::RadioRx, - size: pkt.payload.len(), - snr: Some(pkt.snr), - rssi: Some(pkt.rssi), - peer_name: names.get(our_id), - action: PacketAction::Forwarded, - }) - .await; -} - -#[allow(clippy::too_many_arguments)] -async fn handle_net_packet( - from: &PublicKey, - envelope: Envelope, - iroh_dedup: &mut DedupCache<(PublicKey, u64)>, - echo_dedup: &mut DedupCache<[u8; 16]>, - tx_queue: &mut VecDeque>, - tx_queue_size: usize, - stats: &Stats, - log_tx: &mpsc::Sender, - names: &PeerNames, -) { - // Check iroh-level dedup. - if iroh_dedup.check_and_insert((envelope.origin, envelope.seq)) { - stats.dropped_dedup.fetch_add(1, Ordering::Relaxed); - debug!("net dedup ({} bytes from {})", envelope.payload.len(), fmt_key(from)); - let _ = log_tx - .send(PacketLogEntry { - timestamp: Instant::now(), - direction: PacketDirection::NetRx, - size: envelope.payload.len(), - snr: Some(envelope.snr), - rssi: Some(envelope.rssi), - peer_name: names.get(from), - action: PacketAction::DroppedDedup, - }) - .await; - return; - } - - // Insert content hash for radio echo suppression. - let hash = content_hash(&envelope.payload); - echo_dedup.insert(hash); - - // Enqueue for radio TX. - if tx_queue.len() >= tx_queue_size { - stats.dropped_queue.fetch_add(1, Ordering::Relaxed); - warn!("TX queue full, dropping packet ({} bytes)", envelope.payload.len()); - let _ = log_tx - .send(PacketLogEntry { - timestamp: Instant::now(), - direction: PacketDirection::NetRx, - size: envelope.payload.len(), - snr: Some(envelope.snr), - rssi: Some(envelope.rssi), - peer_name: names.get(from), - action: PacketAction::DroppedQueueFull, - }) - .await; - return; - } - - tx_queue.push_back(envelope.payload); - let _ = log_tx - .send(PacketLogEntry { - timestamp: Instant::now(), - direction: PacketDirection::NetRx, - size: tx_queue.back().map(Vec::len).unwrap_or(0), - snr: Some(envelope.snr), - rssi: Some(envelope.rssi), - peer_name: names.get(from), - action: PacketAction::Transmitted, - }) - .await; -} - -fn fmt_key(pk: &PublicKey) -> String { - let s = pk.to_string(); - if s.len() > 12 { - format!("{}...", &s[..12]) - } else { - s - } -} diff --git a/src/setup.rs b/src/setup.rs new file mode 100644 index 0000000..714d258 --- /dev/null +++ b/src/setup.rs @@ -0,0 +1,177 @@ +//! Interactive console setup wizard for bridge configuration. + +use std::io::{self, Write}; +use std::path::Path; + +use anyhow::{Context, Result}; + +use crate::config::{BridgeSection, Config, RadioSection}; + +/// ANSI escape helpers. +const BOLD: &str = "\x1b[1m"; +const DIM: &str = "\x1b[2m"; +const CYAN: &str = "\x1b[36m"; +const GREEN: &str = "\x1b[32m"; +const YELLOW: &str = "\x1b[33m"; +const RESET: &str = "\x1b[0m"; + +/// Run the interactive setup wizard. If `existing` is Some, use those values as defaults. +pub fn run_wizard(config_path: &Path, existing: Option<&Config>) -> Result<()> { + let defaults = existing.map_or_else( + || (RadioSection::default(), BridgeSection::default()), + |c| (c.radio.clone(), c.bridge.clone()), + ); + let (radio, bridge) = defaults; + + println!(); + println!(" {BOLD}donglora-bridge setup{RESET}"); + println!(" {DIM}Configure your LoRa gossip bridge.{RESET}"); + println!(); + + // ── Passphrase ─────────────────────────────────────────────── + + let default_pass = if bridge.passphrase.is_empty() || bridge.passphrase == "change-me" { + None + } else { + Some(bridge.passphrase.as_str()) + }; + let passphrase = prompt_required("Passphrase", default_pass)?; + + println!(); + println!(" {BOLD}Radio Settings{RESET} {DIM}(Enter to keep default){RESET}"); + println!(" {DIM}────────────────────────────────────{RESET}"); + + // ── Radio ──────────────────────────────────────────────────── + + let freq_default = format!("{:.3}", radio.frequency as f64 / 1_000_000.0); + let freq_mhz: f64 = prompt_parse("Frequency (MHz)", &freq_default)?; + let frequency = (freq_mhz * 1_000_000.0) as u32; + + let bandwidth = prompt_default("Bandwidth", &radio.bandwidth)?; + + let sf_default = format!("{}", radio.spreading_factor); + let spreading_factor: u8 = prompt_parse("Spreading Factor", &sf_default)?; + + let cr_default = format!("{}", radio.coding_rate); + let coding_rate: u8 = prompt_parse("Coding Rate", &cr_default)?; + + let sw_default = format!("0x{:04X}", radio.sync_word); + let sync_word_str = prompt_default("Sync Word", &sw_default)?; + let sync_word = parse_sync_word(&sync_word_str)?; + + let tx_power = prompt_default("TX Power (dBm or 'max')", &radio.tx_power)?; + + let pre_default = format!("{}", radio.preamble); + let preamble: u16 = prompt_parse("Preamble", &pre_default)?; + + let cad_default = if radio.cad { "on" } else { "off" }; + let cad_str = prompt_default("CAD", cad_default)?; + let cad = matches!(cad_str.to_lowercase().as_str(), "on" | "true" | "yes" | "1"); + + let port_default = radio.port.as_deref().unwrap_or("auto"); + let port_str = prompt_default("Serial Port", port_default)?; + let port = if port_str == "auto" || port_str.is_empty() { + None + } else { + Some(port_str) + }; + + // ── Build and save ─────────────────────────────────────────── + + let cfg = Config { + radio: RadioSection { + port, + frequency, + bandwidth, + spreading_factor, + coding_rate, + sync_word, + tx_power, + preamble, + cad, + }, + bridge: BridgeSection { + passphrase, + dedup_window_secs: bridge.dedup_window_secs, + tx_queue_size: bridge.tx_queue_size, + rate_limit_pps: bridge.rate_limit_pps, + log_file: bridge.log_file, + }, + }; + + write_config(config_path, &cfg)?; + + println!(); + println!(" {GREEN}Config saved to {}{RESET}", config_path.display()); + println!(" Run {CYAN}donglora-bridge{RESET} to start!"); + println!(); + + Ok(()) +} + +fn write_config(path: &Path, cfg: &Config) -> Result<()> { + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent) + .with_context(|| format!("creating directory: {}", parent.display()))?; + } + let content = toml::to_string_pretty(cfg).context("serializing config")?; + std::fs::write(path, content) + .with_context(|| format!("writing config: {}", path.display()))?; + Ok(()) +} + +// ── Prompt helpers ─────────────────────────────────────────────── + +fn prompt_default(label: &str, default: &str) -> Result { + print!(" {label} {DIM}[{default}]{RESET}: "); + io::stdout().flush()?; + let mut input = String::new(); + io::stdin().read_line(&mut input)?; + let trimmed = input.trim(); + if trimmed.is_empty() { + Ok(default.to_string()) + } else { + Ok(trimmed.to_string()) + } +} + +fn prompt_required(label: &str, default: Option<&str>) -> Result { + loop { + let prompt = if let Some(d) = default { + format!(" {label} {DIM}[{d}]{RESET}: ") + } else { + format!(" {label}: ") + }; + print!("{prompt}"); + io::stdout().flush()?; + let mut input = String::new(); + io::stdin().read_line(&mut input)?; + let trimmed = input.trim(); + if !trimmed.is_empty() { + return Ok(trimmed.to_string()); + } + if let Some(d) = default { + return Ok(d.to_string()); + } + println!(" {YELLOW}This field is required.{RESET}"); + } +} + +fn prompt_parse(label: &str, default: &str) -> Result +where + T::Err: std::fmt::Display, +{ + loop { + let input = prompt_default(label, default)?; + match input.parse::() { + Ok(v) => return Ok(v), + Err(e) => println!(" {YELLOW}Invalid input: {e}{RESET}"), + } + } +} + +fn parse_sync_word(s: &str) -> Result { + let s = s.trim().to_lowercase(); + let s = s.strip_prefix("0x").unwrap_or(&s); + u16::from_str_radix(s, 16).with_context(|| format!("invalid sync word: {s}")) +} diff --git a/src/tui.rs b/src/tui.rs index 98de167..1fa54b5 100644 --- a/src/tui.rs +++ b/src/tui.rs @@ -7,109 +7,198 @@ use crossterm::terminal::{ EnterAlternateScreen, LeaveAlternateScreen, disable_raw_mode, enable_raw_mode, }; use crossterm::ExecutableCommand; -use iroh::PublicKey; use ratatui::backend::CrosstermBackend; -use ratatui::layout::{Constraint, Direction, Layout, Rect}; +use ratatui::layout::{Alignment, Constraint, Direction, Layout, Rect}; use ratatui::style::{Color, Modifier, Style}; +use ratatui::symbols; use ratatui::text::{Line, Span}; -use ratatui::widgets::{Block, Borders, Paragraph, Wrap}; +use ratatui::widgets::{Block, Borders, Clear, Paragraph, Sparkline}; use ratatui::Terminal; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; -use crate::network::Network; +use crate::gossip::Gossip; use crate::radio::ConfigSource; use crate::router::{PacketAction, PacketDirection, PacketLogEntry, RadioConfigInfo, Stats, StatsSnapshot}; const TICK_RATE: Duration = Duration::from_millis(250); const MAX_LOG_ENTRIES: usize = 200; +const BUCKET_SECS: u64 = 10; +const MAX_BUCKETS: usize = 60; + +// ── Color palette (consistent across panels) ───────────────────── + +const C_RADIO_RX: Color = Color::Green; +const C_RADIO_TX: Color = Color::Blue; +const C_NET_RX: Color = Color::Magenta; +const C_NET_TX: Color = Color::Cyan; +const C_DEDUP: Color = Color::DarkGray; +const C_RATE_DROP: Color = Color::Yellow; +const C_QUEUE_DROP: Color = Color::Red; +const C_SNR: Color = Color::Yellow; +const C_LABEL: Color = Color::DarkGray; +const C_VALUE: Color = Color::White; +const C_ACCENT: Color = Color::Cyan; +const C_BORDER: Color = Color::DarkGray; +const C_TITLE: Color = Color::White; + +// ── Time series ────────────────────────────────────────────────── + +#[derive(Default, Clone, Copy)] +struct Bucket { + radio_rx: u64, + radio_tx: u64, + snr_sum: i64, + snr_count: u64, +} + +struct TimeSeries { + buckets: Vec, + last_snap: StatsSnapshot, + last_bucket_time: Instant, +} + +impl TimeSeries { + fn new(snap: &StatsSnapshot) -> Self { + Self { + buckets: vec![Bucket::default(); MAX_BUCKETS], + last_snap: snap.clone(), + last_bucket_time: Instant::now(), + } + } + + fn update(&mut self, snap: &StatsSnapshot, log_entries: &[PacketLogEntry]) { + let now = Instant::now(); + if now.duration_since(self.last_bucket_time).as_secs() >= BUCKET_SECS { + let cutoff = self.last_bucket_time; + let (snr_sum, snr_count) = snr_stats_since(log_entries, cutoff); + let bucket = Bucket { + radio_rx: snap.radio_rx.saturating_sub(self.last_snap.radio_rx), + radio_tx: snap.radio_tx.saturating_sub(self.last_snap.radio_tx), + snr_sum, + snr_count, + }; + self.buckets.push(bucket); + if self.buckets.len() > MAX_BUCKETS { + self.buckets.remove(0); + } + self.last_snap = snap.clone(); + self.last_bucket_time = now; + } + } + + fn radio_rx(&self) -> Vec { + self.buckets.iter().map(|b| b.radio_rx).collect() + } + fn radio_tx(&self) -> Vec { + self.buckets.iter().map(|b| b.radio_tx).collect() + } + fn avg_snr(&self) -> Vec { + self.buckets + .iter() + .map(|b| { + if b.snr_count > 0 { + let avg = b.snr_sum / b.snr_count as i64; + (avg + 30).max(0) as u64 + } else { + 0 + } + }) + .collect() + } + + fn peak_rx(&self) -> u64 { + self.buckets.iter().map(|b| b.radio_rx).max().unwrap_or(0) + } + fn peak_tx(&self) -> u64 { + self.buckets.iter().map(|b| b.radio_tx).max().unwrap_or(0) + } + fn latest_avg_snr(&self) -> Option { + self.buckets.iter().rev().find(|b| b.snr_count > 0).map(|b| b.snr_sum / b.snr_count as i64) + } +} + +fn snr_stats_since(entries: &[PacketLogEntry], since: Instant) -> (i64, u64) { + let mut sum = 0i64; + let mut count = 0u64; + for e in entries.iter().rev() { + if e.timestamp <= since { + break; + } + if let Some(s) = e.snr { + sum += i64::from(s); + count += 1; + } + } + (sum, count) +} + +// ── Public API ─────────────────────────────────────────────────── + +type Term = Terminal>; -/// Run the TUI event loop. Blocks until `q` is pressed or cancel is triggered. pub async fn run( - node_id: PublicKey, config_watch: tokio::sync::watch::Receiver, - network: Arc, + gossip: Arc, stats: Arc, mut log_rx: mpsc::Receiver, cancel: CancellationToken, start_time: Instant, -) -> anyhow::Result<()> { +) -> anyhow::Result { enable_raw_mode()?; io::stdout().execute(EnterAlternateScreen)?; let backend = CrosstermBackend::new(io::stdout()); let mut terminal = Terminal::new(backend)?; let mut log_entries: Vec = Vec::with_capacity(MAX_LOG_ENTRIES); + let mut series = TimeSeries::new(&stats.snapshot()); - let result = tui_loop( - &mut terminal, - node_id, - config_watch, - &network, - &stats, - &mut log_rx, - &cancel, - start_time, - &mut log_entries, - ) - .await; + let _result = tui_loop( + &mut terminal, config_watch, &gossip, &stats, + &mut log_rx, &cancel, start_time, &mut log_entries, &mut series, + ).await; - disable_raw_mode()?; - io::stdout().execute(LeaveAlternateScreen)?; + terminal.draw(|frame| draw_shutdown_overlay(frame, frame.area()))?; + Ok(terminal) +} - result +pub fn restore_terminal(terminal: &mut Term) { + let _ = disable_raw_mode(); + let _ = io::stdout().execute(LeaveAlternateScreen); + let _ = terminal.show_cursor(); } #[allow(clippy::too_many_arguments)] async fn tui_loop( - terminal: &mut Terminal>, - node_id: PublicKey, + terminal: &mut Term, config_watch: tokio::sync::watch::Receiver, - network: &Network, + gossip: &Gossip, stats: &Stats, log_rx: &mut mpsc::Receiver, cancel: &CancellationToken, start_time: Instant, log_entries: &mut Vec, + series: &mut TimeSeries, ) -> anyhow::Result<()> { let mut last_tick = Instant::now(); - loop { - // Drain log entries. while let Ok(entry) = log_rx.try_recv() { if log_entries.len() >= MAX_LOG_ENTRIES { log_entries.remove(0); } log_entries.push(entry); } - if cancel.is_cancelled() { return Ok(()); } - let snap = stats.snapshot(); - let peers = network.peer_states().await; + series.update(&snap, log_entries); + let swarm_state = gossip.swarm_state(); let config_info = config_watch.borrow().clone(); - terminal.draw(|frame| { - let area = frame.area(); - draw_ui( - frame, - area, - &node_id, - &config_info, - &snap, - &peers, - log_entries, - start_time, - ); + draw_ui(frame, frame.area(), &config_info, &snap, &swarm_state, log_entries, start_time, series); })?; - - // Handle input with timeout for tick rate. - let timeout = TICK_RATE - .checked_sub(last_tick.elapsed()) - .unwrap_or(Duration::ZERO); - + let timeout = TICK_RATE.checked_sub(last_tick.elapsed()).unwrap_or(Duration::ZERO); if event::poll(timeout)? && let Event::Key(key) = event::read()? && key.kind == KeyEventKind::Press @@ -118,324 +207,410 @@ async fn tui_loop( cancel.cancel(); return Ok(()); } - if last_tick.elapsed() >= TICK_RATE { last_tick = Instant::now(); } } } +// ── Layout ─────────────────────────────────────────────────────── + #[allow(clippy::too_many_arguments)] fn draw_ui( frame: &mut ratatui::Frame, area: Rect, - node_id: &PublicKey, config_info: &RadioConfigInfo, stats: &StatsSnapshot, - peers: &[crate::network::PeerState], + swarm_state: &crate::gossip::SwarmState, log_entries: &[PacketLogEntry], start_time: Instant, + series: &TimeSeries, ) { - // Outer layout: header + body. let outer = Layout::default() .direction(Direction::Vertical) - .constraints([Constraint::Length(1), Constraint::Min(0)]) + .constraints([Constraint::Length(1), Constraint::Percentage(65), Constraint::Percentage(35)]) .split(area); - draw_header(frame, outer[0], node_id, start_time, stats); + draw_header(frame, outer[0], stats); - // Body: 2x2 grid. - let rows = Layout::default() - .direction(Direction::Vertical) - .constraints([Constraint::Percentage(40), Constraint::Percentage(60)]) + let cols = Layout::default() + .direction(Direction::Horizontal) + .constraints([Constraint::Length(30), Constraint::Min(40)]) .split(outer[1]); - let top_cols = Layout::default() - .direction(Direction::Horizontal) - .constraints([Constraint::Percentage(50), Constraint::Percentage(50)]) - .split(rows[0]); + let left = Layout::default() + .direction(Direction::Vertical) + .constraints([ + Constraint::Length(6), + Constraint::Length(12), + Constraint::Min(5), + ]) + .split(cols[0]); - let bot_cols = Layout::default() - .direction(Direction::Horizontal) - .constraints([Constraint::Percentage(35), Constraint::Percentage(65)]) - .split(rows[1]); - - draw_peers(frame, top_cols[0], peers); - draw_radio(frame, top_cols[1], config_info); - draw_stats(frame, bot_cols[0], stats, start_time); - draw_log(frame, bot_cols[1], log_entries); + draw_network(frame, left[0], swarm_state); + draw_radio(frame, left[1], config_info); + draw_stats(frame, left[2], stats, start_time); + draw_log(frame, cols[1], log_entries); + draw_activity(frame, outer[2], series); } -fn draw_header( - frame: &mut ratatui::Frame, - area: Rect, - node_id: &PublicKey, - start_time: Instant, - stats: &StatsSnapshot, -) { - let id_str = node_id.to_string(); - let id_short = if id_str.len() > 16 { - format!("{}...", &id_str[..16]) - } else { - id_str - }; - let radio_indicator = if stats.radio_connected { - Span::styled(" RADIO ", Style::default().fg(Color::Black).bg(Color::Green)) - } else { - Span::styled(" RADIO ", Style::default().fg(Color::Black).bg(Color::Red)) - }; - let uptime = format_duration(start_time.elapsed()); - let now = chrono_time(); +// ── Header ─────────────────────────────────────────────────────── +fn draw_header(frame: &mut ratatui::Frame, area: Rect, stats: &StatsSnapshot) { + let now = chrono_time(); let line = Line::from(vec![ - Span::styled( - " donglora-bridge ", - Style::default() - .fg(Color::White) - .add_modifier(Modifier::BOLD), - ), - radio_indicator, + Span::styled(" donglora-bridge ", Style::default().fg(C_TITLE).add_modifier(Modifier::BOLD)), + status_pill("RADIO", stats.radio_connected), + Span::raw(" "), + status_pill("NET", stats.neighbor_count > 0), Span::raw(" "), - Span::styled("ID: ", Style::default().fg(Color::DarkGray)), - Span::styled(id_short, Style::default().fg(Color::Cyan)), - Span::raw(" "), - Span::styled(format!("up {uptime}"), Style::default().fg(Color::DarkGray)), - Span::raw(" "), - Span::styled(now, Style::default().fg(Color::DarkGray)), - Span::styled(" q=quit", Style::default().fg(Color::DarkGray)), + Span::styled(now, Style::default().fg(C_LABEL)), + Span::styled(" q=quit", Style::default().fg(C_LABEL)), ]); frame.render_widget(Paragraph::new(line), area); } -fn draw_peers(frame: &mut ratatui::Frame, area: Rect, peers: &[crate::network::PeerState]) { - let block = Block::default() - .title(" Peers ") - .borders(Borders::ALL) - .border_style(Style::default().fg(Color::DarkGray)); +fn status_pill(label: &str, ok: bool) -> Span<'static> { + let (fg, bg) = if ok { + (Color::Black, Color::Green) + } else { + (Color::White, C_LABEL) + }; + Span::styled(format!(" {label} "), Style::default().fg(fg).bg(bg)) +} - let mut lines: Vec = Vec::new(); - if peers.is_empty() { - lines.push(Line::styled( - " No peers configured", - Style::default().fg(Color::DarkGray), - )); - } - for peer in peers { - let (dot, color) = if peer.connected { - ("●", Color::Green) - } else { - ("○", Color::Red) - }; - let status = if peer.connected { - let ago = peer - .last_seen - .map(|t| format_short_duration(t.elapsed())) - .unwrap_or_default(); - format!(" {ago}") - } else { - " reconnecting".into() - }; - let pk_short = { - let s = peer.public_key.to_string(); - if s.len() > 8 { format!(" ({}...)", &s[..8]) } else { format!(" ({s})") } - }; - let pkts = format!(" in:{} out:{}", peer.packets_in, peer.packets_out); - lines.push(Line::from(vec![ - Span::raw(" "), - Span::styled(dot, Style::default().fg(color)), - Span::raw(" "), - Span::styled( - peer.name.clone(), - Style::default().fg(Color::White).add_modifier(Modifier::BOLD), - ), - Span::styled(pk_short, Style::default().fg(Color::DarkGray)), - Span::styled(status, Style::default().fg(Color::DarkGray)), - Span::styled(pkts, Style::default().fg(Color::DarkGray)), - ])); - } +// ── Network panel (was Swarm) ──────────────────────────────────── + +fn draw_network(frame: &mut ratatui::Frame, area: Rect, state: &crate::gossip::SwarmState) { + let block = panel_block(" Network "); + + let dht_color = match state.dht_status { + crate::gossip::DhtStatus::Ready => Color::Green, + crate::gossip::DhtStatus::Bootstrapping => C_RATE_DROP, + crate::gossip::DhtStatus::PublishFailed => C_QUEUE_DROP, + }; + let neighbor_color = if state.neighbor_count > 0 { Color::Green } else { C_RATE_DROP }; + let last_pub = state + .last_dht_publish + .map(|t| format!("{}s ago", t.elapsed().as_secs())) + .unwrap_or_else(|| "-".into()); + + let lines = vec![ + kv_line("ID", &state.topic_hash, C_ACCENT), + kv_line("Peers", &format!("{}", state.neighbor_count), neighbor_color), + kv_line("DHT", &format!("{}", state.dht_status), dht_color), + kv_line("Published", &last_pub, C_LABEL), + ]; frame.render_widget(Paragraph::new(lines).block(block), area); } +// ── Radio panel ────────────────────────────────────────────────── + fn draw_radio(frame: &mut ratatui::Frame, area: Rect, config_info: &RadioConfigInfo) { - let title = match config_info.source { - ConfigSource::Ours => " Radio ", - ConfigSource::Mux => " Radio (mux config) ", - }; - let border_color = match config_info.source { - ConfigSource::Ours => Color::DarkGray, - ConfigSource::Mux => Color::Yellow, - }; + let is_mux = matches!(config_info.source, ConfigSource::Mux); + let border_color = if is_mux { C_RATE_DROP } else { C_BORDER }; let block = Block::default() - .title(title) + .title(Span::styled( + if is_mux { " Radio (mux) " } else { " Radio " }, + Style::default().fg(C_TITLE).add_modifier(Modifier::BOLD), + )) .borders(Borders::ALL) .border_style(Style::default().fg(border_color)); - let mut lines: Vec = Vec::new(); + let a = &config_info.active; + let r = &config_info.requested; - if matches!(config_info.source, ConfigSource::Mux) { - lines.push(Line::styled( - " Using mux locked config", - Style::default().fg(Color::Yellow), - )); - } + let freq_mhz = a.freq_hz as f64 / 1_000_000.0; + let bw_str = crate::radio::format_bandwidth(a.bw); + let power_str = if a.tx_power_dbm == donglora_client::TX_POWER_MAX { + "max".to_string() + } else { + format!("{} dBm", a.tx_power_dbm) + }; + let preamble = if a.preamble_len == 0 { 16 } else { a.preamble_len }; + let cad = if a.cad != 0 { "on" } else { "off" }; - lines.extend(config_info.config_str.split_whitespace().map(|part| { - let (label, value) = part.split_once(':').unwrap_or(("", part)); - if label.is_empty() { - Line::from(vec![ - Span::raw(" "), - Span::styled(value, Style::default().fg(Color::Cyan)), - ]) - } else { - Line::from(vec![ - Span::raw(" "), - Span::styled(format!("{label}: "), Style::default().fg(Color::DarkGray)), - Span::styled(value, Style::default().fg(Color::Cyan)), - ]) - } - })); + let cf = |label: &str, value: String, matches: bool| -> Line<'static> { + let color = if !is_mux || matches { C_ACCENT } else { C_RATE_DROP }; + let mark = if is_mux && !matches { " !" } else { "" }; + kv_line(label, &format!("{value}{mark}"), color) + }; - frame.render_widget(Paragraph::new(lines).block(block).wrap(Wrap { trim: false }), area); + let dev = if config_info.device.is_empty() { "-" } else { &config_info.device }; + let lines = vec![ + kv_line("Device", dev, C_ACCENT), + cf("Freq", format!("{freq_mhz:.3} MHz"), a.freq_hz == r.freq_hz), + cf("BW", bw_str.to_string(), a.bw == r.bw), + cf("SF", format!("{}", a.sf), a.sf == r.sf), + cf("CR", format!("4/{}", a.cr), a.cr == r.cr), + cf("Sync", format!("0x{:04X}", a.sync_word), a.sync_word == r.sync_word), + cf("TX", power_str, a.tx_power_dbm == r.tx_power_dbm), + cf("Preamble", format!("{preamble}"), a.preamble_len == r.preamble_len), + cf("CAD", cad.to_string(), a.cad == r.cad), + ]; + frame.render_widget(Paragraph::new(lines).block(block), area); } -fn draw_stats( - frame: &mut ratatui::Frame, - area: Rect, - stats: &StatsSnapshot, - start_time: Instant, -) { - let block = Block::default() - .title(" Stats ") - .borders(Borders::ALL) - .border_style(Style::default().fg(Color::DarkGray)); +// ── Stats panel ────────────────────────────────────────────────── - let stat_line = |label: &str, value: u64, color: Color| -> Line { +fn draw_stats(frame: &mut ratatui::Frame, area: Rect, stats: &StatsSnapshot, start_time: Instant) { + let block = panel_block(" Stats "); + + let s = |label: &str, value: u64, color: Color| -> Line<'static> { Line::from(vec![ - Span::raw(" "), - Span::styled( - format!("{label:<14}"), - Style::default().fg(Color::DarkGray), - ), - Span::styled(format!("{value:>8}"), Style::default().fg(color)), + Span::raw(" "), + Span::styled(format!("{label:<9}"), Style::default().fg(C_LABEL)), + Span::styled(format!("{value:>6}"), Style::default().fg(color)), ]) }; let uptime = format_duration(start_time.elapsed()); let lines = vec![ Line::from(vec![ - Span::raw(" "), - Span::styled("Uptime ", Style::default().fg(Color::DarkGray)), - Span::styled(format!("{uptime:>8}"), Style::default().fg(Color::White)), + Span::raw(" "), + Span::styled("Uptime ", Style::default().fg(C_LABEL)), + Span::styled(format!("{uptime:>6}"), Style::default().fg(C_VALUE)), ]), - stat_line("Radio RX", stats.radio_rx, Color::Green), - stat_line("Radio TX", stats.radio_tx, Color::Blue), - stat_line("Forwarded", stats.forwarded, Color::Cyan), - stat_line("Dropped SNR", stats.dropped_snr, Color::Yellow), - stat_line("Deduped", stats.dropped_dedup, Color::DarkGray), - stat_line("Queue drops", stats.dropped_queue, Color::Red), + s("RF RX", stats.radio_rx, C_RADIO_RX), + s("RF TX", stats.radio_tx, C_RADIO_TX), + s("Net RX", stats.gossip_rx, C_NET_RX), + s("Net TX", stats.gossip_tx, C_NET_TX), + s("Deduped", stats.dedup_hits, C_DEDUP), + s("Rate lim", stats.rate_limit_drops, C_RATE_DROP), + s("Q drop", stats.dropped_queue, C_QUEUE_DROP), ]; + frame.render_widget(Paragraph::new(lines).block(block), area); +} + +// ── Packet log ─────────────────────────────────────────────────── + +fn draw_log(frame: &mut ratatui::Frame, area: Rect, entries: &[PacketLogEntry]) { + let block = panel_block(" Packet Log "); + + let inner_height = area.height.saturating_sub(2) as usize; + let data_height = inner_height.saturating_sub(1); // header row + let start = entries.len().saturating_sub(data_height); + let visible = &entries[start..]; + + // Column header with underline effect via bold. + let header = Line::from(vec![ + Span::styled(" Age", Style::default().fg(C_LABEL).add_modifier(Modifier::BOLD)), + Span::styled(" RF", Style::default().fg(C_LABEL).add_modifier(Modifier::BOLD)), + Span::styled(" Net", Style::default().fg(C_LABEL).add_modifier(Modifier::BOLD)), + Span::styled(" Size", Style::default().fg(C_LABEL).add_modifier(Modifier::BOLD)), + Span::styled(" RSSI", Style::default().fg(C_LABEL).add_modifier(Modifier::BOLD)), + Span::styled(" SNR", Style::default().fg(C_LABEL).add_modifier(Modifier::BOLD)), + Span::styled(" Act", Style::default().fg(C_LABEL).add_modifier(Modifier::BOLD)), + ]); + + let mut lines = vec![header]; + + for e in visible { + let age = format_short_duration(e.timestamp.elapsed()); + + let dash = " ───"; + let (rf_col, net_col) = match e.direction { + PacketDirection::RadioRx => ( + Span::styled(" RX", Style::default().fg(C_RADIO_RX)), + Span::styled(dash, Style::default().fg(C_BORDER)), + ), + PacketDirection::RadioTx => ( + Span::styled(" TX", Style::default().fg(C_RADIO_TX)), + Span::styled(dash, Style::default().fg(C_BORDER)), + ), + PacketDirection::GossipRx => ( + Span::styled(dash, Style::default().fg(C_BORDER)), + Span::styled(" RX", Style::default().fg(C_NET_RX)), + ), + }; + + let action_color = match e.action { + PacketAction::Forwarded => C_NET_TX, + PacketAction::Transmitted => C_RADIO_TX, + PacketAction::DroppedDedup => C_DEDUP, + PacketAction::DroppedQueueFull => C_QUEUE_DROP, + PacketAction::DroppedRateLimit => C_RATE_DROP, + }; + + let rssi_span = match e.rssi { + Some(r) => Span::styled(format!("{r:>6}"), Style::default().fg(rssi_color(r))), + None => Span::styled(" -", Style::default().fg(C_LABEL)), + }; + let snr_span = match e.snr { + Some(s) => Span::styled(format!("{s:>5}"), Style::default().fg(snr_color(s))), + None => Span::styled(" -", Style::default().fg(C_LABEL)), + }; + + lines.push(Line::from(vec![ + Span::styled(format!(" {age:>3} "), Style::default().fg(C_LABEL)), + rf_col, + net_col, + Span::styled(format!("{:>5}B", e.size), Style::default().fg(C_VALUE)), + rssi_span, + snr_span, + Span::styled(format!(" {}", e.action), Style::default().fg(action_color)), + ])); + } frame.render_widget(Paragraph::new(lines).block(block), area); } -fn draw_log(frame: &mut ratatui::Frame, area: Rect, entries: &[PacketLogEntry]) { +// ── Activity chart ─────────────────────────────────────────────── + +fn draw_activity(frame: &mut ratatui::Frame, area: Rect, series: &TimeSeries) { + let block = panel_block(" Activity (10s buckets) "); + let inner = block.inner(area); + frame.render_widget(block, area); + + if inner.height < 3 || inner.width < 20 { + return; + } + + let rows = Layout::default() + .direction(Direction::Vertical) + .constraints([ + Constraint::Ratio(1, 3), + Constraint::Ratio(1, 3), + Constraint::Ratio(1, 3), + ]) + .split(inner); + + let rx_data = series.radio_rx(); + let tx_data = series.radio_tx(); + let snr_data = series.avg_snr(); + + let label_w = 12u16; + let spark_w = inner.width.saturating_sub(label_w) as usize; + + let peak_rx = series.peak_rx(); + let peak_tx = series.peak_tx(); + let snr_label = series + .latest_avg_snr() + .map(|s| format!("avg:{s}dB")) + .unwrap_or_else(|| "-".into()); + + let rx_padded = pad_right_align(&rx_data, spark_w); + let tx_padded = pad_right_align(&tx_data, spark_w); + let snr_padded = pad_right_align(&snr_data, spark_w); + + draw_sparkline_row(frame, rows[0], "RF RX", &format!("pk:{peak_rx}"), &rx_padded, C_RADIO_RX, label_w); + draw_sparkline_row(frame, rows[1], "RF TX", &format!("pk:{peak_tx}"), &tx_padded, C_RADIO_TX, label_w); + draw_sparkline_row(frame, rows[2], "SNR", &snr_label, &snr_padded, C_SNR, label_w); +} + +fn draw_sparkline_row( + frame: &mut ratatui::Frame, + area: Rect, + label: &str, + detail: &str, + data: &[u64], + color: Color, + label_width: u16, +) { + if area.width < label_width + 5 || area.height < 1 { + return; + } + + let cols = Layout::default() + .direction(Direction::Horizontal) + .constraints([Constraint::Length(label_width), Constraint::Min(1)]) + .split(area); + + // Label + detail stacked if room, otherwise just label. + let mut label_lines = vec![ + Line::from(Span::styled(format!(" {label}"), Style::default().fg(color).add_modifier(Modifier::BOLD))), + ]; + if cols[0].height > 1 { + label_lines.push(Line::from(Span::styled(format!(" {detail}"), Style::default().fg(C_LABEL)))); + } + frame.render_widget(Paragraph::new(label_lines), cols[0]); + + let spark = Sparkline::default() + .data(data) + .bar_set(symbols::bar::NINE_LEVELS) + .style(Style::default().fg(color)); + frame.render_widget(spark, cols[1]); +} + +/// Pad data with leading zeros so values are right-aligned in `width` columns. +fn pad_right_align(data: &[u64], width: usize) -> Vec { + let tail_start = data.len().saturating_sub(width); + let tail = &data[tail_start..]; + let mut padded = vec![0u64; width.saturating_sub(tail.len())]; + padded.extend_from_slice(tail); + padded +} + +// ── Shutdown overlay ───────────────────────────────────────────── + +fn draw_shutdown_overlay(frame: &mut ratatui::Frame, area: Rect) { + let dim = Paragraph::new("").style(Style::default().bg(Color::Black)); + frame.render_widget(dim, area); + + let w = 36u16; + let h = 3u16; + let x = area.x + area.width.saturating_sub(w) / 2; + let y = area.y + area.height.saturating_sub(h) / 2; + let popup = Rect::new(x, y, w.min(area.width), h.min(area.height)); + + frame.render_widget(Clear, popup); let block = Block::default() - .title(" Packet Log ") .borders(Borders::ALL) - .border_style(Style::default().fg(Color::DarkGray)); + .border_style(Style::default().fg(C_BORDER)); + let text = Paragraph::new(Line::from(vec![ + Span::styled(" Shutting down", Style::default().fg(C_RATE_DROP)), + Span::styled(" — updating DHT...", Style::default().fg(C_LABEL)), + ])) + .alignment(Alignment::Center) + .block(block); + frame.render_widget(text, popup); +} - let inner_height = area.height.saturating_sub(2) as usize; - let start = entries.len().saturating_sub(inner_height); - let visible = &entries[start..]; +// ── Shared helpers ─────────────────────────────────────────────── - let lines: Vec = visible - .iter() - .map(|e| { - let age = format_short_duration(e.timestamp.elapsed()); - let dir_color = match e.direction { - PacketDirection::RadioRx => Color::Green, - PacketDirection::RadioTx => Color::Blue, - PacketDirection::NetRx => Color::Magenta, - }; - let action_color = match e.action { - PacketAction::Forwarded => Color::Cyan, - PacketAction::Transmitted => Color::Blue, - PacketAction::DroppedDedup => Color::DarkGray, - PacketAction::DroppedQueueFull => Color::Red, - }; - let rssi_str = e - .rssi - .map(|r| format!(" {r}dBm")) - .unwrap_or_default(); - let snr_str = e - .snr - .map(|s| format!(" snr:{s}")) - .unwrap_or_default(); - let peer_str = e - .peer_name - .as_deref() - .map(|n| format!(" {n}")) - .unwrap_or_default(); +fn panel_block(title: &str) -> Block<'static> { + Block::default() + .title(Span::styled( + title.to_string(), + Style::default().fg(C_TITLE).add_modifier(Modifier::BOLD), + )) + .borders(Borders::ALL) + .border_style(Style::default().fg(C_BORDER)) +} - Line::from(vec![ - Span::styled( - format!(" {age:>4} "), - Style::default().fg(Color::DarkGray), - ), - Span::styled( - format!("{}", e.direction), - Style::default().fg(dir_color), - ), - Span::styled( - format!(" {:>4}B", e.size), - Style::default().fg(Color::White), - ), - Span::styled(rssi_str, Style::default().fg(Color::DarkGray)), - Span::styled(snr_str, Style::default().fg(Color::DarkGray)), - Span::styled(peer_str, Style::default().fg(Color::DarkGray)), - Span::raw(" "), - Span::styled( - format!("{}", e.action), - Style::default().fg(action_color), - ), - ]) - }) - .collect(); +fn rssi_color(rssi: i16) -> Color { + if rssi >= -70 { Color::Green } else if rssi >= -90 { C_RATE_DROP } else { C_QUEUE_DROP } +} - frame.render_widget(Paragraph::new(lines).block(block), area); +fn snr_color(snr: i16) -> Color { + if snr >= 5 { Color::Green } else if snr >= 0 { C_RATE_DROP } else { C_QUEUE_DROP } +} + +fn kv_line(label: &str, value: &str, color: Color) -> Line<'static> { + Line::from(vec![ + Span::raw(" "), + Span::styled(format!("{label:<10}"), Style::default().fg(C_LABEL)), + Span::styled(value.to_string(), Style::default().fg(color)), + ]) } fn format_duration(d: Duration) -> String { let secs = d.as_secs(); - if secs < 60 { - format!("{secs}s") - } else if secs < 3600 { - format!("{}m {}s", secs / 60, secs % 60) - } else { - format!("{}h {}m", secs / 3600, (secs % 3600) / 60) - } + if secs < 60 { format!("{secs}s") } + else if secs < 3600 { format!("{}m{}s", secs / 60, secs % 60) } + else { format!("{}h{}m", secs / 3600, (secs % 3600) / 60) } } fn format_short_duration(d: Duration) -> String { let secs = d.as_secs(); - if secs < 60 { - format!("{secs}s") - } else if secs < 3600 { - format!("{}m", secs / 60) - } else { - format!("{}h", secs / 3600) - } + if secs < 60 { format!("{secs}s") } + else if secs < 3600 { format!("{}m", secs / 60) } + else { format!("{}h", secs / 3600) } } fn chrono_time() -> String { - // Use a simple approach without chrono dependency. let now = std::time::SystemTime::now(); - let since_epoch = now - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default(); + let since_epoch = now.duration_since(std::time::UNIX_EPOCH).unwrap_or_default(); let secs = since_epoch.as_secs(); - let hours = (secs / 3600) % 24; - let minutes = (secs % 3600) / 60; - let seconds = secs % 60; - format!("{hours:02}:{minutes:02}:{seconds:02}") + format!("{:02}:{:02}:{:02}", (secs / 3600) % 24, (secs % 3600) / 60, secs % 60) }