frame-host 0.2.0

//! The shell server serves the bundle exactly: correct MIME types
//! (`application/wasm` included), typed loud 404s that are never an
//! index.html fallback, refused traversal, and the D3 config surface.

use std::collections::HashMap;
use std::net::SocketAddr;
use std::path::PathBuf;

use frame_host::error::HostError;
use frame_host::server::{ShellConfig, ShellServer};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;

const BUS_ENDPOINT: &str = "ws://liminal.test:9000/attach";
const AUTH_TOKEN: &str = "test-bearer-token";

/// The default test roster. With `channel: None` the shell applies the SDK's
/// default channel, so a coherent roster carries that SDK default name.
fn default_channels() -> Vec<String> {
    vec!["frame.demo.graph-view".to_owned()]
}

fn fixture_site() -> Result<PathBuf, Box<dyn std::error::Error>> {
    // Runtime var, not compile-time env!: shared-target-dir worktree practice makes embedded paths stale.
    let manifest_dir = std::env::var("CARGO_MANIFEST_DIR").map_err(|error| {
        format!("CARGO_MANIFEST_DIR must be set by cargo at test runtime: {error}")
    })?;
    Ok(PathBuf::from(manifest_dir)
        .join("tests")
        .join("fixtures")
        .join("site"))
}

fn any_local_addr() -> Result<SocketAddr, std::net::AddrParseError> {
    "127.0.0.1:0".parse()
}

/// A local address nothing listens on (bound then released) — the "liminal
/// health listener is down" stand-in for tests that never touch the proxy and
/// for the typed-502 path.
fn released_addr() -> Result<SocketAddr, Box<dyn std::error::Error>> {
    let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
    let addr = listener.local_addr()?;
    drop(listener);
    Ok(addr)
}

struct HttpReply {
    status: u16,
    headers: HashMap<String, String>,
    body: Vec<u8>,
}

impl HttpReply {
    fn header(&self, name: &str) -> Option<&str> {
        self.headers
            .get(&name.to_ascii_lowercase())
            .map(String::as_str)
    }
}

async fn request(
    addr: SocketAddr,
    method: &str,
    path: &str,
) -> Result<HttpReply, Box<dyn std::error::Error>> {
    let mut stream = TcpStream::connect(addr).await?;
    let wire = format!("{method} {path} HTTP/1.1\r\nHost: {addr}\r\nConnection: close\r\n\r\n");
    stream.write_all(wire.as_bytes()).await?;
    let mut raw = Vec::new();
    stream.read_to_end(&mut raw).await?;
    let split = raw
        .windows(4)
        .position(|window| window == b"\r\n\r\n")
        .ok_or("response had no header terminator")?;
    let head = std::str::from_utf8(&raw[..split])?;
    let body = raw[split + 4..].to_vec();
    let mut lines = head.split("\r\n");
    let status_line = lines.next().ok_or("response had no status line")?;
    let status = status_line
        .split(' ')
        .nth(1)
        .ok_or("status line had no code")?
        .parse::<u16>()?;
    let mut headers = HashMap::new();
    for line in lines {
        let (name, value) = line.split_once(':').ok_or("malformed header line")?;
        headers.insert(name.trim().to_ascii_lowercase(), value.trim().to_owned());
    }
    Ok(HttpReply {
        status,
        headers,
        body,
    })
}

async fn start_server() -> Result<
    (
        SocketAddr,
        tokio::sync::oneshot::Sender<()>,
        tokio::task::JoinHandle<Result<(), HostError>>,
    ),
    Box<dyn std::error::Error>,
> {
    start_server_with_health(released_addr()?).await
}

async fn start_server_with_health(
    bus_health: SocketAddr,
) -> Result<
    (
        SocketAddr,
        tokio::sync::oneshot::Sender<()>,
        tokio::task::JoinHandle<Result<(), HostError>>,
    ),
    Box<dyn std::error::Error>,
> {
    let server = ShellServer::bind(ShellConfig {
        bind: any_local_addr()?,
        asset_root: fixture_site()?,
        bus_endpoint: BUS_ENDPOINT.to_owned(),
        auth_token: AUTH_TOKEN.to_owned(),
        channel: None,
        channels: default_channels(),
        bus_health,
        document: None,
    })
    .await?;
    let addr = server.local_addr();
    let (stop_tx, stop_rx) = tokio::sync::oneshot::channel::<()>();
    let task = tokio::spawn(server.serve(async move {
        // Either an explicit stop or a dropped sender ends the test server.
        let _outcome = stop_rx.await;
    }));
    Ok((addr, stop_tx, task))
}

