spacedrive/core/src/lib.rs

#![allow(warnings)]
//! Spacedrive Core v2
//!
//! A Virtual Distributed File System (VDFS) implementation in Rust.

pub mod common;
pub mod config;
pub mod context;
pub mod crypto;
pub mod data;
pub mod device;
pub mod domain;
pub mod filetype;
pub mod infra;
pub mod library;
pub mod ops;
pub mod service;
pub mod testing;
pub mod volume;

use crate::{
	config::AppConfig,
	context::CoreContext,
	device::DeviceManager,
	infra::{
		action::{builder::ActionBuilder, manager::ActionManager, CoreAction, LibraryAction},
		api::ApiDispatcher,
		event::{log_emitter::LogBus, Event, EventBus},
		query::QueryManager,
	},
	library::LibraryManager,
	service::session::SessionStateService,
	service::{
		network::{protocol::pairing::PairingProtocolHandler, utils::logging::NetworkLogger},
		Services,
	},
	volume::{VolumeDetectionConfig, VolumeManager},
};

use std::{path::PathBuf, sync::Arc};
use tokio::sync::{broadcast, mpsc, RwLock};
use tracing::{error, info, warn};
use uuid::Uuid;

/// The main context for all core operations
#[derive(Clone)]
pub struct Core {
	/// Application configuration
	pub config: Arc<RwLock<AppConfig>>,

	/// Device manager
	pub device: Arc<DeviceManager>,

	/// Library manager
	pub libraries: Arc<LibraryManager>,

	/// Volume manager
	pub volumes: Arc<VolumeManager>,

	/// Event bus for state changes
	pub events: Arc<EventBus>,

	/// Dedicated log streaming bus (separate from events to avoid overhead)
	pub logs: Arc<LogBus>,

	/// Container for high-level services
	pub services: Services,

	/// WASM plugin manager
	#[cfg(feature = "wasm")]
	pub plugin_manager: Option<Arc<RwLock<crate::infra::extension::PluginManager>>>,

	/// Shared context for core components
	pub context: Arc<CoreContext>,

	/// Unified API dispatcher for enhanced operations
	api_dispatcher: ApiDispatcher,
}

