diff --git a/src/lib.rs b/src/lib.rs index d7e0509f..a8b8fa93 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -36,6 +36,8 @@ pub mod redfish; #[cfg(feature = "desktop")] pub mod rtsp; #[cfg(feature = "desktop")] +pub mod runtime; +#[cfg(feature = "desktop")] pub mod rustdesk; #[cfg(feature = "desktop")] pub mod state; diff --git a/src/main.rs b/src/main.rs index e407bb39..cd64a246 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3,43 +3,20 @@ use std::future::Future; use std::io::Write; use std::net::{IpAddr, SocketAddr}; use std::path::{Path, PathBuf}; -use std::sync::Arc; use axum_server::tls_rustls::RustlsConfig; use clap::{Args, Parser, Subcommand, ValueEnum}; use futures::{stream::FuturesUnordered, StreamExt}; use rustls::crypto::{ring, CryptoProvider}; -use tokio::sync::{broadcast, mpsc}; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; -use one_kvm::atx::AtxController; -use one_kvm::audio::{AudioController, AudioControllerConfig, AudioQuality}; use one_kvm::auth::{SessionStore, TwoFactorService, UserStore}; -use one_kvm::computer_use::ComputerUseManager; -use one_kvm::config::{self, AppConfig, ConfigStore}; +use one_kvm::config; use one_kvm::db::DatabasePool; -use one_kvm::events::EventBus; -use one_kvm::extensions::ExtensionManager; -use one_kvm::hid::{HidBackendType, HidController}; -#[cfg(unix)] -use one_kvm::msd::MsdController; -#[cfg(unix)] -use one_kvm::otg::OtgService; use one_kvm::platform::PlatformCapabilities; -use one_kvm::rtsp::RtspService; -use one_kvm::rustdesk::RustDeskService; -use one_kvm::state::{AppState, ShutdownAction}; -use one_kvm::update::UpdateService; +use one_kvm::runtime::{RuntimeBuilder, WebConfigOverrides}; +use one_kvm::state::ShutdownAction; use one_kvm::utils::bind_tcp_listener; -use one_kvm::video::codec_constraints::{ - enforce_constraints_with_stream_manager, validate_third_party_codec_compatibility, - StreamCodecConstraints, -}; -use one_kvm::video::format::{PixelFormat, Resolution}; -use one_kvm::video::{Streamer, VideoStreamManager}; -use one_kvm::vnc::VncService; -use one_kvm::web; -use one_kvm::webrtc::{WebRtcStreamer, WebRtcStreamerConfig}; #[derive(Debug, Clone, Copy, Default, ValueEnum)] enum LogLevel { @@ -147,28 +124,20 @@ async fn main() -> anyhow::Result<()> { return Ok(()); } - let (db, config_store, mut config) = load_runtime_config(&data_dir).await?; - - if let Some(addr) = args.address { - config.web.bind_address = addr.clone(); - config.web.bind_addresses = vec![addr]; - } - if let Some(port) = args.http_port { - config.web.http_port = port; - } - if let Some(port) = args.https_port { - config.web.https_port = port; - } - if args.enable_https { - config.web.https_enabled = true; - } - - if let Some(cert_path) = args.ssl_cert { - config.web.ssl_cert_path = Some(cert_path.to_string_lossy().to_string()); - } - if let Some(key_path) = args.ssl_key { - config.web.ssl_key_path = Some(key_path.to_string_lossy().to_string()); - } + let overrides = WebConfigOverrides { + address: args.address, + http_port: args.http_port, + https_port: args.https_port, + enable_https: args.enable_https, + ssl_cert: args.ssl_cert, + ssl_key: args.ssl_key, + }; + let mut runtime = RuntimeBuilder::new(data_dir.clone()) + .with_web_overrides(overrides) + .build() + .await?; + let config = runtime.config(); + let state = runtime.state().clone(); let bind_ips = resolve_bind_addresses(&config.web)?; let scheme = if config.web.https_enabled { @@ -187,500 +156,12 @@ async fn main() -> anyhow::Result<()> { tracing::info!("Server will listen on: {}://{}", scheme, addr); } - let session_store = SessionStore::new(config.auth.session_timeout_secs as i64); - - let user_store = UserStore::new(db.clone_pool()); - let two_factor = TwoFactorService::new(db.clone_pool()); - - let (shutdown_tx, _) = broadcast::channel::(1); - - let events = Arc::new(EventBus::new()); - tracing::info!("Event bus initialized"); - - let (video_format, video_resolution) = parse_video_config(&config); - tracing::debug!( - "Parsed video config: {} @ {}x{}", - video_format, - video_resolution.width, - video_resolution.height - ); - - let streamer = Streamer::new(); - streamer.set_event_bus(events.clone()).await; - if let Some(ref device_path) = config.video.device { - if let Err(e) = streamer - .apply_video_config( - device_path, - video_format, - video_resolution, - config.video.fps, - ) - .await - { - tracing::warn!( - "Failed to initialize video with config: {}, will auto-detect", - e - ); - } else { - tracing::info!( - "Video configured: {} @ {}x{} {}", - device_path, - video_resolution.width, - video_resolution.height, - video_format - ); - } - } - - let webrtc_streamer = { - let webrtc_config = WebRtcStreamerConfig { - resolution: video_resolution, - input_format: video_format, - fps: config.video.fps, - bitrate_preset: config.stream.bitrate_preset, - encoder_backend: one_kvm::stream_encoder::encoder_type_to_backend( - config.stream.encoder.clone(), - ), - webrtc: { - let mut stun_servers = vec![]; - let mut turn_servers = vec![]; - - let has_custom_stun = config - .stream - .stun_server - .as_ref() - .map(|s| !s.is_empty()) - .unwrap_or(false); - let has_custom_turn = config - .stream - .turn_server - .as_ref() - .map(|s| !s.is_empty()) - .unwrap_or(false); - - if !has_custom_stun && !has_custom_turn { - use one_kvm::webrtc::config::public_ice; - let stun = public_ice::stun_server().to_string(); - tracing::info!("Using public STUN server: {}", stun); - stun_servers.push(stun); - } else { - if let Some(ref stun) = config.stream.stun_server { - if !stun.is_empty() { - stun_servers.push(stun.clone()); - tracing::info!("Using custom STUN server: {}", stun); - } - } - if let Some(ref turn) = config.stream.turn_server { - if !turn.is_empty() { - let username = config.stream.turn_username.clone().unwrap_or_default(); - let credential = - config.stream.turn_password.clone().unwrap_or_default(); - turn_servers.push(one_kvm::webrtc::config::TurnServer::new( - turn.clone(), - username.clone(), - credential, - )); - tracing::info!( - "Using custom TURN server: {} (user: {})", - turn, - username - ); - } - } - } - - one_kvm::webrtc::config::WebRtcConfig { - stun_servers, - turn_servers, - ..Default::default() - } - }, - ..Default::default() - }; - WebRtcStreamer::with_config(webrtc_config) - }; - tracing::info!("WebRTC streamer created"); - - #[cfg(unix)] - let otg_service = Arc::new(OtgService::new()); - #[cfg(unix)] - tracing::info!("OTG Service created"); - - #[cfg(unix)] - if let Err(e) = otg_service - .apply_config(&config.hid, &config.msd, &config.otg_network, &config.uac) - .await - { - tracing::warn!("Failed to apply OTG config: {}", e); - } - - let hid_backend = match config.hid.backend { - config::HidBackend::Otg => HidBackendType::Otg, - config::HidBackend::Ch9329 => HidBackendType::Ch9329 { - port: config.hid.ch9329_port.clone(), - baud_rate: config.hid.ch9329_baudrate, - hybrid_mouse: config.hid.ch9329_hybrid_mouse, - }, - config::HidBackend::None => HidBackendType::None, - }; - #[cfg(unix)] - let hid = Arc::new(HidController::new(hid_backend, Some(otg_service.clone()))); - #[cfg(not(unix))] - let hid = Arc::new(HidController::new(hid_backend)); - hid.set_event_bus(events.clone()).await; - if let Err(e) = hid.init().await { - tracing::warn!("Failed to initialize HID backend: {}", e); - } - - #[cfg(unix)] - let msd = if config.msd.enabled { - let ventoy_resource_dir = data_dir.join("ventoy"); - let controller = MsdController::new(otg_service.clone(), config.msd.msd_dir_path()); - if let Err(e) = controller.init(&ventoy_resource_dir).await { - tracing::warn!("Failed to initialize MSD controller: {}", e); - None - } else { - controller.set_event_bus(events.clone()).await; - Some(controller) - } - } else { - tracing::info!("MSD disabled in configuration"); - None - }; - - let atx = if config.atx.enabled { - let controller_config = config.atx.to_controller_config(); - let controller = AtxController::new(controller_config); - - if let Err(e) = controller.init().await { - tracing::warn!("Failed to initialize ATX controller: {}", e); - None - } else { - Some(controller) - } - } else { - tracing::info!("ATX disabled in configuration"); - None - }; - - let audio = { - let audio_config = AudioControllerConfig { - enabled: config.audio.enabled, - device: config.audio.device.clone(), - quality: match config.audio.quality.parse::() { - Ok(q) => q, - Err(e) => { - tracing::warn!( - "Invalid audio quality in config (value={:?}): {}, using balanced", - config.audio.quality, - e - ); - AudioQuality::Balanced - } - }, - }; - - let controller = AudioController::new(audio_config); - controller.set_event_bus(events.clone()).await; - - if config.audio.enabled { - tracing::info!( - "Audio enabled: {}, quality={}", - config.audio.device, - config.audio.quality - ); - if let Err(e) = controller.start_streaming().await { - tracing::warn!("Failed to start audio streaming: {}", e); - } - } else { - tracing::info!("Audio disabled in configuration"); - } - - Arc::new(controller) - }; - - let extensions = Arc::new(ExtensionManager::new()); - tracing::info!("Extension manager initialized"); - - webrtc_streamer.set_hid_controller(hid.clone()).await; - - webrtc_streamer.set_audio_controller(audio.clone()).await; - if config.audio.enabled { - if let Err(e) = webrtc_streamer.set_audio_enabled(true).await { - tracing::warn!("Failed to enable WebRTC audio: {}", e); - } else { - tracing::debug!("WebRTC audio enabled"); - } - } - - let (device_path, actual_resolution, actual_format, actual_fps, jpeg_quality) = - streamer.current_capture_config().await; - tracing::debug!( - "Initial video config: {}x{} {:?} @ {}fps", - actual_resolution.width, - actual_resolution.height, - actual_format, - actual_fps - ); - webrtc_streamer - .update_video_config(actual_resolution, actual_format, actual_fps) - .await; - if let Some(device_path) = device_path { - let device_info = streamer.current_device().await; - webrtc_streamer - .set_capture_device(device_path, jpeg_quality, device_info) - .await; - tracing::debug!("WebRTC streamer configured for direct capture"); - } else { - tracing::warn!("No capture device configured for WebRTC"); - } - - let stream_manager = VideoStreamManager::with_webrtc_streamer( - streamer.clone(), - webrtc_streamer.clone() as std::sync::Arc, - ); - stream_manager.set_event_bus(events.clone()).await; - stream_manager.set_config_store(config_store.clone()).await; - { - let stream_manager_weak = Arc::downgrade(&stream_manager); - audio - .set_recovered_callback(Arc::new(move || { - if let Some(stream_manager) = stream_manager_weak.upgrade() { - tokio::spawn(async move { - stream_manager.reconnect_webrtc_audio_sources().await; - }); - } - })) - .await; - } - - let initial_mode = config.stream.mode.clone(); - if let Err(e) = stream_manager.init_with_mode(initial_mode.clone()).await { - tracing::warn!( - "Failed to initialize stream manager with mode {:?}: {}", - initial_mode, - e - ); - } else { - tracing::info!( - "Video stream manager initialized with mode: {:?}", - initial_mode - ); - } - - let third_party_codec_config_valid = match validate_third_party_codec_compatibility(&config) { - Ok(()) => true, - Err(e) => { - tracing::warn!( - "Third-party access codec configuration is invalid; RustDesk/VNC/RTSP will not start: {}", - e - ); - false - } - }; - - let rustdesk = if third_party_codec_config_valid && config.rustdesk.is_valid() { - tracing::info!( - "Initializing RustDesk service: ID={} -> {}", - config.rustdesk.device_id, - config.rustdesk.rendezvous_addr() - ); - let service = RustDeskService::new( - config.rustdesk.clone(), - stream_manager.clone(), - hid.clone(), - audio.clone(), - ); - Some(Arc::new(service)) - } else { - if config.rustdesk.enabled { - tracing::warn!( - "RustDesk enabled but configuration is incomplete (missing server or credentials)" - ); - } else { - tracing::info!("RustDesk disabled in configuration"); - } - None - }; - - let rtsp = if third_party_codec_config_valid && config.rtsp.enabled { - tracing::info!( - "Initializing RTSP service: rtsp://{}:{}/{}", - config.rtsp.bind, - config.rtsp.port, - config.rtsp.path - ); - let service = RtspService::new(config.rtsp.clone(), stream_manager.clone()); - Some(Arc::new(service)) - } else { - tracing::info!("RTSP disabled in configuration"); - None - }; - - let vnc = if third_party_codec_config_valid && config.vnc.enabled { - tracing::info!( - "Initializing VNC service: {}:{} ({:?})", - config.vnc.bind, - config.vnc.port, - config.vnc.encoding - ); - Some(Arc::new(VncService::new( - config.vnc.clone(), - stream_manager.clone(), - hid.clone(), - ))) - } else { - tracing::info!("VNC disabled in configuration"); - None - }; - - let update_service = Arc::new(UpdateService::new()); - let computer_use = ComputerUseManager::new(config_store.clone(), hid.clone()); - - let state = AppState::new( - db.clone(), - config_store.clone(), - session_store, - user_store, - two_factor, - #[cfg(unix)] - otg_service, - stream_manager, - webrtc_streamer.clone(), - hid, - computer_use, - #[cfg(unix)] - msd, - atx, - audio, - rustdesk.clone(), - vnc.clone(), - rtsp.clone(), - extensions.clone(), - events.clone(), - update_service, - shutdown_tx.clone(), - data_dir.clone(), - ); - - #[cfg(unix)] - { - // Initialize UAC playback writer if UAC is enabled. - if config.uac.enabled { - let uac_cfg = one_kvm::audio::uac::UacPlaybackConfig { - sample_rate: config.uac.sample_rate, - channels: config.uac.channels as u16, - ..Default::default() - }; - match one_kvm::audio::uac::UacPlayback::start(uac_cfg) { - Ok(writer) => { - *state.uac_playback.write().await = Some(writer); - tracing::info!("UAC playback writer started"); - } - Err(e) => { - tracing::warn!("Failed to start UAC playback writer: {}", e); - } - } - } - } - - if config.watchdog.enabled { - if let Err(error) = state.watchdog.enable().await { - tracing::error!( - "Configured hardware watchdog failed to start; web service will continue: {}", - error - ); - } else { - tracing::info!("Hardware watchdog started"); - } - } - - extensions.set_event_bus(events.clone()).await; - - if let Some(ref service) = rustdesk { - if let Err(e) = service.start().await { - tracing::error!("Failed to start RustDesk service: {}", e); - } else { - if let Some(updated_config) = service.save_credentials() { - if let Err(e) = config_store - .update(|cfg| { - cfg.rustdesk.public_key = updated_config.public_key.clone(); - cfg.rustdesk.private_key = updated_config.private_key.clone(); - cfg.rustdesk.signing_public_key = updated_config.signing_public_key.clone(); - cfg.rustdesk.signing_private_key = - updated_config.signing_private_key.clone(); - cfg.rustdesk.uuid = updated_config.uuid.clone(); - }) - .await - { - tracing::warn!("Failed to save RustDesk credentials: {}", e); - } - } - tracing::info!("RustDesk service started"); - } - } - if let Some(ref service) = vnc { - if let Err(e) = service.start().await { - tracing::error!("Failed to start VNC service: {}", e); - } else { - tracing::info!("VNC service started"); - } - } - - if let Some(ref service) = rtsp { - if let Err(e) = service.start().await { - tracing::error!("Failed to start RTSP service: {}", e); - } else { - tracing::info!("RTSP service started"); - } - } - - { - let runtime_config = state.runtime_third_party_config().await; - let constraints = StreamCodecConstraints::from_config(&runtime_config); - state - .stream_manager - .set_runtime_codec_constraints(constraints.clone()) - .await; - match enforce_constraints_with_stream_manager(&state.stream_manager, &constraints).await { - Ok(result) if result.changed => { - if let Some(message) = result.message { - tracing::info!("{}", message); - } - } - Ok(_) => {} - Err(e) => tracing::warn!("Failed to enforce startup codec constraints: {}", e), - } - } - - { - let ext_config = config_store.get(); - extensions.start_enabled(&ext_config.extensions).await; - } - - { - let extensions_clone = extensions.clone(); - let config_store_clone = config_store.clone(); - tokio::spawn(async move { - let mut interval = tokio::time::interval(std::time::Duration::from_secs(30)); - loop { - interval.tick().await; - let config = config_store_clone.get(); - extensions_clone.health_check(&config.extensions).await; - } - }); - tracing::info!("Extension health check task started"); - } - - state.publish_device_info().await; - - spawn_device_info_broadcaster(state.clone(), events); - - let app = web::create_router(state.clone()); + let app = runtime.router(); let listeners = bind_tcp_listeners(&bind_ips, bind_port)?; let shutdown_signal = { + let shutdown_tx = state.shutdown_tx.clone(); let mut shutdown_rx = shutdown_tx.subscribe(); async move { tokio::select! { @@ -740,7 +221,7 @@ async fn main() -> anyhow::Result<()> { servers.push(server); } - run_servers_until_shutdown(servers, shutdown_signal, &state, "HTTPS").await + run_servers_until_shutdown(servers, shutdown_signal, "HTTPS").await } else { let servers = FuturesUnordered::new(); for listener in listeners { @@ -752,9 +233,10 @@ async fn main() -> anyhow::Result<()> { servers.push(async move { server.await }); } - run_servers_until_shutdown(servers, shutdown_signal, &state, "HTTP").await + run_servers_until_shutdown(servers, shutdown_signal, "HTTP").await }; + runtime.shutdown().await; tracing::info!("Server shutdown complete"); if let ShutdownAction::Restart { exe_path } = shutdown_action { restart_current_process(exe_path)?; @@ -840,14 +322,13 @@ async fn open_database_pool(data_dir: &Path) -> anyhow::Result { async fn run_servers_until_shutdown( mut servers: FuturesUnordered, shutdown_signal: impl Future, - state: &Arc, protocol: &'static str, ) -> ShutdownAction where F: Future> + Send, E: std::fmt::Display, { - let action = tokio::select! { + tokio::select! { action = shutdown_signal => { action } @@ -857,9 +338,7 @@ where } ShutdownAction::Exit } - }; - cleanup(state).await; - action + } } fn restart_current_process(exe_path: Option) -> anyhow::Result<()> { @@ -896,64 +375,6 @@ async fn run_cli_command(command: CliCommand, data_dir: PathBuf) -> anyhow::Resu } } -async fn load_runtime_config( - data_dir: &Path, -) -> anyhow::Result<(DatabasePool, ConfigStore, AppConfig)> { - tokio::fs::create_dir_all(data_dir).await?; - - let db = open_database_pool(data_dir).await?; - let config_store = ConfigStore::new(db.clone_pool())?; - config_store.load().await?; - let mut config = (*config_store.get()).clone(); - config.apply_platform_defaults(); - - prepare_linux_runtime_dirs(data_dir, &config_store, &mut config).await?; - - Ok((db, config_store, config)) -} - -#[cfg(unix)] -async fn prepare_linux_runtime_dirs( - data_dir: &Path, - config_store: &ConfigStore, - config: &mut AppConfig, -) -> anyhow::Result<()> { - let mut msd_dir_updated = false; - if config.msd.msd_dir.trim().is_empty() { - let msd_dir = data_dir.join("msd"); - config.msd.msd_dir = msd_dir.to_string_lossy().to_string(); - msd_dir_updated = true; - } else if !PathBuf::from(&config.msd.msd_dir).is_absolute() { - let msd_dir = data_dir.join(&config.msd.msd_dir); - tracing::warn!( - "MSD directory is relative, rebasing to {}", - msd_dir.display() - ); - config.msd.msd_dir = msd_dir.to_string_lossy().to_string(); - msd_dir_updated = true; - } - if msd_dir_updated { - config_store.set(config.clone()).await?; - } - let msd_dir = PathBuf::from(&config.msd.msd_dir); - if let Err(e) = tokio::fs::create_dir_all(msd_dir.join("images")).await { - tracing::warn!("Failed to create MSD images directory: {}", e); - } - if let Err(e) = tokio::fs::create_dir_all(msd_dir.join("ventoy")).await { - tracing::warn!("Failed to create MSD ventoy directory: {}", e); - } - Ok(()) -} - -#[cfg(not(unix))] -async fn prepare_linux_runtime_dirs( - _data_dir: &Path, - _config_store: &ConfigStore, - _config: &mut AppConfig, -) -> anyhow::Result<()> { - Ok(()) -} - async fn run_user_action( action: UserAction, users: &UserStore, @@ -1065,17 +486,6 @@ fn bind_tcp_listeners(addrs: &[IpAddr], port: u16) -> anyhow::Result (PixelFormat, Resolution) { - let format = config - .video - .format - .as_ref() - .and_then(|f: &String| f.parse::().ok()) - .unwrap_or(PixelFormat::Mjpeg); - let resolution = Resolution::new(config.video.width, config.video.height); - (format, resolution) -} - fn generate_self_signed_cert() -> anyhow::Result> { use rcgen::generate_simple_self_signed; @@ -1088,197 +498,3 @@ fn generate_self_signed_cert() -> anyhow::Result, events: Arc) { - use std::time::{Duration, Instant}; - - enum DeviceInfoTrigger { - Event, - Lagged { topic: &'static str, count: u64 }, - } - - const DEVICE_INFO_TOPICS: &[&str] = &[ - "stream.state_changed", - "stream.config_applied", - "stream.mode_ready", - ]; - const DEBOUNCE_MS: u64 = 100; - - let (trigger_tx, mut trigger_rx) = mpsc::unbounded_channel(); - - for topic in DEVICE_INFO_TOPICS { - let Some(mut rx) = events.subscribe_topic(topic) else { - tracing::warn!( - "DeviceInfo broadcaster missing topic subscription: {}", - topic - ); - continue; - }; - - let trigger_tx = trigger_tx.clone(); - let topic_name = *topic; - tokio::spawn(async move { - loop { - match rx.recv().await { - Ok(_) => { - if trigger_tx.send(DeviceInfoTrigger::Event).is_err() { - break; - } - } - Err(tokio::sync::broadcast::error::RecvError::Lagged(count)) => { - if trigger_tx - .send(DeviceInfoTrigger::Lagged { - topic: topic_name, - count, - }) - .is_err() - { - break; - } - } - Err(tokio::sync::broadcast::error::RecvError::Closed) => break, - } - } - }); - } - - { - let mut dirty_rx = events.subscribe_device_info_dirty(); - let trigger_tx = trigger_tx.clone(); - tokio::spawn(async move { - loop { - match dirty_rx.recv().await { - Ok(()) => { - if trigger_tx.send(DeviceInfoTrigger::Event).is_err() { - break; - } - } - Err(tokio::sync::broadcast::error::RecvError::Lagged(count)) => { - if trigger_tx - .send(DeviceInfoTrigger::Lagged { - topic: "device_info_dirty", - count, - }) - .is_err() - { - break; - } - } - Err(tokio::sync::broadcast::error::RecvError::Closed) => break, - } - } - }); - } - - tokio::spawn(async move { - let mut last_broadcast = Instant::now() - Duration::from_millis(DEBOUNCE_MS); - let mut pending_broadcast = false; - - loop { - let recv_result = if pending_broadcast { - let remaining = - DEBOUNCE_MS.saturating_sub(last_broadcast.elapsed().as_millis() as u64); - tokio::time::timeout(Duration::from_millis(remaining), trigger_rx.recv()).await - } else { - Ok(trigger_rx.recv().await) - }; - - match recv_result { - Ok(Some(DeviceInfoTrigger::Event)) => { - pending_broadcast = true; - } - Ok(Some(DeviceInfoTrigger::Lagged { topic, count })) => { - tracing::warn!( - "DeviceInfo broadcaster lagged by {} events on topic {}", - count, - topic - ); - pending_broadcast = true; - } - Ok(None) => { - tracing::info!("Event bus closed, stopping DeviceInfo broadcaster"); - break; - } - Err(_timeout) => {} - } - - if pending_broadcast && last_broadcast.elapsed() >= Duration::from_millis(DEBOUNCE_MS) { - state.publish_device_info().await; - tracing::trace!("Broadcasted DeviceInfo (debounced)"); - last_broadcast = Instant::now(); - pending_broadcast = false; - } - } - }); - - tracing::info!( - "DeviceInfo broadcaster task started (debounce: {}ms)", - DEBOUNCE_MS - ); -} - -async fn cleanup(state: &Arc) { - state.extensions.stop_all().await; - tracing::info!("Extensions stopped"); - - if let Some(ref service) = *state.rustdesk.read().await { - if let Err(e) = service.stop().await { - tracing::warn!("Failed to stop RustDesk service: {}", e); - } else { - tracing::info!("RustDesk service stopped"); - } - } - - if let Some(ref service) = *state.vnc.read().await { - if let Err(e) = service.stop().await { - tracing::warn!("Failed to stop VNC service: {}", e); - } else { - tracing::info!("VNC service stopped"); - } - } - - if let Some(ref service) = *state.rtsp.read().await { - if let Err(e) = service.stop().await { - tracing::warn!("Failed to stop RTSP service: {}", e); - } else { - tracing::info!("RTSP service stopped"); - } - } - - if let Err(e) = state.stream_manager.stop().await { - tracing::warn!("Failed to stop streamer: {}", e); - } - - if let Err(e) = state.hid.shutdown().await { - tracing::warn!("Failed to shutdown HID: {}", e); - } - - #[cfg(unix)] - if let Some(msd) = state.msd.write().await.as_mut() { - if let Err(e) = msd.shutdown().await { - tracing::warn!("Failed to shutdown MSD: {}", e); - } - } - - #[cfg(unix)] - if let Err(e) = state.otg_service.shutdown().await { - tracing::warn!("Failed to shutdown OTG: {}", e); - } - - if let Some(atx) = state.atx.write().await.as_mut() { - if let Err(e) = atx.shutdown().await { - tracing::warn!("Failed to shutdown ATX: {}", e); - } - } - - if let Err(e) = state.audio.shutdown().await { - tracing::warn!("Failed to shutdown audio: {}", e); - } - - if let Err(error) = state.watchdog.disable().await { - tracing::error!( - "CRITICAL: failed to disable hardware watchdog during shutdown: {}", - error - ); - } -} diff --git a/src/runtime/builder.rs b/src/runtime/builder.rs new file mode 100644 index 00000000..d576adf2 --- /dev/null +++ b/src/runtime/builder.rs @@ -0,0 +1,598 @@ +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use tokio::sync::broadcast; + +use crate::atx::AtxController; +use crate::audio::{AudioController, AudioControllerConfig, AudioQuality}; +use crate::auth::{SessionStore, TwoFactorService, UserStore}; +use crate::computer_use::ComputerUseManager; +use crate::config::{self, AppConfig, ConfigStore}; +use crate::db::DatabasePool; +use crate::events::EventBus; +use crate::extensions::ExtensionManager; +use crate::hid::{HidBackendType, HidController}; +#[cfg(unix)] +use crate::msd::MsdController; +#[cfg(unix)] +use crate::otg::OtgService; +use crate::state::{AppState, ShutdownAction}; +use crate::update::UpdateService; +use crate::video::format::{PixelFormat, Resolution}; +use crate::video::{Streamer, VideoStreamManager}; +use crate::webrtc::{WebRtcStreamer, WebRtcStreamerConfig}; + +use super::supervisor::RuntimeSupervisor; + +#[derive(Debug, Clone, Default)] +pub struct WebConfigOverrides { + pub address: Option, + pub http_port: Option, + pub https_port: Option, + pub enable_https: bool, + pub ssl_cert: Option, + pub ssl_key: Option, +} + +impl WebConfigOverrides { + fn apply(self, config: &mut AppConfig) { + if let Some(address) = self.address { + config.web.bind_address = address.clone(); + config.web.bind_addresses = vec![address]; + } + if let Some(port) = self.http_port { + config.web.http_port = port; + } + if let Some(port) = self.https_port { + config.web.https_port = port; + } + if self.enable_https { + config.web.https_enabled = true; + } + if let Some(path) = self.ssl_cert { + config.web.ssl_cert_path = Some(path.to_string_lossy().to_string()); + } + if let Some(path) = self.ssl_key { + config.web.ssl_key_path = Some(path.to_string_lossy().to_string()); + } + } +} + +pub struct RuntimeBuilder { + data_dir: PathBuf, + web_overrides: WebConfigOverrides, +} + +impl RuntimeBuilder { + pub fn new(data_dir: PathBuf) -> Self { + Self { + data_dir, + web_overrides: WebConfigOverrides::default(), + } + } + + pub fn with_web_overrides(mut self, overrides: WebConfigOverrides) -> Self { + self.web_overrides = overrides; + self + } + + pub async fn build(self) -> anyhow::Result { + let Self { + data_dir, + web_overrides, + } = self; + let (db, config_store, mut config) = load_runtime_config(&data_dir).await?; + web_overrides.apply(&mut config); + + let sessions = SessionStore::new(config.auth.session_timeout_secs as i64); + let users = UserStore::new(db.clone_pool()); + let two_factor = TwoFactorService::new(db.clone_pool()); + let (shutdown_tx, _) = broadcast::channel::(1); + + let events = Arc::new(EventBus::new()); + tracing::info!("Event bus initialized"); + + let (video_format, video_resolution) = parse_video_config(&config); + let streamer = build_streamer(&config, &events, video_format, video_resolution).await; + let webrtc = build_webrtc(&config, video_format, video_resolution); + + #[cfg(unix)] + let otg_service = build_otg(&config).await; + + let hid_backend = hid_backend_type(&config); + #[cfg(unix)] + let hid = Arc::new(HidController::new(hid_backend, Some(otg_service.clone()))); + #[cfg(not(unix))] + let hid = Arc::new(HidController::new(hid_backend)); + hid.set_event_bus(events.clone()).await; + if let Err(error) = hid.init().await { + tracing::warn!("Failed to initialize HID backend: {}", error); + } + + #[cfg(unix)] + let msd = build_msd(&config, &data_dir, &otg_service, &events).await; + let atx = build_atx(&config).await; + let audio = build_audio(&config, &events).await; + let extensions = Arc::new(ExtensionManager::new()); + tracing::info!("Extension manager initialized"); + + webrtc.set_hid_controller(hid.clone()).await; + webrtc.set_audio_controller(audio.clone()).await; + if config.audio.enabled { + if let Err(error) = webrtc.set_audio_enabled(true).await { + tracing::warn!("Failed to enable WebRTC audio: {}", error); + } else { + tracing::debug!("WebRTC audio enabled"); + } + } + + connect_capture_to_webrtc(&streamer, &webrtc).await; + + let stream_manager = VideoStreamManager::with_webrtc_streamer( + streamer.clone(), + webrtc.clone() as Arc, + ); + stream_manager.set_event_bus(events.clone()).await; + stream_manager.set_config_store(config_store.clone()).await; + connect_audio_recovery(&audio, &stream_manager).await; + + let initial_mode = config.stream.mode.clone(); + if let Err(error) = stream_manager.init_with_mode(initial_mode.clone()).await { + tracing::warn!( + "Failed to initialize stream manager with mode {:?}: {}", + initial_mode, + error + ); + } else { + tracing::info!( + "Video stream manager initialized with mode: {:?}", + initial_mode + ); + } + + let computer_use = ComputerUseManager::new(config_store.clone(), hid.clone()); + let state = AppState::new( + db, + config_store.clone(), + sessions, + users, + two_factor, + #[cfg(unix)] + otg_service, + stream_manager, + webrtc, + hid, + computer_use, + #[cfg(unix)] + msd, + atx, + audio, + extensions.clone(), + events.clone(), + Arc::new(UpdateService::new()), + shutdown_tx, + data_dir.clone(), + ); + + start_uac_playback(&state, &config).await; + start_watchdog(&state, &config).await; + extensions.set_event_bus(events.clone()).await; + state.remote_access.start_configured(&config).await; + + let extension_config = config_store.get(); + extensions.start_enabled(&extension_config.extensions).await; + + state.publish_device_info().await; + let supervisor = RuntimeSupervisor::start(state.clone(), events, extensions, config_store); + + Ok(ApplicationRuntime { + state, + config, + data_dir, + supervisor, + }) + } +} + +pub struct ApplicationRuntime { + state: Arc, + config: AppConfig, + data_dir: PathBuf, + supervisor: RuntimeSupervisor, +} + +impl ApplicationRuntime { + pub fn state(&self) -> &Arc { + &self.state + } + + pub fn config(&self) -> &AppConfig { + &self.config + } + + pub fn data_dir(&self) -> &Path { + &self.data_dir + } + + pub fn router(&self) -> axum::Router { + crate::web::create_router(self.state.clone()) + } + + pub async fn shutdown(&mut self) { + self.supervisor.shutdown(&self.state).await; + } +} + +async fn load_runtime_config( + data_dir: &Path, +) -> anyhow::Result<(DatabasePool, ConfigStore, AppConfig)> { + tokio::fs::create_dir_all(data_dir).await?; + + let db_path = data_dir.join("one-kvm.db"); + let db = DatabasePool::new(&db_path).await?; + db.init_schema().await?; + + let config_store = ConfigStore::new(db.clone_pool())?; + config_store.load().await?; + let mut config = (*config_store.get()).clone(); + config.apply_platform_defaults(); + prepare_linux_runtime_dirs(data_dir, &config_store, &mut config).await?; + + Ok((db, config_store, config)) +} + +#[cfg(unix)] +async fn prepare_linux_runtime_dirs( + data_dir: &Path, + config_store: &ConfigStore, + config: &mut AppConfig, +) -> anyhow::Result<()> { + let mut msd_dir_updated = false; + if config.msd.msd_dir.trim().is_empty() { + config.msd.msd_dir = data_dir.join("msd").to_string_lossy().to_string(); + msd_dir_updated = true; + } else if !PathBuf::from(&config.msd.msd_dir).is_absolute() { + let msd_dir = data_dir.join(&config.msd.msd_dir); + tracing::warn!( + "MSD directory is relative, rebasing to {}", + msd_dir.display() + ); + config.msd.msd_dir = msd_dir.to_string_lossy().to_string(); + msd_dir_updated = true; + } + if msd_dir_updated { + config_store.set(config.clone()).await?; + } + + let msd_dir = PathBuf::from(&config.msd.msd_dir); + if let Err(error) = tokio::fs::create_dir_all(msd_dir.join("images")).await { + tracing::warn!("Failed to create MSD images directory: {}", error); + } + if let Err(error) = tokio::fs::create_dir_all(msd_dir.join("ventoy")).await { + tracing::warn!("Failed to create MSD ventoy directory: {}", error); + } + Ok(()) +} + +#[cfg(not(unix))] +async fn prepare_linux_runtime_dirs( + _data_dir: &Path, + _config_store: &ConfigStore, + _config: &mut AppConfig, +) -> anyhow::Result<()> { + Ok(()) +} + +fn parse_video_config(config: &AppConfig) -> (PixelFormat, Resolution) { + let format = config + .video + .format + .as_ref() + .and_then(|format| format.parse::().ok()) + .unwrap_or(PixelFormat::Mjpeg); + ( + format, + Resolution::new(config.video.width, config.video.height), + ) +} + +async fn build_streamer( + config: &AppConfig, + events: &Arc, + format: PixelFormat, + resolution: Resolution, +) -> Arc { + tracing::debug!( + "Parsed video config: {} @ {}x{}", + format, + resolution.width, + resolution.height + ); + let streamer = Streamer::new(); + streamer.set_event_bus(events.clone()).await; + if let Some(device_path) = config.video.device.as_ref() { + if let Err(error) = streamer + .apply_video_config(device_path, format, resolution, config.video.fps) + .await + { + tracing::warn!( + "Failed to initialize video with config: {}, will auto-detect", + error + ); + } else { + tracing::info!( + "Video configured: {} @ {}x{} {}", + device_path, + resolution.width, + resolution.height, + format + ); + } + } + streamer +} + +fn build_webrtc( + config: &AppConfig, + input_format: PixelFormat, + resolution: Resolution, +) -> Arc { + let webrtc = WebRtcStreamer::with_config(WebRtcStreamerConfig { + resolution, + input_format, + fps: config.video.fps, + bitrate_preset: config.stream.bitrate_preset, + encoder_backend: crate::stream_encoder::encoder_type_to_backend( + config.stream.encoder.clone(), + ), + webrtc: build_ice_config(config), + ..Default::default() + }); + tracing::info!("WebRTC streamer created"); + webrtc +} + +fn build_ice_config(config: &AppConfig) -> crate::webrtc::config::WebRtcConfig { + let mut stun_servers = Vec::new(); + let mut turn_servers = Vec::new(); + let has_custom_stun = config + .stream + .stun_server + .as_ref() + .is_some_and(|server| !server.is_empty()); + let has_custom_turn = config + .stream + .turn_server + .as_ref() + .is_some_and(|server| !server.is_empty()); + + if !has_custom_stun && !has_custom_turn { + let stun = crate::webrtc::config::public_ice::stun_server().to_string(); + tracing::info!("Using public STUN server: {}", stun); + stun_servers.push(stun); + } else { + if let Some(stun) = config + .stream + .stun_server + .as_ref() + .filter(|server| !server.is_empty()) + { + tracing::info!("Using custom STUN server: {}", stun); + stun_servers.push(stun.clone()); + } + if let Some(turn) = config + .stream + .turn_server + .as_ref() + .filter(|server| !server.is_empty()) + { + let username = config.stream.turn_username.clone().unwrap_or_default(); + let credential = config.stream.turn_password.clone().unwrap_or_default(); + turn_servers.push(crate::webrtc::config::TurnServer::new( + turn.clone(), + username.clone(), + credential, + )); + tracing::info!("Using custom TURN server: {} (user: {})", turn, username); + } + } + + crate::webrtc::config::WebRtcConfig { + stun_servers, + turn_servers, + ..Default::default() + } +} + +#[cfg(unix)] +async fn build_otg(config: &AppConfig) -> Arc { + let service = Arc::new(OtgService::new()); + tracing::info!("OTG Service created"); + if let Err(error) = service + .apply_config(&config.hid, &config.msd, &config.otg_network, &config.uac) + .await + { + tracing::warn!("Failed to apply OTG config: {}", error); + } + service +} + +fn hid_backend_type(config: &AppConfig) -> HidBackendType { + match config.hid.backend { + config::HidBackend::Otg => HidBackendType::Otg, + config::HidBackend::Ch9329 => HidBackendType::Ch9329 { + port: config.hid.ch9329_port.clone(), + baud_rate: config.hid.ch9329_baudrate, + hybrid_mouse: config.hid.ch9329_hybrid_mouse, + }, + config::HidBackend::None => HidBackendType::None, + } +} + +#[cfg(unix)] +async fn build_msd( + config: &AppConfig, + data_dir: &Path, + otg: &Arc, + events: &Arc, +) -> Option { + if !config.msd.enabled { + tracing::info!("MSD disabled in configuration"); + return None; + } + + let controller = MsdController::new(otg.clone(), config.msd.msd_dir_path()); + if let Err(error) = controller.init(&data_dir.join("ventoy")).await { + tracing::warn!("Failed to initialize MSD controller: {}", error); + return None; + } + controller.set_event_bus(events.clone()).await; + Some(controller) +} + +async fn build_atx(config: &AppConfig) -> Option { + if !config.atx.enabled { + tracing::info!("ATX disabled in configuration"); + return None; + } + + let controller = AtxController::new(config.atx.to_controller_config()); + if let Err(error) = controller.init().await { + tracing::warn!("Failed to initialize ATX controller: {}", error); + return None; + } + Some(controller) +} + +async fn build_audio(config: &AppConfig, events: &Arc) -> Arc { + let quality = config + .audio + .quality + .parse::() + .unwrap_or_else(|error| { + tracing::warn!( + "Invalid audio quality in config (value={:?}): {}, using balanced", + config.audio.quality, + error + ); + AudioQuality::Balanced + }); + let controller = Arc::new(AudioController::new(AudioControllerConfig { + enabled: config.audio.enabled, + device: config.audio.device.clone(), + quality, + })); + controller.set_event_bus(events.clone()).await; + + if config.audio.enabled { + tracing::info!( + "Audio enabled: {}, quality={}", + config.audio.device, + config.audio.quality + ); + if let Err(error) = controller.start_streaming().await { + tracing::warn!("Failed to start audio streaming: {}", error); + } + } else { + tracing::info!("Audio disabled in configuration"); + } + controller +} + +async fn connect_capture_to_webrtc(streamer: &Arc, webrtc: &Arc) { + let (device_path, resolution, format, fps, jpeg_quality) = + streamer.current_capture_config().await; + tracing::debug!( + "Initial video config: {}x{} {:?} @ {}fps", + resolution.width, + resolution.height, + format, + fps + ); + webrtc.update_video_config(resolution, format, fps).await; + if let Some(device_path) = device_path { + webrtc + .set_capture_device(device_path, jpeg_quality, streamer.current_device().await) + .await; + tracing::debug!("WebRTC streamer configured for direct capture"); + } else { + tracing::warn!("No capture device configured for WebRTC"); + } +} + +async fn connect_audio_recovery( + audio: &Arc, + stream_manager: &Arc, +) { + let stream_manager = Arc::downgrade(stream_manager); + audio + .set_recovered_callback(Arc::new(move || { + if let Some(stream_manager) = stream_manager.upgrade() { + tokio::spawn(async move { + stream_manager.reconnect_webrtc_audio_sources().await; + }); + } + })) + .await; +} + +#[cfg(unix)] +async fn start_uac_playback(state: &Arc, config: &AppConfig) { + if !config.uac.enabled { + return; + } + let playback_config = crate::audio::uac::UacPlaybackConfig { + sample_rate: config.uac.sample_rate, + channels: config.uac.channels as u16, + ..Default::default() + }; + match crate::audio::uac::UacPlayback::start(playback_config) { + Ok(writer) => { + *state.uac_playback.write().await = Some(writer); + tracing::info!("UAC playback writer started"); + } + Err(error) => tracing::warn!("Failed to start UAC playback writer: {}", error), + } +} + +#[cfg(not(unix))] +async fn start_uac_playback(_state: &Arc, _config: &AppConfig) {} + +async fn start_watchdog(state: &Arc, config: &AppConfig) { + if !config.watchdog.enabled { + return; + } + if let Err(error) = state.watchdog.enable().await { + tracing::error!( + "Configured hardware watchdog failed to start; web service will continue: {}", + error + ); + } else { + tracing::info!("Hardware watchdog started"); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn web_overrides_only_replace_explicit_values() { + let mut config = AppConfig::default(); + let original_https_port = config.web.https_port; + + WebConfigOverrides { + address: Some("127.0.0.1".to_string()), + http_port: Some(9000), + enable_https: true, + ..Default::default() + } + .apply(&mut config); + + assert_eq!(config.web.bind_address, "127.0.0.1"); + assert_eq!(config.web.bind_addresses, ["127.0.0.1"]); + assert_eq!(config.web.http_port, 9000); + assert_eq!(config.web.https_port, original_https_port); + assert!(config.web.https_enabled); + } +} diff --git a/src/runtime/config_apply.rs b/src/runtime/config_apply.rs new file mode 100644 index 00000000..1be6a335 --- /dev/null +++ b/src/runtime/config_apply.rs @@ -0,0 +1,44 @@ +use std::sync::Arc; + +use tokio::sync::{Mutex, OwnedMutexGuard}; + +use crate::error::{AppError, Result}; + +#[derive(Debug, Clone, Copy, Default)] +pub struct ConfigApplyOptions { + pub force: bool, + pub preserve_service_state: bool, + pub runtime_only: bool, +} + +impl ConfigApplyOptions { + pub const fn forced() -> Self { + Self { + force: true, + preserve_service_state: false, + runtime_only: false, + } + } + + pub const fn preserving_service_state() -> Self { + Self { + force: false, + preserve_service_state: true, + runtime_only: false, + } + } + + pub const fn runtime_only() -> Self { + Self { + force: false, + preserve_service_state: false, + runtime_only: true, + } + } +} + +pub fn try_apply_lock(lock: &Arc>, domain: &str) -> Result> { + lock.clone().try_lock_owned().map_err(|_| { + AppError::ServiceUnavailable(format!("{domain} configuration is already applying")) + }) +} diff --git a/src/runtime/mod.rs b/src/runtime/mod.rs new file mode 100644 index 00000000..6f12c72b --- /dev/null +++ b/src/runtime/mod.rs @@ -0,0 +1,10 @@ +mod builder; +mod config_apply; +mod remote_access; +mod supervisor; +mod usb; + +pub use builder::{ApplicationRuntime, RuntimeBuilder, WebConfigOverrides}; +pub use config_apply::{try_apply_lock, ConfigApplyOptions}; +pub use remote_access::{RemoteAccessCoordinator, RustDeskRuntimeStatus}; +pub use usb::UsbCoordinator; diff --git a/src/runtime/remote_access.rs b/src/runtime/remote_access.rs new file mode 100644 index 00000000..d5d74605 --- /dev/null +++ b/src/runtime/remote_access.rs @@ -0,0 +1,496 @@ +use std::sync::Arc; + +use tokio::sync::RwLock; + +use crate::audio::AudioController; +use crate::config::{AppConfig, ConfigStore, RtspConfig, VncConfig}; +use crate::error::{AppError, Result}; +use crate::hid::HidController; +use crate::rtsp::{RtspService, RtspServiceStatus}; +use crate::rustdesk::config::RustDeskConfig; +use crate::rustdesk::RustDeskService; +use crate::video::codec_constraints::{ + enforce_constraints_with_stream_manager, validate_third_party_codec_compatibility, + StreamCodecConstraints, +}; +use crate::video::VideoStreamManager; +use crate::vnc::{VncService, VncServiceStatus}; + +use super::ConfigApplyOptions; + +#[derive(Debug, Clone)] +pub struct RustDeskRuntimeStatus { + pub service_status: String, + pub rendezvous_status: Option, +} + +pub struct RemoteAccessCoordinator { + config: ConfigStore, + stream_manager: Arc, + hid: Arc, + audio: Arc, + rustdesk: RwLock>>, + vnc: RwLock>>, + rtsp: RwLock>>, +} + +impl RemoteAccessCoordinator { + pub fn new( + config: ConfigStore, + stream_manager: Arc, + hid: Arc, + audio: Arc, + ) -> Arc { + Arc::new(Self { + config, + stream_manager, + hid, + audio, + rustdesk: RwLock::new(None), + vnc: RwLock::new(None), + rtsp: RwLock::new(None), + }) + } + + pub async fn start_configured(&self, config: &AppConfig) { + if let Err(error) = validate_third_party_codec_compatibility(config) { + tracing::warn!( + "Third-party access codec configuration is invalid; RustDesk/VNC/RTSP will not start: {}", + error + ); + return; + } + + if config.rustdesk.is_valid() { + if let Err(error) = self + .apply_rustdesk( + &RustDeskConfig::default(), + &config.rustdesk, + ConfigApplyOptions::default(), + ) + .await + { + tracing::error!("Failed to start RustDesk service: {}", error); + } + } else if config.rustdesk.enabled { + tracing::warn!( + "RustDesk enabled but configuration is incomplete (missing server or credentials)" + ); + } else { + tracing::info!("RustDesk disabled in configuration"); + } + + if config.vnc.enabled { + if let Err(error) = self + .apply_vnc( + &VncConfig::default(), + &config.vnc, + ConfigApplyOptions::default(), + ) + .await + { + tracing::error!("Failed to start VNC service: {}", error); + } + } else { + tracing::info!("VNC disabled in configuration"); + } + + if config.rtsp.enabled { + if let Err(error) = self + .apply_rtsp( + &RtspConfig::default(), + &config.rtsp, + ConfigApplyOptions::default(), + ) + .await + { + tracing::error!("Failed to start RTSP service: {}", error); + } + } else { + tracing::info!("RTSP disabled in configuration"); + } + + if let Err(error) = self.enforce_codec_constraints().await { + tracing::warn!("Failed to enforce startup codec constraints: {}", error); + } + } + + pub async fn runtime_config(&self) -> AppConfig { + let mut config = self.config.get().as_ref().clone(); + let rustdesk = self.rustdesk.read().await.clone(); + let vnc = self.vnc.read().await.clone(); + let rtsp = self.rtsp.read().await.clone(); + + config.rustdesk.enabled = rustdesk.is_some_and(|service| service.is_listening()); + config.vnc.enabled = match vnc { + Some(service) => matches!( + service.status().await, + VncServiceStatus::Starting | VncServiceStatus::Running + ), + None => false, + }; + config.rtsp.enabled = match rtsp { + Some(service) => matches!( + service.status().await, + RtspServiceStatus::Starting | RtspServiceStatus::Running + ), + None => false, + }; + config + } + + pub async fn rustdesk_status(&self) -> RustDeskRuntimeStatus { + let service = self.rustdesk.read().await.clone(); + match service { + Some(service) => RustDeskRuntimeStatus { + service_status: service.status().to_string(), + rendezvous_status: service.rendezvous_status().map(|status| status.to_string()), + }, + None => RustDeskRuntimeStatus { + service_status: "not_initialized".to_string(), + rendezvous_status: None, + }, + } + } + + pub async fn vnc_status(&self) -> (VncServiceStatus, usize) { + let service = self.vnc.read().await.clone(); + match service { + Some(service) => (service.status().await, service.connection_count()), + None => (VncServiceStatus::Stopped, 0), + } + } + + pub async fn rtsp_status(&self) -> RtspServiceStatus { + let service = self.rtsp.read().await.clone(); + match service { + Some(service) => service.status().await, + None => RtspServiceStatus::Stopped, + } + } + + pub async fn enforce_codec_constraints(&self) -> Result> { + let config = self.runtime_config().await; + let constraints = StreamCodecConstraints::from_config(&config); + self.stream_manager + .set_runtime_codec_constraints(constraints.clone()) + .await; + let enforcement = + enforce_constraints_with_stream_manager(&self.stream_manager, &constraints).await?; + Ok(enforcement.message) + } + + pub async fn apply_rustdesk( + &self, + old_config: &RustDeskConfig, + new_config: &RustDeskConfig, + options: ConfigApplyOptions, + ) -> Result<()> { + tracing::info!("Applying RustDesk config changes..."); + self.validate_rustdesk_candidate(new_config, options.runtime_only) + .await?; + + let need_restart = options.force + || old_config.codec != new_config.codec + || old_config.rendezvous_server != new_config.rendezvous_server + || old_config.device_id != new_config.device_id + || old_config.device_password != new_config.device_password; + let current = self.rustdesk.read().await.clone(); + let mut credentials_to_save = None; + + if !options.preserve_service_state && !new_config.enabled { + if let Some(service) = current.as_ref() { + service.stop().await.map_err(|error| { + AppError::Config(format!("Failed to stop RustDesk service: {error}")) + })?; + tracing::info!("RustDesk service stopped"); + } + *self.rustdesk.write().await = None; + } else if !options.preserve_service_state && new_config.enabled { + match current { + None => { + tracing::info!("Initializing RustDesk service..."); + let service = Arc::new(RustDeskService::new( + new_config.clone(), + self.stream_manager.clone(), + self.hid.clone(), + self.audio.clone(), + )); + *self.rustdesk.write().await = Some(service.clone()); + service.start().await.map_err(|error| { + AppError::Config(format!("Failed to start RustDesk service: {error}")) + })?; + tracing::info!("RustDesk service started with ID: {}", new_config.device_id); + credentials_to_save = service.save_credentials(); + } + Some(service) => { + if service.is_listening() { + if need_restart { + service.restart(new_config.clone()).await.map_err(|error| { + AppError::Config(format!( + "Failed to restart RustDesk service: {error}" + )) + })?; + tracing::info!( + "RustDesk service restarted with ID: {}", + new_config.device_id + ); + } + } else { + service.update_config(new_config.clone()); + service.start().await.map_err(|error| { + AppError::Config(format!("Failed to start RustDesk service: {error}")) + })?; + } + credentials_to_save = service.save_credentials(); + } + } + } else if options.preserve_service_state && need_restart { + if let Some(service) = current { + let mut runtime_config = new_config.clone(); + runtime_config.enabled = true; + service.restart(runtime_config).await.map_err(|error| { + AppError::Config(format!("Failed to restart RustDesk service: {error}")) + })?; + credentials_to_save = service.save_credentials(); + } + } + + if let Some(updated) = credentials_to_save { + tracing::info!("Saving RustDesk credentials to config store..."); + self.config + .update(|config| { + config.rustdesk.public_key = updated.public_key.clone(); + config.rustdesk.private_key = updated.private_key.clone(); + config.rustdesk.signing_public_key = updated.signing_public_key.clone(); + config.rustdesk.signing_private_key = updated.signing_private_key.clone(); + config.rustdesk.uuid = updated.uuid.clone(); + }) + .await?; + tracing::info!("RustDesk credentials saved successfully"); + } + + self.log_enforced_constraints().await?; + Ok(()) + } + + pub async fn apply_vnc( + &self, + old_config: &VncConfig, + new_config: &VncConfig, + options: ConfigApplyOptions, + ) -> Result<()> { + tracing::info!("Applying VNC config changes..."); + self.validate_vnc_candidate(new_config, options.runtime_only) + .await?; + + let runtime_config = self.runtime_config().await; + let will_run = if options.preserve_service_state { + runtime_config.vnc.enabled + } else { + new_config.enabled + }; + if will_run { + let mut candidate = runtime_config; + candidate.vnc = new_config.clone(); + candidate.vnc.enabled = true; + let constraints = StreamCodecConstraints::from_config(&candidate); + match enforce_constraints_with_stream_manager(&self.stream_manager, &constraints).await + { + Ok(result) if result.changed => { + if let Some(message) = result.message { + tracing::info!("{}", message); + } + } + Ok(_) => {} + Err(error) => tracing::warn!( + "Failed to enforce VNC stream constraints before start: {}", + error + ), + } + } + + let need_restart = options.force + || old_config.bind != new_config.bind + || old_config.port != new_config.port + || old_config.encoding != new_config.encoding + || old_config.password != new_config.password + || old_config.allow_one_client != new_config.allow_one_client; + let current = self.vnc.read().await.clone(); + + if !options.preserve_service_state && !new_config.enabled { + if let Some(service) = current.as_ref() { + service.stop().await?; + } + *self.vnc.write().await = None; + } else if !options.preserve_service_state && new_config.enabled { + match current { + None => { + let service = Arc::new(VncService::new( + new_config.clone(), + self.stream_manager.clone(), + self.hid.clone(), + )); + *self.vnc.write().await = Some(service.clone()); + service.start().await?; + tracing::info!("VNC service started"); + } + Some(service) => { + if matches!(service.status().await, VncServiceStatus::Running) { + if need_restart { + service.restart(new_config.clone()).await?; + tracing::info!("VNC service restarted"); + } + } else { + service.update_config(new_config.clone()).await; + service.start().await?; + } + } + } + } else if options.preserve_service_state && need_restart { + if let Some(service) = current { + let mut runtime_config = new_config.clone(); + runtime_config.enabled = true; + service.restart(runtime_config).await?; + } + } + + self.log_enforced_constraints().await?; + Ok(()) + } + + pub async fn apply_rtsp( + &self, + old_config: &RtspConfig, + new_config: &RtspConfig, + options: ConfigApplyOptions, + ) -> Result<()> { + tracing::info!("Applying RTSP config changes..."); + self.validate_rtsp_candidate(new_config, options.runtime_only) + .await?; + + let need_restart = options.force + || old_config.bind != new_config.bind + || old_config.port != new_config.port + || old_config.path != new_config.path + || old_config.codec != new_config.codec + || old_config.username != new_config.username + || old_config.password != new_config.password + || old_config.allow_one_client != new_config.allow_one_client; + let current = self.rtsp.read().await.clone(); + + if !options.preserve_service_state && !new_config.enabled { + if let Some(service) = current.as_ref() { + service.stop().await.map_err(|error| { + AppError::Config(format!("Failed to stop RTSP service: {error}")) + })?; + } + *self.rtsp.write().await = None; + } else if !options.preserve_service_state && new_config.enabled { + match current { + None => { + let service = Arc::new(RtspService::new( + new_config.clone(), + self.stream_manager.clone(), + )); + *self.rtsp.write().await = Some(service.clone()); + service.start().await?; + tracing::info!("RTSP service started"); + } + Some(service) => { + if matches!(service.status().await, RtspServiceStatus::Running) { + if need_restart { + service.restart(new_config.clone()).await?; + tracing::info!("RTSP service restarted"); + } + } else { + service.update_config(new_config.clone()).await; + service.start().await?; + } + } + } + } else if options.preserve_service_state && need_restart { + if let Some(service) = current { + let mut runtime_config = new_config.clone(); + runtime_config.enabled = true; + service.restart(runtime_config).await?; + } + } + + self.log_enforced_constraints().await?; + Ok(()) + } + + pub async fn shutdown(&self) { + let rustdesk = self.rustdesk.write().await.take(); + let vnc = self.vnc.write().await.take(); + let rtsp = self.rtsp.write().await.take(); + + if let Some(service) = rustdesk { + if let Err(error) = service.stop().await { + tracing::warn!("Failed to stop RustDesk service: {}", error); + } else { + tracing::info!("RustDesk service stopped"); + } + } + if let Some(service) = vnc { + if let Err(error) = service.stop().await { + tracing::warn!("Failed to stop VNC service: {}", error); + } else { + tracing::info!("VNC service stopped"); + } + } + if let Some(service) = rtsp { + if let Err(error) = service.stop().await { + tracing::warn!("Failed to stop RTSP service: {}", error); + } else { + tracing::info!("RTSP service stopped"); + } + } + } + + async fn validate_rustdesk_candidate( + &self, + new_config: &RustDeskConfig, + runtime_only: bool, + ) -> Result<()> { + let mut candidate = self.candidate_config(runtime_only).await; + candidate.rustdesk = new_config.clone(); + validate_third_party_codec_compatibility(&candidate) + } + + async fn validate_vnc_candidate( + &self, + new_config: &VncConfig, + runtime_only: bool, + ) -> Result<()> { + let mut candidate = self.candidate_config(runtime_only).await; + candidate.vnc = new_config.clone(); + validate_third_party_codec_compatibility(&candidate) + } + + async fn validate_rtsp_candidate( + &self, + new_config: &RtspConfig, + runtime_only: bool, + ) -> Result<()> { + let mut candidate = self.candidate_config(runtime_only).await; + candidate.rtsp = new_config.clone(); + validate_third_party_codec_compatibility(&candidate) + } + + async fn candidate_config(&self, runtime_only: bool) -> AppConfig { + if runtime_only { + self.runtime_config().await + } else { + self.config.get().as_ref().clone() + } + } + + async fn log_enforced_constraints(&self) -> Result<()> { + if let Some(message) = self.enforce_codec_constraints().await? { + tracing::info!("{}", message); + } + Ok(()) + } +} diff --git a/src/runtime/supervisor.rs b/src/runtime/supervisor.rs new file mode 100644 index 00000000..2b00b8b4 --- /dev/null +++ b/src/runtime/supervisor.rs @@ -0,0 +1,228 @@ +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use tokio::sync::mpsc; +use tokio::task::JoinHandle; + +use crate::config::ConfigStore; +use crate::events::EventBus; +use crate::extensions::ExtensionManager; +use crate::state::AppState; + +pub(super) struct RuntimeSupervisor { + tasks: Vec>, +} + +impl RuntimeSupervisor { + pub(super) fn start( + state: Arc, + events: Arc, + extensions: Arc, + config: ConfigStore, + ) -> Self { + let mut tasks = spawn_device_info_broadcaster(state, events); + tasks.push(spawn_extension_health_check(extensions, config)); + Self { tasks } + } + + pub(super) async fn shutdown(&mut self, state: &Arc) { + for task in self.tasks.drain(..) { + task.abort(); + } + cleanup(state).await; + } +} + +fn spawn_extension_health_check( + extensions: Arc, + config: ConfigStore, +) -> JoinHandle<()> { + let task = tokio::spawn(async move { + let mut interval = tokio::time::interval(Duration::from_secs(30)); + loop { + interval.tick().await; + let config = config.get(); + extensions.health_check(&config.extensions).await; + } + }); + tracing::info!("Extension health check task started"); + task +} + +fn spawn_device_info_broadcaster( + state: Arc, + events: Arc, +) -> Vec> { + enum DeviceInfoTrigger { + Event, + Lagged { topic: &'static str, count: u64 }, + } + + const DEVICE_INFO_TOPICS: &[&str] = &[ + "stream.state_changed", + "stream.config_applied", + "stream.mode_ready", + ]; + const DEBOUNCE_MS: u64 = 100; + + let (trigger_tx, mut trigger_rx) = mpsc::unbounded_channel(); + let mut tasks = Vec::new(); + + for topic in DEVICE_INFO_TOPICS { + let Some(mut rx) = events.subscribe_topic(topic) else { + tracing::warn!( + "DeviceInfo broadcaster missing topic subscription: {}", + topic + ); + continue; + }; + + let trigger_tx = trigger_tx.clone(); + let topic_name = *topic; + tasks.push(tokio::spawn(async move { + loop { + match rx.recv().await { + Ok(_) => { + if trigger_tx.send(DeviceInfoTrigger::Event).is_err() { + break; + } + } + Err(tokio::sync::broadcast::error::RecvError::Lagged(count)) => { + if trigger_tx + .send(DeviceInfoTrigger::Lagged { + topic: topic_name, + count, + }) + .is_err() + { + break; + } + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + } + } + })); + } + + { + let mut dirty_rx = events.subscribe_device_info_dirty(); + let trigger_tx = trigger_tx.clone(); + tasks.push(tokio::spawn(async move { + loop { + match dirty_rx.recv().await { + Ok(()) => { + if trigger_tx.send(DeviceInfoTrigger::Event).is_err() { + break; + } + } + Err(tokio::sync::broadcast::error::RecvError::Lagged(count)) => { + if trigger_tx + .send(DeviceInfoTrigger::Lagged { + topic: "device_info_dirty", + count, + }) + .is_err() + { + break; + } + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + } + } + })); + } + + tasks.push(tokio::spawn(async move { + let mut last_broadcast = Instant::now() - Duration::from_millis(DEBOUNCE_MS); + let mut pending_broadcast = false; + + loop { + let recv_result = if pending_broadcast { + let remaining = + DEBOUNCE_MS.saturating_sub(last_broadcast.elapsed().as_millis() as u64); + tokio::time::timeout(Duration::from_millis(remaining), trigger_rx.recv()).await + } else { + Ok(trigger_rx.recv().await) + }; + + match recv_result { + Ok(Some(DeviceInfoTrigger::Event)) => { + pending_broadcast = true; + } + Ok(Some(DeviceInfoTrigger::Lagged { topic, count })) => { + tracing::warn!( + "DeviceInfo broadcaster lagged by {} events on topic {}", + count, + topic + ); + pending_broadcast = true; + } + Ok(None) => { + tracing::info!("Event bus closed, stopping DeviceInfo broadcaster"); + break; + } + Err(_timeout) => {} + } + + if pending_broadcast && last_broadcast.elapsed() >= Duration::from_millis(DEBOUNCE_MS) { + state.publish_device_info().await; + tracing::trace!("Broadcasted DeviceInfo (debounced)"); + last_broadcast = Instant::now(); + pending_broadcast = false; + } + } + })); + + tracing::info!( + "DeviceInfo broadcaster task started (debounce: {}ms)", + DEBOUNCE_MS + ); + tasks +} + +async fn cleanup(state: &Arc) { + state.extensions.stop_all().await; + tracing::info!("Extensions stopped"); + + state.remote_access.shutdown().await; + + if let Err(error) = state.stream_manager.stop().await { + tracing::warn!("Failed to stop streamer: {}", error); + } + + if let Err(error) = state.hid.shutdown().await { + tracing::warn!("Failed to shutdown HID: {}", error); + } + + #[cfg(unix)] + { + let msd = state.msd.write().await.take(); + if let Some(msd) = msd { + if let Err(error) = msd.shutdown().await { + tracing::warn!("Failed to shutdown MSD: {}", error); + } + } + + if let Err(error) = state.otg_service.shutdown().await { + tracing::warn!("Failed to shutdown OTG: {}", error); + } + } + + let atx = state.atx.write().await.take(); + if let Some(atx) = atx { + if let Err(error) = atx.shutdown().await { + tracing::warn!("Failed to shutdown ATX: {}", error); + } + } + + if let Err(error) = state.audio.shutdown().await { + tracing::warn!("Failed to shutdown audio: {}", error); + } + + if let Err(error) = state.watchdog.disable().await { + tracing::error!( + "CRITICAL: failed to disable hardware watchdog during shutdown: {}", + error + ); + } +} diff --git a/src/runtime/usb.rs b/src/runtime/usb.rs new file mode 100644 index 00000000..2dec7d4a --- /dev/null +++ b/src/runtime/usb.rs @@ -0,0 +1,335 @@ +use std::path::PathBuf; +use std::sync::Arc; + +#[cfg(unix)] +use tokio::sync::RwLock; + +use crate::config::{AppConfig, HidBackend, HidConfig, MsdConfig, OtgNetworkConfig, UacConfig}; +use crate::error::{AppError, Result}; +use crate::events::EventBus; +use crate::hid::{HidBackendType, HidController}; +#[cfg(unix)] +use crate::msd::MsdController; +#[cfg(unix)] +use crate::otg::OtgService; + +use super::ConfigApplyOptions; + +pub struct UsbCoordinator { + hid: Arc, + #[cfg(unix)] + otg: Arc, + #[cfg(unix)] + msd: Arc>>, + #[cfg(unix)] + uac_playback: Arc>>, + events: Arc, + data_dir: PathBuf, +} + +impl UsbCoordinator { + #[allow(clippy::too_many_arguments)] + pub fn new( + hid: Arc, + #[cfg(unix)] otg: Arc, + #[cfg(unix)] msd: Arc>>, + #[cfg(unix)] uac_playback: Arc>>, + events: Arc, + data_dir: PathBuf, + ) -> Arc { + Arc::new(Self { + hid, + #[cfg(unix)] + otg, + #[cfg(unix)] + msd, + #[cfg(unix)] + uac_playback, + events, + data_dir, + }) + } + + pub async fn apply_config(&self, old_config: &AppConfig, new_config: &AppConfig) -> Result<()> { + #[cfg(unix)] + { + let transitioning_away_from_otg = old_config.hid.backend == HidBackend::Otg + && new_config.hid.backend != HidBackend::Otg; + let hid_unchanged = old_config.hid == new_config.hid; + let gadget_rebuilt = old_config.msd != new_config.msd + || old_config.otg_network != new_config.otg_network + || old_config.uac != new_config.uac + || old_config.hid.otg_udc != new_config.hid.otg_udc + || old_config.hid.otg_descriptor != new_config.hid.otg_descriptor + || old_config.hid.backend != new_config.hid.backend + || old_config.hid.constrained_otg_functions() + != new_config.hid.constrained_otg_functions() + || old_config.hid.effective_otg_keyboard_leds() + != new_config.hid.effective_otg_keyboard_leds(); + let restart_uac = + old_config.uac != new_config.uac || (new_config.uac.enabled && gadget_rebuilt); + + if restart_uac { + let playback = self.uac_playback.write().await.take(); + if let Some(playback) = playback { + playback.stop(); + tracing::info!("UAC playback writer stopped before OTG reconcile"); + } + } + + if transitioning_away_from_otg { + self.apply_hid( + &old_config.hid, + &new_config.hid, + &new_config.msd, + &new_config.otg_network, + &new_config.uac, + ConfigApplyOptions::default(), + ) + .await?; + } else { + self.reconcile_otg( + &new_config.hid, + &new_config.msd, + &new_config.otg_network, + &new_config.uac, + ) + .await?; + self.apply_hid( + &old_config.hid, + &new_config.hid, + &new_config.msd, + &new_config.otg_network, + &new_config.uac, + ConfigApplyOptions::default(), + ) + .await?; + } + + if hid_unchanged && gadget_rebuilt && new_config.hid.backend == HidBackend::Otg { + tracing::info!("OTG gadget rebuilt, reloading HID backend for new devices"); + self.hid + .reload(hid_backend_type(&new_config.hid)) + .await + .map_err(|error| { + AppError::Config(format!("HID reload after gadget rebuild failed: {error}")) + })?; + } + + self.apply_msd( + &old_config.msd, + &new_config.msd, + &new_config.hid, + &new_config.otg_network, + &new_config.uac, + ConfigApplyOptions::default(), + ) + .await?; + + if restart_uac && new_config.uac.enabled { + let config = crate::audio::uac::UacPlaybackConfig { + sample_rate: new_config.uac.sample_rate, + channels: new_config.uac.channels as u16, + ..Default::default() + }; + let writer = crate::audio::uac::UacPlayback::start(config).map_err(|error| { + AppError::Config(format!("Failed to start UAC playback: {error}")) + })?; + *self.uac_playback.write().await = Some(writer); + tracing::info!("UAC playback writer started after OTG reconcile"); + } else if restart_uac { + tracing::info!("UAC playback remains disabled"); + } + + Ok(()) + } + + #[cfg(not(unix))] + { + self.apply_hid( + &old_config.hid, + &new_config.hid, + &new_config.msd, + &new_config.otg_network, + &new_config.uac, + ConfigApplyOptions::default(), + ) + .await + } + } + + async fn apply_hid( + &self, + old_config: &HidConfig, + new_config: &HidConfig, + msd_config: &MsdConfig, + network_config: &OtgNetworkConfig, + uac_config: &UacConfig, + options: ConfigApplyOptions, + ) -> Result<()> { + new_config.validate_otg_functions()?; + + let descriptor_changed = old_config.otg_descriptor != new_config.otg_descriptor; + let hid_functions_changed = + old_config.constrained_otg_functions() != new_config.constrained_otg_functions(); + let keyboard_leds_changed = + old_config.effective_otg_keyboard_leds() != new_config.effective_otg_keyboard_leds(); + let ch9329_runtime_changed = + old_config.ch9329_hybrid_mouse != new_config.ch9329_hybrid_mouse; + + if old_config.backend == new_config.backend + && old_config.ch9329_port == new_config.ch9329_port + && old_config.ch9329_baudrate == new_config.ch9329_baudrate + && !ch9329_runtime_changed + && old_config.otg_udc == new_config.otg_udc + && !descriptor_changed + && !hid_functions_changed + && !keyboard_leds_changed + && !options.force + { + tracing::info!("HID config unchanged, skipping reload"); + return Ok(()); + } + + tracing::info!("Applying HID config changes..."); + let backend = hid_backend_type(new_config); + let transitioning_away_from_otg = + old_config.backend == HidBackend::Otg && new_config.backend != HidBackend::Otg; + let otg_changed = hid_otg_config_changed(old_config, new_config); + + if transitioning_away_from_otg { + self.hid + .reload(backend.clone()) + .await + .map_err(|error| AppError::Config(format!("HID reload failed: {error}")))?; + } + if otg_changed { + self.reconcile_otg(new_config, msd_config, network_config, uac_config) + .await?; + } + if !transitioning_away_from_otg { + self.hid + .reload(backend) + .await + .map_err(|error| AppError::Config(format!("HID reload failed: {error}")))?; + } + + tracing::info!( + "HID backend reloaded successfully: {:?}", + new_config.backend + ); + Ok(()) + } + + async fn reconcile_otg( + &self, + hid: &HidConfig, + msd: &MsdConfig, + network: &OtgNetworkConfig, + uac: &UacConfig, + ) -> Result<()> { + #[cfg(unix)] + { + self.otg + .apply_config(hid, msd, network, uac) + .await + .map_err(|error| AppError::Config(format!("OTG reconcile failed: {error}"))) + } + #[cfg(not(unix))] + { + let _ = (hid, msd, network, uac); + Ok(()) + } + } + + #[cfg(unix)] + async fn apply_msd( + &self, + old_config: &MsdConfig, + new_config: &MsdConfig, + hid_config: &HidConfig, + network_config: &OtgNetworkConfig, + uac_config: &UacConfig, + options: ConfigApplyOptions, + ) -> Result<()> { + let old_enabled = old_config.enabled; + let new_enabled = new_config.enabled && hid_config.backend == HidBackend::Otg; + let directory_changed = old_config.msd_dir != new_config.msd_dir; + let inquiry_changed = old_config.flash_inquiry_string != new_config.flash_inquiry_string + || old_config.cdrom_inquiry_string != new_config.cdrom_inquiry_string; + + let msd_dir = new_config.msd_dir_path(); + if let Err(error) = std::fs::create_dir_all(msd_dir.join("images")) { + tracing::warn!("Failed to create MSD images directory: {}", error); + } + if let Err(error) = std::fs::create_dir_all(msd_dir.join("ventoy")) { + tracing::warn!("Failed to create MSD ventoy directory: {}", error); + } + + if !options.force && old_enabled == new_enabled && !directory_changed && !inquiry_changed { + tracing::info!("MSD configuration unchanged, no reload needed"); + return Ok(()); + } + + if new_enabled { + tracing::info!("(Re)initializing MSD..."); + self.reconcile_otg(hid_config, new_config, network_config, uac_config) + .await?; + + let old_msd = self.msd.write().await.take(); + if let Some(msd) = old_msd { + msd.shutdown() + .await + .map_err(|error| AppError::Config(format!("MSD shutdown failed: {error}")))?; + } + + let msd = MsdController::new(self.otg.clone(), new_config.msd_dir_path()); + msd.init(&self.data_dir.join("ventoy")) + .await + .map_err(|error| AppError::Config(format!("MSD initialization failed: {error}")))?; + msd.set_event_bus(self.events.clone()).await; + *self.msd.write().await = Some(msd); + tracing::info!("MSD initialized successfully"); + } else { + tracing::info!("MSD disabled in config, shutting down..."); + let old_msd = self.msd.write().await.take(); + if let Some(msd) = old_msd { + msd.shutdown() + .await + .map_err(|error| AppError::Config(format!("MSD shutdown failed: {error}")))?; + } + tracing::info!("MSD shutdown complete"); + self.reconcile_otg(hid_config, new_config, network_config, uac_config) + .await?; + } + + if hid_config.backend == HidBackend::Otg && (options.force || old_enabled != new_enabled) { + self.hid + .reload(HidBackendType::Otg) + .await + .map_err(|error| AppError::Config(format!("OTG HID reload failed: {error}")))?; + } + Ok(()) + } +} + +fn hid_backend_type(config: &HidConfig) -> HidBackendType { + match config.backend { + HidBackend::Otg => HidBackendType::Otg, + HidBackend::Ch9329 => HidBackendType::Ch9329 { + port: config.ch9329_port.clone(), + baud_rate: config.ch9329_baudrate, + hybrid_mouse: config.ch9329_hybrid_mouse, + }, + HidBackend::None => HidBackendType::None, + } +} + +fn hid_otg_config_changed(old_config: &HidConfig, new_config: &HidConfig) -> bool { + old_config.backend == HidBackend::Otg + || new_config.backend == HidBackend::Otg + || old_config.otg_udc != new_config.otg_udc + || old_config.otg_descriptor != new_config.otg_descriptor + || old_config.constrained_otg_functions() != new_config.constrained_otg_functions() + || old_config.effective_otg_keyboard_leds() != new_config.effective_otg_keyboard_leds() +} diff --git a/src/state.rs b/src/state.rs index 9afc162c..1d1e1389 100644 --- a/src/state.rs +++ b/src/state.rs @@ -19,11 +19,9 @@ use crate::hid::HidController; use crate::msd::MsdController; #[cfg(unix)] use crate::otg::OtgService; -use crate::rtsp::RtspService; -use crate::rustdesk::RustDeskService; +use crate::runtime::{RemoteAccessCoordinator, UsbCoordinator}; use crate::update::UpdateService; use crate::video::VideoStreamManager; -use crate::vnc::VncService; use crate::watchdog::WatchdogController; use crate::webrtc::WebRtcStreamer; @@ -81,9 +79,8 @@ pub struct AppState { pub audio: Arc, #[cfg(unix)] pub uac_playback: Arc>>, - pub rustdesk: Arc>>>, - pub vnc: Arc>>>, - pub rtsp: Arc>>>, + pub usb: Arc, + pub remote_access: Arc, pub extensions: Arc, pub events: Arc, device_info_tx: watch::Sender>, @@ -111,9 +108,6 @@ impl AppState { #[cfg(unix)] msd: Option, atx: Option, audio: Arc, - rustdesk: Option>, - vnc: Option>, - rtsp: Option>, extensions: Arc, events: Arc, update: Arc, @@ -122,6 +116,28 @@ impl AppState { ) -> Arc { let (device_info_tx, _device_info_rx) = watch::channel(None); + let remote_access = RemoteAccessCoordinator::new( + config.clone(), + stream_manager.clone(), + hid.clone(), + audio.clone(), + ); + #[cfg(unix)] + let msd = Arc::new(RwLock::new(msd)); + #[cfg(unix)] + let uac_playback = Arc::new(RwLock::new(None)); + let usb = UsbCoordinator::new( + hid.clone(), + #[cfg(unix)] + otg_service.clone(), + #[cfg(unix)] + msd.clone(), + #[cfg(unix)] + uac_playback.clone(), + events.clone(), + data_dir.clone(), + ); + Arc::new(Self { db, config, @@ -135,12 +151,11 @@ impl AppState { hid, computer_use, #[cfg(unix)] - msd: Arc::new(RwLock::new(msd)), + msd, atx: Arc::new(RwLock::new(atx)), audio, - rustdesk: Arc::new(RwLock::new(rustdesk)), - vnc: Arc::new(RwLock::new(vnc)), - rtsp: Arc::new(RwLock::new(rtsp)), + usb, + remote_access, extensions, events, device_info_tx, @@ -151,7 +166,7 @@ impl AppState { config_apply_locks: ConfigApplyLocks::new(), data_dir, #[cfg(unix)] - uac_playback: Arc::new(RwLock::new(None)), + uac_playback, }) } @@ -159,33 +174,6 @@ impl AppState { &self.data_dir } - pub async fn runtime_third_party_config(&self) -> crate::config::AppConfig { - let mut config = self.config.get().as_ref().clone(); - - config.rustdesk.enabled = self - .rustdesk - .read() - .await - .as_ref() - .is_some_and(|service| service.is_listening()); - config.vnc.enabled = match self.vnc.read().await.as_ref() { - Some(service) => matches!( - service.status().await, - crate::vnc::VncServiceStatus::Starting | crate::vnc::VncServiceStatus::Running - ), - None => false, - }; - config.rtsp.enabled = match self.rtsp.read().await.as_ref() { - Some(service) => matches!( - service.status().await, - crate::rtsp::RtspServiceStatus::Starting | crate::rtsp::RtspServiceStatus::Running - ), - None => false, - }; - - config - } - pub fn subscribe_device_info(&self) -> watch::Receiver> { self.device_info_tx.subscribe() } diff --git a/src/web/handlers/config/apply.rs b/src/web/handlers/config/apply.rs index be073e4c..bed1afb1 100644 --- a/src/web/handlers/config/apply.rs +++ b/src/web/handlers/config/apply.rs @@ -2,96 +2,9 @@ use std::sync::Arc; use crate::config::*; use crate::error::{AppError, Result}; -use crate::rtsp::RtspService; +pub use crate::runtime::{try_apply_lock, ConfigApplyOptions}; use crate::state::AppState; use crate::stream_encoder::encoder_type_to_backend; -use crate::video::codec_constraints::{ - enforce_constraints_with_stream_manager, validate_third_party_codec_compatibility, - StreamCodecConstraints, -}; -use tokio::sync::{Mutex, OwnedMutexGuard}; - -#[derive(Debug, Clone, Copy, Default)] -pub struct ConfigApplyOptions { - pub force: bool, - pub preserve_service_state: bool, - pub runtime_only: bool, -} - -impl ConfigApplyOptions { - pub const fn forced() -> Self { - Self { - force: true, - preserve_service_state: false, - runtime_only: false, - } - } - - pub const fn preserving_service_state() -> Self { - Self { - force: false, - preserve_service_state: true, - runtime_only: false, - } - } - - pub const fn runtime_only() -> Self { - Self { - force: false, - preserve_service_state: false, - runtime_only: true, - } - } -} - -pub fn try_apply_lock(lock: &Arc>, domain: &str) -> Result> { - lock.clone().try_lock_owned().map_err(|_| { - AppError::ServiceUnavailable(format!("{domain} configuration is already applying")) - }) -} - -fn hid_backend_type(config: &HidConfig) -> crate::hid::HidBackendType { - match config.backend { - HidBackend::Otg => crate::hid::HidBackendType::Otg, - HidBackend::Ch9329 => crate::hid::HidBackendType::Ch9329 { - port: config.ch9329_port.clone(), - baud_rate: config.ch9329_baudrate, - hybrid_mouse: config.ch9329_hybrid_mouse, - }, - HidBackend::None => crate::hid::HidBackendType::None, - } -} - -fn hid_otg_config_changed(old_config: &HidConfig, new_config: &HidConfig) -> bool { - old_config.backend == HidBackend::Otg - || new_config.backend == HidBackend::Otg - || old_config.otg_udc != new_config.otg_udc - || old_config.otg_descriptor != new_config.otg_descriptor - || old_config.constrained_otg_functions() != new_config.constrained_otg_functions() - || old_config.effective_otg_keyboard_leds() != new_config.effective_otg_keyboard_leds() -} - -async fn reconcile_otg_config( - state: &Arc, - hid: &HidConfig, - msd: &MsdConfig, - network: &OtgNetworkConfig, - uac: &UacConfig, -) -> Result<()> { - #[cfg(not(unix))] - { - let _ = (state, hid, msd, network, uac); - Ok(()) - } - #[cfg(unix)] - { - state - .otg_service - .apply_config(hid, msd, network, uac) - .await - .map_err(|e| AppError::Config(format!("OTG reconcile failed: {}", e))) - } -} pub async fn apply_video_config( state: &Arc, @@ -189,305 +102,6 @@ pub async fn apply_stream_config( Ok(()) } -pub async fn apply_hid_config( - state: &Arc, - old_config: &HidConfig, - new_config: &HidConfig, - msd_config: &MsdConfig, - network_config: &OtgNetworkConfig, - uac_config: &UacConfig, - options: ConfigApplyOptions, -) -> Result<()> { - new_config.validate_otg_functions()?; - - let descriptor_changed = old_config.otg_descriptor != new_config.otg_descriptor; - let old_hid_functions = old_config.constrained_otg_functions(); - let new_hid_functions = new_config.constrained_otg_functions(); - let hid_functions_changed = old_hid_functions != new_hid_functions; - let keyboard_leds_changed = - old_config.effective_otg_keyboard_leds() != new_config.effective_otg_keyboard_leds(); - let ch9329_runtime_changed = old_config.ch9329_hybrid_mouse != new_config.ch9329_hybrid_mouse; - - if old_config.backend == new_config.backend - && old_config.ch9329_port == new_config.ch9329_port - && old_config.ch9329_baudrate == new_config.ch9329_baudrate - && !ch9329_runtime_changed - && old_config.otg_udc == new_config.otg_udc - && !descriptor_changed - && !hid_functions_changed - && !keyboard_leds_changed - && !options.force - { - tracing::info!("HID config unchanged, skipping reload"); - return Ok(()); - } - - tracing::info!("Applying HID config changes..."); - - let new_hid_backend = hid_backend_type(new_config); - let transitioning_away_from_otg = - old_config.backend == HidBackend::Otg && new_config.backend != HidBackend::Otg; - let otg_config_changed = hid_otg_config_changed(old_config, new_config); - - if transitioning_away_from_otg { - state - .hid - .reload(new_hid_backend.clone()) - .await - .map_err(|e| AppError::Config(format!("HID reload failed: {}", e)))?; - } - - if otg_config_changed { - reconcile_otg_config(state, new_config, msd_config, network_config, uac_config).await?; - } - - if !transitioning_away_from_otg { - state - .hid - .reload(new_hid_backend) - .await - .map_err(|e| AppError::Config(format!("HID reload failed: {}", e)))?; - } - - tracing::info!( - "HID backend reloaded successfully: {:?}", - new_config.backend - ); - - Ok(()) -} - -#[cfg(unix)] -pub async fn apply_msd_config( - state: &Arc, - old_config: &MsdConfig, - new_config: &MsdConfig, - hid_config: &HidConfig, - network_config: &OtgNetworkConfig, - uac_config: &UacConfig, - options: ConfigApplyOptions, -) -> Result<()> { - let hid_backend_is_otg = hid_config.backend == HidBackend::Otg; - let effective_new_msd_enabled = new_config.enabled && hid_backend_is_otg; - - tracing::info!("MSD config sent, checking if reload needed..."); - tracing::debug!("Old MSD config: {:?}", old_config); - tracing::debug!("New MSD config: {:?}", new_config); - - let old_msd_enabled = old_config.enabled; - let new_msd_enabled = effective_new_msd_enabled; - let msd_dir_changed = old_config.msd_dir != new_config.msd_dir; - let inquiry_strings_changed = old_config.flash_inquiry_string - != new_config.flash_inquiry_string - || old_config.cdrom_inquiry_string != new_config.cdrom_inquiry_string; - - tracing::info!( - "MSD enabled: old={}, new={}", - old_msd_enabled, - new_msd_enabled - ); - if msd_dir_changed { - tracing::info!("MSD directory changed: {}", new_config.msd_dir); - } - if inquiry_strings_changed { - tracing::info!("MSD inquiry strings changed"); - } - - let msd_dir = new_config.msd_dir_path(); - if let Err(e) = std::fs::create_dir_all(msd_dir.join("images")) { - tracing::warn!("Failed to create MSD images directory: {}", e); - } - if let Err(e) = std::fs::create_dir_all(msd_dir.join("ventoy")) { - tracing::warn!("Failed to create MSD ventoy directory: {}", e); - } - - let needs_reload = options.force - || old_msd_enabled != new_msd_enabled - || msd_dir_changed - || inquiry_strings_changed; - if !needs_reload { - tracing::info!("MSD configuration unchanged, no reload needed"); - return Ok(()); - } - - if new_msd_enabled { - tracing::info!("(Re)initializing MSD..."); - - reconcile_otg_config(state, hid_config, new_config, network_config, uac_config).await?; - - let mut msd_guard = state.msd.write().await; - if let Some(msd) = msd_guard.as_mut() { - msd.shutdown() - .await - .map_err(|e| AppError::Config(format!("MSD shutdown failed: {e}")))?; - } - *msd_guard = None; - drop(msd_guard); - - let msd = - crate::msd::MsdController::new(state.otg_service.clone(), new_config.msd_dir_path()); - let ventoy_resource_dir = state.data_dir().join("ventoy"); - msd.init(&ventoy_resource_dir) - .await - .map_err(|e| AppError::Config(format!("MSD initialization failed: {}", e)))?; - - let events = state.events.clone(); - msd.set_event_bus(events).await; - - *state.msd.write().await = Some(msd); - tracing::info!("MSD initialized successfully"); - } else { - tracing::info!("MSD disabled in config, shutting down..."); - - let mut msd_guard = state.msd.write().await; - if let Some(msd) = msd_guard.as_mut() { - msd.shutdown() - .await - .map_err(|e| AppError::Config(format!("MSD shutdown failed: {e}")))?; - } - *msd_guard = None; - tracing::info!("MSD shutdown complete"); - - reconcile_otg_config(state, hid_config, new_config, network_config, uac_config).await?; - } - - if hid_config.backend == HidBackend::Otg - && (options.force || old_msd_enabled != new_msd_enabled) - { - state - .hid - .reload(crate::hid::HidBackendType::Otg) - .await - .map_err(|e| AppError::Config(format!("OTG HID reload failed: {}", e)))?; - } - - Ok(()) -} - -pub async fn apply_usb_config( - state: &Arc, - old_config: &AppConfig, - new_config: &AppConfig, -) -> Result<()> { - #[cfg(unix)] - { - let transitioning_away_from_otg = - old_config.hid.backend == HidBackend::Otg && new_config.hid.backend != HidBackend::Otg; - - let hid_unchanged = old_config.hid == new_config.hid; - let otg_gadget_rebuilt = old_config.msd != new_config.msd - || old_config.otg_network != new_config.otg_network - || old_config.uac != new_config.uac - || old_config.hid.otg_udc != new_config.hid.otg_udc - || old_config.hid.otg_descriptor != new_config.hid.otg_descriptor - || old_config.hid.backend != new_config.hid.backend - || old_config.hid.constrained_otg_functions() - != new_config.hid.constrained_otg_functions() - || old_config.hid.effective_otg_keyboard_leds() - != new_config.hid.effective_otg_keyboard_leds(); - let restart_uac_playback = - old_config.uac != new_config.uac || (new_config.uac.enabled && otg_gadget_rebuilt); - - // A bound ALSA handle refers to the old configfs function. Stop it - // before any gadget teardown so the worker cannot write through a - // disappearing PCM node. It is restarted only after every reconcile. - if restart_uac_playback { - let playback = state.uac_playback.write().await.take(); - if let Some(playback) = playback { - playback.stop(); - tracing::info!("UAC playback writer stopped before OTG reconcile"); - } - } - - if transitioning_away_from_otg { - apply_hid_config( - state, - &old_config.hid, - &new_config.hid, - &new_config.msd, - &new_config.otg_network, - &new_config.uac, - ConfigApplyOptions::default(), - ) - .await?; - } else { - reconcile_otg_config( - state, - &new_config.hid, - &new_config.msd, - &new_config.otg_network, - &new_config.uac, - ) - .await?; - apply_hid_config( - state, - &old_config.hid, - &new_config.hid, - &new_config.msd, - &new_config.otg_network, - &new_config.uac, - ConfigApplyOptions::default(), - ) - .await?; - } - - // When the OTG gadget was rebuilt due to MSD or network config changes - // while HID config stayed the same, the /dev/hidg* devices are new and - // the HID backend must be reloaded to reopen them. - if hid_unchanged && otg_gadget_rebuilt && new_config.hid.backend == HidBackend::Otg { - tracing::info!("OTG gadget rebuilt, reloading HID backend for new devices"); - let hid_backend = hid_backend_type(&new_config.hid); - state.hid.reload(hid_backend).await.map_err(|e| { - AppError::Config(format!("HID reload after gadget rebuild failed: {}", e)) - })?; - } - - apply_msd_config( - state, - &old_config.msd, - &new_config.msd, - &new_config.hid, - &new_config.otg_network, - &new_config.uac, - ConfigApplyOptions::default(), - ) - .await?; - - // apply_msd_config may perform a second gadget reconcile. Resolve the - // new ALSA card only after that final rebuild, then publish the worker. - if restart_uac_playback && new_config.uac.enabled { - let config = crate::audio::uac::UacPlaybackConfig { - sample_rate: new_config.uac.sample_rate, - channels: new_config.uac.channels as u16, - ..Default::default() - }; - let writer = crate::audio::uac::UacPlayback::start(config).map_err(|error| { - AppError::Config(format!("Failed to start UAC playback: {error}")) - })?; - *state.uac_playback.write().await = Some(writer); - tracing::info!("UAC playback writer started after OTG reconcile"); - } else if restart_uac_playback { - tracing::info!("UAC playback remains disabled"); - } - - Ok(()) - } - - #[cfg(not(unix))] - { - apply_hid_config( - state, - &old_config.hid, - &new_config.hid, - &new_config.msd, - &new_config.otg_network, - &new_config.uac, - ConfigApplyOptions::default(), - ) - .await - } -} - pub async fn apply_atx_config( state: &Arc, _old_config: &AtxConfig, @@ -554,336 +168,3 @@ pub async fn apply_audio_config( Ok(()) } - -pub async fn enforce_stream_codec_constraints(state: &Arc) -> Result> { - let config = state.runtime_third_party_config().await; - let constraints = StreamCodecConstraints::from_config(&config); - state - .stream_manager - .set_runtime_codec_constraints(constraints.clone()) - .await; - let enforcement = - enforce_constraints_with_stream_manager(&state.stream_manager, &constraints).await?; - Ok(enforcement.message) -} - -async fn validate_runtime_candidate( - state: &Arc, - apply: impl FnOnce(&mut crate::config::AppConfig, T), - config: T, -) -> Result<()> { - let mut candidate = state.runtime_third_party_config().await; - apply(&mut candidate, config); - validate_third_party_codec_compatibility(&candidate) -} - -fn validate_rustdesk_candidate( - state: &Arc, - new_config: &crate::rustdesk::config::RustDeskConfig, -) -> Result<()> { - let mut candidate = state.config.get().as_ref().clone(); - candidate.rustdesk = new_config.clone(); - validate_third_party_codec_compatibility(&candidate) -} - -fn validate_vnc_candidate(state: &Arc, new_config: &VncConfig) -> Result<()> { - let mut candidate = state.config.get().as_ref().clone(); - candidate.vnc = new_config.clone(); - validate_third_party_codec_compatibility(&candidate) -} - -fn validate_rtsp_candidate(state: &Arc, new_config: &RtspConfig) -> Result<()> { - let mut candidate = state.config.get().as_ref().clone(); - candidate.rtsp = new_config.clone(); - validate_third_party_codec_compatibility(&candidate) -} - -pub async fn apply_rustdesk_config( - state: &Arc, - old_config: &crate::rustdesk::config::RustDeskConfig, - new_config: &crate::rustdesk::config::RustDeskConfig, - options: ConfigApplyOptions, -) -> Result<()> { - tracing::info!("Applying RustDesk config changes..."); - - if options.runtime_only { - validate_runtime_candidate( - state, - |candidate, config| candidate.rustdesk = config, - new_config.clone(), - ) - .await?; - } else { - validate_rustdesk_candidate(state, new_config)?; - } - - let mut rustdesk_guard = state.rustdesk.write().await; - let mut credentials_to_save = None; - let need_restart = options.force - || old_config.codec != new_config.codec - || old_config.rendezvous_server != new_config.rendezvous_server - || old_config.device_id != new_config.device_id - || old_config.device_password != new_config.device_password; - - if !options.preserve_service_state && !new_config.enabled { - if let Some(ref service) = *rustdesk_guard { - service - .stop() - .await - .map_err(|e| AppError::Config(format!("Failed to stop RustDesk service: {}", e)))?; - tracing::info!("RustDesk service stopped"); - } - *rustdesk_guard = None; - } - - if !options.preserve_service_state && new_config.enabled { - if rustdesk_guard.is_none() { - tracing::info!("Initializing RustDesk service..."); - let service = std::sync::Arc::new(crate::rustdesk::RustDeskService::new( - new_config.clone(), - state.stream_manager.clone(), - state.hid.clone(), - state.audio.clone(), - )); - *rustdesk_guard = Some(service.clone()); - service.start().await.map_err(|e| { - AppError::Config(format!("Failed to start RustDesk service: {}", e)) - })?; - tracing::info!("RustDesk service started with ID: {}", new_config.device_id); - credentials_to_save = service.save_credentials(); - } else { - if let Some(ref service) = *rustdesk_guard { - if service.is_listening() { - if need_restart { - service.restart(new_config.clone()).await.map_err(|e| { - AppError::Config(format!("Failed to restart RustDesk service: {}", e)) - })?; - tracing::info!( - "RustDesk service restarted with ID: {}", - new_config.device_id - ); - } - } else { - service.update_config(new_config.clone()); - service.start().await.map_err(|e| { - AppError::Config(format!("Failed to start RustDesk service: {}", e)) - })?; - } - credentials_to_save = service.save_credentials(); - } - } - } else if options.preserve_service_state && need_restart { - if let Some(ref service) = *rustdesk_guard { - let mut runtime_config = new_config.clone(); - runtime_config.enabled = true; - service.restart(runtime_config).await.map_err(|e| { - AppError::Config(format!("Failed to restart RustDesk service: {}", e)) - })?; - credentials_to_save = service.save_credentials(); - } - } - - drop(rustdesk_guard); - if let Some(updated_config) = credentials_to_save { - tracing::info!("Saving RustDesk credentials to config store..."); - state - .config - .update(|cfg| { - cfg.rustdesk.public_key = updated_config.public_key.clone(); - cfg.rustdesk.private_key = updated_config.private_key.clone(); - cfg.rustdesk.signing_public_key = updated_config.signing_public_key.clone(); - cfg.rustdesk.signing_private_key = updated_config.signing_private_key.clone(); - cfg.rustdesk.uuid = updated_config.uuid.clone(); - }) - .await?; - tracing::info!("RustDesk credentials saved successfully"); - } - - if let Some(message) = enforce_stream_codec_constraints(state).await? { - tracing::info!("{}", message); - } - - Ok(()) -} - -pub async fn apply_vnc_config( - state: &Arc, - old_config: &VncConfig, - new_config: &VncConfig, - options: ConfigApplyOptions, -) -> Result<()> { - tracing::info!("Applying VNC config changes..."); - - if options.runtime_only { - validate_runtime_candidate( - state, - |candidate, config| candidate.vnc = config, - new_config.clone(), - ) - .await?; - } else { - validate_vnc_candidate(state, new_config)?; - } - - let runtime_config = state.runtime_third_party_config().await; - let will_run = if options.preserve_service_state { - runtime_config.vnc.enabled - } else { - new_config.enabled - }; - if will_run { - let mut candidate = runtime_config; - candidate.vnc = new_config.clone(); - candidate.vnc.enabled = true; - let constraints = StreamCodecConstraints::from_config(&candidate); - match enforce_constraints_with_stream_manager(&state.stream_manager, &constraints).await { - Ok(result) if result.changed => { - if let Some(message) = result.message { - tracing::info!("{}", message); - } - } - Ok(_) => {} - Err(e) => tracing::warn!( - "Failed to enforce VNC stream constraints before start: {}", - e - ), - } - } - - let mut vnc_guard = state.vnc.write().await; - let need_restart = options.force - || old_config.bind != new_config.bind - || old_config.port != new_config.port - || old_config.encoding != new_config.encoding - || old_config.password != new_config.password - || old_config.allow_one_client != new_config.allow_one_client; - - if !options.preserve_service_state && !new_config.enabled { - if let Some(ref service) = *vnc_guard { - service.stop().await?; - } - *vnc_guard = None; - } - - if !options.preserve_service_state && new_config.enabled { - if vnc_guard.is_none() { - let service = Arc::new(crate::vnc::VncService::new( - new_config.clone(), - state.stream_manager.clone(), - state.hid.clone(), - )); - *vnc_guard = Some(service.clone()); - service.start().await?; - tracing::info!("VNC service started"); - } else { - if let Some(ref service) = *vnc_guard { - if matches!( - service.status().await, - crate::vnc::VncServiceStatus::Running - ) { - if need_restart { - service.restart(new_config.clone()).await?; - tracing::info!("VNC service restarted"); - } - } else { - service.update_config(new_config.clone()).await; - service.start().await?; - } - } - } - } else if options.preserve_service_state && need_restart { - if let Some(ref service) = *vnc_guard { - let mut runtime_config = new_config.clone(); - runtime_config.enabled = true; - service.restart(runtime_config).await?; - } - } - - drop(vnc_guard); - if let Some(message) = enforce_stream_codec_constraints(state).await? { - tracing::info!("{}", message); - } - - Ok(()) -} - -pub async fn apply_rtsp_config( - state: &Arc, - old_config: &RtspConfig, - new_config: &RtspConfig, - options: ConfigApplyOptions, -) -> Result<()> { - tracing::info!("Applying RTSP config changes..."); - - if options.runtime_only { - validate_runtime_candidate( - state, - |candidate, config| candidate.rtsp = config, - new_config.clone(), - ) - .await?; - } else { - validate_rtsp_candidate(state, new_config)?; - } - - let mut rtsp_guard = state.rtsp.write().await; - let need_restart = options.force - || old_config.bind != new_config.bind - || old_config.port != new_config.port - || old_config.path != new_config.path - || old_config.codec != new_config.codec - || old_config.username != new_config.username - || old_config.password != new_config.password - || old_config.allow_one_client != new_config.allow_one_client; - - if !options.preserve_service_state && !new_config.enabled { - if let Some(ref service) = *rtsp_guard { - service - .stop() - .await - .map_err(|e| AppError::Config(format!("Failed to stop RTSP service: {}", e)))?; - } - *rtsp_guard = None; - } - - if !options.preserve_service_state && new_config.enabled { - if rtsp_guard.is_none() { - let service = Arc::new(RtspService::new( - new_config.clone(), - state.stream_manager.clone(), - )); - *rtsp_guard = Some(service.clone()); - service.start().await?; - tracing::info!("RTSP service started"); - } else { - if let Some(ref service) = *rtsp_guard { - if matches!( - service.status().await, - crate::rtsp::RtspServiceStatus::Running - ) { - if need_restart { - service.restart(new_config.clone()).await?; - tracing::info!("RTSP service restarted"); - } - } else { - service.update_config(new_config.clone()).await; - service.start().await?; - } - } - } - } else if options.preserve_service_state && need_restart { - if let Some(ref service) = *rtsp_guard { - let mut runtime_config = new_config.clone(); - runtime_config.enabled = true; - service.restart(runtime_config).await?; - } - } - - drop(rtsp_guard); - - if let Some(message) = enforce_stream_codec_constraints(state).await? { - tracing::info!("{}", message); - } - - Ok(()) -} diff --git a/src/web/handlers/config/hid.rs b/src/web/handlers/config/hid.rs index 3dcf45f6..7e4d88db 100644 --- a/src/web/handlers/config/hid.rs +++ b/src/web/handlers/config/hid.rs @@ -1,19 +1,18 @@ use axum::{extract::State, Json}; -use std::sync::Arc; use crate::config::HidConfig; use crate::error::Result; -use crate::state::AppState; +use crate::web::state::UsbApiState; use super::types::HidConfigUpdate; use super::usb_update::{stage_hid_config_update, update_usb_config}; -pub async fn get_hid_config(State(state): State>) -> Json { +pub async fn get_hid_config(State(state): State) -> Json { Json(state.config.get().hid.clone()) } pub async fn update_hid_config( - State(state): State>, + State(state): State, Json(req): Json, ) -> Result> { let config = update_usb_config(&state, move |staged| { diff --git a/src/web/handlers/config/msd.rs b/src/web/handlers/config/msd.rs index 47a37431..a975a004 100644 --- a/src/web/handlers/config/msd.rs +++ b/src/web/handlers/config/msd.rs @@ -1,19 +1,18 @@ use axum::{extract::State, Json}; -use std::sync::Arc; use crate::config::MsdConfig; use crate::error::Result; -use crate::state::AppState; +use crate::web::state::UsbApiState; use super::otg::update_otg_config_inner; use super::types::{MsdConfigUpdate, OtgConfigUpdate}; -pub async fn get_msd_config(State(state): State>) -> Json { +pub async fn get_msd_config(State(state): State) -> Json { Json(state.config.get().msd.clone()) } pub async fn update_msd_config( - State(state): State>, + State(state): State, Json(req): Json, ) -> Result> { let response = update_otg_config_inner( diff --git a/src/web/handlers/config/otg.rs b/src/web/handlers/config/otg.rs index b06bee9b..9b9f4d56 100644 --- a/src/web/handlers/config/otg.rs +++ b/src/web/handlers/config/otg.rs @@ -1,5 +1,3 @@ -use std::sync::Arc; - use axum::{extract::State, Json}; use serde::Serialize; use typeshare::typeshare; @@ -7,7 +5,7 @@ use typeshare::typeshare; use crate::config::{HidConfig, MsdConfig, OtgNetworkConfig}; use crate::error::Result; use crate::otg::OtgNetworkStatus; -use crate::state::AppState; +use crate::web::state::UsbApiState; use super::types::OtgConfigUpdate; use super::usb_update::{stage_hid_config_update, update_usb_config}; @@ -22,14 +20,14 @@ pub struct OtgConfigResponse { } pub async fn update_otg_config( - State(state): State>, + State(state): State, Json(request): Json, ) -> Result> { update_otg_config_inner(&state, request).await.map(Json) } pub(super) async fn update_otg_config_inner( - state: &Arc, + state: &UsbApiState, request: OtgConfigUpdate, ) -> Result { let staged_config = update_usb_config(state, move |staged| { @@ -54,6 +52,6 @@ pub(super) async fn update_otg_config_inner( hid: staged_config.hid, msd: staged_config.msd, network: staged_config.otg_network, - status: state.otg_service.network_status().await, + status: state.otg.network_status().await, }) } diff --git a/src/web/handlers/config/otg_network.rs b/src/web/handlers/config/otg_network.rs index d9da27dc..4f0abae9 100644 --- a/src/web/handlers/config/otg_network.rs +++ b/src/web/handlers/config/otg_network.rs @@ -1,21 +1,19 @@ -use std::sync::Arc; - use axum::{extract::State, Json}; use crate::config::OtgNetworkConfig; use crate::error::Result; use crate::otg::OtgNetworkStatus; -use crate::state::AppState; +use crate::web::state::UsbApiState; use super::otg::update_otg_config_inner; use super::types::{OtgConfigUpdate, OtgNetworkConfigUpdate}; -pub async fn get_otg_network_config(State(state): State>) -> Json { +pub async fn get_otg_network_config(State(state): State) -> Json { Json(state.config.get().otg_network.clone()) } pub async fn update_otg_network_config( - State(state): State>, + State(state): State, Json(request): Json, ) -> Result> { let response = update_otg_config_inner( @@ -29,6 +27,6 @@ pub async fn update_otg_network_config( Ok(Json(response.network)) } -pub async fn get_otg_network_status(State(state): State>) -> Json { - Json(state.otg_service.network_status().await) +pub async fn get_otg_network_status(State(state): State) -> Json { + Json(state.otg.network_status().await) } diff --git a/src/web/handlers/config/rtsp.rs b/src/web/handlers/config/rtsp.rs index 7a5a5fbf..f94f03c4 100644 --- a/src/web/handlers/config/rtsp.rs +++ b/src/web/handlers/config/rtsp.rs @@ -1,20 +1,22 @@ use axum::{extract::State, Json}; -use std::sync::Arc; use crate::error::Result; -use crate::state::AppState; +use crate::web::state::RemoteAccessApiState; -use super::apply::{apply_rtsp_config, try_apply_lock, ConfigApplyOptions}; use super::types::{RtspConfigResponse, RtspConfigUpdate, RtspStatusResponse}; +use crate::runtime::{try_apply_lock, ConfigApplyOptions}; -fn validate_candidate(state: &Arc, config: &crate::config::RtspConfig) -> Result<()> { +fn validate_candidate( + state: &RemoteAccessApiState, + config: &crate::config::RtspConfig, +) -> Result<()> { let mut candidate = state.config.get().as_ref().clone(); candidate.rtsp = config.clone(); crate::video::codec_constraints::validate_third_party_codec_compatibility(&candidate) } async fn persist_and_apply( - state: &Arc, + state: &RemoteAccessApiState, old_config: crate::config::RtspConfig, new_config: crate::config::RtspConfig, ) -> Result { @@ -26,31 +28,31 @@ async fn persist_and_apply( }) .await?; let stored_config = state.config.get().rtsp.clone(); - apply_rtsp_config( - state, - &old_config, - &stored_config, - ConfigApplyOptions::preserving_service_state(), - ) - .await?; + state + .coordinator + .apply_rtsp( + &old_config, + &stored_config, + ConfigApplyOptions::preserving_service_state(), + ) + .await?; Ok(stored_config) } -async fn current_status(state: &Arc) -> crate::rtsp::RtspServiceStatus { - let guard = state.rtsp.read().await; - if let Some(ref service) = *guard { - service.status().await - } else { - crate::rtsp::RtspServiceStatus::Stopped - } +async fn current_status(state: &RemoteAccessApiState) -> crate::rtsp::RtspServiceStatus { + state.coordinator.rtsp_status().await } -pub async fn get_rtsp_config(State(state): State>) -> Json { +pub async fn get_rtsp_config( + State(state): State, +) -> Json { let config = state.config.get(); Json(RtspConfigResponse::from(&config.rtsp)) } -pub async fn get_rtsp_status(State(state): State>) -> Json { +pub async fn get_rtsp_status( + State(state): State, +) -> Json { let config = state.config.get().rtsp.clone(); let status = current_status(&state).await; @@ -58,12 +60,12 @@ pub async fn get_rtsp_status(State(state): State>) -> Json>, + State(state): State, Json(req): Json, ) -> Result> { req.validate()?; - let _apply_guard = try_apply_lock(&state.config_apply_locks.rtsp, "rtsp")?; + let _apply_guard = try_apply_lock(&state.rtsp_apply_lock, "rtsp")?; let old_config = state.config.get().rtsp.clone(); let mut merged_config = old_config.clone(); req.apply_to(&mut merged_config); @@ -73,40 +75,42 @@ pub async fn update_rtsp_config( } pub async fn start_rtsp_service( - State(state): State>, + State(state): State, ) -> Result> { - let _apply_guard = try_apply_lock(&state.config_apply_locks.rtsp, "rtsp")?; + let _apply_guard = try_apply_lock(&state.rtsp_apply_lock, "rtsp")?; let stored_config = state.config.get().rtsp.clone(); - let runtime_config = state.runtime_third_party_config().await.rtsp; + let runtime_config = state.coordinator.runtime_config().await.rtsp; let mut start_config = stored_config.clone(); start_config.enabled = true; - apply_rtsp_config( - &state, - &runtime_config, - &start_config, - ConfigApplyOptions::runtime_only(), - ) - .await?; + state + .coordinator + .apply_rtsp( + &runtime_config, + &start_config, + ConfigApplyOptions::runtime_only(), + ) + .await?; let status = current_status(&state).await; Ok(Json(RtspStatusResponse::new(&stored_config, status))) } pub async fn stop_rtsp_service( - State(state): State>, + State(state): State, ) -> Result> { - let _apply_guard = try_apply_lock(&state.config_apply_locks.rtsp, "rtsp")?; + let _apply_guard = try_apply_lock(&state.rtsp_apply_lock, "rtsp")?; let stored_config = state.config.get().rtsp.clone(); - let runtime_config = state.runtime_third_party_config().await.rtsp; + let runtime_config = state.coordinator.runtime_config().await.rtsp; let mut stop_config = stored_config.clone(); stop_config.enabled = false; - apply_rtsp_config( - &state, - &runtime_config, - &stop_config, - ConfigApplyOptions::runtime_only(), - ) - .await?; + state + .coordinator + .apply_rtsp( + &runtime_config, + &stop_config, + ConfigApplyOptions::runtime_only(), + ) + .await?; let status = current_status(&state).await; Ok(Json(RtspStatusResponse::new(&stored_config, status))) diff --git a/src/web/handlers/config/rustdesk.rs b/src/web/handlers/config/rustdesk.rs index ce08d920..7399ece0 100644 --- a/src/web/handlers/config/rustdesk.rs +++ b/src/web/handlers/config/rustdesk.rs @@ -1,21 +1,20 @@ use axum::{extract::State, Json}; -use std::sync::Arc; use crate::error::Result; use crate::rustdesk::config::RustDeskConfig; -use crate::state::AppState; +use crate::web::state::RemoteAccessApiState; -use super::apply::{apply_rustdesk_config, try_apply_lock, ConfigApplyOptions}; use super::types::RustDeskConfigUpdate; +use crate::runtime::{try_apply_lock, ConfigApplyOptions}; -fn validate_candidate(state: &Arc, config: &RustDeskConfig) -> Result<()> { +fn validate_candidate(state: &RemoteAccessApiState, config: &RustDeskConfig) -> Result<()> { let mut candidate = state.config.get().as_ref().clone(); candidate.rustdesk = config.clone(); crate::video::codec_constraints::validate_third_party_codec_compatibility(&candidate) } async fn persist_and_apply( - state: &Arc, + state: &RemoteAccessApiState, old_config: RustDeskConfig, new_config: RustDeskConfig, ) -> Result { @@ -27,32 +26,27 @@ async fn persist_and_apply( }) .await?; let stored_config = state.config.get().rustdesk.clone(); - apply_rustdesk_config( - state, - &old_config, - &stored_config, - ConfigApplyOptions::preserving_service_state(), - ) - .await?; + state + .coordinator + .apply_rustdesk( + &old_config, + &stored_config, + ConfigApplyOptions::preserving_service_state(), + ) + .await?; Ok(stored_config) } -async fn current_status(state: &Arc, config: RustDeskConfig) -> RustDeskStatusResponse { - let (service_status, rendezvous_status) = { - let guard = state.rustdesk.read().await; - if let Some(ref service) = *guard { - let status = format!("{}", service.status()); - let rv_status = service.rendezvous_status().map(|s| format!("{}", s)); - (status, rv_status) - } else { - ("not_initialized".to_string(), None) - } - }; +async fn current_status( + state: &RemoteAccessApiState, + config: RustDeskConfig, +) -> RustDeskStatusResponse { + let runtime = state.coordinator.rustdesk_status().await; RustDeskStatusResponse { config: RustDeskConfigResponse::from(&config), - service_status, - rendezvous_status, + service_status: runtime.service_status, + rendezvous_status: runtime.rendezvous_status, } } @@ -91,25 +85,25 @@ pub struct RustDeskStatusResponse { } pub async fn get_rustdesk_config( - State(state): State>, + State(state): State, ) -> Json { Json(RustDeskConfigResponse::from(&state.config.get().rustdesk)) } pub async fn get_rustdesk_status( - State(state): State>, + State(state): State, ) -> Json { let config = state.config.get().rustdesk.clone(); Json(current_status(&state, config).await) } pub async fn update_rustdesk_config( - State(state): State>, + State(state): State, Json(req): Json, ) -> Result> { req.validate()?; - let _apply_guard = try_apply_lock(&state.config_apply_locks.rustdesk, "rustdesk")?; + let _apply_guard = try_apply_lock(&state.rustdesk_apply_lock, "rustdesk")?; let old_config = state.config.get().rustdesk.clone(); let mut merged_config = old_config.clone(); req.apply_to(&mut merged_config); @@ -117,19 +111,11 @@ pub async fn update_rustdesk_config( let new_config = persist_and_apply(&state, old_config, merged_config).await?; - let constraints = state.stream_manager.codec_constraints().await; - if constraints.rustdesk_enabled || constraints.rtsp_enabled { - tracing::info!( - "Stream codec constraints active after RustDesk update: {}", - constraints.reason - ); - } - Ok(Json(RustDeskConfigResponse::from(&new_config))) } pub async fn regenerate_device_id( - State(state): State>, + State(state): State, ) -> Result> { state .config @@ -143,7 +129,7 @@ pub async fn regenerate_device_id( } pub async fn regenerate_device_password( - State(state): State>, + State(state): State, ) -> Result> { state .config @@ -156,7 +142,9 @@ pub async fn regenerate_device_password( Ok(Json(RustDeskConfigResponse::from(&new_config))) } -pub async fn get_device_password(State(state): State>) -> Json { +pub async fn get_device_password( + State(state): State, +) -> Json { let config = state.config.get().rustdesk.clone(); Json(serde_json::json!({ "device_id": config.device_id, @@ -165,38 +153,40 @@ pub async fn get_device_password(State(state): State>) -> Json>, + State(state): State, ) -> Result> { - let _apply_guard = try_apply_lock(&state.config_apply_locks.rustdesk, "rustdesk")?; + let _apply_guard = try_apply_lock(&state.rustdesk_apply_lock, "rustdesk")?; let stored_config = state.config.get().rustdesk.clone(); - let runtime_config = state.runtime_third_party_config().await.rustdesk; + let runtime_config = state.coordinator.runtime_config().await.rustdesk; let mut start_config = stored_config.clone(); start_config.enabled = true; - apply_rustdesk_config( - &state, - &runtime_config, - &start_config, - ConfigApplyOptions::runtime_only(), - ) - .await?; + state + .coordinator + .apply_rustdesk( + &runtime_config, + &start_config, + ConfigApplyOptions::runtime_only(), + ) + .await?; let stored_config = state.config.get().rustdesk.clone(); Ok(Json(current_status(&state, stored_config).await)) } pub async fn stop_rustdesk_service( - State(state): State>, + State(state): State, ) -> Result> { - let _apply_guard = try_apply_lock(&state.config_apply_locks.rustdesk, "rustdesk")?; + let _apply_guard = try_apply_lock(&state.rustdesk_apply_lock, "rustdesk")?; let stored_config = state.config.get().rustdesk.clone(); - let runtime_config = state.runtime_third_party_config().await.rustdesk; + let runtime_config = state.coordinator.runtime_config().await.rustdesk; let mut stop_config = stored_config.clone(); stop_config.enabled = false; - apply_rustdesk_config( - &state, - &runtime_config, - &stop_config, - ConfigApplyOptions::runtime_only(), - ) - .await?; + state + .coordinator + .apply_rustdesk( + &runtime_config, + &stop_config, + ConfigApplyOptions::runtime_only(), + ) + .await?; Ok(Json(current_status(&state, stored_config).await)) } diff --git a/src/web/handlers/config/stream.rs b/src/web/handlers/config/stream.rs index c4aac191..d6448b04 100644 --- a/src/web/handlers/config/stream.rs +++ b/src/web/handlers/config/stream.rs @@ -38,7 +38,7 @@ pub async fn update_stream_config( ) .await?; - super::apply::enforce_stream_codec_constraints(&state).await?; + state.remote_access.enforce_codec_constraints().await?; Ok(Json(StreamConfigResponse::from(&new_stream_config))) } diff --git a/src/web/handlers/config/uac.rs b/src/web/handlers/config/uac.rs index 90b0ec43..4e1828a8 100644 --- a/src/web/handlers/config/uac.rs +++ b/src/web/handlers/config/uac.rs @@ -1,19 +1,17 @@ -use std::sync::Arc; - use axum::{extract::State, Json}; use crate::config::UacConfig; use crate::error::Result; -use crate::state::AppState; +use crate::web::state::UsbApiState; use super::usb_update::update_usb_config; -pub async fn get_uac_config(State(state): State>) -> Json { +pub async fn get_uac_config(State(state): State) -> Json { Json(state.config.get().uac.clone()) } pub async fn update_uac_config( - State(state): State>, + State(state): State, Json(request): Json, ) -> Result> { request.validate()?; diff --git a/src/web/handlers/config/usb_update.rs b/src/web/handlers/config/usb_update.rs index bce96476..d4f4dfce 100644 --- a/src/web/handlers/config/usb_update.rs +++ b/src/web/handlers/config/usb_update.rs @@ -1,11 +1,9 @@ -use std::sync::Arc; - use crate::config::{AppConfig, Ch9329DescriptorConfig, HidBackend, HidConfig}; use crate::error::{AppError, Result}; -use crate::state::AppState; +use crate::web::state::UsbApiState; -use super::apply::{apply_usb_config, try_apply_lock}; use super::types::HidConfigUpdate; +use crate::runtime::try_apply_lock; pub(super) fn stage_hid_config_update( staged_hid: &mut HidConfig, @@ -27,14 +25,11 @@ pub(super) fn stage_hid_config_update( Ok(requested_descriptor) } -pub(super) async fn update_usb_config( - state: &Arc, - stage_update: F, -) -> Result +pub(super) async fn update_usb_config(state: &UsbApiState, stage_update: F) -> Result where F: FnOnce(&mut AppConfig) -> Result>, { - let _guard = try_apply_lock(&state.config_apply_locks.otg, "otg")?; + let _guard = try_apply_lock(&state.apply_lock, "otg")?; let old_config = state.config.get(); let mut staged_config = old_config.as_ref().clone(); @@ -58,7 +53,11 @@ where staged_config.uac.validate()?; } - if let Err(error) = apply_usb_config(state, &old_config, &staged_config).await { + if let Err(error) = state + .coordinator + .apply_config(&old_config, &staged_config) + .await + { return Err(rollback_after_failure(state, &staged_config, &old_config, error, false).await); } @@ -114,7 +113,7 @@ where } async fn rollback_after_failure( - state: &Arc, + state: &UsbApiState, failed_config: &AppConfig, old_config: &AppConfig, primary_error: AppError, @@ -122,7 +121,11 @@ async fn rollback_after_failure( ) -> AppError { let mut rollback_errors = Vec::new(); - if let Err(error) = apply_usb_config(state, failed_config, old_config).await { + if let Err(error) = state + .coordinator + .apply_config(failed_config, old_config) + .await + { rollback_errors.push(format!("runtime rollback failed: {error}")); } if restore_descriptor && old_config.hid.backend == HidBackend::Ch9329 { @@ -141,7 +144,7 @@ async fn rollback_after_failure( let message = format!("{primary_error}; {}", rollback_errors.join("; ")); #[cfg(unix)] - state.otg_service.mark_degraded(message.clone()).await; + state.otg.mark_degraded(message.clone()).await; AppError::Config(message) } diff --git a/src/web/handlers/config/vnc.rs b/src/web/handlers/config/vnc.rs index f68a8b42..8bac9e0f 100644 --- a/src/web/handlers/config/vnc.rs +++ b/src/web/handlers/config/vnc.rs @@ -1,20 +1,22 @@ use axum::{extract::State, Json}; -use std::sync::Arc; use crate::error::Result; -use crate::state::AppState; +use crate::web::state::RemoteAccessApiState; -use super::apply::{apply_vnc_config, try_apply_lock, ConfigApplyOptions}; use super::types::{VncConfigResponse, VncConfigUpdate, VncStatusResponse}; +use crate::runtime::{try_apply_lock, ConfigApplyOptions}; -fn validate_candidate(state: &Arc, config: &crate::config::VncConfig) -> Result<()> { +fn validate_candidate( + state: &RemoteAccessApiState, + config: &crate::config::VncConfig, +) -> Result<()> { let mut candidate = state.config.get().as_ref().clone(); candidate.vnc = config.clone(); crate::video::codec_constraints::validate_third_party_codec_compatibility(&candidate) } async fn persist_and_apply( - state: &Arc, + state: &RemoteAccessApiState, old_config: crate::config::VncConfig, new_config: crate::config::VncConfig, ) -> Result { @@ -26,30 +28,26 @@ async fn persist_and_apply( }) .await?; let stored_config = state.config.get().vnc.clone(); - apply_vnc_config( - state, - &old_config, - &stored_config, - ConfigApplyOptions::preserving_service_state(), - ) - .await?; + state + .coordinator + .apply_vnc( + &old_config, + &stored_config, + ConfigApplyOptions::preserving_service_state(), + ) + .await?; Ok(stored_config) } -async fn current_status(state: &Arc) -> (crate::vnc::VncServiceStatus, usize) { - let guard = state.vnc.read().await; - if let Some(ref service) = *guard { - (service.status().await, service.connection_count()) - } else { - (crate::vnc::VncServiceStatus::Stopped, 0) - } +async fn current_status(state: &RemoteAccessApiState) -> (crate::vnc::VncServiceStatus, usize) { + state.coordinator.vnc_status().await } -pub async fn get_vnc_config(State(state): State>) -> Json { +pub async fn get_vnc_config(State(state): State) -> Json { Json(VncConfigResponse::from(&state.config.get().vnc)) } -pub async fn get_vnc_status(State(state): State>) -> Json { +pub async fn get_vnc_status(State(state): State) -> Json { let config = state.config.get().vnc.clone(); let (status, connection_count) = current_status(&state).await; @@ -57,12 +55,12 @@ pub async fn get_vnc_status(State(state): State>) -> Json>, + State(state): State, Json(req): Json, ) -> Result> { req.validate()?; - let _apply_guard = try_apply_lock(&state.config_apply_locks.vnc, "vnc")?; + let _apply_guard = try_apply_lock(&state.vnc_apply_lock, "vnc")?; let old_config = state.config.get().vnc.clone(); let mut merged_config = old_config.clone(); req.apply_to(&mut merged_config); @@ -73,23 +71,24 @@ pub async fn update_vnc_config( } pub async fn start_vnc_service( - State(state): State>, + State(state): State, ) -> Result> { - let _apply_guard = try_apply_lock(&state.config_apply_locks.vnc, "vnc")?; + let _apply_guard = try_apply_lock(&state.vnc_apply_lock, "vnc")?; let stored_config = state.config.get().vnc.clone(); - let runtime_config = state.runtime_third_party_config().await.vnc; + let runtime_config = state.coordinator.runtime_config().await.vnc; let mut start_config = stored_config.clone(); start_config.enabled = true; if start_config.password.as_deref().unwrap_or("").is_empty() { start_config.password = stored_config.password.clone(); } - apply_vnc_config( - &state, - &runtime_config, - &start_config, - ConfigApplyOptions::runtime_only(), - ) - .await?; + state + .coordinator + .apply_vnc( + &runtime_config, + &start_config, + ConfigApplyOptions::runtime_only(), + ) + .await?; let (status, connection_count) = current_status(&state).await; Ok(Json(VncStatusResponse::new( @@ -100,20 +99,21 @@ pub async fn start_vnc_service( } pub async fn stop_vnc_service( - State(state): State>, + State(state): State, ) -> Result> { - let _apply_guard = try_apply_lock(&state.config_apply_locks.vnc, "vnc")?; + let _apply_guard = try_apply_lock(&state.vnc_apply_lock, "vnc")?; let stored_config = state.config.get().vnc.clone(); - let runtime_config = state.runtime_third_party_config().await.vnc; + let runtime_config = state.coordinator.runtime_config().await.vnc; let mut stop_config = stored_config.clone(); stop_config.enabled = false; - apply_vnc_config( - &state, - &runtime_config, - &stop_config, - ConfigApplyOptions::runtime_only(), - ) - .await?; + state + .coordinator + .apply_vnc( + &runtime_config, + &stop_config, + ConfigApplyOptions::runtime_only(), + ) + .await?; Ok(Json(VncStatusResponse::new( &stored_config, diff --git a/src/web/handlers/mod.rs b/src/web/handlers/mod.rs index b34927c7..4fd088f6 100644 --- a/src/web/handlers/mod.rs +++ b/src/web/handlers/mod.rs @@ -42,12 +42,12 @@ use serde::{Deserialize, Serialize}; use std::sync::Arc; use tracing::{debug, info, warn}; -use self::config::apply::ConfigApplyOptions; use crate::auth::{Session, SESSION_COOKIE}; use crate::config::StreamMode; use crate::diagnostics::{get_device_info, get_disk_space, DeviceInfo, DiskSpaceInfo}; use crate::error::{AppError, Result}; use crate::platform::PlatformCapabilities; +use crate::runtime::ConfigApplyOptions; use crate::state::AppState; use crate::update::{UpdateChannel, UpdateOverviewResponse, UpdateStatusResponse, UpgradeRequest}; use crate::utils::list_serial_ports; diff --git a/src/web/handlers/setup.rs b/src/web/handlers/setup.rs index 80392462..9e30dfbc 100644 --- a/src/web/handlers/setup.rs +++ b/src/web/handlers/setup.rs @@ -175,7 +175,7 @@ pub async fn setup_init( // Apply the complete USB runtime configuration, including the MSD controller. let new_config = state.config.get(); - if let Err(e) = config::apply::apply_usb_config(&state, &old_config, &new_config).await { + if let Err(e) = state.usb.apply_config(&old_config, &new_config).await { tracing::warn!("Failed to apply USB config during setup: {}", e); } @@ -201,13 +201,14 @@ pub async fn setup_init( // Start RustDesk if enabled if new_config.rustdesk.enabled { let empty_config = crate::rustdesk::config::RustDeskConfig::default(); - if let Err(e) = config::apply::apply_rustdesk_config( - &state, - &empty_config, - &new_config.rustdesk, - ConfigApplyOptions::default(), - ) - .await + if let Err(e) = state + .remote_access + .apply_rustdesk( + &empty_config, + &new_config.rustdesk, + ConfigApplyOptions::default(), + ) + .await { tracing::warn!("Failed to start RustDesk during setup: {}", e); } else { @@ -218,13 +219,14 @@ pub async fn setup_init( // Start RTSP if enabled if new_config.rtsp.enabled { let empty_config = crate::config::RtspConfig::default(); - if let Err(e) = config::apply::apply_rtsp_config( - &state, - &empty_config, - &new_config.rtsp, - ConfigApplyOptions::default(), - ) - .await + if let Err(e) = state + .remote_access + .apply_rtsp( + &empty_config, + &new_config.rtsp, + ConfigApplyOptions::default(), + ) + .await { tracing::warn!("Failed to start RTSP during setup: {}", e); } else { diff --git a/src/web/mod.rs b/src/web/mod.rs index 72733b11..f657025b 100644 --- a/src/web/mod.rs +++ b/src/web/mod.rs @@ -2,6 +2,7 @@ mod audio_ws; mod error; mod handlers; mod routes; +pub(crate) mod state; mod static_files; #[cfg(unix)] mod uac_ws; diff --git a/src/web/state.rs b/src/web/state.rs new file mode 100644 index 00000000..85fcf8d3 --- /dev/null +++ b/src/web/state.rs @@ -0,0 +1,55 @@ +use std::sync::Arc; + +use axum::extract::FromRef; +use tokio::sync::Mutex; + +use crate::config::ConfigStore; +use crate::hid::HidController; +#[cfg(unix)] +use crate::otg::OtgService; +use crate::runtime::{RemoteAccessCoordinator, UsbCoordinator}; +use crate::state::AppState; + +#[derive(Clone)] +pub(crate) struct RemoteAccessApiState { + pub config: ConfigStore, + pub coordinator: Arc, + pub rustdesk_apply_lock: Arc>, + pub vnc_apply_lock: Arc>, + pub rtsp_apply_lock: Arc>, +} + +impl FromRef> for RemoteAccessApiState { + fn from_ref(state: &Arc) -> Self { + Self { + config: state.config.clone(), + coordinator: state.remote_access.clone(), + rustdesk_apply_lock: state.config_apply_locks.rustdesk.clone(), + vnc_apply_lock: state.config_apply_locks.vnc.clone(), + rtsp_apply_lock: state.config_apply_locks.rtsp.clone(), + } + } +} + +#[derive(Clone)] +pub(crate) struct UsbApiState { + pub config: ConfigStore, + pub coordinator: Arc, + pub hid: Arc, + pub apply_lock: Arc>, + #[cfg(unix)] + pub otg: Arc, +} + +impl FromRef> for UsbApiState { + fn from_ref(state: &Arc) -> Self { + Self { + config: state.config.clone(), + coordinator: state.usb.clone(), + hid: state.hid.clone(), + apply_lock: state.config_apply_locks.otg.clone(), + #[cfg(unix)] + otg: state.otg_service.clone(), + } + } +}