//! The website-facing HTTP surface: a WebSocket live feed, and REST reads backed by the store.
//!
//! Two rules shape everything here, and both are inherited rather than invented:
//!
//! 1. **Authentication is always on.** `/health` is the only unauthenticated route, because
//!    monitoring must be able to reach it. Everything else is behind a shared token that the config
//!    generates on first run, so there is no state in which this process is listening without one.
//! 2. **Every response advertises the protocol version**, and a client that declares a different
//!    one is refused `409` rather than served something it will mis-parse. A version mismatch is a
//!    deployment fault, and it should look like one.
//!
//! The reads are store-backed on purpose. `/server` answers the last thing the game said about
//! itself even while the game is off, which is what lets the website render a server list during a
//! wipe or a restart. `/status` is the one route that round-trips the plugin, and it is the one
//! route that fails when the game is down — which is the honest answer to "what is it doing *right
//! now*".

use std::sync::atomic::{AtomicI64, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

use axum::{
    extract::ws::{Message, WebSocket, WebSocketUpgrade},
    extract::{Query, Request, State},
    http::{HeaderValue, StatusCode},
    middleware::{self, Next},
    response::{IntoResponse, Response},
    routing::{get, post},
    Json, Router,
};
use serde::Deserialize;
use serde_json::{json, Value};
use tokio::sync::broadcast;
use tracing::{debug, info, warn};

use crate::game::GameHandle;
use crate::rpc::{Rpc, RpcError};
use crate::store::Store;
use crate::PROTOCOL_VERSION;

/// The header every response carries, and the one a client may send to declare its own version.
pub const VERSION_HEADER: &str = "X-RustLink-Version";

/// Where the live feed lives. Named rather than spelled inline because `--print-config` reports it
/// to the installer, and the two must not drift.
pub const WS_PATH: &str = "/ws";

/// Shared state handed to each request handler.
#[derive(Clone)]
pub struct AppState {
    pub events: broadcast::Sender<String>,
    pub game: GameHandle,
    pub rpc: Rpc,
    pub store: Store,
    /// Shared secret the website must present. Always set (config guarantees non-empty).
    pub token: Arc<String>,
    pub started: Instant,
    /// Epoch ms of the last line received from the plugin, 0 if none yet.
    pub last_event: Arc<AtomicI64>,
    /// Frames this build could not file, because they carried no usable `type`. See `/health`.
    pub untyped: Arc<AtomicU64>,
}

pub async fn serve(addr: &str, state: AppState) -> anyhow::Result<()> {
    let protected = Router::new()
        .route(WS_PATH, get(ws_upgrade))
        // The board: the last `server.hello`. Store-backed, so it answers while the game is off.
        .route("/server", get(server_board))
        // Every board at once — what a consumer reads on connect to know the present before it
        // starts following the story.
        .route("/boards", get(boards))
        // Event history, newest first, optionally filtered by kind and wipe. For a human.
        .route("/events", get(events))
        // The ingest cursor: oldest first, strictly after an id. For a consumer that must not miss
        // a row. Deliberately a second route rather than a flag on the first — see `feed`.
        .route("/feed", get(feed))
        // Live: a correlated round trip to the plugin. Fails when the game is down, by design.
        .route("/status", get(status))
        // **The first route on this sidecar that is not a GET** (protocol 3). Everything before it
        // was the website reading what the game had already said; this is the website asking the
        // game a question only the game can answer.
        //
        // It is still a forwarder and holds no authority of its own: it does not mint codes, does
        // not store them, does not know what a website user is, and cannot tell a good code from a
        // bad one. It moves one string to the plugin and one reply back.
        .route("/link/confirm", post(link_confirm))
        // Protocol 4, and the first pair that exists so the website can WRITE to the game (R2).
        //
        // The catalogue is a live read of what this server's loaded plugins have registered — the
        // option source the website's authoring form is built from, so a grant can only name a
        // permission that will actually resolve.
        .route("/permissions/catalogue", get(perm_catalogue))
        // The sync is the website's whole desired permission set for this server. This sidecar
        // reads none of it: it does not know what a group is, which names are managed, or what the
        // plugin will do with any of it. It puts `cmd` and `reqId` on the object and forwards it,
        // exactly as it forwards a link code.
        .route("/permissions/sync", post(perm_sync))
        // Protocol 5 (R18): the plugin's own view of the game host's configuration tree. All
        // three are correlated round trips and all three fail when the game is down, because
        // "what is on that host's disk" has no stale answer worth giving — and, unlike a board,
        // nothing here is ever filed in the store. A config this sidecar cached would be a config
        // an operator edited over SSH and the website then overwrote.
        .route("/config/files", get(config_files))
        .route("/config/file", get(config_file))
        // The only route on this sidecar that causes a WRITE on the game host, and the only one
        // whose reply can take most of the RPC budget: the plugin holds it open across a reload
        // and, at worst, across a rollback as well. See `CONFIG_RELOAD_WINDOW`.
        .route("/config/write", post(config_write))
        // Protocol 8 (the module's PLAN.md §27): an event borrowing a value and giving it back.
        // Three correlated round trips, and the sidecar knows nothing about any of them — not
        // which keys exist, not their bounds, not what a deadline is. The plugin holds the
        // allowlist, the ceiling and the timer; the website holds the ledger. This moves lines.
        .route("/lease", get(lease_list).post(lease_apply))
        .route("/lease/release", post(lease_release))
        .route_layer(middleware::from_fn_with_state(state.clone(), gate));

    let app = Router::new()
        .route("/health", get(health))
        .merge(protected)
        // Every response advertises the protocol version, so a client can notice a mismatch even
        // on /health or on an error response.
        .layer(middleware::from_fn(version_header))
        .with_state(state);

    let listener = tokio::net::TcpListener::bind(addr).await?;
    info!(addr = %listener.local_addr()?, "web server listening");
    axum::serve(listener, app).await?;
    Ok(())
}

// ---- health ----

/// Protocol version, whether the plugin is connected, database reachability, uptime, and when the
/// plugin last sent anything. Unauthenticated, so monitoring can reach it.
async fn health(State(st): State<AppState>) -> impl IntoResponse {
    let plugin = st.game.is_connected().await;
    let db_ok = st.store.ping().await.is_ok();
    let last_ms = st.last_event.load(Ordering::Relaxed);

    let status = if plugin && db_ok { "ok" } else { "degraded" };

    let untyped = st.untyped.load(Ordering::Relaxed);

    Json(json!({
        "status": status,
        "protocol": PROTOCOL_VERSION,
        "plugin_connected": plugin,
        "database": if db_ok { "ok" } else { "error" },
        "uptime": format_uptime(st.started.elapsed()),
        "last_event": iso_ms(last_ms),
        // Non-zero means the plugin is speaking a protocol this build cannot file. It is reported
        // here rather than only in the log because the symptom an operator sees — a website that
        // shows nothing while the game is plainly up — names nothing at all.
        "untyped_frames": untyped,
    }))
}

fn format_uptime(d: Duration) -> String {
    let secs = d.as_secs();
    let (days, hours, mins) = (secs / 86400, (secs % 86400) / 3600, (secs % 3600) / 60);
    if days > 0 {
        format!("{days}d {hours}h")
    } else if hours > 0 {
        format!("{hours}h {mins}m")
    } else {
        format!("{mins}m")
    }
}

fn iso_ms(ms: i64) -> Option<String> {
    if ms <= 0 {
        return None;
    }
    chrono::DateTime::from_timestamp_millis(ms)
        .map(|dt| dt.format("%Y-%m-%dT%H:%M:%SZ").to_string())
}

// ---- reads ----

/// The last `server.hello`, or `204` if the game has never connected.
///
/// `204` rather than `200` with a null: "we have never heard from this server" and "this server
/// reports nothing" are different answers, and a client that cannot tell them apart renders a
/// server that does not exist.
async fn server_board(State(st): State<AppState>) -> Response {
    match st.store.board(crate::store::SERVER_BOARD).await {
        Ok(Some(v)) => Json(v).into_response(),
        Ok(None) => StatusCode::NO_CONTENT.into_response(),
        Err(e) => {
            warn!(error = %e, "server board read failed");
            store_error()
        }
    }
}

/// Every board, keyed by kind. `200` with an empty object when the game has never connected —
/// unlike `/server`, which answers `204`, because "no boards yet" is a complete answer to "give me
/// all of them" and an empty map renders correctly where a `204` has to be special-cased.
async fn boards(State(st): State<AppState>) -> Response {
    match st.store.all_boards().await {
        Ok(map) => Json(json!({ "boards": map })).into_response(),
        Err(e) => {
            warn!(error = %e, "board read failed");
            store_error()
        }
    }
}

#[derive(Debug, Deserialize)]
struct EventsQuery {
    kind: Option<String>,
    wipe: Option<String>,
    limit: Option<i64>,
}

async fn events(State(st): State<AppState>, Query(q): Query<EventsQuery>) -> Response {
    match st
        .store
        .recent(q.kind.as_deref(), q.wipe.as_deref(), q.limit.unwrap_or(50))
        .await
    {
        Ok(rows) => Json(json!({ "events": rows })).into_response(),
        Err(e) => {
            warn!(error = %e, "event read failed");
            store_error()
        }
    }
}

#[derive(Debug, Deserialize)]
struct FeedQuery {
    since: Option<i64>,
    limit: Option<i64>,
}

/// The ingest cursor: everything after `since`, oldest first.
///
/// **Omitting `since` asks where the end is** — it answers with no rows and the current `lastId`,
/// which is what a consumer with no cursor of its own needs. `since=0` is the other question, and
/// the one nobody should ask by accident: replay everything retained. A module installed today
/// against a sidecar that has been running a month wants what happens next, not a fortnight of
/// deaths it has no rollups for, and the difference between those two intentions must not be the
/// difference between typing a parameter and forgetting it.
///
/// The response always carries `lastId`, so a caller advances without inspecting the rows, and
/// `more`, so one that has fallen behind comes straight back rather than waiting out its poll
/// interval.
async fn feed(State(st): State<AppState>, Query(q): Query<FeedQuery>) -> Response {
    let since = match q.since {
        Some(n) => n.max(0),
        None => match st.store.last_event_id().await {
            Ok(id) => {
                return Json(json!({ "items": [], "lastId": id, "more": false })).into_response()
            }
            Err(e) => {
                warn!(error = %e, "feed tail read failed");
                return store_error();
            }
        },
    };

    match st.store.feed(since, q.limit.unwrap_or(200)).await {
        Ok((items, more)) => {
            // The cursor a caller should send next. When the page is empty that is the cursor it
            // sent — never 0, which would silently replay the whole retained history on the next
            // poll of a quiet server.
            let last_id = items.last().map(|i| i.id).unwrap_or(since);
            let rows: Vec<Value> = items.iter().map(|i| i.to_json()).collect();

            Json(json!({ "items": rows, "lastId": last_id, "more": more })).into_response()
        }
        Err(e) => {
            warn!(error = %e, "feed read failed");
            store_error()
        }
    }
}

fn store_error() -> Response {
    (
        StatusCode::INTERNAL_SERVER_ERROR,
        Json(json!({"error": "store read failed"})),
    )
        .into_response()
}

/// A correlated round trip to the plugin: what the server is doing right now.
async fn status(State(st): State<AppState>) -> Response {
    let req_id = st.rpc.next_req_id();
    let command = json!({ "cmd": "server.status", "reqId": req_id });
    respond(st.rpc.call(&st.game, command, &req_id).await)
}

// ---- link confirm ----

/// What the website sends to redeem a link code.
#[derive(serde::Deserialize)]
struct LinkConfirm {
    code: String,
}

/// Longer than any code the plugin mints, short enough that nothing else gets forwarded.
const MAX_CODE_LEN: usize = 32;

/// Redeem a one-time link code the plugin minted for a player who ran `/link`.
///
/// **The sidecar validates nothing here beyond the shape**, deliberately. Only the game server
/// holds the pending codes, and only it knows which Steam id a code belongs to — so the whole of
/// this function is "forward it, hand back what came out". Putting the code table here instead
/// would give the sidecar a credential and an opinion, and it is designed to have neither.
///
/// The reply is whatever the plugin said: `link.ok` carrying a `steamId`, or `link.error` carrying
/// a reason. **Both are 200s.** A refused code is an answer, not a transport failure, and the
/// website needs to tell "that code is wrong" from "the game never replied" to say the right thing
/// to a player. The two transport failures keep their own codes through [`respond`] — `503` when
/// the game is down, `504` when it is up and silent.
async fn link_confirm(State(st): State<AppState>, Json(body): Json<LinkConfirm>) -> Response {
    let code = match usable_code(&body.code) {
        Some(c) => c,
        None => {
            return (
                StatusCode::BAD_REQUEST,
                Json(json!({"error": "a link code is required"})),
            )
                .into_response()
        }
    };

    let req_id = st.rpc.next_req_id();
    let command = json!({ "cmd": "link.confirm", "reqId": req_id, "code": code });
    respond(st.rpc.call(&st.game, command, &req_id).await)
}

/// The submitted code, trimmed, or `None` when there is nothing worth forwarding.
///
/// Split out so it can be tested without an [`AppState`], and because the two rejections are
/// easy to get subtly wrong. **Trim first, then measure**: a player pasting a code out of the
/// game chat brings whitespace with it, and a field of nothing but spaces is empty rather than
/// four characters long. The length bound is on the trimmed value for the same reason.
fn usable_code(raw: &str) -> Option<&str> {
    let code = raw.trim();

    if code.is_empty() || code.len() > MAX_CODE_LEN {
        return None;
    }

    Some(code)
}

// ---- permissions (protocol 4) ----

/// The largest command this sidecar will put on the game link.
///
/// Both ends discard an over-long line rather than buffering it (PROTOCOL.md §3.1), and a discarded
/// command is indistinguishable from a plugin that never answered: the caller waits out its whole
/// timeout and is told `504`, which sends an operator to look at the game server. Refusing here
/// costs one comparison and says which limit was actually hit.
///
/// It is the game link's cap rather than a smaller number of our own, because the line this
/// function builds is the line that has to fit.
const MAX_COMMAND_BYTES: usize = 1024 * 1024;

/// What this server's loaded plugins have registered, and the groups its store holds.
async fn perm_catalogue(State(st): State<AppState>) -> Response {
    let req_id = st.rpc.next_req_id();
    let command = json!({ "cmd": "perm.catalogue", "reqId": req_id });
    respond(st.rpc.call(&st.game, command, &req_id).await)
}

/// Forward the website's desired permission set to the plugin, and hand back its report.
///
/// **The body is opaque here.** The sidecar defines no schema for a frame's contents — that is the
/// dumb-forwarder property this whole bridge is built on, and it is why protocol 4 adds a shape
/// this large without touching the store or the feed. What this function does own is the envelope:
/// `cmd` and `reqId` are inserted **after** the caller's object is taken, so they overwrite
/// anything a caller put there and no request can arrive claiming to be a different command.
async fn perm_sync(State(st): State<AppState>, Json(body): Json<Value>) -> Response {
    let mut command = match body {
        Value::Object(map) => map,
        _ => {
            return (
                StatusCode::BAD_REQUEST,
                Json(json!({"error": "a permission set must be an object"})),
            )
                .into_response()
        }
    };

    let req_id = st.rpc.next_req_id();
    command.insert("cmd".into(), Value::String("perm.sync".into()));
    command.insert("reqId".into(), Value::String(req_id.clone()));

    let command = Value::Object(command);
    let encoded = command.to_string();

    if encoded.len() > MAX_COMMAND_BYTES {
        warn!(bytes = encoded.len(), "permission sync refused: too large");
        return (
            StatusCode::PAYLOAD_TOO_LARGE,
            Json(json!({
                "error": "the permission set is larger than the game link will carry",
                "bytes": encoded.len(),
                "limit": MAX_COMMAND_BYTES,
            })),
        )
            .into_response();
    }

    respond(st.rpc.call(&st.game, command, &req_id).await)
}

#[derive(Debug, Deserialize)]
struct ConfigPathQuery {
    path: String,
}

/// What `GET /lease` may narrow to. Both optional, both forwarded as they are.
#[derive(Debug, Deserialize)]
struct LeaseQuery {
    key: Option<String>,
    target: Option<String>,
}

/// Everything this server lends, what it holds now, and every hold in force (protocol 8).
///
/// A live round trip, never a cached board, for the reason `/permissions/catalogue` is one: the
/// website asks this immediately before taking a lease, to record the baseline it will later give
/// back, and a baseline that was a minute stale would be restored over whatever happened in that
/// minute.
async fn lease_list(State(st): State<AppState>, Query(q): Query<LeaseQuery>) -> Response {
    let req_id = st.rpc.next_req_id();
    let mut command = json!({ "cmd": "lease.list", "reqId": req_id });

    if let Some(key) = q.key {
        command["key"] = Value::String(key);
    }
    if let Some(target) = q.target {
        command["target"] = Value::String(target);
    }

    respond(st.rpc.call(&st.game, command, &req_id).await)
}

/// Take a value for a while. The plugin answers `lease.ok` or `lease.error` with a reason.
async fn lease_apply(State(st): State<AppState>, Json(body): Json<Value>) -> Response {
    forward_object(&st, body, "lease.apply", "a lease").await
}

/// Give a value back. **A drifted value is a `200` carrying `lease.drifted`**, not an error: the
/// plugin did exactly what it was asked — it compared, and declined to overwrite somebody's
/// deliberate change — and the website records that as its own outcome.
async fn lease_release(State(st): State<AppState>, Json(body): Json<Value>) -> Response {
    forward_object(&st, body, "lease.release", "a lease release").await
}

/// Put `cmd` and `reqId` on a caller's object, **after** it is taken, so they overwrite anything
/// the caller put there. `None` when the body is not an object.
fn stamp(body: Value, cmd: &str, req_id: &str) -> Option<Value> {
    let mut map = match body {
        Value::Object(map) => map,
        _ => return None,
    };

    map.insert("cmd".into(), Value::String(cmd.into()));
    map.insert("reqId".into(), Value::String(req_id.into()));
    Some(Value::Object(map))
}

/// The opaque-object forward the lease routes share: stamp the envelope, refuse what the game
/// link cannot carry, and hand back whatever the plugin answered.
async fn forward_object(st: &AppState, body: Value, cmd: &str, what: &str) -> Response {
    let req_id = st.rpc.next_req_id();

    let command = match stamp(body, cmd, &req_id) {
        Some(command) => command,
        None => {
            return (
                StatusCode::BAD_REQUEST,
                Json(json!({ "error": format!("{what} must be an object") })),
            )
                .into_response()
        }
    };

    let encoded = command.to_string();

    if encoded.len() > MAX_COMMAND_BYTES {
        warn!(bytes = encoded.len(), cmd, "command refused: too large");
        return (
            StatusCode::PAYLOAD_TOO_LARGE,
            Json(json!({
                "error": format!("{what} is larger than the game link will carry"),
                "bytes": encoded.len(),
                "limit": MAX_COMMAND_BYTES,
            })),
        )
            .into_response();
    }

    respond(st.rpc.call(&st.game, command, &req_id).await)
}

/// Every settings file on the game host, and every plugin loaded to reload one.
///
/// The sidecar knows nothing about either: not where the tree is (the framework decides, and it is
/// `oxide/config` on one framework and `carbon/configs` on the other), not which files are
/// editable, not what a plugin is called. It asks and forwards.
async fn config_files(State(st): State<AppState>) -> Response {
    let req_id = st.rpc.next_req_id();
    let command = json!({ "cmd": "config.list", "reqId": req_id });
    respond(st.rpc.call(&st.game, command, &req_id).await)
}

/// One settings file, as text, with the version a write must present back.
///
/// **The path is forwarded verbatim and judged at the far end.** That is not laziness: the only
/// process that can decide whether a path resolves inside the configuration directory is the one
/// that has the directory and the filesystem. A guard here would be a second, weaker opinion in a
/// place with no way to check it, and the kind of guard that makes the real one feel optional.
async fn config_file(State(st): State<AppState>, Query(q): Query<ConfigPathQuery>) -> Response {
    let req_id = st.rpc.next_req_id();
    let command = json!({ "cmd": "config.read", "reqId": req_id, "path": q.path });
    respond(st.rpc.call(&st.game, command, &req_id).await)
}

/// The window the plugin gives a reload to announce itself, mirrored here from
/// `ConfigReloadWindowSeconds` in `overlay/oxide/plugins/RunicGateway.cs`.
///
/// It is duplicated rather than negotiated because the two numbers are a *pairing*, like the
/// protocol version: the plugin owns the behaviour and this end owns the budget it has to fit in.
/// The test below is what keeps them honest — a worst-case write is two windows, and a
/// [`REPLY_TIMEOUT`](crate::rpc::REPLY_TIMEOUT) that does not cover both would abandon the caller
/// precisely when a rollback had just saved them, leaving the website to report a timeout over a
/// server that is perfectly healthy.
pub const CONFIG_RELOAD_WINDOW: Duration = Duration::from_secs(4);

/// Replace a set of configuration files and reload whatever owns them.
///
/// Opaque body, exactly as `/permissions/sync`: `cmd` and `reqId` are written over whatever the
/// caller sent, and nothing else about the object is read here. What the website sends is whole
/// file *text* rather than a key and a value (D35), so there is nothing on this hop that could
/// misunderstand a number — which is the point of composing the bytes on the side that will be
/// blamed for them.
async fn config_write(State(st): State<AppState>, Json(body): Json<Value>) -> Response {
    let mut command = match body {
        Value::Object(map) => map,
        _ => {
            return (
                StatusCode::BAD_REQUEST,
                Json(json!({"error": "a configuration write must be an object"})),
            )
                .into_response()
        }
    };

    let req_id = st.rpc.next_req_id();
    command.insert("cmd".into(), Value::String("config.write".into()));
    command.insert("reqId".into(), Value::String(req_id.clone()));

    let command = Value::Object(command);
    let encoded = command.to_string();

    if encoded.len() > MAX_COMMAND_BYTES {
        warn!(
            bytes = encoded.len(),
            "configuration write refused: too large"
        );
        return (
            StatusCode::PAYLOAD_TOO_LARGE,
            Json(json!({
                "error": "the configuration write is larger than the game link will carry",
                "bytes": encoded.len(),
                "limit": MAX_COMMAND_BYTES,
            })),
        )
            .into_response();
    }

    let result = st.rpc.call(&st.game, command, &req_id).await;

    if matches!(result, Err(RpcError::Timeout)) {
        // A bare `504` on a route that writes reads as "did my change land, and is the plugin
        // still up?" — and unlike every other timeout on this sidecar, that question has a good
        // answer. The plugin writes a whole set or restores a whole set, never half of either, so
        // a re-read settles it; and a write that is still in flight is most likely inside the
        // rollback this window pays for.
        return (
            StatusCode::GATEWAY_TIMEOUT,
            Json(json!({
                "error": "the plugin did not report within the write budget",
                "reloadWindowSeconds": CONFIG_RELOAD_WINDOW.as_secs(),
                "hint": "re-read the files: the plugin writes the whole set or restores it",
            })),
        )
            .into_response();
    }

    respond(result)
}

/// Maps an RPC outcome onto a status code.
///
/// The two failures are deliberately different codes. `NoPlugin` is `503`: the game is down and the
/// caller should stop asking for a moment. `Timeout` is `504`: the game is up and did not answer in
/// time, which is a different operational problem with a different fix.
fn respond(result: Result<Value, RpcError>) -> Response {
    match result {
        Ok(v) => Json(v).into_response(),
        Err(RpcError::NoPlugin) => (
            StatusCode::SERVICE_UNAVAILABLE,
            Json(json!({"error": "no plugin connected"})),
        )
            .into_response(),
        Err(RpcError::Timeout) => (
            StatusCode::GATEWAY_TIMEOUT,
            Json(json!({"error": "the plugin did not reply in time"})),
        )
            .into_response(),
    }
}

// ---- gate: protocol check + auth ----

/// Adds the version header to every response.
async fn version_header(req: Request, next: Next) -> Response {
    let mut resp = next.run(req).await;
    if let Ok(v) = HeaderValue::from_str(&PROTOCOL_VERSION.to_string()) {
        resp.headers_mut().insert(VERSION_HEADER, v);
    }
    resp
}

/// Guards every non-health route: first a protocol-version check (if the client declares one), then
/// authentication. The token may arrive as `Authorization: Bearer <t>`, `X-Api-Key: <t>`, or
/// `?token=<t>` (the last so browser WebSocket clients, which cannot set handshake headers, still
/// authenticate). The token compare is constant-time.
async fn gate(State(st): State<AppState>, req: Request, next: Next) -> Response {
    // Protocol version: if the client states one and it disagrees, fail loudly and specifically.
    if let Some(v) = req
        .headers()
        .get(VERSION_HEADER.to_ascii_lowercase().as_str())
        .and_then(|h| h.to_str().ok())
    {
        if v.trim() != PROTOCOL_VERSION.to_string() {
            return (
                StatusCode::CONFLICT,
                Json(json!({
                    "error": "protocol version mismatch",
                    "sidecar_protocol": PROTOCOL_VERSION,
                    "client_protocol": v.trim(),
                })),
            )
                .into_response();
        }
    }

    match extract_token(&req) {
        Some(t) if constant_time_eq(t.as_bytes(), st.token.as_bytes()) => next.run(req).await,
        _ => (
            StatusCode::UNAUTHORIZED,
            Json(json!({"error": "missing or invalid auth token"})),
        )
            .into_response(),
    }
}

fn extract_token(req: &Request) -> Option<String> {
    // Authorization: Bearer <token>
    if let Some(v) = req
        .headers()
        .get("authorization")
        .and_then(|h| h.to_str().ok())
    {
        if let Some(rest) = v
            .strip_prefix("Bearer ")
            .or_else(|| v.strip_prefix("bearer "))
        {
            return Some(rest.trim().to_string());
        }
    }
    // X-Api-Key: <token>
    if let Some(v) = req.headers().get("x-api-key").and_then(|h| h.to_str().ok()) {
        return Some(v.trim().to_string());
    }
    // ?token=<token>
    if let Some(q) = req.uri().query() {
        for pair in q.split('&') {
            if let Some(v) = pair.strip_prefix("token=") {
                return Some(v.to_string());
            }
        }
    }
    None
}

/// Length-independent, early-return-free comparison, so a wrong token leaks no timing signal.
fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
    if a.len() != b.len() {
        return false;
    }
    let mut diff = 0u8;
    for (x, y) in a.iter().zip(b.iter()) {
        diff |= x ^ y;
    }
    diff == 0
}

