cellos-ctl 0.5.0

//! `cellctl webui` — localhost reverse proxy + static bundle host.
//!
//! Per ADR-0017, the web view ships *with cellctl*, not with cellos-server.
//! Operators invoke `cellctl webui` to spin up a foreground localhost proxy:
//!
//!   - Serves `crates/cellos-ctl/static/` (the Vite build output) at `/`.
//!   - Reverse-proxies `GET /v1/*` and `GET /ws/events` upstream to the
//!     cellos-server URL configured in `~/.cellctl/config`, injecting the
//!     bearer token on the way out. The bundle never sees the token.
//!   - Refuses any non-`GET` method with HTTP 405 (`Allow: GET`). This is the
//!     structural enforcement of ADR-0016's read-only browser boundary.
//!   - Binds a session token in a URL fragment (`/#sess=<base64>`) and
//!     swaps it for an `HttpOnly; SameSite=Strict` cookie via
//!     `POST /auth/exchange`. Subsequent proxy + WS requests require the
//!     cookie. The fragment is cleared by the bundle's bootstrap once
//!     the cookie is set.
//!   - Exits cleanly on SIGINT.
//!
//! Bind modes (ADR-0017 §Decision 4):
//!
//! - `--bind auto` (default, on Unix): bind BOTH a loopback TCP port (for the
//!   browser) AND a Unix socket at
//!   `${XDG_RUNTIME_DIR:-/tmp}/cellctl-webui-<pid>.sock` (for inter-process
//!   tooling that wants to bypass loopback). On Windows, `auto` degrades to
//!   loopback-only.
//! - `--bind loopback`: TCP loopback only.
//! - `--bind unix` (Unix only): Unix socket only. The browser cannot reach a
//!   Unix socket directly — this mode is for inter-process forwarders (e.g.
//!   `socat` / `ssh -L`). On Windows this errors out.
//!
//! The Unix socket is created with mode `0600` (operator-owned only) and is
//! removed on graceful shutdown (SIGINT).

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

use axum::body::{Body, Bytes};
use axum::extract::{ws::WebSocketUpgrade, Path as AxumPath, State};
use axum::http::{header, HeaderMap, HeaderName, HeaderValue, Method, StatusCode, Uri};
use axum::response::{IntoResponse, Response};
use axum::routing::{any, post};
use axum::{Json, Router};
use base64::Engine as _;
use clap::ValueEnum;
use futures_util::{SinkExt, StreamExt};
use rand::RngCore;
use serde::Deserialize;
use tokio::sync::RwLock;
use tower_http::services::ServeDir;

use crate::client::CellosClient;
use crate::config::Config;
use crate::exit::{CtlError, CtlResult};

/// Where the bundle lives on disk. Resolved relative to the cellctl binary's
/// crate at compile time; we look up the actual path at runtime via the
/// `CARGO_MANIFEST_DIR` env var (set in tests) or `./static` (when the
/// binary is installed alongside its bundle).
const BUNDLE_DIR_RELATIVE: &str = "static";

/// Bind mode for the webui proxy.
#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, ValueEnum)]
pub enum BindMode {
    /// Default: on Unix, bind BOTH a loopback TCP port and a Unix socket.
    /// On Windows, degrade to loopback-only.
    #[default]
    Auto,
    /// Force 127.0.0.1 on a random high port. No Unix socket.
    Loopback,
    /// Force a Unix socket only (no TCP). Browsers cannot reach a Unix
    /// socket directly; this mode is for inter-process forwarders. Errors
    /// on Windows.
    Unix,
}

/// Per-process shared state: upstream client + the one valid session token
/// (until it's exchanged) and the resulting session cookie value.
#[derive(Clone)]
struct AppState {
    /// Upstream cellos-server base URL (e.g. `http://127.0.0.1:8080`).
    upstream_base: Arc<String>,
    /// Upstream Bearer token to inject on every proxied request, if any.
    upstream_bearer: Arc<Option<String>>,
    /// The unguessable, single-use session token printed in the URL fragment.
    session_token: Arc<String>,
    /// Once `/auth/exchange` succeeds, this holds the cookie value the
    /// browser is expected to present on every subsequent request.
    session_cookie: Arc<RwLock<Option<String>>>,
    /// Filesystem path of the static bundle (for `ServeDir`).
    bundle_dir: Arc<PathBuf>,
}

#[derive(Deserialize)]
struct ExchangeRequest {
    sess: String,
}

