pulpod 0.0.42

pub mod auth;
pub mod config;
mod embed;
pub mod event_push;
pub mod events;
pub mod fleet;
pub mod health;
pub mod inks;
pub mod node;
pub mod notifications;
pub mod peers;
pub mod push;

pub mod node_auth;
pub mod node_commands;
pub mod routes;
pub mod schedules;
pub mod secrets;
mod session_remote;
pub mod sessions;
pub mod static_files;
pub mod watchdog;
pub mod ws;

use std::path::PathBuf;
use std::sync::Arc;

use axum::Router;
use pulpo_common::event::PulpoEvent;
use tokio::sync::{RwLock, broadcast};

use crate::config::Config;
use crate::controller::{CommandQueue, SessionIndex};
use crate::peers::PeerRegistry;
use crate::session::manager::SessionManager;
use crate::store::Store;
use crate::watchdog::WatchdogRuntimeConfig;

const EVENT_CHANNEL_CAPACITY: usize = 256;

pub struct AppState {
    pub config: Arc<RwLock<Config>>,
    pub config_path: PathBuf,
    pub session_manager: SessionManager,
    pub peer_registry: PeerRegistry,
    pub store: Store,
    /// On-demand peer prober with TTL cache. Only present in production builds;
    /// excluded under coverage because the monomorphized
    /// `CachedProber<HttpPeerProber>` methods are never exercised (the handler
    /// call site is also gated) and would produce uncoverable lines.
    #[cfg(not(coverage))]
    pub cached_prober:
        Option<crate::peers::health::CachedProber<crate::peers::health::HttpPeerProber>>,
    pub event_tx: broadcast::Sender<PulpoEvent>,
    /// Watch channel sender for pushing watchdog config changes to the running loop.
    pub watchdog_config_tx: Option<tokio::sync::watch::Sender<WatchdogRuntimeConfig>>,
    /// Controller session index, hydrated from `SQLite` on startup and kept hot in memory.
    pub session_index: Option<Arc<SessionIndex>>,
    /// Command queue for controller mode (pending commands for nodes).
    pub command_queue: Option<Arc<CommandQueue>>,
}

impl AppState {
    pub fn new(
        config: Config,
        session_manager: SessionManager,
        peer_registry: PeerRegistry,
        store: Store,
    ) -> Arc<Self> {
        let (event_tx, _) = broadcast::channel(EVENT_CHANNEL_CAPACITY);
        Arc::new(Self {
            config: Arc::new(RwLock::new(config)),
            config_path: PathBuf::new(),
            session_manager,
            peer_registry,
            store,
            #[cfg(not(coverage))]
            cached_prober: None,
            event_tx,
            watchdog_config_tx: None,
            session_index: None,
            command_queue: None,
        })
    }

    pub fn with_event_tx(
        config: Config,
        config_path: PathBuf,
        session_manager: SessionManager,
        peer_registry: PeerRegistry,
        event_tx: broadcast::Sender<PulpoEvent>,
        store: Store,
    ) -> Arc<Self> {
        Arc::new(Self {
            config: Arc::new(RwLock::new(config)),
            config_path,
            session_manager,
            peer_registry,
            store,
            #[cfg(not(coverage))]
            cached_prober: Some(crate::peers::health::CachedProber::new(
                crate::peers::health::HttpPeerProber::new(),
                std::time::Duration::from_secs(60),
            )),
            event_tx,
            watchdog_config_tx: None,
            session_index: None,
            command_queue: None,
        })
    }

    pub fn with_event_tx_controller(
        config: Config,
        config_path: PathBuf,
        session_manager: SessionManager,
        peer_registry: PeerRegistry,
        event_tx: broadcast::Sender<PulpoEvent>,
        store: Store,
        session_index: Option<Arc<SessionIndex>>,
        command_queue: Option<Arc<CommandQueue>>,
    ) -> Arc<Self> {
        Arc::new(Self {
            config: Arc::new(RwLock::new(config)),
            config_path,
            session_manager,
            peer_registry,
            store,
            #[cfg(not(coverage))]
            cached_prober: Some(crate::peers::health::CachedProber::new(
                crate::peers::health::HttpPeerProber::new(),
                std::time::Duration::from_secs(60),
            )),
            event_tx,
            watchdog_config_tx: None,
            session_index,
            command_queue,
        })
    }