// ---- live feed ----

async fn ws_upgrade(ws: WebSocketUpgrade, State(state): State<AppState>) -> impl IntoResponse {
    ws.on_upgrade(move |socket| ws_client(socket, state))
}

async fn ws_client(mut socket: WebSocket, state: AppState) {
    let mut rx = state.events.subscribe();
    info!("ws client connected");

    let hello = json!({"kind": "ws.hello", "protocol": PROTOCOL_VERSION}).to_string();
    if socket.send(Message::Text(hello)).await.is_err() {
        return;
    }

    loop {
        tokio::select! {
            recv = rx.recv() => {
                match recv {
                    Ok(line) => {
                        if socket.send(Message::Text(line)).await.is_err() {
                            break;
                        }
                    }
                    Err(broadcast::error::RecvError::Lagged(n)) => {
                        warn!(skipped = n, "ws client lagged; dropping missed events");
                    }
                    Err(broadcast::error::RecvError::Closed) => break,
                }
            }
            msg = socket.recv() => {
                match msg {
                    Some(Ok(Message::Close(_))) | None => break,
                    Some(Ok(Message::Ping(p))) => { let _ = socket.send(Message::Pong(p)).await; }
                    Some(Ok(other)) => debug!(?other, "ws client message ignored"),
                    Some(Err(e)) => { debug!(error = %e, "ws client error"); break; }
                }
            }
            _ = tokio::time::sleep(Duration::from_secs(30)) => {
                if socket.send(Message::Ping(Vec::new())).await.is_err() {
                    break;
                }
            }
        }
    }

    info!("ws client disconnected");
}