/// Entry point invoked by `main.rs`.
pub async fn run(cfg: &Config, open: bool, bind: BindMode) -> CtlResult<()> {
    // Reuse the same effective config CellosClient sees, so the URL/token
    // injected into the proxy matches what the rest of cellctl would use.
    // Touching CellosClient also validates that the token (if any) parses as
    // a header value.
    let _ = CellosClient::new(cfg)?;

    let upstream_base = cfg.effective_server().trim_end_matches('/').to_string();
    let upstream_bearer = cfg.effective_token();

    let bundle_dir = resolve_bundle_dir()?;
    if !bundle_dir.join("index.html").exists() {
        return Err(CtlError::usage(format!(
            "webui bundle not found at {}/index.html — run `npm --prefix web run build` first",
            bundle_dir.display()
        )));
    }

    let session_token = mint_session_token();

    let state = AppState {
        upstream_base: Arc::new(upstream_base),
        upstream_bearer: Arc::new(upstream_bearer),
        session_token: Arc::new(session_token.clone()),
        session_cookie: Arc::new(RwLock::new(None)),
        bundle_dir: Arc::new(bundle_dir.clone()),
    };

    let app = build_router(state);

    // Decide which listeners we actually want for this run.
    let (want_tcp, want_unix) = resolve_bind_plan(bind)?;

    let tcp_listener = if want_tcp {
        let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
        Some(
            tokio::net::TcpListener::bind(addr)
                .await
                .map_err(|e| CtlError::usage(format!("bind 127.0.0.1: {e}")))?,
        )
    } else {
        None
    };

    #[cfg(unix)]
    let unix_socket_path: Option<PathBuf> = if want_unix {
        Some(unix_socket_path_for_pid(std::process::id()))
    } else {
        None
    };
    #[cfg(not(unix))]
    let unix_socket_path: Option<PathBuf> = None;

    let browser_url = if let Some(l) = tcp_listener.as_ref() {
        let local_addr = l
            .local_addr()
            .map_err(|e| CtlError::usage(format!("local_addr: {e}")))?;
        Some(format!("http://{}/#sess={}", local_addr, session_token))
    } else {
        None
    };

    if let Some(url) = browser_url.as_ref() {
        println!("cellctl webui: {}", url);
    }
    if let Some(p) = unix_socket_path.as_ref() {
        println!("cellctl webui: unix://{}", p.display());
    }
    if browser_url.is_none() {
        // unix-only mode: be loud that the browser can't reach this.
        eprintln!(
            "cellctl webui: --bind unix has no browser-reachable URL; \
             use a forwarder (e.g. `socat TCP-LISTEN:0,reuseaddr,fork UNIX-CONNECT:{}`) \
             or rerun with `--bind auto` for a loopback URL.",
            unix_socket_path
                .as_ref()
                .map(|p| p.display().to_string())
                .unwrap_or_else(|| "<socket>".to_string())
        );
    }
    println!("upstream: {}", state_upstream_for_log(&app));
    println!("press Ctrl-C to stop");

    if open {
        if let Some(url) = browser_url.as_ref() {
            if let Err(e) = opener::open(url) {
                eprintln!("cellctl webui: could not launch browser: {e}");
            }
        } else {
            eprintln!("cellctl webui: --open ignored: no loopback URL bound (use --bind auto)");
        }
    }

    // Build a shutdown signal future shared by both listeners. We resolve on
    // SIGINT and (best-effort) clean up the socket file before either listener
    // unbinds it.
    let (shutdown_tx, _shutdown_rx) = tokio::sync::broadcast::channel::<()>(1);
    {
        let shutdown_tx = shutdown_tx.clone();
        tokio::spawn(async move {
            let _ = tokio::signal::ctrl_c().await;
            eprintln!("shutting down");
            let _ = shutdown_tx.send(());
        });
    }

    // Spawn the TCP server if requested.
    let tcp_task = if let Some(listener) = tcp_listener {
        let app = app.clone();
        let mut rx = shutdown_tx.subscribe();
        Some(tokio::spawn(async move {
            axum::serve(listener, app)
                .with_graceful_shutdown(async move {
                    let _ = rx.recv().await;
                })
                .await
        }))
    } else {
        None
    };

    // Spawn the Unix server if requested.
    #[cfg(unix)]
    let unix_task = if let Some(path) = unix_socket_path.clone() {
        let app = app.clone();
        let mut rx = shutdown_tx.subscribe();
        let listener = bind_unix_listener(&path)?;
        Some(tokio::spawn(async move {
            serve_unix(listener, app, async move {
                let _ = rx.recv().await;
            })
            .await
        }))
    } else {
        None
    };
    #[cfg(not(unix))]
    let unix_task: Option<tokio::task::JoinHandle<std::io::Result<()>>> = None;

    // Wait for both servers (whichever exist). Capture the first error.
    let mut first_err: Option<String> = None;
    if let Some(t) = tcp_task {
        match t.await {
            Ok(Ok(())) => {}
            Ok(Err(e)) => {
                let _ = first_err.get_or_insert_with(|| format!("tcp: {e}"));
            }
            Err(e) => {
                let _ = first_err.get_or_insert_with(|| format!("tcp join: {e}"));
            }
        }
    }
    if let Some(t) = unix_task {
        match t.await {
            Ok(Ok(())) => {}
            Ok(Err(e)) => {
                let _ = first_err.get_or_insert_with(|| format!("unix: {e}"));
            }
            Err(e) => {
                let _ = first_err.get_or_insert_with(|| format!("unix join: {e}"));
            }
        }
    }

    // Best-effort cleanup: remove the socket file. (UnixListener does not unlink
    // on drop.)
    if let Some(p) = unix_socket_path.as_ref() {
        let _ = std::fs::remove_file(p);
    }

    if let Some(e) = first_err {
        return Err(CtlError::api(format!("webui server: {e}")));
    }
    Ok(())
}

