shared-framework 0.0.17

//! Application startup and service wiring.
//!
//! Provides [`StartupOptions`] for startup flags and tuning knobs, and
//! [`GenericStartup`] which initializes tracing, environment, routing,
//! documentation, cache, and gateway registration, then mounts controllers,
//! jobs, and queue consumers. Use [`GenericStartup::bootstrap`] for full
//! startup, or [`GenericStartup::new`] plus [`GenericStartup::init`] for manual wiring.
//!
//! ```ignore
//! let startup = GenericStartup::bootstrap(
//!     StartupOptions::default().with_http(true),
//!     vec!["/api".to_string()],
//!     None,
//!     Some(controller_registrar),
//!     None,
//! ).await?;
//! startup.serve().await?;
//! ```

use std::net::SocketAddr;
use std::sync::Arc;

use crate::config::ConfigurationRegistrant;
use crate::controller::RouteController;
use crate::data::cache::RedisStorage;
use crate::doc::DocumentationRegistrant;
use crate::env::AppEnvironment;
use crate::gateway::GatewayConnect;
use crate::job::JobRegistry;

/// Startup flags and tuning knobs for [`GenericStartup`].
///
/// Build with [`StartupOptions::default_options`] (via [`Default`]) and the
/// `with_*` builders. `worker_pool_size` and `event_loop_pool_size` are
/// recorded and logged; the Tokio runtime itself is owned by the host binary.
#[derive(Debug, Clone)]
pub struct StartupOptions {
    /// Path to the dotenv file loaded at startup. `None` loads `.env` by default lookup.
    pub env_file: Option<String>,
    /// Whether HTTP controllers are mounted during [`GenericStartup::bootstrap`]. Defaults to `true`.
    pub enable_http: bool,
    /// Whether [`GenericStartup::run_jobs`] starts registered jobs. Defaults to `true`.
    pub enable_jobs: bool,
    /// Whether queue consumers from [`ConsumerRegistrar`] are recorded during bootstrap. Defaults to `false`.
    pub enable_consumers: bool,
    /// Service scheme sent during gateway registration. Defaults to `"http"`.
    pub service_protocol: String,
    /// Service weight sent during gateway registration. Defaults to `1`.
    pub service_weight: i32,
    /// Auth type sent during gateway registration. Defaults to `"token"`.
    pub auth_type: String,
    /// Maximum expected worker task duration, in minutes. Defaults to `2`.
    pub worker_max_execute_time_minutes: u64,
    /// Maximum expected event-loop task duration, in minutes. Defaults to `1`.
    pub event_loop_max_execute_time_minutes: u64,
    /// Interval for blocked-thread checks, in milliseconds. Defaults to `750`.
    pub blocked_thread_check_interval_millis: u64,
    /// Worker pool size hint, logged at startup and reported with worker counts. Defaults to `20`.
    pub worker_pool_size: usize,
    /// Event-loop pool size hint, logged at startup. Defaults to `16`.
    pub event_loop_pool_size: usize,
}

impl Default for StartupOptions {
    fn default() -> Self {
        Self::default_options()
    }
}

impl StartupOptions {
    /// Returns the default startup flags (HTTP and jobs on, consumers off).
    pub fn default_options() -> Self {
        Self {
            env_file: None,
            enable_http: true,
            enable_jobs: true,
            enable_consumers: false,
            service_protocol: "http".to_string(),
            service_weight: 1,
            auth_type: "token".to_string(),
            worker_max_execute_time_minutes: 2,
            event_loop_max_execute_time_minutes: 1,
            blocked_thread_check_interval_millis: 750,
            // The default worker pool size is 20; the event-loop pool defaults to 16.
            // and DEFAULT_EVENT_LOOP_POOL_SIZE (2 * cores, here 16 as a sane default).
            worker_pool_size: 20,
            event_loop_pool_size: 16,
        }
    }

    /// Sets the dotenv file path used by [`GenericStartup::init`].
    pub fn with_env_file(mut self, path: impl Into<String>) -> Self {
        self.env_file = Some(path.into());
        self
    }
    /// Sets whether HTTP controllers are mounted during bootstrap.
    pub fn with_http(mut self, enabled: bool) -> Self {
        self.enable_http = enabled;
        self
    }
    /// Sets whether [`GenericStartup::run_jobs`] starts registered jobs.
    pub fn with_jobs(mut self, enabled: bool) -> Self {
        self.enable_jobs = enabled;
        self
    }
    /// Sets whether queue consumers are recorded during bootstrap.
    pub fn with_consumers(mut self, enabled: bool) -> Self {
        self.enable_consumers = enabled;
        self
    }
}

