//! Shared state and command types for the daemon (supervisor + web server). //! //! The supervisor owns all mutable state. The web server gets read-only access //! to status via `Arc>` and sends commands via an mpsc channel. use std::collections::HashMap; use std::path::PathBuf; use std::sync::atomic::AtomicU64; use std::sync::mpsc; use std::sync::{Arc, RwLock}; use std::time::{Instant, SystemTime}; use crate::config::Config; /// Default port for the built-in web UI / API server. pub const DEFAULT_WEB_PORT: u16 = 8090; /// Shared application state accessible by both the supervisor and web server. pub struct AppState { /// Current active configuration (read by UI, updated on reload). pub config: Arc>, /// Live daemon status (updated by supervisor every poll cycle). pub status: Arc>, /// Command channel: web server → supervisor. pub cmd_tx: mpsc::Sender, /// Path to the config file on disk. pub config_path: PathBuf, /// SSE broadcast: supervisor sends `()` after each status update; /// web server subscribers render partials and push to connected clients. pub sse_tx: tokio::sync::broadcast::Sender<()>, } /// Overall daemon status, updated by the supervisor loop. pub struct DaemonStatus { /// When the daemon started. pub started_at: Instant, /// Per-share mount status. pub shares: Vec, /// Whether smbd is running. pub smbd_running: bool, /// Whether WebDAV is running. pub webdav_running: bool, /// Whether NFS exports are active. pub nfs_exported: bool, /// Per-rule warmup status (populated when warmup is triggered). pub warmup: Vec, /// Generation counter — incremented each time warmup is (re)started. /// Workers check this to detect when they've been superseded. pub warmup_generation: u64, /// Whether any dir-refresh threads are active. pub dir_refresh_running: bool, /// Timestamp of the last successful dir-refresh call, keyed by share name. pub last_dir_refresh: HashMap, /// Generation counter for dir-refresh threads (incremented on each re-spawn). pub dir_refresh_generation: u64, /// Shared atomic for generation — dir-refresh threads hold a clone and check /// this directly (without taking the RwLock) to detect supersession. pub dir_refresh_gen_arc: Arc, /// Number of subdirectories successfully refreshed in the last cycle, keyed by share name. pub dir_refresh_dirs_ok: HashMap, /// Number of subdirectories that failed to refresh in the last cycle, keyed by share name. pub dir_refresh_dirs_failed: HashMap, } impl DaemonStatus { /// Create initial status for a set of share names. pub fn new(share_names: &[String]) -> Self { Self { started_at: Instant::now(), shares: share_names .iter() .map(|name| ShareStatus { name: name.clone(), mounted: false, rc_port: 0, cache_bytes: 0, dirty_count: 0, errored_files: 0, speed: 0.0, transfers: 0, errors: 0, health: ShareHealth::Pending, }) .collect(), smbd_running: false, webdav_running: false, nfs_exported: false, warmup: Vec::new(), warmup_generation: 0, dir_refresh_running: false, last_dir_refresh: HashMap::new(), dir_refresh_generation: 0, dir_refresh_gen_arc: Arc::new(AtomicU64::new(0)), dir_refresh_dirs_ok: HashMap::new(), dir_refresh_dirs_failed: HashMap::new(), } } /// Aggregate warmup summary for a share. /// /// Returns `(state_label, done_count, total_count)` where state_label is /// one of "none", "pending", "running", "complete", "failed". pub fn warmup_summary_for(&self, share_name: &str) -> (&str, usize, usize) { let rules: Vec<&WarmupRuleStatus> = self .warmup .iter() .filter(|r| r.share == share_name) .collect(); if rules.is_empty() { return ("none", 0, 0); } let total: usize = rules.iter().map(|r| r.total_files).sum(); let done: usize = rules.iter().map(|r| r.cached + r.skipped).sum(); let any_failed = rules .iter() .any(|r| matches!(r.state, WarmupRuleState::Failed(_))); let any_running = rules .iter() .any(|r| matches!(r.state, WarmupRuleState::Listing | WarmupRuleState::Caching)); let all_complete = rules .iter() .all(|r| matches!(r.state, WarmupRuleState::Complete)); let label = if any_failed { "failed" } else if any_running { "running" } else if all_complete { "complete" } else { "pending" }; (label, done, total) } /// How long ago the last dir-refresh ran for this share. /// /// Returns `None` if no refresh has completed yet for the share. pub fn dir_refresh_ago_for(&self, share_name: &str) -> Option { let ts = self.last_dir_refresh.get(share_name)?; let secs = ts.elapsed().unwrap_or_default().as_secs(); Some(format_ago(secs)) } /// Format uptime as a human-readable string. pub fn uptime_string(&self) -> String { let secs = self.started_at.elapsed().as_secs(); let days = secs / 86400; let hours = (secs % 86400) / 3600; let mins = (secs % 3600) / 60; if days > 0 { format!("{days}d {hours}h {mins}m") } else if hours > 0 { format!("{hours}h {mins}m") } else { format!("{mins}m") } } } /// Per-share runtime status. pub struct ShareStatus { /// Share name (matches ShareConfig.name). pub name: String, /// Whether the FUSE mount is active. pub mounted: bool, /// RC API port for this share's rclone instance. pub rc_port: u16, /// Bytes used in VFS disk cache. pub cache_bytes: u64, /// Number of dirty files (uploads_in_progress + uploads_queued). pub dirty_count: u64, /// Number of errored cache files. pub errored_files: u64, /// Current transfer speed (bytes/sec from core/stats). pub speed: f64, /// Active transfer count. pub transfers: u64, /// Cumulative error count. pub errors: u64, /// Pre-mount probe result. pub health: ShareHealth, } impl ShareStatus { /// Format cache size as human-readable string. pub fn cache_display(&self) -> String { format_bytes(self.cache_bytes) } /// Format speed as human-readable string. pub fn speed_display(&self) -> String { if self.speed < 1.0 { "-".to_string() } else { format!("{}/s", format_bytes(self.speed as u64)) } } /// Human-readable health label: "PENDING", "PROBING", "OK", or "FAILED". pub fn health_label(&self) -> &str { match &self.health { ShareHealth::Pending => "PENDING", ShareHealth::Probing => "PROBING", ShareHealth::Healthy => "OK", ShareHealth::Failed(_) => "FAILED", } } /// Error message when health is Failed, None otherwise. pub fn health_message(&self) -> Option<&str> { match &self.health { ShareHealth::Failed(msg) => Some(msg), _ => None, } } /// Whether the share is healthy (probe succeeded). pub fn is_healthy(&self) -> bool { self.health == ShareHealth::Healthy } } /// Format an elapsed-seconds count as "5s ago", "3m ago", "2h ago". fn format_ago(secs: u64) -> String { if secs < 60 { format!("{secs}s ago") } else if secs < 3600 { format!("{}m ago", secs / 60) } else { format!("{}h ago", secs / 3600) } } /// Format bytes as human-readable (e.g. "45.2 GiB"). fn format_bytes(bytes: u64) -> String { const KIB: f64 = 1024.0; const MIB: f64 = KIB * 1024.0; const GIB: f64 = MIB * 1024.0; const TIB: f64 = GIB * 1024.0; let b = bytes as f64; if b >= TIB { format!("{:.1} TiB", b / TIB) } else if b >= GIB { format!("{:.1} GiB", b / GIB) } else if b >= MIB { format!("{:.1} MiB", b / MIB) } else if b >= KIB { format!("{:.1} KiB", b / KIB) } else { format!("{bytes} B") } } /// Per-share health state from pre-mount probing. #[derive(Clone, Debug, PartialEq)] pub enum ShareHealth { /// Not yet probed (initial state). Pending, /// Probe in progress. Probing, /// Remote path verified, ready to mount. Healthy, /// Probe failed — share will not be mounted. Failed(String), } /// Per-rule warmup progress, updated by the warmup worker thread. #[derive(Clone, Debug, serde::Serialize)] pub struct WarmupRuleStatus { pub share: String, pub path: String, pub newer_than: Option, pub state: WarmupRuleState, pub total_files: usize, pub skipped: usize, pub cached: usize, pub errors: usize, } /// State machine for a single warmup rule. #[derive(Clone, Debug, PartialEq, serde::Serialize)] pub enum WarmupRuleState { Pending, Listing, Caching, Complete, Failed(String), } /// Commands sent from the web server (or CLI) to the supervisor. pub enum SupervisorCmd { /// Apply a new configuration (triggers tiered reload). Reload(Config), /// Graceful shutdown. Shutdown, /// Live bandwidth adjustment (Tier A — no restart needed). BwLimit { up: String, down: String }, /// Reconnect (re-probe + re-mount) a single share by name. Reconnect(String), } #[cfg(test)] mod tests { use super::*; #[test] fn test_format_bytes() { assert_eq!(format_bytes(0), "0 B"); assert_eq!(format_bytes(500), "500 B"); assert_eq!(format_bytes(1024), "1.0 KiB"); assert_eq!(format_bytes(1536), "1.5 KiB"); assert_eq!(format_bytes(1048576), "1.0 MiB"); assert_eq!(format_bytes(1073741824), "1.0 GiB"); assert_eq!(format_bytes(1099511627776), "1.0 TiB"); } #[test] fn test_daemon_status_new() { let names = vec!["photos".to_string(), "projects".to_string()]; let status = DaemonStatus::new(&names); assert_eq!(status.shares.len(), 2); assert_eq!(status.shares[0].name, "photos"); assert_eq!(status.shares[1].name, "projects"); assert!(!status.smbd_running); assert!(!status.webdav_running); assert!(!status.nfs_exported); assert!(status.warmup.is_empty()); assert_eq!(status.warmup_generation, 0); } #[test] fn test_uptime_string() { let status = DaemonStatus::new(&[]); let uptime = status.uptime_string(); assert!(uptime.contains('m')); } #[test] fn test_share_status_display() { let share = ShareStatus { name: "test".into(), mounted: true, rc_port: 5572, cache_bytes: 48_578_891_776, // ~45.2 GiB dirty_count: 3, errored_files: 0, speed: 2_200_000.0, transfers: 2, errors: 0, health: ShareHealth::Healthy, }; assert!(share.cache_display().contains("GiB")); assert!(share.speed_display().contains("/s")); } #[test] fn test_share_status_no_speed() { let share = ShareStatus { name: "test".into(), mounted: true, rc_port: 5572, cache_bytes: 0, dirty_count: 0, errored_files: 0, speed: 0.0, transfers: 0, errors: 0, health: ShareHealth::Pending, }; assert_eq!(share.speed_display(), "-"); } #[test] fn test_share_health_labels() { let mut share = ShareStatus { name: "test".into(), mounted: false, rc_port: 0, cache_bytes: 0, dirty_count: 0, errored_files: 0, speed: 0.0, transfers: 0, errors: 0, health: ShareHealth::Pending, }; assert_eq!(share.health_label(), "PENDING"); assert!(share.health_message().is_none()); assert!(!share.is_healthy()); share.health = ShareHealth::Probing; assert_eq!(share.health_label(), "PROBING"); share.health = ShareHealth::Healthy; assert_eq!(share.health_label(), "OK"); assert!(share.is_healthy()); share.health = ShareHealth::Failed("remote path not found".into()); assert_eq!(share.health_label(), "FAILED"); assert_eq!(share.health_message(), Some("remote path not found")); assert!(!share.is_healthy()); } #[test] fn test_warmup_summary_no_rules() { let status = DaemonStatus::new(&["photos".to_string()]); let (label, done, total) = status.warmup_summary_for("photos"); assert_eq!(label, "none"); assert_eq!(done, 0); assert_eq!(total, 0); } #[test] fn test_warmup_summary_pending() { let mut status = DaemonStatus::new(&["photos".to_string()]); status.warmup.push(WarmupRuleStatus { share: "photos".into(), path: "/2024".into(), newer_than: None, state: WarmupRuleState::Pending, total_files: 0, skipped: 0, cached: 0, errors: 0, }); let (label, _, _) = status.warmup_summary_for("photos"); assert_eq!(label, "pending"); } #[test] fn test_warmup_summary_running() { let mut status = DaemonStatus::new(&["photos".to_string()]); status.warmup.push(WarmupRuleStatus { share: "photos".into(), path: "/2024".into(), newer_than: None, state: WarmupRuleState::Caching, total_files: 100, skipped: 10, cached: 40, errors: 0, }); let (label, done, total) = status.warmup_summary_for("photos"); assert_eq!(label, "running"); assert_eq!(done, 50); assert_eq!(total, 100); } #[test] fn test_warmup_summary_complete() { let mut status = DaemonStatus::new(&["photos".to_string()]); status.warmup.push(WarmupRuleStatus { share: "photos".into(), path: "/2024".into(), newer_than: None, state: WarmupRuleState::Complete, total_files: 100, skipped: 30, cached: 70, errors: 0, }); let (label, done, total) = status.warmup_summary_for("photos"); assert_eq!(label, "complete"); assert_eq!(done, 100); assert_eq!(total, 100); } #[test] fn test_warmup_summary_wrong_share() { let mut status = DaemonStatus::new(&["photos".to_string()]); status.warmup.push(WarmupRuleStatus { share: "photos".into(), path: "/2024".into(), newer_than: None, state: WarmupRuleState::Caching, total_files: 50, skipped: 0, cached: 10, errors: 0, }); let (label, _, _) = status.warmup_summary_for("videos"); assert_eq!(label, "none"); } }