/// Decide which listeners to spawn for a given bind mode.
///
/// Returns `(want_tcp, want_unix)`. On non-Unix platforms, `want_unix` is
/// always false and `BindMode::Unix` errors.
fn resolve_bind_plan(bind: BindMode) -> CtlResult<(bool, bool)> {
    #[cfg(unix)]
    {
        Ok(match bind {
            BindMode::Auto => (true, true),
            BindMode::Loopback => (true, false),
            BindMode::Unix => (false, true),
        })
    }
    #[cfg(not(unix))]
    {
        Ok(match bind {
            BindMode::Auto => (true, false),
            BindMode::Loopback => (true, false),
            BindMode::Unix => {
                return Err(CtlError::usage(
                    "--bind unix is not supported on Windows; use --bind loopback (the default)",
                ));
            }
        })
    }
}

/// Compute the Unix socket path for a given pid:
/// `${XDG_RUNTIME_DIR:-/tmp}/cellctl-webui-<pid>.sock`.
#[cfg(unix)]
fn unix_socket_path_for_pid(pid: u32) -> PathBuf {
    let dir = std::env::var_os("XDG_RUNTIME_DIR")
        .map(PathBuf::from)
        .unwrap_or_else(|| PathBuf::from("/tmp"));
    dir.join(format!("cellctl-webui-{pid}.sock"))
}

/// Bind a Unix socket at `path` with mode 0600. If the path already exists
/// (stale socket from a previous run with the same pid), we remove and
/// re-bind.
#[cfg(unix)]
fn bind_unix_listener(path: &std::path::Path) -> CtlResult<tokio::net::UnixListener> {
    use std::os::unix::fs::PermissionsExt;

    // Remove a stale socket file from a prior crashed run. Best-effort.
    let _ = std::fs::remove_file(path);

    let listener = tokio::net::UnixListener::bind(path)
        .map_err(|e| CtlError::usage(format!("bind {}: {e}", path.display())))?;

    // Restrict to operator-only (0600).
    let perms = std::fs::Permissions::from_mode(0o600);
    std::fs::set_permissions(path, perms)
        .map_err(|e| CtlError::usage(format!("chmod 0600 {}: {e}", path.display())))?;

    Ok(listener)
}

/// Serve `app` over a Unix-domain `UnixListener` with graceful shutdown.
///
/// `axum::serve` requires a `tokio::net::TcpListener`, so for Unix sockets we
/// drop down to the lower-level `accept_loop`-style approach using `hyper`
/// directly via `tower::Service::call`.
#[cfg(unix)]
async fn serve_unix(
    listener: tokio::net::UnixListener,
    app: Router,
    shutdown: impl std::future::Future<Output = ()>,
) -> std::io::Result<()> {
    use std::convert::Infallible;
    use tower::Service;

    // axum::Router implements tower::Service<Request<Body>>. We accept Unix
    // connections and hand each one to a hyper http1 connection driving the
    // router. This mirrors what axum::serve does internally for TCP, minus the
    // TCP-specific bits.
    tokio::pin!(shutdown);

    loop {
        tokio::select! {
            _ = &mut shutdown => return Ok(()),
            accepted = listener.accept() => {
                let (stream, _peer) = match accepted {
                    Ok(s) => s,
                    Err(e) => {
                        // EMFILE / transient — log and continue, do not crash.
                        eprintln!("webui: unix accept error: {e}");
                        continue;
                    }
                };
                let app = app.clone();
                tokio::spawn(async move {
                    // `service_fn` requires `Fn`, but Router::call needs
                    // `&mut self`. Clone the router per request so each
                    // invocation owns its own mutable handle.
                    let svc = hyper::service::service_fn(move |req: http::Request<hyper::body::Incoming>| {
                        let mut router = app.clone();
                        async move {
                            let resp: Response = match router.call(req.map(Body::new)).await {
                                Ok(r) => r,
                                Err(never) => match never {},
                            };
                            Ok::<_, Infallible>(resp)
                        }
                    });
                    let io = hyper_util::rt::TokioIo::new(stream);
                    let _ = hyper_util::server::conn::auto::Builder::new(hyper_util::rt::TokioExecutor::new())
                        .serve_connection_with_upgrades(io, svc)
                        .await;
                });
            }
        }
    }
}

