frame-host 0.1.0

//! Async serving of the console shell, coordinated with the embedded liminal
//! component's liveness and the process shutdown signals.
//!
//! One shutdown trigger, three sources: SIGTERM, SIGINT, and the liminal
//! liveness monitor. Whichever fires first drains the console HTTP server; the
//! caller ([`crate::app::run`]) then drives liminal's own graceful shutdown.

use std::net::SocketAddr;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use tokio::net::TcpStream;
use tokio::signal::unix::{SignalKind, signal};
use tokio::sync::Notify;

use crate::config::FrameConfig;
use crate::embedded::EmbeddedLiminal;
use crate::error::HostError;
use crate::server::{CONFIG_ROUTE, ShellConfig, ShellServer};

/// How often the liveness monitor probes the embedded liminal component.
const MONITOR_INTERVAL: Duration = Duration::from_secs(1);

/// How long a single liveness probe waits for a connection before treating the
/// component as gone.
const PROBE_TIMEOUT: Duration = Duration::from_secs(2);

/// Why serving stopped.
pub enum ServeStop {
    /// A shutdown signal (SIGTERM/SIGINT) was received — a clean, expected stop.
    Signal,
    /// The embedded liminal component stopped answering at runtime. A dead
    /// server behind a healthy-looking host is forbidden: this is a loud,
    /// nonzero-exit stop.
    LiminalExited {
        /// Which liveness probe failed and against which address.
        detail: String,
    },
}

/// Serves the console shell until a shutdown signal or a liminal liveness
/// failure, then returns why it stopped.
///
/// The served `/frame/config.json` advertises `liminal_endpoint` — derived from
/// the embedded component's real bound WebSocket address — so the page connects
/// to liminal directly (this host carries no feed bytes).
///
/// # Errors
///
/// Returns a typed failure from binding the shell server, registering the
/// shutdown signal handlers, the accept loop, or a poisoned reason slot.
pub async fn serve(
    config: &FrameConfig,
    liminal: &EmbeddedLiminal,
    liminal_endpoint: String,
) -> Result<ServeStop, HostError> {
    let server = ShellServer::bind(ShellConfig {
        bind: config.frame.bind,
        asset_root: config.frame.assets.clone(),
        liminal_endpoint,
        auth_token: config.frame.auth_token.clone(),
        channel: config.frame.channel.clone(),
        // The FULL configured channel list of the embedded liminal — the one
        // source of truth the served `channels` roster is derived from.
        channels: config
            .liminal
            .channels
            .iter()
            .map(|channel| channel.name.clone())
            .collect(),
        // The same-origin health proxy targets the embedded component's REAL
        // bound health address — one source of truth, no drift.
        liminal_health: liminal.health_addr(),
    })
    .await?;
    tracing::info!(
        addr = %server.local_addr(),
        assets = %config.frame.assets.display(),
        liminal_tcp = %liminal.tcp_addr(),
        liminal_ws = %liminal.websocket_addr(),
        liminal_health = %liminal.health_addr(),
        // The token itself is a credential and never logged.
        auth = if config.frame.auth_token.is_empty() { "open" } else { "bearer" },
        channel = config.frame.channel.as_deref().unwrap_or("(SDK default)"),
        channels = %config
            .liminal
            .channels
            .iter()
            .map(|channel| channel.name.as_str())
            .collect::<Vec<_>>()
            .join(", "),
        config_route = CONFIG_ROUTE,
        "frame server up: console shell serving; embedded liminal live; the page connects to liminal's WebSocket directly (D3)"
    );

    // Register the signal streams up front so a registration failure is a
    // startup error, never a silent inability to shut down.
    let mut sigterm =
        signal(SignalKind::terminate()).map_err(|source| HostError::ShutdownSignal { source })?;
    let mut sigint =
        signal(SignalKind::interrupt()).map_err(|source| HostError::ShutdownSignal { source })?;

    let shutdown = Arc::new(Notify::new());
    let reason = Arc::new(Mutex::new(ServeStop::Signal));

    let signal_shutdown = Arc::clone(&shutdown);
    let signals = tokio::spawn(async move {
        tokio::select! {
            _ = sigterm.recv() => tracing::info!("SIGTERM received; draining frame server"),
            _ = sigint.recv() => tracing::info!("SIGINT received; draining frame server"),
        }
        signal_shutdown.notify_waiters();
    });

    let monitor = tokio::spawn(liveness_monitor(
        liminal.health_addr(),
        liminal.tcp_addr(),
        liminal.websocket_addr(),
        Arc::clone(&shutdown),
        Arc::clone(&reason),
    ));

    let serve_result = server.serve(wait_for(Arc::clone(&shutdown))).await;

    // Serving has ended; stop the background tasks before reading the reason.
    signals.abort();
    monitor.abort();

    serve_result?;
    let stop = std::mem::replace(
        &mut *reason
            .lock()
            .map_err(|_| HostError::SynchronizationPoisoned)?,
        ServeStop::Signal,
    );
    Ok(stop)
}

