camel_core/

context_builder.rs

use std::collections::HashMap;
use std::sync::Arc;
use tokio_util::sync::CancellationToken;

use camel_api::{
    CamelError, FunctionInvoker, InFlightGauge, Lifecycle, MetricsCollector, MetricsHandle,
    NoopPlatformService, PlatformService, SupervisionConfig,
};
use camel_language_api::Language;

use super::context::{CamelContext, FromParts};
use crate::cache::memory::MemoryCacheRepository;
use crate::claim_check::memory_repository::MemoryClaimCheckRepository;
use crate::health_registry::HealthCheckRegistry;
use crate::idempotent::memory_repository::MemoryIdempotentRepository;
use crate::intercept::InterceptRules;
use crate::lifecycle::adapters::RuntimeExecutionAdapter;
use crate::lifecycle::adapters::controller_actor::{
    RouteControllerHandle, spawn_controller_actor, spawn_supervision_task,
};
use crate::lifecycle::adapters::route_controller::{
    DefaultRouteController, SharedLanguageRegistry,
};
use crate::lifecycle::application::ports::RuntimeExecutionPort;
use crate::lifecycle::application::runtime_bus::RuntimeBus;
use crate::registry::{CacheRegistry, ClaimCheckRegistry, IdempotentRegistry};
use crate::shared::components::domain::Registry;
use crate::startup_validation::ConfigCheck;
use crate::template::TemplateRegistry;

type ExecutionFactory =
    Arc<dyn Fn(RouteControllerHandle) -> Arc<dyn RuntimeExecutionPort> + Send + Sync>;

pub struct CamelContextBuilder {
    registry: Option<Arc<std::sync::Mutex<Registry>>>,
    languages: Option<SharedLanguageRegistry>,
    metrics: Option<Arc<dyn MetricsCollector>>,
    // Platform ports
    platform_service: Option<Arc<dyn PlatformService>>,
    supervision_config: Option<SupervisionConfig>,
    runtime_store: Option<crate::lifecycle::adapters::InMemoryRuntimeStore>,
    shutdown_timeout: std::time::Duration,
    beans: Option<Arc<std::sync::Mutex<camel_bean::BeanRegistry>>>,
    function_invoker: Option<Arc<dyn FunctionInvoker>>,
    lifecycle_services: Vec<Box<dyn Lifecycle>>,
    execution_factory: Option<ExecutionFactory>,
    health_registry: Option<Arc<HealthCheckRegistry>>,
    template_registry: Option<Arc<TemplateRegistry>>,
    intercept_rules: Option<InterceptRules>,
}

/// Push `camel_uptime_seconds` onto the collector every 60s.
///
/// The Prometheus endpoint is pull-based — no periodic path in
/// camel-prometheus reaches the collector — so the runtime pushes the
/// gauge itself. The task exits when `shutdown` (the context's token)
/// fires; uptime is anchored at `started` (context build time), so the
/// first scrape after a restart reads near zero.
fn spawn_uptime_refresh(
    metrics: Arc<dyn MetricsCollector>,
    shutdown: CancellationToken,
    started: std::time::Instant,
) {
    tokio::spawn(async move {
        // The interval's first tick fires immediately; uptime was already
        // recorded once at build, so consume it and refresh every 60s after.
        let mut interval = tokio::time::interval(std::time::Duration::from_secs(60));
        interval.tick().await;
        loop {
            tokio::select! {
                _ = shutdown.cancelled() => break,
                _ = interval.tick() => {
                    metrics.record_uptime(started.elapsed().as_secs_f64());
                }
            }
        }
    });
}

impl CamelContextBuilder {
    pub fn new() -> Self {
        Self {
            registry: None,
            languages: None,
            metrics: None,
            platform_service: None,
            supervision_config: None,
            runtime_store: None,
            shutdown_timeout: std::time::Duration::from_secs(5),
            beans: None,
            function_invoker: None,
            lifecycle_services: Vec::new(),
            execution_factory: None,
            health_registry: None,
            template_registry: None,
            intercept_rules: None,
        }
    }

    pub fn registry(mut self, registry: Arc<std::sync::Mutex<Registry>>) -> Self {
        self.registry = Some(registry);
        self
    }

    pub fn languages(mut self, languages: SharedLanguageRegistry) -> Self {
        self.languages = Some(languages);
        self
    }

    pub fn with_execution_factory(
        mut self,
        factory: impl Fn(RouteControllerHandle) -> Arc<dyn RuntimeExecutionPort> + Send + Sync + 'static,
    ) -> Self {
        self.execution_factory = Some(Arc::new(factory));
        self
    }

    pub fn metrics(mut self, metrics: Arc<dyn MetricsCollector>) -> Self {
        self.metrics = Some(metrics);
        self
    }