/// Mint a 32-byte random session token and base64-url-encode it.
fn mint_session_token() -> String {
    let mut buf = [0u8; 32];
    rand::thread_rng().fill_bytes(&mut buf);
    base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(buf)
}

/// Resolve the bundle directory. Tests set `CELLCTL_WEBUI_BUNDLE_DIR`
/// explicitly. Installed binaries look next to `crates/cellos-ctl/static/`
/// via `CARGO_MANIFEST_DIR` (build-time embed) or the current working dir.
fn resolve_bundle_dir() -> CtlResult<PathBuf> {
    if let Ok(p) = std::env::var("CELLCTL_WEBUI_BUNDLE_DIR") {
        return Ok(PathBuf::from(p));
    }
    // At build time we record the crate root; installed binaries can still
    // override via the env var above.
    let from_manifest = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(BUNDLE_DIR_RELATIVE);
    Ok(from_manifest)
}

/// Build the axum router with the static fileserver, /auth/exchange handler,
/// and the catch-all proxy route.
fn build_router(state: AppState) -> Router {
    // Static bundle: ServeDir handles GET only — non-GET on the static tree
    // falls through to our catch-all 405 layer.
    let bundle_dir = state.bundle_dir.as_ref().clone();
    let serve_dir = ServeDir::new(&bundle_dir).append_index_html_on_directories(true);

    Router::new()
        .route("/auth/exchange", post(auth_exchange))
        .route("/v1/*rest", any(proxy_v1))
        .route("/ws/events", any(ws_events))
        // Fallback: serve the bundle (ServeDir). ServeDir already 405s on
        // non-GET; we intercept earlier in the middleware chain to apply our
        // own canonical Allow: GET header.
        .fallback_service(serve_dir)
        .layer(axum::middleware::from_fn(reject_non_get))
        .with_state(state)
}

/// Middleware: every method except `GET` (and the special POST to
/// `/auth/exchange`) returns 405 with `Allow: GET`. This is the structural
/// enforcement of ADR-0016's read-only browser boundary.
async fn reject_non_get(req: axum::http::Request<Body>, next: axum::middleware::Next) -> Response {
    let method = req.method().clone();
    let path = req.uri().path().to_string();

    // /auth/exchange is the one exception — the bundle must POST it once to
    // swap the fragment token for a cookie. WebSocket upgrade requests come
    // through as GET, so that path is fine.
    let is_auth_exchange = method == Method::POST && path == "/auth/exchange";
    if method != Method::GET && !is_auth_exchange {
        return method_not_allowed();
    }
    next.run(req).await
}

fn method_not_allowed() -> Response {
    let mut resp = (StatusCode::METHOD_NOT_ALLOWED, "method not allowed\n").into_response();
    resp.headers_mut()
        .insert(header::ALLOW, HeaderValue::from_static("GET"));
    resp
}

/// `POST /auth/exchange` — body is `{"sess": "<base64-token>"}`. On match
/// against the in-process session token, mint a cookie value and set
/// `Set-Cookie: cellctl_session=<value>; HttpOnly; SameSite=Strict; Path=/`.
async fn auth_exchange(
    State(state): State<AppState>,
    Json(body): Json<ExchangeRequest>,
) -> Response {
    // Constant-time-ish comparison; for this MVP the token is also random
    // 32 bytes so timing leaks ~zero useful info, but be tidy.
    let provided = body.sess.as_bytes();
    let expected = state.session_token.as_bytes();
    if provided.len() != expected.len()
        || !provided
            .iter()
            .zip(expected.iter())
            .fold(true, |acc, (a, b)| acc & (a == b))
    {
        return (StatusCode::UNAUTHORIZED, "bad session token\n").into_response();
    }

    // Mint the cookie value. Distinct from the URL token so we never reuse
    // a value that might have appeared anywhere in history.
    let mut buf = [0u8; 32];
    rand::thread_rng().fill_bytes(&mut buf);
    let cookie_value = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(buf);

    {
        let mut slot = state.session_cookie.write().await;
        *slot = Some(cookie_value.clone());
    }

    let cookie_header = format!(
        "cellctl_session={}; HttpOnly; SameSite=Strict; Path=/",
        cookie_value
    );

    let mut resp = (StatusCode::OK, "ok\n").into_response();
    resp.headers_mut().insert(
        header::SET_COOKIE,
        HeaderValue::from_str(&cookie_header).unwrap(),
    );
    resp
}