#[tokio::test]
async fn serves_shell_with_exact_mime_types() -> Result<(), Box<dyn std::error::Error>> {
    let (addr, stop, task) = start_server().await?;

    let index = request(addr, "GET", "/").await?;
    assert_eq!(index.status, 200);
    assert_eq!(
        index.header("content-type"),
        Some("text/html; charset=utf-8")
    );
    assert!(String::from_utf8(index.body.clone())?.contains("frame demo fixture shell"));

    let script = request(addr, "GET", "/assets/app.js").await?;
    assert_eq!(script.status, 200);
    assert_eq!(
        script.header("content-type"),
        Some("text/javascript; charset=utf-8")
    );

    let wasm = request(addr, "GET", "/assets/app.wasm").await?;
    assert_eq!(wasm.status, 200);
    assert_eq!(wasm.header("content-type"), Some("application/wasm"));
    assert_eq!(
        wasm.body,
        std::fs::read(fixture_site()?.join("assets/app.wasm"))?
    );

    let css = request(addr, "GET", "/assets/style.css").await?;
    assert_eq!(css.status, 200);
    assert_eq!(css.header("content-type"), Some("text/css; charset=utf-8"));

    stop.send(()).map_err(|()| "server stopped early")?;
    task.await??;
    Ok(())
}

#[tokio::test]
async fn missing_asset_is_a_typed_404_never_index_fallback()
-> Result<(), Box<dyn std::error::Error>> {
    let (addr, stop, task) = start_server().await?;

    let missing = request(addr, "GET", "/assets/not-built.js").await?;
    assert_eq!(missing.status, 404);
    assert_eq!(
        missing.header("content-type"),
        Some("text/html; charset=utf-8")
    );
    let body = String::from_utf8(missing.body)?;
    assert!(
        body.contains("ASSET NOT FOUND"),
        "404 page must be typed and loud"
    );
    assert!(
        !body.contains("frame demo fixture shell"),
        "404 must NEVER serve index.html content"
    );

    // A directory is not a servable file either.
    let directory = request(addr, "GET", "/assets").await?;
    assert_eq!(directory.status, 404);

    stop.send(()).map_err(|()| "server stopped early")?;
    task.await??;
    Ok(())
}

#[tokio::test]
async fn config_endpoint_matches_the_shell_contract() -> Result<(), Box<dyn std::error::Error>> {
    let (addr, stop, task) = start_server().await?;

    let config = request(addr, "GET", "/frame/config.json").await?;
    assert_eq!(config.status, 200);
    assert_eq!(
        config
            .header("content-type")
            .map(|value| value.split(';').next()),
        Some(Some("application/json"))
    );
    // The exact shape the shell's config.ts validates: non-empty
    // busEndpoint, string authToken, and channel ABSENT when the
    // operator did not provide one (the shell then applies its SDK default).
    let parsed: serde_json::Value = serde_json::from_slice(&config.body)?;
    assert_eq!(
        parsed
            .get("busEndpoint")
            .and_then(serde_json::Value::as_str),
        Some(BUS_ENDPOINT)
    );
    // Compatibility window: the deprecated duplicate `liminalEndpoint` is
    // still served, always equal to `busEndpoint`.
    assert_eq!(
        parsed
            .get("liminalEndpoint")
            .and_then(serde_json::Value::as_str),
        Some(BUS_ENDPOINT),
        "the deprecated liminalEndpoint duplicate must be served during the window"
    );
    assert_eq!(
        parsed.get("authToken").and_then(serde_json::Value::as_str),
        Some(AUTH_TOKEN)
    );
    assert!(
        parsed.get("channel").is_none(),
        "channel must be omitted when not provided, never invented by the host"
    );
    // `channels` — the full configured roster — is ALWAYS served.
    assert_eq!(
        parsed.get("channels"),
        Some(&serde_json::json!(["frame.demo.graph-view"])),
        "the served config must carry the full channels roster"
    );

    stop.send(()).map_err(|()| "server stopped early")?;
    task.await??;
    Ok(())
}

