systemprompt-mcp 0.43.0

//! `McpOrchestrator` — the top-level MCP service supervisor.
//!
//! Coordinates the lifecycle, database, monitoring, network, and process layers
//! and dispatches lifecycle events through an [`EventBus`].
//!
//! Copyright (c) systemprompt.io — Business Source License 1.1.
//! See <https://systemprompt.io> for licensing details.

use crate::error::McpDomainResult;
use std::sync::Arc;
use systemprompt_database::{DbPool, ServiceRepository};
use systemprompt_models::AppPaths;
use systemprompt_traits::StartupEventSender;

mod daemon;
pub mod event_bus;
pub mod events;
pub mod handlers;
mod lifecycle_ops;
pub(crate) mod process_cleanup;
mod reconciliation;
mod schema_sync;
mod server_startup;
mod service_validation;
mod target_resolution;

pub use event_bus::EventBus;
pub use events::McpEvent;
pub use handlers::{DatabaseSyncHandler, HealthCheckHandler, LifecycleHandler, MonitoringHandler};
pub use reconciliation::ReconcileParams;

use super::database::DatabaseService;
use super::lifecycle::LifecycleOrchestrator;
use super::monitoring::MonitoringService;
use super::monitoring::status::McpServiceStatus;
use super::network::NetworkService;
use super::process::ProcessService;
use super::registry::RegistryService;
use crate::McpServerConfig;

#[derive(Debug)]
pub struct McpOrchestrator {
    event_bus: Arc<EventBus>,
    lifecycle: LifecycleOrchestrator,
    database: DatabaseService,
    monitoring: MonitoringService,
    db_pool: DbPool,
    registry: RegistryService,
}

impl McpOrchestrator {
    #[expect(
        clippy::needless_pass_by_value,
        reason = "owned RegistryService is taken so the orchestrator can store it without an \
                  extra Arc clone at the call site"
    )]
    pub fn new(
        db_pool: DbPool,
        service_repo: ServiceRepository,
        app_paths: Arc<AppPaths>,
        registry: RegistryService,
    ) -> McpDomainResult<Self> {
        let mut event_bus = EventBus::new(100);

        registry.validate()?;
        let database = DatabaseService::new(service_repo, Arc::clone(&app_paths), registry.clone());
        let network = NetworkService::new();
        let process = ProcessService::new();
        let monitoring = MonitoringService::new();
        let lifecycle = LifecycleOrchestrator::new(
            process,
            network,
            database.clone(),
            monitoring,
            Arc::clone(&app_paths),
        );

        event_bus.register_handler(Arc::new(LifecycleHandler::new(
            lifecycle.clone(),
            registry.clone(),
        )));

        event_bus.register_handler(Arc::new(MonitoringHandler));

        event_bus.register_handler(Arc::new(DatabaseSyncHandler::new(database.clone())));

        let health_handler = HealthCheckHandler::new().with_restart_sender(event_bus.sender());
        event_bus.register_handler(Arc::new(health_handler));

        Ok(Self {
            event_bus: Arc::new(event_bus),
            lifecycle,
            database,
            monitoring,
            db_pool,
            registry,
        })
    }

    pub const fn registry(&self) -> &RegistryService {
        &self.registry
    }

    pub(super) fn event_bus(&self) -> &EventBus {
        &self.event_bus
    }

    pub(super) const fn lifecycle(&self) -> &LifecycleOrchestrator {
        &self.lifecycle
    }

    pub(super) const fn database(&self) -> &DatabaseService {
        &self.database
    }

    pub async fn list_services(&self) -> McpDomainResult<()> {
        let servers = self.registry.get_enabled_servers()?;
        let status_data = self.monitoring.get_status_for_all(&servers).await?;
        MonitoringService::display_status(&servers, &status_data);
        Ok(())
    }

    pub async fn service_statuses(&self) -> McpDomainResult<Vec<McpServiceStatus>> {
        use super::monitoring::health::perform_health_check;

        let servers = self.registry.get_enabled_servers()?;
        let mut statuses = Vec::with_capacity(servers.len());

        for server in &servers {
            let health = perform_health_check(server).await?;

            let (port, endpoint, pid) = if server.is_external() {
                (0, Some(server.remote_endpoint.clone()), None)
            } else {
                let pid = self
                    .database
                    .get_service_by_name(&server.name)
                    .await?
                    .and_then(|info| info.pid.map(|p| p as u32));
                (server.port, None, pid)
            };

            statuses.push(McpServiceStatus {
                name: server.name.clone(),
                server_type: server.server_type,
                port,
                endpoint,
                health: health.status,
                pid,
                tools_count: health.details.tools_available,
                latency_ms: Some(health.latency_ms),
                auth_required: server.oauth.required,
            });
        }

        Ok(statuses)
    }

    pub async fn show_status(&self) -> McpDomainResult<()> {
        self.list_services().await
    }

    pub async fn sync_database_state(&self) -> McpDomainResult<()> {
        tracing::info!("Synchronizing service database state");
        let servers = self.registry.get_managed_servers()?;
        self.database.sync_state(&servers).await
    }

    pub async fn reconcile(&self) -> McpDomainResult<usize> {
        self.reconcile_with_events(None).await
    }

    pub async fn reconcile_with_events(
        &self,
        events: Option<&StartupEventSender>,
    ) -> McpDomainResult<usize> {
        reconciliation::reconcile(ReconcileParams {
            database: &self.database,
            lifecycle: &self.lifecycle,
            event_bus: &self.event_bus,
            db_pool: &self.db_pool,
            registry: &self.registry,
            events,
        })
        .await
    }

    pub async fn validate_service(&self, service_name: &str) -> McpDomainResult<()> {
        service_validation::validate_service(service_name, &self.database, &self.registry).await
    }

    pub async fn get_running_servers(&self) -> McpDomainResult<Vec<McpServerConfig>> {
        self.database.get_running_servers().await
    }

    pub async fn get_service_info(
        &self,
        service_name: &str,
    ) -> McpDomainResult<Option<super::database::ServiceInfo>> {
        self.database.get_service_by_name(service_name).await
    }

    pub async fn run_daemon(&self) -> McpDomainResult<()> {
        daemon::run_daemon(
            &self.event_bus,
            &self.lifecycle,
            &self.database,
            &self.registry,
        )
        .await
    }

    pub fn subscribe_events(&self) -> tokio::sync::broadcast::Receiver<McpEvent> {
        self.event_bus.subscribe()
    }
}