/// Verify the request carries the `cellctl_session` cookie we set in
/// `/auth/exchange`. Returns Some(()) if OK, None if missing/mismatch.
async fn require_session_cookie(state: &AppState, headers: &HeaderMap) -> bool {
    let expected = match state.session_cookie.read().await.clone() {
        Some(v) => v,
        None => return false,
    };
    let Some(cookie_hdr) = headers.get(header::COOKIE) else {
        return false;
    };
    let Ok(cookie_str) = cookie_hdr.to_str() else {
        return false;
    };
    // Cookies are `name=value; name=value` — find our entry.
    for entry in cookie_str.split(';') {
        let entry = entry.trim();
        if let Some(v) = entry.strip_prefix("cellctl_session=") {
            return v == expected;
        }
    }
    false
}

/// `GET /v1/*` — reverse-proxy to upstream with Bearer injection.
async fn proxy_v1(
    State(state): State<AppState>,
    AxumPath(rest): AxumPath<String>,
    uri: Uri,
    headers: HeaderMap,
) -> Response {
    if !require_session_cookie(&state, &headers).await {
        return unauthorized();
    }

    let query = uri.query().map(|q| format!("?{q}")).unwrap_or_default();
    let upstream_url = format!("{}/v1/{}{}", state.upstream_base, rest, query);

    let client = match reqwest::Client::builder().build() {
        Ok(c) => c,
        Err(e) => return upstream_error(format!("client: {e}")),
    };

    let mut req = client.get(&upstream_url);
    if let Some(tok) = state.upstream_bearer.as_ref() {
        req = req.header(reqwest::header::AUTHORIZATION, format!("Bearer {tok}"));
    }

    let upstream_resp = match req.send().await {
        Ok(r) => r,
        Err(e) => return upstream_error(format!("send: {e}")),
    };

    let status = upstream_resp.status();
    let resp_headers = upstream_resp.headers().clone();
    let body_bytes = match upstream_resp.bytes().await {
        Ok(b) => b,
        Err(e) => return upstream_error(format!("read body: {e}")),
    };

    let mut out = Response::new(Body::from(body_bytes));
    *out.status_mut() = StatusCode::from_u16(status.as_u16()).unwrap_or(StatusCode::BAD_GATEWAY);
    // Forward content-type only (avoid leaking upstream Set-Cookie etc.).
    if let Some(ct) = resp_headers.get(reqwest::header::CONTENT_TYPE) {
        if let Ok(v) = HeaderValue::from_bytes(ct.as_bytes()) {
            out.headers_mut().insert(header::CONTENT_TYPE, v);
        }
    }
    out
}