    /// Full constructor with all optional fields (watchdog + controller mode).
    #[allow(clippy::too_many_arguments)]
    pub fn with_all(
        config: Config,
        config_path: PathBuf,
        session_manager: SessionManager,
        peer_registry: PeerRegistry,
        event_tx: broadcast::Sender<PulpoEvent>,
        watchdog_config_tx: Option<tokio::sync::watch::Sender<WatchdogRuntimeConfig>>,
        store: Store,
        session_index: Option<Arc<SessionIndex>>,
        command_queue: Option<Arc<CommandQueue>>,
    ) -> Arc<Self> {
        Arc::new(Self {
            config: Arc::new(RwLock::new(config)),
            config_path,
            session_manager,
            peer_registry,
            store,
            #[cfg(not(coverage))]
            cached_prober: Some(crate::peers::health::CachedProber::new(
                crate::peers::health::HttpPeerProber::new(),
                std::time::Duration::from_secs(60),
            )),
            event_tx,
            watchdog_config_tx,
            session_index,
            command_queue,
        })
    }

    pub fn with_watchdog_tx(
        config: Config,
        config_path: PathBuf,
        session_manager: SessionManager,
        peer_registry: PeerRegistry,
        event_tx: broadcast::Sender<PulpoEvent>,
        watchdog_config_tx: Option<tokio::sync::watch::Sender<WatchdogRuntimeConfig>>,
        store: Store,
    ) -> Arc<Self> {
        Arc::new(Self {
            config: Arc::new(RwLock::new(config)),
            config_path,
            session_manager,
            peer_registry,
            store,
            #[cfg(not(coverage))]
            cached_prober: Some(crate::peers::health::CachedProber::new(
                crate::peers::health::HttpPeerProber::new(),
                std::time::Duration::from_secs(60),
            )),
            event_tx,
            watchdog_config_tx,
            session_index: None,
            command_queue: None,
        })
    }
}