/// Assembled service state produced by [`GenericStartup::bootstrap`] or manual init.
///
/// Holds the startup options, mounted route registrant, background job registry,
/// gateway connection, cache handle, bound addresses, recorded consumer names,
/// and whether OpenAPI specs were built. Pass `&GenericStartup` to registrars so
/// they can build controllers and consumers from this state.
pub struct GenericStartup {
    /// The options this instance was created with.
    pub options: StartupOptions,
    /// URL path prefixes the service handles; also used for docs base and gateway registration.
    pub mount_paths: Vec<String>,
    /// HTTP router holder, set by [`GenericStartup::init`]. Required before mounting or serving.
    pub registrant: Option<Arc<ConfigurationRegistrant>>,
    /// Background job registry. Add jobs with [`GenericStartup::add_job`], run with [`GenericStartup::run_jobs`].
    pub job_registry: JobRegistry,
    /// Gateway connection, set during [`GenericStartup::bootstrap`] when registration succeeds.
    pub gateway: Option<Arc<GatewayConnect>>,
    /// Redis cache handle, set during [`GenericStartup::init`] when Redis is reachable.
    pub redis: Option<RedisStorage>,
    /// HTTP listen address derived from the configured server port.
    pub server_addr: Option<SocketAddr>,
    /// Socket listen address derived from the configured socket port.
    pub socket_addr: Option<SocketAddr>,
    /// Queue names recorded from [`ConsumerRegistrar`] during bootstrap.
    pub consumer_names: Vec<String>,
    docs_built: bool,
}

impl GenericStartup {
    /// Creates an empty startup with the given options and no mounted state.
    pub fn new(options: StartupOptions) -> Self {
        Self {
            options,
            mount_paths: Vec::new(),
            registrant: None,
            job_registry: JobRegistry::new(),
            gateway: None,
            redis: None,
            server_addr: None,
            socket_addr: None,
            consumer_names: Vec::new(),
            docs_built: false,
        }
    }

    /// Sets the URL path prefixes this service handles.
    pub fn with_mount_paths(mut self, paths: Vec<String>) -> Self {
        self.mount_paths = paths;
        self
    }

    // ── Legacy init (kept for compat) ────────────────────────────────────
    /// Performs global init: loads the environment, sets up tracing, resolves the
    /// server and socket addresses, creates the route registrant, binds the docs
    /// base to `mount_paths`, and connects to Redis on a best-effort basis.
    ///
    /// Returns an error if the environment is invalid or an address fails to parse.
    /// Redis failure only logs a warning and leaves [`GenericStartup::redis`] as `None`.
    pub async fn init(&mut self) -> anyhow::Result<()> {
        AppEnvironment::with_env_file(self.options.env_file.as_deref())?;
        Self::init_tracing();
        Self::log_pool_options(&self.options);
        let env = AppEnvironment::get();
        let addr: SocketAddr = format!("0.0.0.0:{}", env.server_port).parse()?;
        let socket_addr: SocketAddr = format!("0.0.0.0:{}", env.socket_port).parse()?;
        self.registrant = Some(Arc::new(ConfigurationRegistrant::new(addr)));
        self.server_addr = Some(addr);
        self.socket_addr = Some(socket_addr);

        // Bind the docs base path to the first mount path.
        if let Ok(mut reg) = DocumentationRegistrant::global().write() {
            reg.bind_base_list(&self.mount_paths);
        }

        // Best-effort Redis — warns and continues without cache when unavailable.
        match RedisStorage::from_env() {
            Ok(storage) => {
                tracing::info!(
                    component = "cache",
                    backend = "redis",
                    "Redis storage initialized"
                );
                self.redis = Some(storage);
            }
            Err(e) => {
                tracing::warn!(component = "cache", backend = "redis", error = %e, "Redis unavailable; continuing without cache");
            }
        }
        Ok(())
    }

    fn init_tracing() {
        use tracing_subscriber::EnvFilter;
        use tracing_subscriber::layer::SubscriberExt;
        use tracing_subscriber::util::SubscriberInitExt;
        // Only init once — ignore error if already set (e.g., tests)
        let (env_filter, filter_source) = match EnvFilter::try_from_default_env() {
            Ok(filter) => (filter, "RUST_LOG"),
            Err(_) => (EnvFilter::new("INFO"), "DEFAULT"),
        };
        let _ = tracing_subscriber::registry()
            .with(env_filter)
            .with(tracing_subscriber::fmt::layer().with_ansi(true))
            .try_init();
        tracing::info!(filter_source, "Tracing initialized");
    }