/// `GET /ws/events` — WebSocket upgrade, forwarded to upstream.
async fn ws_events(
    State(state): State<AppState>,
    uri: Uri,
    headers: HeaderMap,
    ws: Option<WebSocketUpgrade>,
) -> Response {
    if !require_session_cookie(&state, &headers).await {
        return unauthorized();
    }
    let Some(ws) = ws else {
        return (StatusCode::BAD_REQUEST, "expected websocket upgrade\n").into_response();
    };

    let query = uri.query().map(|q| format!("?{q}")).unwrap_or_default();
    let ws_url = {
        // http:// → ws://, https:// → wss://
        let base = state.upstream_base.as_str();
        let ws_base = if let Some(rest) = base.strip_prefix("https://") {
            format!("wss://{rest}")
        } else if let Some(rest) = base.strip_prefix("http://") {
            format!("ws://{rest}")
        } else {
            base.to_string()
        };
        format!("{ws_base}/ws/events{query}")
    };
    let bearer = state.upstream_bearer.as_ref().clone();

    ws.on_upgrade(move |client_ws| async move {
        let (mut client_tx, mut client_rx) = client_ws.split();

        // Build a request to upstream with the bearer header, since
        // `connect_async` accepts a raw URL but loses our custom headers.
        let mut request =
            match tokio_tungstenite::tungstenite::client::IntoClientRequest::into_client_request(
                ws_url.as_str(),
            ) {
                Ok(r) => r,
                Err(_) => return,
            };
        if let Some(tok) = bearer {
            if let Ok(v) = tokio_tungstenite::tungstenite::http::HeaderValue::from_str(&format!(
                "Bearer {tok}"
            )) {
                request.headers_mut().insert(
                    tokio_tungstenite::tungstenite::http::header::AUTHORIZATION,
                    v,
                );
            }
        }
        let (upstream_ws, _) = match tokio_tungstenite::connect_async(request).await {
            Ok(p) => p,
            Err(_) => return,
        };
        let (mut up_tx, mut up_rx) = upstream_ws.split();

        loop {
            tokio::select! {
                msg = client_rx.next() => match msg {
                    Some(Ok(axum::extract::ws::Message::Text(t))) => {
                        let _ = up_tx
                            .send(tokio_tungstenite::tungstenite::Message::Text(t))
                            .await;
                    }
                    Some(Ok(axum::extract::ws::Message::Binary(b))) => {
                        let _ = up_tx
                            .send(tokio_tungstenite::tungstenite::Message::Binary(b))
                            .await;
                    }
                    Some(Ok(axum::extract::ws::Message::Ping(p))) => {
                        let _ = up_tx
                            .send(tokio_tungstenite::tungstenite::Message::Ping(p))
                            .await;
                    }
                    Some(Ok(axum::extract::ws::Message::Pong(p))) => {
                        let _ = up_tx
                            .send(tokio_tungstenite::tungstenite::Message::Pong(p))
                            .await;
                    }
                    Some(Ok(axum::extract::ws::Message::Close(_))) | None => break,
                    Some(Err(_)) => break,
                },
                msg = up_rx.next() => match msg {
                    Some(Ok(tokio_tungstenite::tungstenite::Message::Text(t))) => {
                        let _ = client_tx
                            .send(axum::extract::ws::Message::Text(t))
                            .await;
                    }
                    Some(Ok(tokio_tungstenite::tungstenite::Message::Binary(b))) => {
                        let _ = client_tx
                            .send(axum::extract::ws::Message::Binary(b))
                            .await;
                    }
                    Some(Ok(tokio_tungstenite::tungstenite::Message::Ping(p))) => {
                        let _ = client_tx
                            .send(axum::extract::ws::Message::Ping(p))
                            .await;
                    }
                    Some(Ok(tokio_tungstenite::tungstenite::Message::Pong(p))) => {
                        let _ = client_tx
                            .send(axum::extract::ws::Message::Pong(p))
                            .await;
                    }
                    Some(Ok(tokio_tungstenite::tungstenite::Message::Close(_))) | None => break,
                    Some(Ok(tokio_tungstenite::tungstenite::Message::Frame(_))) => {}
                    Some(Err(_)) => break,
                },
            }
        }
    })
}

fn unauthorized() -> Response {
    (StatusCode::UNAUTHORIZED, "missing session cookie\n").into_response()
}

fn upstream_error(msg: String) -> Response {
    (StatusCode::BAD_GATEWAY, format!("upstream: {msg}\n")).into_response()
}

/// Tiny helper: format the upstream URL for the startup banner. (Reads back
/// from the router via state, which is awkward; we accept a Router-shaped
/// argument and don't actually consult it — kept as a hook for richer logs.)
fn state_upstream_for_log(_app: &Router) -> &'static str {
    // The actual upstream URL is held in AppState which is consumed by axum;
    // the caller printed it via state.upstream_base before this — leaving
    // this function as a stable hook means callers can extend the banner
    // later without changing the call site.
    "(see state)"
}