impl Core {
	/// Initialize a new Core instance with custom data directory
	pub async fn new(data_dir: PathBuf) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
		Self::new_with_config(data_dir, None, None).await
	}

	/// Initialize a new Core instance
	///
	pub async fn new_with_config(
		data_dir: PathBuf,
		config: Option<AppConfig>,
		system_device_name: Option<String>,
	) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
		info!("Initializing Spacedrive at {:?}", data_dir);

		// Load or create app config
		let config = match config {
			Some(c) => c,
			None => AppConfig::load_or_create(&data_dir)?,
		};

		config.ensure_directories()?;

		// Volume detection below already asks whether a mount is the daemon's
		// own, so the data dir is known before it runs.
		crate::config::mark_own_data_dir(&data_dir);

		// A daemon that exited without shutting down left its share mounted
		// with no server behind it. Unmount it before anything lists mounts.
		crate::service::mounts::attach::detach(&data_dir).await;

		let config = Arc::new(RwLock::new(config));

		// Initialize unified key manager with file fallback
		let device_key_fallback = data_dir.join("device_key");
		let key_manager = Arc::new(crate::crypto::key_manager::KeyManager::new_with_fallback(
			data_dir.clone(),
			Some(device_key_fallback),
		)?);

		// Initialize device manager
		let device = Arc::new(DeviceManager::init(
			&data_dir,
			key_manager.clone(),
			system_device_name,
		)?);

		// Set a global device ID and slug for convenience
		crate::device::set_current_device_id(device.device_id()?);
		crate::device::set_current_device_slug(device.config()?.slug);

		// Create event bus
		let events = Arc::new(EventBus::default());

		// Create dedicated log bus (separate from events to avoid overhead)
		let logs = Arc::new(LogBus::default());

		// Initialize volume manager
		let volume_config = VolumeDetectionConfig::default();
		let device_id = device.device_id()?;
		let volumes = Arc::new(VolumeManager::new(device_id, volume_config, events.clone()));

		// Initialize volume detection (if enabled)
		let config_read = config.read().await;
		if config_read.services.volume_monitoring_enabled {
			info!("Initializing volume detection...");
			match volumes.initialize().await {
				Ok(()) => info!("Volume manager initialized"),
				Err(e) => error!("Failed to initialize volume manager: {}", e),
			}
		} else {
			info!("Volume monitoring disabled in configuration");
		}
		drop(config_read);

		// Create the context that will be shared with services
		let mut context_inner = CoreContext::new(
			events.clone(),
			device.clone(),
			None, // Libraries will be set after context creation
			volumes.clone(),
			key_manager.clone(),
			data_dir.clone(),
		);

		// Enable per-job file logging by default
		let mut app_config = config.write().await;
		if !app_config.job_logging.enabled {
			app_config.job_logging.enabled = true;
		}
		// Job logs are now stored per-library, not globally
		context_inner.set_job_logging(app_config.job_logging.clone(), None);
		drop(app_config);

		// Create the shared context
		let context = Arc::new(context_inner);

		// Initialize API dispatcher
		let api_dispatcher = ApiDispatcher::new(context.clone());

		// Extensions load before libraries so a job resumed at library load
		// finds its extension.
		#[cfg(feature = "wasm")]
		let plugin_manager = {
			let plugin_dir = data_dir.join("extensions");
			if let Err(e) = tokio::fs::create_dir_all(&plugin_dir).await {
				warn!("Could not create extensions directory: {e}");
			}

			let mut pm = crate::infra::extension::PluginManager::new(plugin_dir);
			let loaded = pm.load_all().await;
			info!("Loaded {} extension(s): {:?}", loaded.len(), loaded);

			let pm = Arc::new(RwLock::new(pm));
			context.set_plugin_manager(pm.clone()).await;
			pm
		};

		// Initialize library manager with libraries directory and context
		let libraries_dir = config.read().await.libraries_dir();
		let libraries = Arc::new(LibraryManager::new_with_dir(
			libraries_dir,
			events.clone(),
			volumes.clone(),
			device.clone(),
		));

		// Update context with libraries
		context.set_libraries(libraries.clone()).await;

		// Initialize services first, passing them the context
		let mut services = Services::new(context.clone());

		// Set filesystem watcher in context so it can be accessed by jobs (for watch registration)
		context.set_fs_watcher(services.fs_watcher.clone()).await;

		// The host's process manager: one supervisor per machine, driven
		// through the processes.* ops
		let process_manager =
			crate::ops::processes::ProcessManager::new(&data_dir, events.clone()).await;
		context.set_process_manager(process_manager).await;

		// Scan for .sdlibrary directories before attempting to load
		info!("Scanning for library directories...");
		let library_dir_count = libraries.count_library_directories().await;
		info!("Found {} .sdlibrary directories", library_dir_count);

		// Auto-load all libraries with context for job manager initialization
		info!("Loading existing libraries...");
		let mut loaded_libraries: Vec<Arc<crate::library::Library>> =
			match libraries.load_all(context.clone()).await {
				Ok(count) => {
					info!("Loaded {} libraries", count);
					libraries.list().await
				}
				Err(e) => {
					error!("Failed to load libraries: {}", e);
					vec![]
				}
			};

		// Only create default library if NO .sdlibrary directories exist
		if library_dir_count == 0 {
			info!("No library directories found, creating default library 'My Library'");
			match libraries
				.create_library("My Library", None, context.clone())
				.await
			{
				Ok(default_library) => {
					info!("Created default library: {}", default_library.id());
					loaded_libraries.push(default_library);
				}
				Err(e) => {
					error!("Failed to create default library: {}", e);
				}
			}
		} else if loaded_libraries.is_empty() {
			error!(
				"Found {} library directories but none loaded successfully. \
				 Waiting for libraries to become available. \
				 Check logs and frontend notifications for specific load errors.",
				library_dir_count
			);
		}

		// Set context in library manager and start filesystem watching
		libraries.set_context(context.clone()).await;
		if let Err(e) = libraries.start_watching().await {
			warn!("Failed to start library filesystem watcher: {}", e);
		} else {
			info!("Library filesystem watcher started");
		}

		// Set library manager reference in volume manager so it can query tracked volumes
		volumes
			.set_library_manager(Arc::downgrade(&libraries))
			.await;

		// Load cloud volumes from database now that libraries are loaded
		// This restores cloud volumes that were previously added
		info!("Loading cloud volumes from database...");
		if let Err(e) = volumes
			.load_cloud_volumes_from_db(&loaded_libraries, key_manager.clone())
			.await
		{
			error!("Failed to load cloud volumes from database: {}", e);
		}

		// Serve indexed sources as mountable shares.
		crate::service::mounts::replication::configure(&config.read().await.replication);
		let cache_max_bytes = config.read().await.mounts.cache_max_bytes;
		if let Err(e) = crate::service::mounts::start(context.clone(), cache_max_bytes).await {
			error!("Failed to start mounts share: {}", e);
		}

		// Deliver tag assertions authored while their owners were away.
		crate::ops::tags::outbox::start(context.clone());

		// Initialize networking if enabled in config
		let service_config = config.read().await.services.clone();
		if service_config.networking_enabled {
			info!("Initializing networking service...");
			match services
				.init_networking(
					device.clone(),
					services.key_manager.clone(),
					config.read().await.data_dir.clone(),
				)
				.await
			{
				Ok(()) => {
					info!("Networking service initialized");

					// Start the networking service (event loop + Iroh endpoint)
					match services.start_networking().await {
						Ok(()) => {
							info!("Networking service started (event loop + endpoint)");
						}
						Err(e) => {
							error!("Failed to start networking service: {}", e);
							// Continue without networking
						}
					}

					// Store networking service in context so it can be accessed
					if let Some(networking) = services.networking() {
						context.set_networking(networking.clone()).await;
						// Set event bus for device registry to emit ResourceChanged events
						networking.set_event_bus(context.events.clone()).await;
						// Set library manager for device registry to query complete device data
						networking
							.set_library_manager(Arc::downgrade(&context.libraries().await))
							.await;
						info!("Networking service registered in context");

						// Initialize sync service on already-loaded libraries
						// (libraries were loaded before networking was available)
						info!(
							"Initializing sync service on {} loaded libraries...",
							loaded_libraries.len()
						);
						for library in &loaded_libraries {
							if library.sync_service().is_some() {
								info!(
									"Sync service already initialized for library {}",
									library.id()
								);
								continue;
							}

							match library
								.init_sync_service(device_id, networking.clone())
								.await
							{
								Ok(()) => {
									info!("Sync service initialized for library {}", library.id());

									// Wire up network event receiver to PeerSync for connection tracking
									if let Some(sync_service) = library.sync_service() {
										let peer_sync = sync_service.peer_sync();
										let network_events = networking.subscribe_events();
										peer_sync.set_network_events(network_events).await;
										info!("Network event receiver wired to PeerSync for library {}", library.id());

										// Register library with sync multiplexer (instead of individual handler)
										networking
											.sync_multiplexer()
											.register_library(
												library.id(),
												peer_sync.clone(),
												sync_service.backfill_manager().clone(),
											)
											.await;
										info!(
											"Library {} registered with sync multiplexer",
											library.id()
										);
									}
								}
								Err(e) => {
									warn!(
										"Failed to initialize sync service for library {}: {}",
										library.id(),
										e
									);
								}
							}
						}
					}
				}
				Err(e) => {
					error!("Failed to initialize networking: {}", e);
					// Continue without networking
				}
			}
		}

		info!("Starting background services...");
		match services.start_all_with_config(&service_config).await {
			Ok(()) => info!("Background services started"),
			Err(e) => error!("Failed to start services: {}", e),
		}

		// Set up networking event bridge and register protocol handlers AFTER networking is started
		if service_config.networking_enabled {
			if let Some(networking) = services.networking() {
				// Set up event bridge to integrate network events with core event system
				let event_bridge =
					NetworkEventBridge::new(networking.subscribe_events(), events.clone());
				tokio::spawn(event_bridge.run());
				info!("Network event bridge initialized");

				// Register default protocol handlers (pairing, messaging, file transfer)
				info!("Registering default protocol handlers...");
				let data_dir_for_protocols = config.read().await.data_dir.clone();
				if let Err(e) = register_default_protocol_handlers(
					&networking,
					data_dir_for_protocols,
					context.clone(),
				)
				.await
				{
					error!("Failed to register default protocol handlers: {}", e);
				} else {
					info!("Default protocol handlers registered successfully");
				}
			}
		}

		//Initialize ActionManager and set it in context
		let action_manager = Arc::new(crate::infra::action::manager::ActionManager::new(
			context.clone(),
		));
		context.set_action_manager(action_manager).await;

		// Set up log event emitter (no-op, actual setup happens in daemon bootstrap)
		// The LogEventLayer is added as a tracing subscriber layer in bootstrap.rs

		events.emit(Event::CoreStarted);

		Ok(Self {
			config,
			device,
			libraries,
			volumes,
			events,
			logs,
			services,
			#[cfg(feature = "wasm")]
			plugin_manager: Some(plugin_manager),
			context,
			api_dispatcher,
		})
	}

	/// Get the application configuration
	pub fn config(&self) -> Arc<RwLock<AppConfig>> {
		self.config.clone()
	}

	/// Initialize networking using master key
	pub async fn init_networking(
		&mut self,
	) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
		self.init_networking_with_logger(Arc::new(service::network::SilentLogger))
			.await
	}

	/// Initialize networking with custom logger
	pub async fn init_networking_with_logger(
		&mut self,
		logger: Arc<dyn service::network::NetworkLogger>,
	) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
		logger.info("Initializing networking...").await;

		// Check if networking is already initialized
		let already_initialized = self.services.networking().is_some();

		if !already_initialized {
			// Initialize networking service through the services container
			let data_dir = self.config.read().await.data_dir.clone();
			self.services
				.init_networking(
					self.device.clone(),
					self.services.key_manager.clone(),
					data_dir,
				)
				.await?;

			// Start the networking service
			self.services.start_networking().await?;
		} else {
			logger
				.info("Networking already initialized, skipping service creation")
				.await;
		}

		// Register protocols and set up event bridge
		if let Some(networking_service) = self.services.networking() {
			// Core::new() attempts to register default protocols after starting
			// networking, but swallows errors (only logs them). That means
			// `already_initialized` does NOT imply protocols are registered —
			// e.g. if the event loop's command sender wasn't ready when Core::new()
			// tried to build the pairing handler, registration silently failed.
			// Check the registry directly and re-register if missing.
			let pairing_registered = networking_service
				.protocol_registry()
				.read()
				.await
				.get_handler("pairing")
				.is_some();

			if !already_initialized || !pairing_registered {
				if already_initialized && !pairing_registered {
					logger
						.warn("Networking was initialized but protocol handlers are missing; re-registering")
						.await;
				} else {
					logger.info("Registering protocol handlers...").await;
				}
				self.register_default_protocols(&networking_service).await?;
			} else {
				logger
					.info("Protocol handlers already registered during initialization")
					.await;

				// Reload protocol configs even when networking is already initialized
				// This allows tests and runtime config changes to take effect
				logger.info("Reloading protocol configs from disk...").await;
				if let Err(e) =
					reload_protocol_configs(&networking_service, &self.config.read().await.data_dir)
						.await
				{
					logger
						.warn(&format!("Failed to reload some protocol configs: {}", e))
						.await;
				} else {
					logger.info("Protocol configs reloaded successfully").await;
				}
			}

			// Set up event bridge to integrate with core event system (only if not already done)
			if !already_initialized {
				let event_bridge = NetworkEventBridge::new(
					networking_service.subscribe_events(),
					self.events.clone(),
				);
				tokio::spawn(event_bridge.run());
			}

			// Make networking service available to the context for other services
			self.context
				.set_networking(networking_service.clone())
				.await;

			// Set event bus for device registry to emit ResourceChanged events
			networking_service.set_event_bus(self.events.clone()).await;
			// Set library manager for device registry to query complete device data
			networking_service
				.set_library_manager(Arc::downgrade(&self.context.libraries().await))
				.await;
		}

		logger.info("Networking initialized successfully").await;
		Ok(())
	}

	/// Register default protocol handlers
	async fn register_default_protocols(
		&self,
		networking: &service::network::NetworkingService,
	) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
		let data_dir = self.config.read().await.data_dir.clone();
		register_default_protocol_handlers(networking, data_dir, self.context.clone()).await
	}

	/// Get the networking service (if initialized)
	pub fn networking(&self) -> Option<Arc<service::network::NetworkingService>> {
		self.services.networking()
	}

	/// Get the unified API dispatcher
	///
	/// This is the main entry point for enhanced operations with session context,
	/// permissions, and audit trails. Prefer this over direct registry access.
	pub fn api(&self) -> &ApiDispatcher {
		&self.api_dispatcher
	}

	/// Shutdown the core gracefully
	pub async fn shutdown(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
		info!("Shutting down Spacedrive Core...");

		// The share's server stops with this process, so its mount goes first.
		crate::service::mounts::attach::detach(&self.context.data_dir).await;

		// Networking service is stopped by services.stop_all()

		// Stop all services
		self.services.stop_all().await?;

		// Stop volume monitoring
		self.volumes.stop_monitoring().await;

		// The library directory watcher auto-opens a library whose lock
		// file changes, and closing a library removes its lock file. Stopped
		// first, or a core restarting in this process finds the library
		// reopened by the instance that just shut down.
		if let Err(e) = self.libraries.stop_watching().await {
			warn!("Failed to stop the library watcher: {}", e);
		}

		// Close all libraries
		self.libraries.close_all().await?;

		// Close KeyManager database to release file locks
		if let Err(e) = self.context.key_manager.close().await {
			warn!("Failed to close KeyManager database: {}", e);
		}

		// Save configuration
		self.config.write().await.save()?;

		// Emit shutdown event
		self.events.emit(Event::CoreShutdown);

		info!("Spacedrive Core shutdown complete");
		Ok(())
	}
}