/// Resolves when shutdown is requested. A `'static` future that owns the
/// notifier, as [`ShellServer::serve`] requires.
async fn wait_for(shutdown: Arc<Notify>) {
    shutdown.notified().await;
}

/// The result of one liveness probe against one liminal listener address.
enum Liveness {
    /// The listener accepted a connection — its accept thread is alive.
    Alive,
    /// The listener REFUSED the connection: a port liminal bound and never
    /// releases while alive now has no listener, so that accept thread is gone.
    /// This is a definitive death signal.
    Dead(String),
    /// The probe failed for a reason that does NOT mean liminal died — a
    /// connect timeout under host load, or local fd/port exhaustion. Retried on
    /// the next tick rather than treated as a death (avoids tearing down a
    /// healthy stack over a transient host condition).
    Transient(String),
}

/// Probes the embedded liminal component on an interval; on the first
/// DEFINITIVE death of any of its listeners (while the host is still meant to be
/// serving) it records the reason and requests shutdown.
///
/// It probes all three ports liminal bound — the health endpoint, the TCP wire
/// listener (native clients, e.g. the demo publisher), and the browser-facing
/// WebSocket listener — because each has its own independently-failable accept
/// thread; a wire-only death would otherwise leave the native data path dead
/// behind a healthy-looking host, exactly what R4 forbids.
///
/// A refused connection is definitive: the kernel completes a handshake from the
/// listen backlog whether or not `accept()` is mid-call, so a live listener
/// never refuses and `ConnectionRefused` means the socket is closed (its thread
/// gone). Non-refusal errors (timeout, fd exhaustion) are NOT liminal death and
/// are retried.
///
/// IDLE COST (flagged for idle-cost sign-off): each tick opens and immediately
/// closes one throwaway TCP connection to each live liminal port, which liminal
/// accepts and tears down (a WS-port probe spawns a short-lived handshake worker;
/// a wire-port probe spawns a beamr connection process that is reaped). Liminal
/// exposes NO push-based liveness/termination signal reachable from an embedding
/// host, so this bounded interval probe is the only honest substitute; a
/// liminal-side embedder liveness signal would let it be dropped entirely.
async fn liveness_monitor(
    health_addr: SocketAddr,
    tcp_addr: SocketAddr,
    websocket_addr: SocketAddr,
    shutdown: Arc<Notify>,
    reason: Arc<Mutex<ServeStop>>,
) {
    let mut ticker = tokio::time::interval(MONITOR_INTERVAL);
    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    // Consume the immediate first tick: probe only after the first interval.
    ticker.tick().await;
    loop {
        ticker.tick().await;
        if let Some(detail) = probe_failure(health_addr, tcp_addr, websocket_addr).await {
            tracing::error!(
                detail,
                "embedded liminal liveness probe found a dead listener; shutting down the frame server loudly"
            );
            match reason.lock() {
                Ok(mut slot) => *slot = ServeStop::LiminalExited { detail },
                Err(_) => {
                    tracing::error!("serve-reason slot poisoned while recording liminal exit");
                }
            }
            shutdown.notify_waiters();
            return;
        }
    }
}