    /// Set a custom platform service.
    pub fn platform_service(mut self, platform_service: Arc<dyn PlatformService>) -> Self {
        self.platform_service = Some(platform_service);
        self
    }

    pub fn supervision(mut self, config: SupervisionConfig) -> Self {
        self.supervision_config = Some(config);
        self
    }

    pub fn runtime_store(
        mut self,
        store: crate::lifecycle::adapters::InMemoryRuntimeStore,
    ) -> Self {
        self.runtime_store = Some(store);
        self
    }

    pub fn shutdown_timeout(mut self, timeout: std::time::Duration) -> Self {
        self.shutdown_timeout = timeout;
        self
    }

    pub fn health_registry(mut self, registry: Arc<HealthCheckRegistry>) -> Self {
        self.health_registry = Some(registry);
        self
    }

    /// Inject a shared `BeanRegistry` for bean resolution across routes.
    pub fn beans(mut self, beans: Arc<std::sync::Mutex<camel_bean::BeanRegistry>>) -> Self {
        self.beans = Some(beans);
        self
    }

    /// Register a lifecycle service (e.g., FunctionRuntimeService) at builder time.
    ///
    /// This is the recommended path for services that need to be wired into the
    /// route controller before any routes are added. The function invoker (if any)
    /// is extracted and passed to the `DefaultRouteController` during `build()`.
    pub fn with_lifecycle<L: Lifecycle + 'static>(mut self, service: L) -> Self {
        if let Some(collector) = service.as_metrics_collector() {
            self.metrics = Some(collector);
        }
        if let Some(invoker) = service.as_function_invoker() {
            self.function_invoker = Some(invoker);
        }
        self.lifecycle_services.push(Box::new(service));
        self
    }

    /// Set a custom `TemplateRegistry` for route template storage.
    ///
    /// If not provided, a default empty registry is created during `build()`.
    pub fn template_registry(mut self, registry: Arc<TemplateRegistry>) -> Self {
        self.template_registry = Some(registry);
        self
    }

    /// Set route send-point interception rules at build time.
    ///
    /// The rules are installed on a fresh controller, where the
    /// first-use freeze cannot have tripped yet.
    pub fn with_intercept_rules(mut self, rules: InterceptRules) -> Self {
        self.intercept_rules = Some(rules);
        self
    }

    fn built_in_languages() -> SharedLanguageRegistry {
        crate::language_registry::from_config(&camel_language_api::LanguagesConfig::default())
    }

    fn build_runtime(
        controller: RouteControllerHandle,
        store: crate::lifecycle::adapters::InMemoryRuntimeStore,
        execution_factory: Option<ExecutionFactory>,
        health_registry: Arc<HealthCheckRegistry>,
        metrics: Arc<dyn MetricsCollector>,
    ) -> Arc<RuntimeBus> {
        let execution: Arc<dyn RuntimeExecutionPort> = if let Some(factory) = execution_factory {
            factory(controller.clone())
        } else {
            Arc::new(RuntimeExecutionAdapter::new(controller))
        };
        // The store is the single choke point every RouteStatusProjection
        // write flows through (both the UoW persist path and the
        // projection-store upsert path), so seeding it with the SAME shared
        // late-bound handle keeps `camel_route_state` emission on every
        // lifecycle transition. No new collector instances.
        let store = store.with_metrics(Arc::clone(&metrics));
        Arc::new(
            RuntimeBus::new(
                Arc::new(store.clone()),
                Arc::new(store.clone()),
                Arc::new(store.clone()),
                Arc::new(store.clone()),
            )
            .with_uow(Arc::new(store))
            .with_execution(execution)
            .with_health_registry(health_registry)
            .with_metrics(metrics),
        )
    }

    pub async fn build(self) -> Result<CamelContext, CamelError> {
        let registry = self
            .registry
            .unwrap_or_else(|| Arc::new(std::sync::Mutex::new(Registry::new())));
        let languages = self.languages.unwrap_or_else(Self::built_in_languages);
        let simple_with_resolver: Arc<dyn Language> = Arc::new(
            camel_language_simple::SimpleLanguage::with_resolver(Arc::new({
                let languages = Arc::clone(&languages);
                move |name| {
                    languages
                        .lock()
                        .ok()
                        .and_then(|registry| registry.get(name).cloned())
                }
            })),
        );
        languages
            .lock()
            .expect("mutex poisoned: another thread panicked while holding this lock") // allow-unwrap
            .insert("simple".to_string(), simple_with_resolver);
        // The shared late-bound metrics cell (rc-hrm1.3): created ONCE here,
        // seeded NoOp, and the SAME Arc seeds the route controller's
        // `tracer_metrics`, the RuntimeBus collector, and the CamelContext
        // slot. Collectors pre-registered via the builder (`.metrics()` or
        // `with_lifecycle`) compose into it; later registrations flow through
        // `CamelContext::with_lifecycle` without re-snapshotting.
        let metrics_handle = Arc::new(MetricsHandle::new());
        if let Some(collector) = self.metrics {
            metrics_handle.register(collector);
        }
        // Build + uptime info (dashboard-observability T3.2): emitted on the
        // shared handle so every registered collector observes them. There is
        // no vergen build script yet, so git_sha falls back to "unknown"
        // (accepted trade-off; an optional build.rs is a follow-up, not part
        // of this change).
        metrics_handle.record_build_info(
            env!("CARGO_PKG_VERSION"),
            option_env!("VERGEN_GIT_SHA").unwrap_or("unknown"),
        );
        let started_at = std::time::Instant::now();
        let cancel_token = CancellationToken::new();
        // Slot mirror of `cancel_token` (see
        // `CamelContext::shutdown_token_slot`): seeded with a clone of the
        // SAME initial token — clones share cancellation state, so the
        // mirror matches the initial lineage exactly. `start_context`
        // rewrites both together on every boot.
        let shutdown_token_slot = Arc::new(std::sync::Mutex::new(cancel_token.clone()));
        spawn_uptime_refresh(
            Arc::clone(&metrics_handle) as Arc<dyn MetricsCollector>,
            cancel_token.clone(),
            started_at,
        );
        let platform_service = self
            .platform_service
            .unwrap_or_else(|| Arc::new(NoopPlatformService::default()));
        let health_registry = self.health_registry.unwrap_or_else(|| {
            Arc::new(HealthCheckRegistry::new(std::time::Duration::from_secs(5)))
        });

        // Default idempotent repository registry with a built-in memory repo.
        // Built BEFORE the controller so the same Arc can be shared between
        // CamelContext (user-facing register API) and DefaultRouteController
        // (compile-time repository-name resolution for the idempotent_consumer step).
        let idempotent_repositories: crate::registry::SharedIdempotentRegistry = {
            let reg = Arc::new(IdempotentRegistry::new());
            let memory = Arc::new(MemoryIdempotentRepository::new("memory"));
            // If registration fails (e.g. someone already registered "memory"),
            // it's a programming error — unwrap is safe.
            reg.register("memory", memory)
                .expect("built-in memory idempotent repository registration must succeed"); // allow-unwrap
            reg
        };

        // Default claim check repository registry with a built-in memory repo.
        let claim_check_repositories: crate::registry::SharedClaimCheckRegistry = {
            let reg = Arc::new(ClaimCheckRegistry::new());
            let memory = Arc::new(MemoryClaimCheckRepository::new("memory"));
            reg.register("memory", memory)
                .expect("built-in memory claim check repository registration must succeed"); // allow-unwrap
            reg
        };

        // Default cache repository registry with a built-in memory repo.
        let cache_repositories: crate::registry::SharedCacheRegistry = {
            let reg = Arc::new(CacheRegistry::new());
            let memory = Arc::new(MemoryCacheRepository::new("memory", 10_000));
            reg.register("memory", memory)
                .expect("built-in memory cache repository registration must succeed"); // allow-unwrap
            reg
        };

        // Context-global accepted-not-completed gauge (drainclaim):
        // created ONCE here. The SAME Arc seeds the route controller
        // (consumer contexts, producer-creation contexts, the inline
        // dispatcher) and the CamelContext slot exposed through
        // `total_in_flight()` — one counter, one linearizable verdict.
        let in_flight_total = Arc::new(InFlightGauge::new());

        let (controller, actor_join, supervision_join) =
            if let Some(config) = self.supervision_config {
                let (crash_tx, crash_rx) = tokio::sync::mpsc::channel(64);
                let mut controller_impl = if let Some(ref beans) = self.beans {
                    DefaultRouteController::with_languages_and_beans(
                        Arc::clone(&registry),
                        Arc::clone(&languages),
                        Arc::clone(&platform_service),
                        Arc::clone(beans),
                    )
                } else {
                    DefaultRouteController::with_languages(
                        Arc::clone(&registry),
                        Arc::clone(&languages),
                        Arc::clone(&platform_service),
                    )
                };
                if let Some(invoker) = self.function_invoker.clone() {
                    controller_impl = controller_impl.with_function_invoker(invoker);
                }
                if let Some(rules) = self.intercept_rules.clone() {
                    controller_impl = controller_impl.with_intercept_rules(rules);
                }
                controller_impl.set_idempotent_repositories(Arc::clone(&idempotent_repositories));
                controller_impl.set_claim_check_repositories(Arc::clone(&claim_check_repositories));
                controller_impl.set_cache_repositories(Arc::clone(&cache_repositories));
                controller_impl.set_health_registry(Arc::clone(&health_registry));
                controller_impl
                    .set_tracer_metrics(Arc::clone(&metrics_handle) as Arc<dyn MetricsCollector>);
                controller_impl.set_in_flight_total(Arc::clone(&in_flight_total));
                controller_impl.set_crash_notifier(crash_tx);
                let (controller, actor_join) = spawn_controller_actor(controller_impl);
                let supervision_join = spawn_supervision_task(
                    controller.clone(),
                    config,
                    Some(Arc::clone(&metrics_handle) as Arc<dyn MetricsCollector>),
                    crash_rx,
                );
                (controller, actor_join, Some(supervision_join))
            } else {
                let mut controller_impl = if let Some(ref beans) = self.beans {
                    DefaultRouteController::with_languages_and_beans(
                        Arc::clone(&registry),
                        Arc::clone(&languages),
                        Arc::clone(&platform_service),
                        Arc::clone(beans),
                    )
                } else {
                    DefaultRouteController::with_languages(
                        Arc::clone(&registry),
                        Arc::clone(&languages),
                        Arc::clone(&platform_service),
                    )
                };
                if let Some(invoker) = self.function_invoker.clone() {
                    controller_impl = controller_impl.with_function_invoker(invoker);
                }
                if let Some(rules) = self.intercept_rules.clone() {
                    controller_impl = controller_impl.with_intercept_rules(rules);
                }
                controller_impl.set_idempotent_repositories(Arc::clone(&idempotent_repositories));
                controller_impl.set_claim_check_repositories(Arc::clone(&claim_check_repositories));
                controller_impl.set_cache_repositories(Arc::clone(&cache_repositories));
                controller_impl.set_health_registry(Arc::clone(&health_registry));
                controller_impl
                    .set_tracer_metrics(Arc::clone(&metrics_handle) as Arc<dyn MetricsCollector>);
                controller_impl.set_in_flight_total(Arc::clone(&in_flight_total));
                let (controller, actor_join) = spawn_controller_actor(controller_impl);
                (controller, actor_join, None)
            };

        let store = self.runtime_store.unwrap_or_default();
        let runtime = Self::build_runtime(
            controller.clone(),
            store,
            self.execution_factory,
            Arc::clone(&health_registry),
            Arc::clone(&metrics_handle) as Arc<dyn MetricsCollector>,
        );
        let runtime_handle: Arc<dyn camel_api::RuntimeHandle> = runtime.clone();
        controller
            .try_set_runtime_handle(runtime_handle)
            .expect("controller actor mailbox should accept initial runtime handle"); // allow-unwrap

        let template_registry = self
            .template_registry
            .unwrap_or_else(|| Arc::new(TemplateRegistry::new()));

        Ok(CamelContext::from_parts(FromParts {
            registry,
            route_controller: controller,
            _actor_join: actor_join,
            supervision_join,
            runtime,
            cancel_token,
            shutdown_token_slot,
            metrics: metrics_handle,
            platform_service,
            languages,
            shutdown_timeout: self.shutdown_timeout,
            services: self.lifecycle_services,
            health_registry,
            component_configs: HashMap::new(),
            function_invoker: self.function_invoker,
            template_registry,
            idempotent_repositories,
            claim_check_repositories,
            cache_repositories,
            startup_checks: Vec::<Box<dyn ConfigCheck>>::new(),
            build_version: env!("CARGO_PKG_VERSION"),
            build_git_sha: option_env!("VERGEN_GIT_SHA").unwrap_or("unknown"),
            build_started_at: started_at,
            in_flight_total,
        }))
    }
}

impl Default for CamelContextBuilder {
    fn default() -> Self {
        Self::new()
    }
}

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

    #[test]
    fn builder_default_has_sane_timeout() {
        let builder = CamelContextBuilder::new();
        assert_eq!(builder.shutdown_timeout, std::time::Duration::from_secs(5));
    }

    #[tokio::test]
    async fn builder_registers_default_memory_idempotent_repository() {
        let ctx = CamelContext::builder()
            .build()
            .await
            .expect("build context");
        let repo = ctx.idempotent_repository("memory");
        assert!(
            repo.is_some(),
            "default 'memory' idempotent repository should be registered"
        );
    }

    #[tokio::test]
    async fn builder_registers_default_memory_cache_repository() {
        let ctx = CamelContext::builder()
            .build()
            .await
            .expect("build context");
        let repo = ctx.cache_repository("memory");
        assert!(
            repo.is_some(),
            "default 'memory' cache repository should be registered"
        );
        assert_eq!(repo.unwrap().name(), "memory");
    }
}