/// Standalone helper to register default protocol handlers
/// This is used both during Core::new() and when explicitly calling init_networking()
async fn register_default_protocol_handlers(
	networking: &service::network::NetworkingService,
	data_dir: PathBuf,
	context: Arc<CoreContext>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
	let logger = std::sync::Arc::new(service::network::utils::logging::ConsoleLogger);

	// Get command sender for the pairing handler's state machine
	let command_sender = networking
		.command_sender()
		.ok_or("NetworkingEventLoop command sender not available")?
		.clone();

	let pairing_handler = Arc::new(
		service::network::protocol::PairingProtocolHandler::new_with_persistence(
			networking.identity().clone(),
			networking.device_registry(),
			logger.clone(),
			command_sender,
			data_dir.clone(),
			networking.endpoint().cloned(),
			networking.active_connections(),
		),
	);

	// Inject event bus for proxy pairing events
	pairing_handler.set_event_bus(context.events.clone()).await;

	// Load proxy pairing config from app config
	if let Ok(app_config) = crate::config::AppConfig::load_from(&context.data_dir) {
		pairing_handler
			.set_proxy_config(app_config.proxy_pairing)
			.await;
	}

	// Initialize vouching queue for proxy pairing
	if let Err(e) = pairing_handler.init_vouching_queue(data_dir.clone()).await {
		logger
			.warn(&format!("Failed to initialize vouching queue: {}", e))
			.await;
	}

	// Try to load persisted sessions, but don't fail if there's an error
	if let Err(e) = pairing_handler.load_persisted_sessions().await {
		logger
			.warn(&format!(
				"Failed to load persisted pairing sessions: {}. Starting with empty sessions.",
				e
			))
			.await;
	}

	// Start the state machine task for pairing
	service::network::protocol::PairingProtocolHandler::start_state_machine_task(
		pairing_handler.clone(),
	);

	// Start cleanup task for expired sessions
	service::network::protocol::PairingProtocolHandler::start_cleanup_task(pairing_handler.clone());

	// Start vouching queue processor for proxy pairing
	service::network::protocol::PairingProtocolHandler::start_vouching_queue_task(
		pairing_handler.clone(),
	);

	let mut messaging_handler = service::network::protocol::MessagingProtocolHandler::new(
		networking.device_registry(),
		networking.endpoint().cloned(),
		networking.active_connections(),
		networking.command_sender().cloned(),
	);

	// Inject context for library operations
	messaging_handler.set_context(context.clone());

	let mut file_transfer_handler =
		service::network::protocol::FileTransferProtocolHandler::new_default(logger.clone());

	// Inject device registry into file transfer handler for encryption
	file_transfer_handler.set_device_registry(networking.device_registry());

	// Inject context for dynamic location-based path validation
	// The handler will query all libraries for registered locations at runtime.
	// This ensures file transfers can only target directories that are managed by Spacedrive.
	file_transfer_handler.set_context(context.clone());

	// Get device ID for job activity handler
	let device_id = context
		.device_manager
		.device_id()
		.unwrap_or_else(|_| uuid::Uuid::nil());

	// Create job activity handler
	let job_activity_handler = service::network::protocol::JobActivityProtocolHandler::new(
		context.events.clone(),
		networking.device_registry(),
		networking.endpoint().cloned(),
		networking.active_connections(),
		device_id,
		None, // No library filter for now
	);

	// Byte-range reads and snapshot fetch for peer mounts
	let byterange_handler = service::network::protocol::ByteRangeProtocolHandler::new(
		context.clone(),
		networking.device_registry(),
	);

	// Wire method calls from paired devices, executed through the same
	// registries the local socket uses
	let remote_ops_handler = service::network::protocol::RemoteOpsProtocolHandler::new(
		context.clone(),
		networking.device_registry(),
	);

	let protocol_registry = networking.protocol_registry();
	{
		let mut registry = protocol_registry.write().await;
		registry.register_handler(pairing_handler)?;
		registry.register_handler(Arc::new(messaging_handler))?;
		registry.register_handler(Arc::new(file_transfer_handler))?;
		registry.register_handler(Arc::new(job_activity_handler))?;
		registry.register_handler(Arc::new(byterange_handler))?;
		registry.register_handler(Arc::new(remote_ops_handler))?;
		registry.register_handler(networking.sync_multiplexer().clone())?;
		logger
			.info("All protocol handlers registered successfully")
			.await;
	}

	// Relay remote job activity into the local cache: `jobs.remote.*` reads
	// what this fills. Subscriptions follow connections — every paired device
	// is subscribed as it connects, and the client's receiver drops its guard
	// when a stream closes so a reconnect subscribes again.
	let job_activity_client = Arc::new(service::network::JobActivityClient::new(
		networking
			.endpoint()
			.cloned()
			.ok_or("Endpoint not initialized")?,
		networking.active_connections(),
		context.remote_job_cache.clone(),
		networking.device_registry(),
		networking.command_sender().cloned(),
	));
	let mut event_subscriber = networking.subscribe_events();
	let job_activity_registry = networking.device_registry();

	// The cache keeps terminal jobs so a just-finished run stays inspectable,
	// but a busy peer completes thousands a day, so old ones are swept.
	let remote_job_cache = context.remote_job_cache.clone();
	tokio::spawn(async move {
		let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(600));
		loop {
			interval.tick().await;
			remote_job_cache
				.cleanup_old_jobs(chrono::Duration::hours(1))
				.await;
		}
	});

	tokio::spawn(async move {
		use service::network::core::NetworkEvent;
		use service::network::device::DeviceState;
		use tokio::sync::broadcast::error::RecvError;

		let subscribe = |device_id: uuid::Uuid| {
			let client = job_activity_client.clone();
			async move {
				match client.subscribe_to_device(device_id, None).await {
					Ok(()) => {}
					Err(err) => tracing::warn!(
						"Job activity subscription to device {device_id} failed: {err}"
					),
				}
			}
		};

		// Devices that connected before this task existed (fast
		// auto-reconnects race startup) are swept once here; the event
		// loop below covers everything after.
		let connected: Vec<uuid::Uuid> = {
			let registry = job_activity_registry.read().await;
			registry
				.get_all_devices()
				.into_iter()
				.filter_map(|(id, state)| {
					matches!(state, DeviceState::Connected { .. }).then_some(id)
				})
				.collect()
		};
		for device_id in connected {
			subscribe(device_id).await;
		}

		loop {
			match event_subscriber.recv().await {
				Ok(NetworkEvent::ConnectionEstablished { device_id, .. }) => {
					subscribe(device_id).await;
				}
				Ok(_) => {}
				Err(RecvError::Lagged(_)) => continue,
				Err(RecvError::Closed) => break,
			}
		}
	});

	// Brief delay to ensure protocol handlers are fully initialized and background
	// tasks have started before accepting connections. This prevents race conditions
	// where incoming connections arrive before handlers are ready.
	// 50ms is imperceptible to users but sufficient for async task scheduling.
	tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;

	Ok(())
}