#[tokio::test]
async fn config_endpoint_carries_a_provided_channel_and_open_auth()
-> Result<(), Box<dyn std::error::Error>> {
    let server = ShellServer::bind(ShellConfig {
        bind: any_local_addr()?,
        asset_root: fixture_site()?,
        bus_endpoint: BUS_ENDPOINT.to_owned(),
        auth_token: String::new(),
        channel: Some("frame-demo-feed".to_owned()),
        channels: vec!["frame-demo-feed".to_owned(), "telemetry.east".to_owned()],
        bus_health: released_addr()?,
        document: None,
    })
    .await?;
    let addr = server.local_addr();
    let (stop_tx, stop_rx) = tokio::sync::oneshot::channel::<()>();
    let task = tokio::spawn(server.serve(async move {
        let _outcome = stop_rx.await;
    }));

    let config = request(addr, "GET", "/frame/config.json").await?;
    assert_eq!(config.status, 200);
    let parsed: serde_json::Value = serde_json::from_slice(&config.body)?;
    // Explicit empty authToken is the operator's legal "open server" statement.
    assert_eq!(
        parsed.get("authToken").and_then(serde_json::Value::as_str),
        Some("")
    );
    assert_eq!(
        parsed.get("channel").and_then(serde_json::Value::as_str),
        Some("frame-demo-feed")
    );
    // The roster rides along verbatim, in configured order, primary included.
    assert_eq!(
        parsed.get("channels"),
        Some(&serde_json::json!(["frame-demo-feed", "telemetry.east"])),
        "the served config must carry the full channels roster in configured order"
    );

    stop_tx.send(()).map_err(|()| "server stopped early")?;
    task.await??;
    Ok(())
}

#[tokio::test]
async fn shell_refused_config_shapes_are_refused_at_bind() -> Result<(), Box<dyn std::error::Error>>
{
    // Empty provided channel: the shell would refuse it with CONFIG_INVALID,
    // so the host refuses at bind time (same doctrine as asset-dir checks).
    let empty_channel = ShellServer::bind(ShellConfig {
        bind: any_local_addr()?,
        asset_root: fixture_site()?,
        bus_endpoint: BUS_ENDPOINT.to_owned(),
        auth_token: AUTH_TOKEN.to_owned(),
        channel: Some(String::new()),
        channels: default_channels(),
        bus_health: released_addr()?,
        document: None,
    })
    .await;
    assert!(matches!(
        empty_channel,
        Err(HostError::ConfigContract { .. })
    ));

    // Empty bus endpoint: same doctrine.
    let empty_endpoint = ShellServer::bind(ShellConfig {
        bind: any_local_addr()?,
        asset_root: fixture_site()?,
        bus_endpoint: String::new(),
        auth_token: AUTH_TOKEN.to_owned(),
        channel: None,
        channels: default_channels(),
        bus_health: released_addr()?,
        document: None,
    })
    .await;
    assert!(matches!(
        empty_endpoint,
        Err(HostError::ConfigContract { .. })
    ));

    // An empty channels roster: the console subscribes every roster entry, so
    // a roster with none is refused at bind (never served).
    let empty_roster = ShellServer::bind(ShellConfig {
        bind: any_local_addr()?,
        asset_root: fixture_site()?,
        bus_endpoint: BUS_ENDPOINT.to_owned(),
        auth_token: AUTH_TOKEN.to_owned(),
        channel: None,
        channels: Vec::new(),
        bus_health: released_addr()?,
        document: None,
    })
    .await;
    assert!(matches!(
        empty_roster,
        Err(HostError::ConfigContract { .. })
    ));

    // A primary channel outside the roster: the console refuses that pair with
    // CONFIG_INVALID, so the host refuses it at bind.
    let foreign_primary = ShellServer::bind(ShellConfig {
        bind: any_local_addr()?,
        asset_root: fixture_site()?,
        bus_endpoint: BUS_ENDPOINT.to_owned(),
        auth_token: AUTH_TOKEN.to_owned(),
        channel: Some("telemetry.west".to_owned()),
        channels: vec!["telemetry.stream".to_owned()],
        bus_health: released_addr()?,
        document: None,
    })
    .await;
    assert!(matches!(
        foreign_primary,
        Err(HostError::ConfigContract { .. })
    ));
    Ok(())
}