#[cfg(test)]
mod tests {
    use super::*;
    use axum::body::Body;

    fn req_with(headers: &[(&str, &str)], uri: &str) -> Request {
        let mut b = Request::builder().uri(uri);
        for (k, v) in headers {
            b = b.header(*k, *v);
        }
        b.body(Body::empty()).unwrap()
    }

    #[test]
    fn the_token_is_read_from_all_three_places() {
        assert_eq!(
            extract_token(&req_with(&[("authorization", "Bearer abc")], "/events")).as_deref(),
            Some("abc")
        );
        assert_eq!(
            extract_token(&req_with(&[("authorization", "bearer abc")], "/events")).as_deref(),
            Some("abc")
        );
        assert_eq!(
            extract_token(&req_with(&[("x-api-key", "abc")], "/events")).as_deref(),
            Some("abc")
        );
        // The query form exists for browser WebSocket clients, which cannot set a handshake header.
        assert_eq!(
            extract_token(&req_with(&[], "/ws?token=abc")).as_deref(),
            Some("abc")
        );
        assert_eq!(extract_token(&req_with(&[], "/events")), None);
    }

    /// `Authorization: abc` with no scheme is not a token. Accepting it would make the header's
    /// grammar optional, and a caller that got it wrong would work here and nowhere else.
    #[test]
    fn a_bare_authorization_value_is_not_a_token() {
        assert_eq!(
            extract_token(&req_with(&[("authorization", "abc")], "/x")),
            None
        );
    }