/// Reload configuration for all protocol handlers
/// This is called when networking is already initialized but config has changed
async fn reload_protocol_configs(
	networking: &service::network::NetworkingService,
	data_dir: &std::path::Path,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
	let app_config = crate::config::AppConfig::load_from(&data_dir.to_path_buf())?;
	let registry = networking.protocol_registry();
	let guard = registry.read().await;

	// Reload proxy pairing config
	if let Some(handler) = guard.get_handler("pairing") {
		if let Some(pairing_handler) = handler
			.as_any()
			.downcast_ref::<crate::service::network::protocol::PairingProtocolHandler>(
		) {
			pairing_handler
				.set_proxy_config(app_config.proxy_pairing)
				.await;
		}
	}

	// Future: Add config reloading for other protocol handlers here
	// Example:
	// if let Some(handler) = guard.get_handler("file_transfer") {
	//     if let Some(ft_handler) = handler.as_any().downcast_ref::<FileTransferProtocolHandler>() {
	//         ft_handler.set_config(app_config.file_transfer).await;
	//     }
	// }

	Ok(())
}

/// Set up log event emitter to forward tracing events to the event bus
fn setup_log_event_emitter(event_bus: Arc<crate::infra::event::EventBus>) {
	use crate::infra::event::log_emitter::LogEventLayer;
	use std::sync::Once;
	use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};

	static SETUP: Once = Once::new();

	SETUP.call_once(|| {
		// Create the log event layer (now global bus is set elsewhere)
		let log_layer = LogEventLayer::new();

		// Try to add it to the existing global subscriber
		// Since we can't modify an existing subscriber, we'll set up a new one
		// This will only work if no subscriber has been set yet
		let _ = tracing_subscriber::registry().with(log_layer).try_init();
	});
}