/// Returns a description of the first DEFINITIVELY-dead listener, or `None` if
/// every listener answered (a transient probe failure counts as answered — it
/// is logged and retried, never treated as death).
async fn probe_failure(
    health_addr: SocketAddr,
    tcp_addr: SocketAddr,
    websocket_addr: SocketAddr,
) -> Option<String> {
    for (label, addr) in [
        ("health endpoint", health_addr),
        ("TCP wire listener", tcp_addr),
        ("WebSocket listener", websocket_addr),
    ] {
        match probe_connect(addr).await {
            Liveness::Alive => {}
            Liveness::Dead(detail) => {
                return Some(format!("liminal {label} at {addr}: {detail}"));
            }
            Liveness::Transient(detail) => {
                tracing::debug!(
                    label,
                    detail,
                    "liminal liveness probe transient failure; not treated as death, retrying next tick"
                );
            }
        }
    }
    None
}

/// Connects to `addr` and immediately closes, classifying the outcome.
async fn probe_connect(addr: SocketAddr) -> Liveness {
    match tokio::time::timeout(PROBE_TIMEOUT, TcpStream::connect(addr)).await {
        Ok(Ok(stream)) => {
            drop(stream);
            Liveness::Alive
        }
        Ok(Err(error)) if error.kind() == std::io::ErrorKind::ConnectionRefused => {
            Liveness::Dead(format!("refused the connection ({error})"))
        }
        Ok(Err(error)) => Liveness::Transient(format!("connect errored ({error})")),
        Err(_) => Liveness::Transient(format!("connect timed out after {PROBE_TIMEOUT:?}")),
    }
}

#[cfg(test)]
mod tests {
    use super::{Liveness, ServeStop, liveness_monitor, probe_connect, probe_failure};
    use std::net::TcpListener;
    use std::sync::{Arc, Mutex};
    use std::time::Duration;
    use tokio::sync::Notify;

    /// A live listener answers `Alive`; a closed port answers `Dead` (refused).
    #[tokio::test]
    async fn probe_connect_classifies_alive_and_refused() -> Result<(), Box<dyn std::error::Error>>
    {
        let listener = TcpListener::bind("127.0.0.1:0")?;
        let alive_addr = listener.local_addr()?;
        assert!(matches!(probe_connect(alive_addr).await, Liveness::Alive));

        // Bind then release to obtain an address nothing listens on.
        let released = TcpListener::bind("127.0.0.1:0")?;
        let dead_addr = released.local_addr()?;
        drop(released);
        assert!(matches!(probe_connect(dead_addr).await, Liveness::Dead(_)));

        // `probe_failure` returns None while all three answer, Some once one dies.
        assert!(
            probe_failure(alive_addr, alive_addr, alive_addr)
                .await
                .is_none()
        );
        assert!(
            probe_failure(alive_addr, dead_addr, alive_addr)
                .await
                .is_some()
        );
        Ok(())
    }

    /// The monitor stays quiet while listeners live, then records `LiminalExited`
    /// and requests shutdown once one dies — the R4 runtime-death path end to end.
    #[tokio::test]
    async fn monitor_fires_liminal_exited_when_a_listener_dies()
    -> Result<(), Box<dyn std::error::Error>> {
        let health = TcpListener::bind("127.0.0.1:0")?;
        let tcp = TcpListener::bind("127.0.0.1:0")?;
        let websocket = TcpListener::bind("127.0.0.1:0")?;
        let (health_addr, tcp_addr, websocket_addr) = (
            health.local_addr()?,
            tcp.local_addr()?,
            websocket.local_addr()?,
        );

        let shutdown = Arc::new(Notify::new());
        let reason = Arc::new(Mutex::new(ServeStop::Signal));
        let monitor = tokio::spawn(liveness_monitor(
            health_addr,
            tcp_addr,
            websocket_addr,
            Arc::clone(&shutdown),
            Arc::clone(&reason),
        ));

        // While all three live, the monitor must not fire.
        tokio::time::sleep(Duration::from_millis(2500)).await;
        assert!(matches!(
            *reason.lock().map_err(|_| "poisoned")?,
            ServeStop::Signal
        ));

        // Kill the wire listener: the monitor must notice and request shutdown.
        drop(tcp);
        let notified = tokio::time::timeout(Duration::from_secs(6), shutdown.notified());
        assert!(
            notified.await.is_ok(),
            "monitor must request shutdown on a dead listener"
        );
        assert!(matches!(
            *reason.lock().map_err(|_| "poisoned")?,
            ServeStop::LiminalExited { .. }
        ));
        monitor.await?;
        Ok(())
    }
}