    #[test]
    fn constant_time_eq_still_compares_correctly() {
        assert!(constant_time_eq(b"abc", b"abc"));
        assert!(!constant_time_eq(b"abc", b"abd"));
        assert!(!constant_time_eq(b"abc", b"abcd"));
        assert!(constant_time_eq(b"", b""));
    }

    /// The two RPC failures must not collapse into one code: "the game is down" and "the game is up
    /// and slow" have different fixes, and the website's client branches on the status.
    #[test]
    fn a_submitted_code_is_trimmed_before_it_is_judged() {
        // A player pastes out of game chat and brings whitespace with them. Trimming after the
        // length check would forward the padding; checking emptiness before trimming would accept
        // a field of spaces and send the plugin nothing to look up.
        assert_eq!(usable_code("  ABC123 "), Some("ABC123"));
        assert_eq!(usable_code("ABC123"), Some("ABC123"));
        assert_eq!(usable_code("   "), None);
        assert_eq!(usable_code(""), None);
    }

    #[test]
    fn a_code_longer_than_any_the_plugin_mints_is_refused_here() {
        // The plugin's alphabet is six characters. A caller sending a megabyte does not have a
        // code, and the game link should never carry the attempt.
        let long = "A".repeat(MAX_CODE_LEN + 1);
        assert_eq!(usable_code(&long), None);

        let at_bound = "A".repeat(MAX_CODE_LEN);
        assert_eq!(usable_code(&at_bound), Some(at_bound.as_str()));
    }