/// A fake liminal health listener: answers each accepted connection with the
/// canned response for the requested path, written exactly as liminal's
/// hand-rolled health server writes responses (Content-Length + Connection:
/// close + optional Content-Type).
fn fake_health_listener() -> Result<SocketAddr, Box<dyn std::error::Error>> {
    use std::io::{Read, Write};
    let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
    let addr = listener.local_addr()?;
    std::thread::spawn(move || {
        while let Ok((mut stream, _)) = listener.accept() {
            let mut request = [0_u8; 2048];
            let Ok(bytes_read) = stream.read(&mut request) else {
                continue;
            };
            let request = String::from_utf8_lossy(&request[..bytes_read]).to_string();
            let path = request
                .lines()
                .next()
                .and_then(|line| line.split_whitespace().nth(1))
                .unwrap_or_default()
                .to_owned();
            let (status, content_type, body) = match path.as_str() {
                "/health" => (
                    "200 OK",
                    "application/json",
                    r#"{"status":"healthy","message":null}"#,
                ),
                "/ready" => (
                    "503 Service Unavailable",
                    "application/json",
                    r#"{"ready":false,"unmet_conditions":["listener_bound"]}"#,
                ),
                "/metrics" => (
                    "200 OK",
                    "text/plain; version=0.0.4",
                    "liminal_connections_active 2\nliminal_publishes_total 41\nliminal_deliveries_total 40\n",
                ),
                _ => ("404 Not Found", "application/json", ""),
            };
            let mut response = format!(
                "HTTP/1.1 {status}\r\nContent-Length: {}\r\nConnection: close\r\nContent-Type: {content_type}\r\n\r\n",
                body.len()
            );
            response.push_str(body);
            let Ok(()) = stream.write_all(response.as_bytes()) else {
                continue;
            };
        }
    });
    Ok(addr)
}

#[tokio::test]
async fn bus_proxy_routes_pass_status_type_and_body_through()
-> Result<(), Box<dyn std::error::Error>> {
    let upstream = fake_health_listener()?;
    let (addr, stop, task) = start_server_with_health(upstream).await?;

    let health = request(addr, "GET", "/frame/bus/health").await?;
    assert_eq!(health.status, 200);
    assert_eq!(health.header("content-type"), Some("application/json"));
    assert_eq!(
        String::from_utf8(health.body)?,
        r#"{"status":"healthy","message":null}"#
    );

    // A drain-window 503 from /ready stays a 503 in the page — pass-through,
    // never softened to a 200.
    let ready = request(addr, "GET", "/frame/bus/ready").await?;
    assert_eq!(ready.status, 503);
    let ready_body: serde_json::Value = serde_json::from_slice(&ready.body)?;
    assert_eq!(
        ready_body.get("ready").and_then(serde_json::Value::as_bool),
        Some(false)
    );

    // The metrics body is re-vocabularied: the upstream `liminal_*` families
    // are served under the frame-public `bus_*` names FIRST, with the
    // original `liminal_*` lines retained as deprecated duplicates.
    let metrics = request(addr, "GET", "/frame/bus/metrics").await?;
    assert_eq!(metrics.status, 200);
    assert_eq!(
        metrics.header("content-type"),
        Some("text/plain; version=0.0.4")
    );
    let metrics_text = String::from_utf8(metrics.body)?;
    assert!(metrics_text.contains("bus_connections_active 2"));
    assert!(metrics_text.contains("bus_publishes_total 41"));
    assert!(metrics_text.contains("bus_deliveries_total 40"));
    assert!(metrics_text.contains("liminal_connections_active 2"));
    assert!(metrics_text.contains("liminal_publishes_total 41"));
    assert!(metrics_text.contains("liminal_deliveries_total 40"));

    stop.send(()).map_err(|()| "server stopped early")?;
    task.await??;
    Ok(())
}