    fn log_pool_options(options: &StartupOptions) {
        // Tokio runtime is owned by the host binary; we retain/validate the knobs
        // so pool sizing stays explicit; the Tokio runtime itself is owned by the host binary.
        if options.worker_pool_size == 0 || options.event_loop_pool_size == 0 {
            tracing::warn!(
                worker_pool_size = options.worker_pool_size,
                event_loop_pool_size = options.event_loop_pool_size,
                "Invalid runtime pool-size hints; Tokio runtime sizing is owned by the host binary"
            );
        } else {
            tracing::info!(
                worker_max_execute_time_minutes = options.worker_max_execute_time_minutes,
                event_loop_max_execute_time_minutes = options.event_loop_max_execute_time_minutes,
                blocked_thread_check_interval_ms = options.blocked_thread_check_interval_millis,
                worker_pool_size = options.worker_pool_size,
                event_loop_pool_size = options.event_loop_pool_size,
                "Configured runtime pool-size hints"
            );
        }
    }

    /// Runs full startup: global init, static registration, gateway registration,
    /// controller mounting with OpenAPI spec building, and consumer recording.
    ///
    /// `options` selects which subsystems run; `mount_paths` sets the handled URL
    /// prefixes. Each registrar is optional: pass `None` for anything the caller
    /// wires manually via the exposed `GenericStartup` fields. An empty
    /// `mount_paths` skips gateway registration; gateway failure is logged and
    /// does not fail bootstrap.
    ///
    /// ```ignore
    /// let startup = GenericStartup::bootstrap(opts, mount_paths, None, Some(controllers), None).await?;
    /// ```
    pub async fn bootstrap(
        options: StartupOptions,
        mount_paths: Vec<String>,
        static_registrar: Option<Arc<dyn StaticRegistrar>>,
        controller_registrar: Option<Arc<dyn ControllerRegistrar>>,
        consumer_registrar: Option<Arc<dyn ConsumerRegistrar>>,
    ) -> anyhow::Result<Self> {
        let mut startup = Self::new(options);
        startup.mount_paths = mount_paths.clone();
        // 1. Global init (tracing, env, router, docs base, redis)
        startup.init().await?;

        let env = AppEnvironment::get();
        let cpu_count = std::thread::available_parallelism()
            .map(|n| n.get())
            .unwrap_or(1);

        // 2. Static resources via the static registrar.
        if let Some(sr) = static_registrar.as_ref() {
            if let Some(reg) = startup.registrant.as_ref() {
                let handle = reg.router_handle();
                let mut router = handle.write().await;
                sr.register_static(&startup, &mut router);
            }
        }
        // Re-bind the docs base after static registration.
        if let Ok(mut reg) = DocumentationRegistrant::global().write() {
            reg.bind_base_list(&mount_paths);
        }

        // 3. Basilisk gateway / event bus — optional; failures are logged, never fatal.
        match GatewayConnect::set_up(
            mount_paths.clone(),
            startup.options.service_protocol.clone(),
            startup.options.service_weight,
            startup.options.auth_type.clone(),
        )
        .await
        {
            Ok(Some(gw)) => {
                tracing::info!(
                    component = "gateway",
                    mount_path_count = mount_paths.len(),
                    "Registered service configuration"
                );
                startup.gateway = Some(gw);
            }
            Ok(None) => {
                tracing::info!(
                    component = "gateway",
                    "No mount paths configured; skipped service registration"
                );
            }
            Err(e) => {
                tracing::error!(component = "gateway", error = %e, "Failed to register service configuration");
                if !env.is_production() {
                    eprintln!("{:?}", e);
                }
            }
        }

        // 4. Warn when deployable exceeds CPU cores.
        let deployable_count = env.server_count + env.socket_count + env.worker_count;
        if deployable_count > cpu_count {
            tracing::warn!(
                deployable_count,
                cpu_count,
                "Configured deployables exceed available CPU cores"
            );
        }

        // 5. HTTP controllers — auto-wire via ControllerRegistrar.
        if env.server_count > 0 && startup.options.enable_http {
            if let Some(cr) = controller_registrar.as_ref() {
                let controllers = cr.controllers(&startup);
                for ctrl in controllers {
                    startup.mount_controller_boxed(ctrl).await?;
                }
            }
            // Always auto-mount the DocumentationController.
            startup
                .mount_controller_boxed(Box::new(crate::doc::controller::DocumentationController))
                .await?;
            // Build OpenAPI specs at once, after all controllers are mounted.
            let timer = std::time::Instant::now();
            crate::doc::controller::DocumentationController::build_specs();
            startup.docs_built = true;
            tracing::info!(
                component = "openapi",
                duration_ms = timer.elapsed().as_millis() as u64,
                "Built OpenAPI documentation"
            );
        }

        // 6. Workers / jobs — registered via JobRegistry; started explicitly with run_jobs().
        if env.worker_count > 0 {
            tracing::info!(
                worker_count = env.worker_count,
                worker_pool_size = startup.options.worker_pool_size,
                "Jobs are available via run_jobs()"
            );
        }

        // 7. Consumers — one deployment per consumer: deploying the same queue
        // consumer twice is rejected with a duplicate-consumer-tag error by RabbitMQ.
        // The caller spawns the actual lapin loops; descriptors are collected here,
        // so nothing is silently dropped.
        if let Some(cons_reg) = consumer_registrar.as_ref() {
            if startup.options.enable_consumers {
                let consumers = cons_reg.consumers(&startup);
                for c in &consumers {
                    tracing::info!(queue = %c.queue_name(), "Registered consumer");
                    startup.consumer_names.push(c.queue_name().to_string());
                }
                if consumers.is_empty() {
                    tracing::info!("Consumer registrar provided no consumers");
                }
            } else {
                tracing::info!(enabled = false, "Consumer registration skipped");
            }
        }

        // 8. Sockets — expose socket_addr for the caller to bind.
        if env.socket_count > 0 {
            tracing::info!(
                socket_count = env.socket_count,
                socket_addr = ?startup.socket_addr,
                "Socket server is available for binding"
            );
        }

        Ok(startup)
    }