    #[test]
    fn rpc_failures_map_to_distinct_codes() {
        assert_eq!(
            respond(Err(RpcError::NoPlugin)).status(),
            StatusCode::SERVICE_UNAVAILABLE
        );
        assert_eq!(
            respond(Err(RpcError::Timeout)).status(),
            StatusCode::GATEWAY_TIMEOUT
        );
        assert_eq!(respond(Ok(json!({"ok": true}))).status(), StatusCode::OK);
    }

    /// The envelope is this side's, not the caller's. A website that sent `cmd` or `reqId` in its
    /// own body must not be able to aim the forwarded line at a different command, or at a
    /// correlation id somebody else is waiting on.
    #[test]
    fn a_synced_body_cannot_choose_its_own_command_or_correlation_id() {
        let body = json!({ "cmd": "link.confirm", "reqId": "r-1", "grants": [] });

        let mut command = match body {
            Value::Object(map) => map,
            _ => unreachable!(),
        };

        // The two lines `perm_sync` runs, in its order.
        command.insert("cmd".into(), Value::String("perm.sync".into()));
        command.insert("reqId".into(), Value::String("r-42".into()));

        assert_eq!(command["cmd"], json!("perm.sync"));
        assert_eq!(command["reqId"], json!("r-42"));
        // And everything the caller actually meant is still there, untouched and unread.
        assert_eq!(command["grants"], json!([]));
    }

