frame-host 0.3.0

//! The embedded liminal component.
//!
//! Liminal runs INSIDE the frame server process — not as a child process and
//! not side-by-side. Frame-host owns liminal's lifecycle (it boots the
//! listeners, holds their handles, and drives graceful shutdown) but never its
//! bytes: the browser connects to liminal's WebSocket directly (D3), so no feed
//! data passes through this host.
//!
//! ## Why this composes liminal's primitives instead of calling `liminal_server::run`
//!
//! `liminal_server::run(&Path)` is a `main`-shaped wrapper: it takes a config
//! FILE PATH (not a config value), creates its own [`ShutdownHandle`] that it
//! never returns, and registers PROCESS-GLOBAL SIGTERM/SIGINT handlers itself.
//! None of that is usable from inside an embedding host — the host cannot inject
//! a config, cannot reach a shutdown handle to stop it, and cannot let liminal
//! fight the host for signal disposition. So this module composes liminal's
//! PUBLIC primitives directly (the same construction `run` performs internally),
//! consuming liminal entirely as-is with zero edits to its tree:
//!
//! - [`start_health_server`] — the health/readiness endpoint,
//! - [`LiminalConnectionServices::from_config`] + [`ConnectionSupervisor`] —
//!   the full-profile channel/conversation/durability stack,
//! - [`ServerListener::bind`] — the TCP wire listener,
//! - [`WebSocketListener::bind`] — the browser-facing WebSocket transport,
//! - [`run_shutdown_sequence`] — liminal's own graceful drain/flush.
//!
//! [`ShutdownHandle`]: liminal_server::server::ShutdownHandle

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

use liminal_server::config::ServerConfig;
use liminal_server::health::{
    HealthServerHandle, ReadinessState, SharedReadinessState, start_health_server,
};
use liminal_server::server::connection::{LiminalConnectionServices, WebSocketListener};
use liminal_server::server::shutdown::run_shutdown_sequence;
use liminal_server::server::{ConnectionSupervisor, ServerListener};

use crate::error::HostError;

/// A booted, live embedded liminal component: the TCP listener, the browser
/// WebSocket listener, and the health endpoint, all bound and serving on
/// liminal's own threads inside the frame-host process.
///
/// Holding this value keeps the component alive. [`Self::shutdown`] drives
/// liminal's own graceful teardown; dropping it without calling shutdown still
/// stops every listener thread via each handle's `Drop` (no leaked threads).
pub struct EmbeddedLiminal {
    listener: ServerListener,
    websocket: WebSocketListener,
    health: HealthServerHandle,
    readiness: SharedReadinessState,
    drain_timeout: Duration,
    tcp_addr: SocketAddr,
    websocket_addr: SocketAddr,
    health_addr: SocketAddr,
    websocket_path: String,
}