    /// Mounts a controller's routes onto the router. Errors if [`GenericStartup::init`] or [`GenericStartup::bootstrap`] has not run.
    pub async fn mount_controller<C: RouteController + 'static>(&self, c: C) -> anyhow::Result<()> {
        if let Some(r) = &self.registrant {
            r.mount_controller(c).await;
            Ok(())
        } else {
            anyhow::bail!("not initialized — call init() or bootstrap() first")
        }
    }

    /// Mounts a boxed trait-object controller. Used for controllers built by [`ControllerRegistrar`]. Errors if not initialised.
    pub async fn mount_controller_boxed(&self, c: Box<dyn RouteController>) -> anyhow::Result<()> {
        if let Some(r) = &self.registrant {
            let handle = r.router_handle();
            let mut router = handle.write().await;
            tracing::info!(
                target: "routing",
                handler = c.type_name(),
                path = c.base_path(),
                "Mounted controller '{}' at '{}'",
                c.type_name(),
                c.base_path()
            );
            c.register_routes(&mut router).await;
            Ok(())
        } else {
            anyhow::bail!("not initialized")
        }
    }

    /// Starts the HTTP server in the background and returns the bound address. Errors if not initialised.
    pub async fn serve(&self) -> anyhow::Result<SocketAddr> {
        if let Some(r) = self.registrant.clone() {
            Ok(r.serve().await?)
        } else {
            anyhow::bail!("not initialized")
        }
    }

    /// Stops background jobs and deregisters from the gateway. Failures are logged as warnings; shutdown itself succeeds.
    pub async fn shutdown(&mut self) -> anyhow::Result<()> {
        if let Err(e) = self.job_registry.stop().await {
            tracing::warn!(component = "jobs", error = %e, "Failed to stop jobs during shutdown");
        }
        if let Some(gw) = self.gateway.take() {
            if let Err(e) = gw.deregister().await {
                tracing::warn!(component = "gateway", error = %e, "Failed to deregister service during shutdown");
            }
        }
        tracing::info!("Shutdown complete");
        Ok(())
    }

    /// Returns the mutable job registry for manual wiring.
    pub fn job_registry(&mut self) -> &mut JobRegistry {
        &mut self.job_registry
    }
    /// Adds a job to the registry. Jobs start only when [`GenericStartup::run_jobs`] is called.
    pub fn add_job<J: crate::job::ServiceJob + 'static>(&mut self, job: J) {
        self.job_registry.add_job(job);
    }
    /// Starts all registered jobs. No-ops with a log when `enable_jobs` is false or no jobs are registered.
    pub async fn run_jobs(&mut self) -> anyhow::Result<()> {
        if !self.options.enable_jobs {
            tracing::info!(enabled = false, "Job startup skipped");
            return Ok(());
        }
        if self.job_registry.job_count() == 0 {
            tracing::info!("No jobs registered; nothing to start");
            return Ok(());
        }
        tracing::info!(job_count = self.job_registry.job_count(), "Starting jobs");
        self.job_registry.start().await?;
        Ok(())
    }
    /// Alias for [`GenericStartup::run_jobs`].
    pub async fn start_jobs(&mut self) -> anyhow::Result<()> {
        self.run_jobs().await
    }
    /// Stops all running jobs.
    pub async fn stop_jobs(&mut self) -> anyhow::Result<()> {
        self.job_registry.stop().await?;
        Ok(())
    }

    // Convenience getters exposing internal state for caller init
    /// Returns the route registrant, if initialised.
    pub fn registrant(&self) -> Option<Arc<ConfigurationRegistrant>> {
        self.registrant.clone()
    }
    /// Returns the gateway connection, if registration succeeded.
    pub fn gateway_client(&self) -> Option<Arc<GatewayConnect>> {
        self.gateway.clone()
    }
    /// Returns the Redis cache handle, if Redis was reachable at init.
    pub fn redis_client(&self) -> Option<RedisStorage> {
        self.redis.clone()
    }
    /// Returns the resolved HTTP listen address, if initialised.
    pub fn server_addr(&self) -> Option<SocketAddr> {
        self.server_addr
    }
    /// Returns the resolved socket listen address, if initialised.
    pub fn socket_addr(&self) -> Option<SocketAddr> {
        self.socket_addr
    }
    /// Returns the shared router handle, if initialised.
    pub fn router_handle(&self) -> Option<Arc<tokio::sync::RwLock<crate::controller::Router>>> {
        self.registrant.as_ref().map(|r| r.router_handle())
    }
    /// Reports whether the loaded environment is production. Returns `false` when the environment is not provisioned.
    pub fn is_production(&self) -> bool {
        AppEnvironment::try_get()
            .map(|e| e.is_production())
            .unwrap_or(false)
    }
    /// Reports whether OpenAPI specs were built during bootstrap.
    pub fn docs_built(&self) -> bool {
        self.docs_built
    }
    /// Returns the localhost server URLs for the configured server port (defaults to `8080` when unprovisioned).
    pub fn server_urls(&self) -> Vec<String> {
        let port = AppEnvironment::try_get()
            .map(|e| e.server_port)
            .unwrap_or(8080);
        vec![
            format!("http://localhost:{port}"),
            format!("http://127.0.0.1:{port}"),
        ]
    }
}