    /// A command larger than the game link's own line cap is refused here, where the caller learns
    /// why. Forwarded, it would be discarded by both ends without a word and present as a `504`.
    /// The pairing in `CONFIG_RELOAD_WINDOW`'s own words, asserted rather than trusted.
    ///
    /// The worst path through one configuration write is two windows — wait for the edited
    /// plugin's reload, give up, restore the files, reload again — plus the write, the reads and
    /// a line each way. If that does not fit inside the RPC timeout, the caller is abandoned at
    /// exactly the moment the rollback saved it, and the website reports a timeout over a server
    /// that is healthy and has the old config back.
    #[test]
    fn a_rollback_fits_inside_the_rpc_budget() {
        let worst = CONFIG_RELOAD_WINDOW * 2;
        assert!(
            worst + Duration::from_secs(1) <= crate::rpc::REPLY_TIMEOUT,
            "two reload windows ({worst:?}) plus a second of slack must fit in {:?}",
            crate::rpc::REPLY_TIMEOUT
        );
    }

    /// Protocol 8's routes cannot choose their own command or correlation id either, and they keep
    /// everything else the caller sent, unread.
    #[test]
    fn a_lease_cannot_choose_its_own_command() {
        let body =
            json!({ "cmd": "config.write", "reqId": "r-1", "key": "decay.scale", "value": "0" });

        let command = stamp(body, "lease.apply", "r-9").expect("an object is stamped");

        assert_eq!(command["cmd"], json!("lease.apply"));
        assert_eq!(command["reqId"], json!("r-9"));
        assert_eq!(command["key"], json!("decay.scale"));
        assert_eq!(command["value"], json!("0"));
    }