#[allow(dead_code)]
const _BYTES_KEEP: fn() = || {
    // Suppress unused import warnings for `Bytes` until proxy body streaming
    // is upgraded to use it directly.
    let _: Bytes = Bytes::new();
    let _: HeaderName = HeaderName::from_static("x-keep");
};

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use axum::body::to_bytes;
    use axum::http::Request;
    use tower::ServiceExt; // for `oneshot`

    fn test_state(bundle_dir: PathBuf) -> AppState {
        AppState {
            upstream_base: Arc::new("http://127.0.0.1:0".to_string()),
            upstream_bearer: Arc::new(None),
            session_token: Arc::new("test-token".to_string()),
            session_cookie: Arc::new(RwLock::new(None)),
            bundle_dir: Arc::new(bundle_dir),
        }
    }

    /// Use the real bundle if it's been built; otherwise create a fake one
    /// in a tempdir with a minimal index.html.
    fn ensure_bundle_dir() -> PathBuf {
        let manifest_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        let real = manifest_dir.join("static");
        if real.join("index.html").exists() {
            return real;
        }
        // Fall back to a freshly-minted temp dir.
        let tmp = std::env::temp_dir().join(format!("cellctl-webui-test-{}", std::process::id()));
        std::fs::create_dir_all(&tmp).expect("mkdir tmp bundle");
        std::fs::write(
            tmp.join("index.html"),
            "<!doctype html><title>cellctl webui</title>",
        )
        .expect("write index.html");
        tmp
    }

    #[tokio::test]
    async fn serves_index_at_root() {
        let bundle = ensure_bundle_dir();
        let app = build_router(test_state(bundle));

        let resp = app
            .oneshot(
                Request::builder()
                    .method("GET")
                    .uri("/")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let body = to_bytes(resp.into_body(), 64 * 1024).await.unwrap();
        let body_str = std::str::from_utf8(&body).unwrap();
        assert!(
            body_str.to_ascii_lowercase().contains("<!doctype html")
                || body_str.to_ascii_lowercase().contains("<html"),
            "expected HTML at /, got: {body_str:.200}"
        );
    }

    #[tokio::test]
    async fn non_get_returns_405() {
        let bundle = ensure_bundle_dir();
        let app = build_router(test_state(bundle));

        let resp = app
            .oneshot(
                Request::builder()
                    .method("DELETE")
                    .uri("/v1/formations")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::METHOD_NOT_ALLOWED);
        assert_eq!(
            resp.headers().get(header::ALLOW).map(|v| v.as_bytes()),
            Some(b"GET" as &[u8]),
        );
    }

    #[tokio::test]
    async fn put_to_v1_returns_405() {
        let bundle = ensure_bundle_dir();
        let app = build_router(test_state(bundle));

        let resp = app
            .oneshot(
                Request::builder()
                    .method("PUT")
                    .uri("/v1/formations/foo")
                    .body(Body::from("body"))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::METHOD_NOT_ALLOWED);
    }

    #[tokio::test]
    async fn proxy_without_cookie_returns_401() {
        let bundle = ensure_bundle_dir();
        let app = build_router(test_state(bundle));

        let resp = app
            .oneshot(
                Request::builder()
                    .method("GET")
                    .uri("/v1/formations")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
    }

    #[tokio::test]
    async fn auth_exchange_with_wrong_token_returns_401() {
        let bundle = ensure_bundle_dir();
        let app = build_router(test_state(bundle));

        let resp = app
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/auth/exchange")
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(Body::from(r#"{"sess":"wrong"}"#))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
    }

    #[tokio::test]
    async fn auth_exchange_with_right_token_sets_cookie() {
        let bundle = ensure_bundle_dir();
        let state = test_state(bundle);
        let app = build_router(state.clone());

        let resp = app
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/auth/exchange")
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(Body::from(r#"{"sess":"test-token"}"#))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let cookie = resp
            .headers()
            .get(header::SET_COOKIE)
            .expect("Set-Cookie header present")
            .to_str()
            .unwrap()
            .to_string();
        assert!(cookie.starts_with("cellctl_session="));
        assert!(cookie.contains("HttpOnly"));
        assert!(cookie.contains("SameSite=Strict"));

        // And the state should now hold a matching cookie value.
        let stored = state.session_cookie.read().await.clone();
        assert!(stored.is_some());
        let stored = stored.unwrap();
        assert!(cookie.contains(&format!("cellctl_session={stored}")));
    }

    #[tokio::test]
    async fn proxy_with_valid_cookie_attempts_upstream() {
        // After /auth/exchange succeeds, a subsequent GET /v1/* should *not*
        // 401 — it'll instead 502 (bad gateway) because there's no real
        // upstream listening, which is exactly what proves the cookie gate
        // passed.
        let bundle = ensure_bundle_dir();
        let state = test_state(bundle);
        let app = build_router(state.clone());

        let exch = app
            .clone()
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/auth/exchange")
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(Body::from(r#"{"sess":"test-token"}"#))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(exch.status(), StatusCode::OK);
        let cookie_hdr = exch
            .headers()
            .get(header::SET_COOKIE)
            .unwrap()
            .to_str()
            .unwrap()
            .to_string();
        // Strip attributes; keep just `cellctl_session=<value>`.
        let cookie_pair = cookie_hdr.split(';').next().unwrap().trim().to_string();

        let resp = app
            .oneshot(
                Request::builder()
                    .method("GET")
                    .uri("/v1/formations")
                    .header(header::COOKIE, cookie_pair)
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        // No live upstream → bad gateway, not 401.
        assert_ne!(resp.status(), StatusCode::UNAUTHORIZED);
    }

    /// Proxy invariant: no matter what HTTP method the browser sends, the
    /// upstream cellos-server must NEVER see a non-GET. The proxy is the
    /// structural enforcement of ADR-0016's read-only browser boundary
    /// (ADR-0017 §Decision 4, item 5: "Refuse non-GET methods with HTTP 405").
    ///
    /// We spin up a tiny inline mock upstream on a real loopback port that
    /// counts the methods it sees, point AppState at it, and fire a battery of
    /// inbound methods through the proxy. The assertion is bidirectional:
    ///
    ///   1. The mock saw ZERO non-GET requests.
    ///   2. Every non-GET inbound request returned 405 with `Allow: GET`.
    #[tokio::test]
    async fn proxy_only_emits_get_to_upstream() {
        use std::sync::Mutex;
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        use tokio::net::TcpListener;

        // ---- inline mock upstream ----
        let mock = TcpListener::bind("127.0.0.1:0").await.expect("bind mock");
        let mock_addr = mock.local_addr().expect("mock addr");
        let methods_seen: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
        let methods_for_task = methods_seen.clone();

        // Spawn a tiny accept loop that reads the request line, records the
        // method, then writes a minimal `HTTP/1.1 200 OK` and closes. We do
        // not implement keep-alive, content-length parsing, or anything else
        // — the proxy doesn't keep connections open across requests against
        // this mock anyway (each `reqwest::Client::builder().build()` builds a
        // fresh client).
        tokio::spawn(async move {
            loop {
                let (mut stream, _) = match mock.accept().await {
                    Ok(p) => p,
                    Err(_) => return,
                };
                let methods = methods_for_task.clone();
                tokio::spawn(async move {
                    let mut buf = [0u8; 4096];
                    let n = match stream.read(&mut buf).await {
                        Ok(n) => n,
                        Err(_) => return,
                    };
                    let head = String::from_utf8_lossy(&buf[..n]).to_string();
                    if let Some(first_line) = head.lines().next() {
                        if let Some(method) = first_line.split_whitespace().next() {
                            methods.lock().unwrap().push(method.to_string());
                        }
                    }
                    let _ = stream
                        .write_all(
                            b"HTTP/1.1 200 OK\r\n\
                              Content-Type: application/json\r\n\
                              Content-Length: 2\r\n\
                              Connection: close\r\n\
                              \r\n\
                              {}",
                        )
                        .await;
                    let _ = stream.shutdown().await;
                });
            }
        });

        // ---- proxy under test ----
        let bundle = ensure_bundle_dir();
        let state = AppState {
            upstream_base: Arc::new(format!("http://{}", mock_addr)),
            upstream_bearer: Arc::new(Some("test-bearer".to_string())),
            session_token: Arc::new("test-token".to_string()),
            session_cookie: Arc::new(RwLock::new(None)),
            bundle_dir: Arc::new(bundle),
        };
        let app = build_router(state.clone());

        // Mint a session cookie so /v1/* GETs can pass the auth gate. (Non-GET
        // requests will be rejected at the 405 middleware *before* the cookie
        // check, so this only matters for the GET-passes-through assertion.)
        let exch = app
            .clone()
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/auth/exchange")
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(Body::from(r#"{"sess":"test-token"}"#))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(exch.status(), StatusCode::OK);
        let cookie_pair = exch
            .headers()
            .get(header::SET_COOKIE)
            .unwrap()
            .to_str()
            .unwrap()
            .split(';')
            .next()
            .unwrap()
            .trim()
            .to_string();

        // Fire a battery of inbound methods. Each non-GET MUST return 405 and
        // MUST NOT touch the upstream.
        let non_get_methods = ["POST", "PUT", "DELETE", "PATCH"];
        for m in non_get_methods {
            let resp = app
                .clone()
                .oneshot(
                    Request::builder()
                        .method(m)
                        .uri("/v1/formations")
                        .header(header::COOKIE, cookie_pair.clone())
                        .body(Body::from("payload that must never reach upstream"))
                        .unwrap(),
                )
                .await
                .unwrap();
            assert_eq!(
                resp.status(),
                StatusCode::METHOD_NOT_ALLOWED,
                "inbound {m} should be 405"
            );
            assert_eq!(
                resp.headers().get(header::ALLOW).map(|v| v.as_bytes()),
                Some(b"GET" as &[u8]),
                "405 response for {m} must carry `Allow: GET`"
            );
        }

        // Now fire an actual GET, which SHOULD reach the mock upstream.
        let get_resp = app
            .clone()
            .oneshot(
                Request::builder()
                    .method("GET")
                    .uri("/v1/formations")
                    .header(header::COOKIE, cookie_pair.clone())
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        // The GET must NOT be rejected by the proxy's read-only gate.
        assert_ne!(
            get_resp.status(),
            StatusCode::METHOD_NOT_ALLOWED,
            "GET must pass the 405 middleware"
        );

        // Give the mock a moment to record the inbound connection.
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        // ---- assertions on the mock's observation log ----
        let observed = methods_seen.lock().unwrap().clone();
        // Exactly one upstream request (the GET) — the four non-GETs were
        // all stopped at the proxy.
        let non_get_count = observed.iter().filter(|m| m.as_str() != "GET").count();
        assert_eq!(
            non_get_count, 0,
            "upstream saw non-GET method(s): {observed:?}"
        );
        assert!(
            observed.iter().any(|m| m == "GET"),
            "expected at least one GET to reach upstream, saw {observed:?}"
        );
    }
}