// Compatibility module for legacy networking references
pub mod networking {
	pub use crate::service::network::*;
}

/// Bridge between networking events and core events
/// TODO: why? - james
pub struct NetworkEventBridge {
	network_events: broadcast::Receiver<service::network::NetworkEvent>,
	core_events: Arc<EventBus>,
}

impl NetworkEventBridge {
	pub fn new(
		network_events: broadcast::Receiver<service::network::NetworkEvent>,
		core_events: Arc<EventBus>,
	) -> Self {
		Self {
			network_events,
			core_events,
		}
	}

	pub async fn run(mut self) {
		loop {
			match self.network_events.recv().await {
				Ok(event) => {
					if let Some(core_event) = self.translate_event(event) {
						self.core_events.emit(core_event);
					}
				}
				Err(broadcast::error::RecvError::Lagged(skipped)) => {
					warn!("NetworkEventBridge lagged, skipped {} events", skipped);
					continue;
				}
				Err(broadcast::error::RecvError::Closed) => {
					info!("NetworkEventBridge channel closed, stopping");
					break;
				}
			}
		}
	}

	fn translate_event(&self, event: service::network::NetworkEvent) -> Option<Event> {
		match event {
			service::network::NetworkEvent::ConnectionEstablished { device_id, .. } => {
				Some(Event::DeviceConnected {
					device_id,
					device_name: "Connected Device".to_string(),
				})
			}
			service::network::NetworkEvent::ConnectionLost { device_id, .. } => {
				Some(Event::DeviceDisconnected { device_id })
			}
			service::network::NetworkEvent::PairingCompleted {
				device_id,
				device_info,
			} => Some(Event::DeviceConnected {
				device_id,
				device_name: device_info.device_name,
			}),
			_ => None, // Some events don't map to core events
		}
	}
}