/// Compatibility window: the legacy `/frame/liminal/*` routes answer as
/// aliases of `/frame/bus/*` — same handlers, same bodies.
#[tokio::test]
async fn legacy_liminal_routes_answer_as_aliases_of_the_bus_routes()
-> Result<(), Box<dyn std::error::Error>> {
    let upstream = fake_health_listener()?;
    let (addr, stop, task) = start_server_with_health(upstream).await?;

    let legacy_health = request(addr, "GET", "/frame/liminal/health").await?;
    assert_eq!(legacy_health.status, 200);
    assert_eq!(
        String::from_utf8(legacy_health.body)?,
        r#"{"status":"healthy","message":null}"#
    );

    let legacy_ready = request(addr, "GET", "/frame/liminal/ready").await?;
    assert_eq!(legacy_ready.status, 503, "alias passes the 503 through too");

    let legacy_metrics = request(addr, "GET", "/frame/liminal/metrics").await?;
    assert_eq!(legacy_metrics.status, 200);
    let metrics_text = String::from_utf8(legacy_metrics.body)?;
    assert!(
        metrics_text.contains("bus_connections_active 2"),
        "the alias serves the SAME transformed metrics body as the bus route"
    );
    assert!(metrics_text.contains("liminal_connections_active 2"));

    stop.send(()).map_err(|()| "server stopped early")?;
    task.await??;
    Ok(())
}

#[tokio::test]
async fn bus_proxy_reports_a_down_listener_as_typed_502_json()
-> Result<(), Box<dyn std::error::Error>> {
    // The default test server targets a released address — the bus "down".
    let (addr, stop, task) = start_server().await?;

    let down = request(addr, "GET", "/frame/bus/health").await?;
    assert_eq!(
        down.status, 502,
        "a down health listener is a 502, never a hang or a fake 200"
    );
    assert_eq!(
        down.header("content-type"),
        Some("application/json; charset=utf-8")
    );
    let body: serde_json::Value = serde_json::from_slice(&down.body)?;
    assert_eq!(
        body.get("error").and_then(serde_json::Value::as_str),
        Some("BUS_UNREACHABLE")
    );
    assert_eq!(
        body.get("upstream_path")
            .and_then(serde_json::Value::as_str),
        Some("/health")
    );
    assert!(
        body.get("detail")
            .and_then(serde_json::Value::as_str)
            .is_some_and(|detail| !detail.is_empty()),
        "the 502 body must carry the exact failure detail"
    );

    stop.send(()).map_err(|()| "server stopped early")?;
    task.await??;
    Ok(())
}

#[test]
fn missing_config_flag_is_a_startup_refusal_naming_the_flag()
-> Result<(), Box<dyn std::error::Error>> {
    use clap::Parser;
    // The full frame server takes one flag: `--config <frame.toml>`. Every
    // deployment value now lives in that file, so parsing without it must
    // refuse and name the missing flag.
    let error = frame_host::Cli::try_parse_from(["frame-host"])
        .err()
        .ok_or("parsing without --config must refuse")?
        .to_string();
    assert!(
        error.contains("--config"),
        "the refusal must name the missing flag, got: {error}"
    );
    Ok(())
}

#[tokio::test]
async fn traversal_and_bad_methods_are_refused() -> Result<(), Box<dyn std::error::Error>> {
    let (addr, stop, task) = start_server().await?;

    let encoded_traversal = request(addr, "GET", "/%2e%2e/Cargo.toml").await?;
    assert_eq!(encoded_traversal.status, 400);

    let doubled = request(addr, "GET", "//index.html").await?;
    assert_eq!(doubled.status, 400);

    let posted = request(addr, "POST", "/").await?;
    assert_eq!(posted.status, 405);

    stop.send(()).map_err(|()| "server stopped early")?;
    task.await??;
    Ok(())
}

#[tokio::test]
async fn unusable_asset_roots_are_refused_at_bind() -> Result<(), Box<dyn std::error::Error>> {
    let absent = ShellServer::bind(ShellConfig {
        bind: any_local_addr()?,
        asset_root: fixture_site()?.join("does-not-exist"),
        bus_endpoint: BUS_ENDPOINT.to_owned(),
        auth_token: AUTH_TOKEN.to_owned(),
        channel: None,
        channels: default_channels(),
        bus_health: released_addr()?,
        document: None,
    })
    .await;
    assert!(matches!(absent, Err(HostError::AssetRoot { .. })));

    let no_index = ShellServer::bind(ShellConfig {
        bind: any_local_addr()?,
        asset_root: fixture_site()?.join("assets"),
        bus_endpoint: BUS_ENDPOINT.to_owned(),
        auth_token: AUTH_TOKEN.to_owned(),
        channel: None,
        channels: default_channels(),
        bus_health: released_addr()?,
        document: None,
    })
    .await;
    assert!(matches!(no_index, Err(HostError::MissingIndex { .. })));
    Ok(())
}