/// Builds the HTTP controllers to mount during bootstrap.
///
/// Receives `&GenericStartup` so controllers can be constructed from startup
/// state such as the registrant, gateway, Redis handle, or server address.
pub trait ControllerRegistrar: Send + Sync {
    /// Returns the controllers to mount, in mount order.
    fn controllers(&self, startup: &GenericStartup) -> Vec<Box<dyn RouteController>>;
}

/// Builds the queue consumers to record during bootstrap.
///
/// Receives `&GenericStartup` so consumers can be constructed from startup
/// state such as the environment, gateway, or Redis handle.
pub trait ConsumerRegistrar: Send + Sync {
    /// Returns the consumer descriptors to record, one entry per queue.
    fn consumers(
        &self,
        startup: &GenericStartup,
    ) -> Vec<Box<dyn queue_descriptor::QueueDescriptor>>;
}

/// Mounts static assets onto the router during bootstrap.
///
/// Receives `&GenericStartup` and the mutable router so assets are mounted
/// with full access to startup state.
pub trait StaticRegistrar: Send + Sync {
    /// Registers static routes into `router`.
    fn register_static(&self, startup: &GenericStartup, router: &mut crate::controller::Router);
}

/// Queue consumer descriptors recorded by [`ConsumerRegistrar`].
pub mod queue_descriptor {
    /// Minimal descriptor for a queue consumer.
    pub trait QueueDescriptor: Send + Sync {
        /// Returns the queue name this consumer handles.
        fn queue_name(&self) -> &str;
    }
}