    #[test]
    fn a_lease_that_is_not_an_object_is_not_stamped() {
        assert!(stamp(json!(["decay.scale"]), "lease.apply", "r-1").is_none());
        assert!(stamp(json!("decay.scale"), "lease.release", "r-1").is_none());
    }

    /// The same envelope rule as `perm_sync`, on the route that writes to a filesystem.
    #[test]
    fn a_configuration_write_cannot_choose_its_own_command() {
        let body = json!({ "cmd": "perm.sync", "reqId": "r-1", "files": [], "reload": "Kits" });

        let mut command = match body {
            Value::Object(map) => map,
            _ => unreachable!(),
        };

        command.insert("cmd".into(), Value::String("config.write".into()));
        command.insert("reqId".into(), Value::String("r-7".into()));

        assert_eq!(command["cmd"], json!("config.write"));
        assert_eq!(command["reqId"], json!("r-7"));
        assert_eq!(command["reload"], json!("Kits"));
    }

    #[test]
    fn the_command_cap_is_the_game_links_line_cap() {
        assert_eq!(MAX_COMMAND_BYTES, 1024 * 1024);

        let big = json!({ "note": "x".repeat(MAX_COMMAND_BYTES) }).to_string();
        assert!(big.len() > MAX_COMMAND_BYTES);

        let small = json!({ "grants": [] }).to_string();
        assert!(small.len() <= MAX_COMMAND_BYTES);
    }

    #[test]
    fn uptime_reads_as_a_human_would_write_it() {
        assert_eq!(format_uptime(Duration::from_secs(90)), "1m");
        assert_eq!(format_uptime(Duration::from_secs(3 * 3600 + 120)), "3h 2m");
        assert_eq!(
            format_uptime(Duration::from_secs(2 * 86400 + 3600)),
            "2d 1h"
        );
    }

    /// A zero timestamp means "never", and must not render as 1970.
    #[test]
    fn a_never_timestamp_is_none_rather_than_the_epoch() {
        assert_eq!(iso_ms(0), None);
        assert_eq!(iso_ms(-1), None);
        assert!(iso_ms(1_757_000_000_000).unwrap().starts_with("2025-"));
    }
}