pub fn router(state: Arc<AppState>) -> Router {
    routes::build(state)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::HashMap;

    use crate::backend::StubBackend;
    use crate::config::{Config, NodeConfig};
    use crate::peers::PeerRegistry;
    use crate::store::Store;

    #[tokio::test]
    async fn test_app_state_new() {
        let tmpdir = tempfile::tempdir().unwrap();
        let store = Store::new(tmpdir.path().to_str().unwrap()).await.unwrap();
        store.migrate().await.unwrap();
        let config = Config {
            node: NodeConfig {
                name: "test".into(),
                port: 7433,
                data_dir: tmpdir.path().to_str().unwrap().into(),
                ..NodeConfig::default()
            },
            auth: crate::config::AuthConfig::default(),
            peers: HashMap::new(),

            watchdog: crate::config::WatchdogConfig::default(),
            inks: HashMap::new(),
            notifications: crate::config::NotificationsConfig::default(),
            docker: crate::config::DockerConfig::default(),
            controller: crate::config::ControllerConfig::default(),
        };
        let backend = Arc::new(StubBackend);
        let manager =
            SessionManager::new(backend, store.clone(), HashMap::new(), None).with_no_stale_grace();
        let peer_registry = PeerRegistry::new(&HashMap::new());
        let state = AppState::new(config, manager, peer_registry, store);
        assert_eq!(state.config.read().await.node.name, "test");
        assert!(state.config_path.as_os_str().is_empty());
    }

    #[tokio::test]
    async fn test_app_state_with_event_tx() {
        let tmpdir = tempfile::tempdir().unwrap();
        let store = Store::new(tmpdir.path().to_str().unwrap()).await.unwrap();
        store.migrate().await.unwrap();
        let config = Config {
            node: NodeConfig {
                name: "test".into(),
                port: 7433,
                data_dir: tmpdir.path().to_str().unwrap().into(),
                ..NodeConfig::default()
            },
            auth: crate::config::AuthConfig::default(),
            peers: HashMap::new(),

            watchdog: crate::config::WatchdogConfig::default(),
            inks: HashMap::new(),
            notifications: crate::config::NotificationsConfig::default(),
            docker: crate::config::DockerConfig::default(),
            controller: crate::config::ControllerConfig::default(),
        };
        let config_path = tmpdir.path().join("config.toml");
        let backend = Arc::new(StubBackend);
        let manager =
            SessionManager::new(backend, store.clone(), HashMap::new(), None).with_no_stale_grace();
        let peer_registry = PeerRegistry::new(&HashMap::new());
        let (event_tx, _) = tokio::sync::broadcast::channel(16);
        let state = AppState::with_event_tx(
            config,
            config_path.clone(),
            manager,
            peer_registry,
            event_tx,
            store,
        );
        assert_eq!(state.config.read().await.node.name, "test");
        assert_eq!(state.config_path, config_path);
    }

    #[tokio::test]
    async fn test_app_state_with_watchdog_tx() {
        let tmpdir = tempfile::tempdir().unwrap();
        let store = Store::new(tmpdir.path().to_str().unwrap()).await.unwrap();
        store.migrate().await.unwrap();
        let config = Config {
            node: NodeConfig {
                name: "test".into(),
                port: 7433,
                data_dir: tmpdir.path().to_str().unwrap().into(),
                ..NodeConfig::default()
            },
            auth: crate::config::AuthConfig::default(),
            peers: HashMap::new(),

            watchdog: crate::config::WatchdogConfig::default(),
            inks: HashMap::new(),
            notifications: crate::config::NotificationsConfig::default(),
            docker: crate::config::DockerConfig::default(),
            controller: crate::config::ControllerConfig::default(),
        };
        let backend = Arc::new(StubBackend);
        let manager =
            SessionManager::new(backend, store.clone(), HashMap::new(), None).with_no_stale_grace();
        let peer_registry = PeerRegistry::new(&HashMap::new());
        let (event_tx, _) = tokio::sync::broadcast::channel(16);
        let initial = crate::watchdog::WatchdogRuntimeConfig {
            threshold: 90,
            interval: std::time::Duration::from_secs(10),
            breach_count: 3,
            idle: crate::watchdog::IdleConfig::default(),
            ready_ttl_secs: 0,
            adopt_tmux: true,
            extra_waiting_patterns: Vec::new(),
        };
        let (config_tx, _config_rx) = tokio::sync::watch::channel(initial);
        let state = AppState::with_watchdog_tx(
            config,
            tmpdir.path().join("config.toml"),
            manager,
            peer_registry,
            event_tx,
            Some(config_tx),
            store,
        );
        assert!(state.watchdog_config_tx.is_some());
    }

    #[tokio::test]
    async fn test_app_state_with_watchdog_tx_none() {
        let tmpdir = tempfile::tempdir().unwrap();
        let store = Store::new(tmpdir.path().to_str().unwrap()).await.unwrap();
        store.migrate().await.unwrap();
        let config = Config {
            node: NodeConfig {
                name: "test".into(),
                port: 7433,
                data_dir: tmpdir.path().to_str().unwrap().into(),
                ..NodeConfig::default()
            },
            auth: crate::config::AuthConfig::default(),
            peers: HashMap::new(),

            watchdog: crate::config::WatchdogConfig::default(),
            inks: HashMap::new(),
            notifications: crate::config::NotificationsConfig::default(),
            docker: crate::config::DockerConfig::default(),
            controller: crate::config::ControllerConfig::default(),
        };
        let backend = Arc::new(StubBackend);
        let manager =
            SessionManager::new(backend, store.clone(), HashMap::new(), None).with_no_stale_grace();
        let peer_registry = PeerRegistry::new(&HashMap::new());
        let (event_tx, _) = tokio::sync::broadcast::channel(16);
        let state = AppState::with_watchdog_tx(
            config,
            tmpdir.path().join("config.toml"),
            manager,
            peer_registry,
            event_tx,
            None,
            store,
        );
        assert!(state.watchdog_config_tx.is_none());
    }

    #[tokio::test]
    async fn test_app_state_with_event_tx_controller() {
        let tmpdir = tempfile::tempdir().unwrap();
        let tmpdir = Box::leak(Box::new(tmpdir));
        let store = Store::new(tmpdir.path().to_str().unwrap()).await.unwrap();
        store.migrate().await.unwrap();
        let config = Config {
            node: NodeConfig {
                name: "test".into(),
                port: 7433,
                data_dir: tmpdir.path().to_str().unwrap().into(),
                ..NodeConfig::default()
            },
            auth: crate::config::AuthConfig::default(),
            peers: HashMap::new(),
            watchdog: crate::config::WatchdogConfig::default(),
            inks: HashMap::new(),
            notifications: crate::config::NotificationsConfig::default(),
            docker: crate::config::DockerConfig::default(),
            controller: crate::config::ControllerConfig::default(),
        };
        let config_path = tmpdir.path().join("config.toml");
        let backend = Arc::new(StubBackend);
        let manager =
            SessionManager::new(backend, store.clone(), HashMap::new(), None).with_no_stale_grace();
        let peer_registry = PeerRegistry::new(&HashMap::new());
        let (event_tx, _) = tokio::sync::broadcast::channel(16);
        let session_index = Arc::new(crate::controller::SessionIndex::new());
        let command_queue = Arc::new(crate::controller::CommandQueue::new());
        let state = AppState::with_event_tx_controller(
            config,
            config_path.clone(),
            manager,
            peer_registry,
            event_tx,
            store,
            Some(session_index),
            Some(command_queue),
        );
        assert_eq!(state.config.read().await.node.name, "test");
        assert_eq!(state.config_path, config_path);
        assert!(state.session_index.is_some());
        assert!(state.command_queue.is_some());
    }

    #[tokio::test]
    async fn test_app_state_with_event_tx_controller_none() {
        let tmpdir = tempfile::tempdir().unwrap();
        let tmpdir = Box::leak(Box::new(tmpdir));
        let store = Store::new(tmpdir.path().to_str().unwrap()).await.unwrap();
        store.migrate().await.unwrap();
        let config = Config {
            node: NodeConfig {
                name: "test".into(),
                port: 7433,
                data_dir: tmpdir.path().to_str().unwrap().into(),
                ..NodeConfig::default()
            },
            auth: crate::config::AuthConfig::default(),
            peers: HashMap::new(),
            watchdog: crate::config::WatchdogConfig::default(),
            inks: HashMap::new(),
            notifications: crate::config::NotificationsConfig::default(),
            docker: crate::config::DockerConfig::default(),
            controller: crate::config::ControllerConfig::default(),
        };
        let backend = Arc::new(StubBackend);
        let manager =
            SessionManager::new(backend, store.clone(), HashMap::new(), None).with_no_stale_grace();
        let peer_registry = PeerRegistry::new(&HashMap::new());
        let (event_tx, _) = tokio::sync::broadcast::channel(16);
        let state = AppState::with_event_tx_controller(
            config,
            tmpdir.path().join("config.toml"),
            manager,
            peer_registry,
            event_tx,
            store,
            None,
            None,
        );
        assert!(state.session_index.is_none());
        assert!(state.command_queue.is_none());
    }

    #[tokio::test]
    async fn test_router_builds() {
        let tmpdir = tempfile::tempdir().unwrap();
        let store = Store::new(tmpdir.path().to_str().unwrap()).await.unwrap();
        store.migrate().await.unwrap();
        let config = Config {
            node: NodeConfig {
                name: "test".into(),
                port: 7433,
                data_dir: tmpdir.path().to_str().unwrap().into(),
                ..NodeConfig::default()
            },
            auth: crate::config::AuthConfig::default(),
            peers: HashMap::new(),

            watchdog: crate::config::WatchdogConfig::default(),
            inks: HashMap::new(),
            notifications: crate::config::NotificationsConfig::default(),
            docker: crate::config::DockerConfig::default(),
            controller: crate::config::ControllerConfig::default(),
        };
        let backend = Arc::new(StubBackend);
        let manager =
            SessionManager::new(backend, store.clone(), HashMap::new(), None).with_no_stale_grace();
        let peer_registry = PeerRegistry::new(&HashMap::new());
        let state = AppState::new(config, manager, peer_registry, store);
        let _router = router(state);
    }
}