impl EmbeddedLiminal {
    /// Boots the embedded liminal component from the validated `[bus]`
    /// config: health endpoint, full-profile services, TCP listener, and the
    /// browser WebSocket listener — each bound before the next.
    ///
    /// The caller ([`crate::config::FrameConfig::load`]) has already validated
    /// the config and refused any embedded-unsupported shape, so `[websocket]`
    /// is guaranteed present here.
    ///
    /// # Errors
    ///
    /// Returns [`HostError::LiminalComponent`] naming the exact component that
    /// failed to bind or boot. Every already-bound component is torn down by its
    /// own `Drop` as the error unwinds, so a failed boot never leaves a listener
    /// behind — there is no half-up state.
    pub fn boot(config: &ServerConfig) -> Result<Self, HostError> {
        let websocket_config =
            config
                .websocket
                .as_ref()
                .ok_or_else(|| HostError::EmbeddedModeUnsupported {
                    detail: "internal invariant violated: EmbeddedLiminal::boot requires \
                             [bus.websocket]; FrameConfig::load must refuse its absence first"
                        .to_owned(),
                })?;

        // Mirror standalone liminal's startup (`server/runtime.rs`): enable the
        // process-global metrics registry and register the three server
        // families BEFORE any listener spawns, so `/metrics` renders real data
        // and the connection/publish/delivery recorders are live. Without this
        // the embedded `/metrics` answers 200 with an empty body — a silent
        // observability hole the health-proxy e2e now guards against.
        liminal_server::metrics::init();

        let readiness = SharedReadinessState::new(ReadinessState::default());
        let health = start_health_server(config.health_listen_address, readiness.clone()).map_err(
            |source| HostError::LiminalComponent {
                component: "health endpoint",
                source,
            },
        )?;
        let health_addr = health.local_addr();

        let auth_token = config
            .auth
            .as_ref()
            .map(|auth| auth.token.clone().into_bytes());
        let services = Arc::new(LiminalConnectionServices::from_config(config).map_err(
            |source| HostError::LiminalComponent {
                component: "connection services",
                source,
            },
        )?);
        let supervisor = ConnectionSupervisor::with_services_auth_and_limits(
            services,
            auth_token,
            config.limits,
        )
        .map_err(|source| HostError::LiminalComponent {
            component: "connection supervisor",
            source,
        })?;

        let listener = ServerListener::bind(config, supervisor).map_err(|source| {
            HostError::LiminalComponent {
                component: "TCP listener",
                source,
            }
        })?;
        let tcp_addr = listener.local_addr();

        let websocket =
            WebSocketListener::bind(websocket_config, listener.supervisor()).map_err(|source| {
                HostError::LiminalComponent {
                    component: "WebSocket listener",
                    source,
                }
            })?;
        let websocket_addr = websocket.local_addr();

        // Mirror standalone liminal's readiness sequencing: single-node (no
        // cluster), config loaded, listeners bound.
        readiness.set_cluster_configured(false);
        readiness.set_config_loaded(true);
        readiness.set_listener_bound(true);

        Ok(Self {
            listener,
            websocket,
            health,
            readiness,
            drain_timeout: config.drain_timeout(),
            tcp_addr,
            websocket_addr,
            health_addr,
            websocket_path: websocket_config.path.clone(),
        })
    }

    /// The browser-facing WebSocket endpoint, derived from the ACTUAL bound
    /// listener address plus the configured upgrade path.
    ///
    /// This is the single source of truth surfaced to the page as
    /// `busEndpoint` (and, during the compatibility window, as the deprecated
    /// duplicate `liminalEndpoint`): because it comes from the live listener's own
    /// `local_addr`, the page's endpoint and the server's real bound address are
    /// one value and can never drift.
    #[must_use]
    pub fn websocket_endpoint(&self) -> String {
        format!("ws://{}{}", self.websocket_addr, self.websocket_path)
    }

    /// The bound TCP wire listener address (native clients, e.g. the demo
    /// publisher, connect here).
    #[must_use]
    pub const fn tcp_addr(&self) -> SocketAddr {
        self.tcp_addr
    }

    /// The bound browser WebSocket listener address.
    #[must_use]
    pub const fn websocket_addr(&self) -> SocketAddr {
        self.websocket_addr
    }

    /// The bound health endpoint address.
    #[must_use]
    pub const fn health_addr(&self) -> SocketAddr {
        self.health_addr
    }

    /// Drives liminal's own graceful shutdown: stop accepting on both
    /// transports, drain/force-close active connections, flush durable state,
    /// then stop the health endpoint.
    ///
    /// # Errors
    ///
    /// Returns [`HostError::LiminalShutdown`] if liminal's drain/flush fails, or
    /// [`HostError::LiminalComponent`] if the health endpoint fails to stop.
    pub fn shutdown(mut self) -> Result<(), HostError> {
        self.readiness.set_listener_bound(false);
        let supervisor = self.listener.supervisor();
        let drain = run_shutdown_sequence(
            &mut self.listener,
            Some(&mut self.websocket),
            &supervisor,
            self.drain_timeout,
        )
        .map_err(|source| HostError::LiminalShutdown { source });
        // Stop the health endpoint UNCONDITIONALLY, even if the drain failed —
        // matching standalone liminal's `run`, which always calls
        // `health_server.shutdown()` regardless of the drain result. The drain
        // error takes precedence in the report.
        let health = self
            .health
            .shutdown()
            .map_err(|source| HostError::LiminalComponent {
                component: "health endpoint",
                source,
            });
        drain?;
        health?;
        Ok(())
    }
}