link/sidecar/src/web.rs

//! The website-facing HTTP surface: a WebSocket live feed and REST queries.
//!
//! Unlike the shard link, this side *may* be exposed beyond loopback — it is the website's entry
//! point, and the sidecar is the gatekeeper. It defaults to loopback anyway; widen the bind address
//! deliberately (and add auth) when the website runs on another host.

use std::time::Duration;

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

use axum::{
    extract::ws::{Message, WebSocket, WebSocketUpgrade},
    extract::{Path, 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::rpc::{Rpc, RpcError};
use crate::shard::ShardHandle;
use crate::store::Store;
use crate::PROTOCOL_VERSION;

/// Shared state handed to each request handler.
#[derive(Clone)]
pub struct AppState {
    pub events: broadcast::Sender<String>,
    pub shard: ShardHandle,
    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 shard, 0 if none yet.
    pub last_event: Arc<AtomicI64>,
}

pub async fn serve(addr: &str, state: AppState) -> anyhow::Result<()> {
    // Everything except /health is behind the auth check.
    let protected = Router::new()
        .route("/ws", get(ws_upgrade))
        // Queries (shard reply correlated by reqId).
        .route("/char/:account/:slot", get(char_by_slot))
        .route("/char/serial/:serial", get(char_by_serial))
        .route("/roster/:account", get(roster))
        .route("/vendors/:account", get(vendors))
        // Inbound commands (correlated by code / id).
        .route("/link/confirm", post(link_confirm))
        // Account provisioning (Protocol 2.0). Create is correlated by reqId; the DELETE unlinks.
        .route("/accounts/create", post(account_create))
        .route("/link/:account", get(link_lookup).delete(link_delete))
        .route("/towncrier", post(towncrier_add))
        .route("/towncrier/:id", axum::routing::delete(towncrier_remove))
        // Staff write plane (correlated by reqId). The shard enforces the real authorization;
        // the website must gate these behind admin/moderator roles before calling.
        .route("/admin/kick", post(admin_kick))
        .route("/admin/ban", post(admin_ban))
        .route("/admin/unban", post(admin_unban))
        .route("/admin/broadcast", post(admin_broadcast))
        // Help-page (support) queue: snapshot the open queue, respond to / close a page.
        .route("/pages", get(pages_list))
        .route("/pages/:id/respond", post(page_respond))
        .route("/pages/:id/close", post(page_close))
        // History, read from SQLite rather than the shard.
        .route("/history", get(history))
        .route("/economy", get(economy))
        .route("/champs", get(champs))
        // World-state boards (Protocol 2.0), served from the store so they answer without the shard
        // and survive an outage with the last-known snapshot (docs/PROTOCOL_2.md §12.2).
        .route("/guilds", get(guilds))
        .route("/governors", get(governors))
        .route("/online", get(online))
        .route_layer(middleware::from_fn_with_state(state.clone(), gate));

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

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

// ---- health ----

/// Rich health: protocol version, whether the shard plugin is connected, database reachability,
/// uptime, and when the shard last sent anything. Unauthenticated, so monitoring can reach it.
async fn health(State(st): State<AppState>) -> impl IntoResponse {
    let plugin = st.shard.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" };

    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),
    }))
}

fn format_uptime(d: std::time::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())
}

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

/// Adds `X-UOLink-Version` 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("X-UOLink-Version", 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 can't 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("x-uolink-version")
        .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();
        }
    }

    let provided = extract_token(&req);
    match provided {
        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
}

// ---- shared reply handling ----

/// Turns an RPC result into an HTTP response. A `bridge.error` reply from the shard becomes a 4xx;
/// a real reply is returned as-is; transport failures map to 503/504.
fn respond(result: Result<Value, RpcError>) -> (StatusCode, Json<Value>) {
    match result {
        Ok(value) => {
            let kind = value.get("kind").and_then(|k| k.as_str()).unwrap_or("");
            if kind == "bridge.error" || kind.ends_with(".error") {
                let reason = value
                    .get("reason")
                    .and_then(|r| r.as_str())
                    .unwrap_or("request rejected");
                // "unknown account" / "no character" are not-founds; the rest are bad requests.
                let code = if reason.contains("unknown") || reason.contains("no ") {
                    StatusCode::NOT_FOUND
                } else {
                    StatusCode::BAD_REQUEST
                };
                (code, Json(value))
            } else {
                (StatusCode::OK, Json(value))
            }
        }
        Err(RpcError::NoShard) => (
            StatusCode::SERVICE_UNAVAILABLE,
            Json(json!({"error": "shard not connected"})),
        ),
        Err(RpcError::Timeout) => (
            StatusCode::GATEWAY_TIMEOUT,
            Json(json!({"error": "shard did not reply in time"})),
        ),
    }
}

/// Like `respond`, but for the admin write plane, where a rejection is not a not-found. Maps an
/// `admin.error` reply to a status by its reason: an unknown target is a 404, a floor/authorization
/// refusal (protected target, or the write plane being disabled) is a 403, anything else a 400.
fn respond_admin(result: Result<Value, RpcError>) -> (StatusCode, Json<Value>) {
    match result {
        Ok(value) => {
            let kind = value.get("kind").and_then(|k| k.as_str()).unwrap_or("");
            if kind == "admin.error" {
                let reason = value
                    .get("reason")
                    .and_then(|r| r.as_str())
                    .unwrap_or("request rejected");
                let code = if reason.contains("unknown") {
                    StatusCode::NOT_FOUND
                } else if reason.contains("protected")
                    || reason.contains("refused")
                    || reason.contains("disabled")
                {
                    StatusCode::FORBIDDEN
                } else {
                    StatusCode::BAD_REQUEST
                };
                (code, Json(value))
            } else {
                (StatusCode::OK, Json(value))
            }
        }
        Err(RpcError::NoShard) => (
            StatusCode::SERVICE_UNAVAILABLE,
            Json(json!({"error": "shard not connected"})),
        ),
        Err(RpcError::Timeout) => (
            StatusCode::GATEWAY_TIMEOUT,
            Json(json!({"error": "shard did not reply in time"})),
        ),
    }
}

/// Like `respond`, but for the account-provisioning plane. Maps an `account.error` reply to a
/// status by its reason: a name clash is a 409, the per-IP cap is a 429, a disabled/protected/
/// refused action is a 403, an unknown target or "not linked" is a 404, anything else a 400.
fn respond_account(result: Result<Value, RpcError>) -> (StatusCode, Json<Value>) {
    match result {
        Ok(value) => {
            let kind = value.get("kind").and_then(|k| k.as_str()).unwrap_or("");
            if kind == "account.error" {
                let reason = value
                    .get("reason")
                    .and_then(|r| r.as_str())
                    .unwrap_or("request rejected");
                let code = if reason.contains("already exists") {
                    StatusCode::CONFLICT
                } else if reason.contains("ip account limit") {
                    StatusCode::TOO_MANY_REQUESTS
                } else if reason.contains("disabled")
                    || reason.contains("protected")
                    || reason.contains("refused")
                {
                    StatusCode::FORBIDDEN
                } else if reason.contains("unknown") || reason.contains("not linked") {
                    StatusCode::NOT_FOUND
                } else {
                    StatusCode::BAD_REQUEST
                };
                (code, Json(value))
            } else {
                (StatusCode::OK, Json(value))
            }
        }
        Err(RpcError::NoShard) => (
            StatusCode::SERVICE_UNAVAILABLE,
            Json(json!({"error": "shard not connected"})),
        ),
        Err(RpcError::Timeout) => (
            StatusCode::GATEWAY_TIMEOUT,
            Json(json!({"error": "shard did not reply in time"})),
        ),
    }
}

// ---- account-provisioning handlers ----

/// Body: {"actor","account","password","websiteUserId","ip"}. Creates and links a game account.
/// Correlated on a fresh reqId. The password is forwarded to the shard (loopback) but never logged
/// here and never appears in the reply; a successful create mirrors the link into the store.
async fn account_create(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
    let mut obj = match body {
        Value::Object(m) => m,
        _ => {
            return (
                StatusCode::BAD_REQUEST,
                Json(json!({"error": "body must be a JSON object"})),
            )
        }
    };

    // Required, non-empty. `ip` is validated on the shard (which owns the cap), not here.
    for field in ["actor", "account", "password", "websiteUserId"] {
        let present = obj
            .get(field)
            .and_then(|v| v.as_str())
            .map(|s| !s.trim().is_empty())
            .unwrap_or(false);
        if !present {
            return (
                StatusCode::BAD_REQUEST,
                Json(json!({ "error": format!("{field} is required") })),
            );
        }
    }

    let req_id = st.rpc.next_req_id();
    obj.insert("kind".to_string(), json!("account.create"));
    obj.insert("reqId".to_string(), json!(req_id));

    let result = st.rpc.call(&st.shard, Value::Object(obj), &req_id).await;

    // Mirror a successful create's link into the store, so events are attributable without the
    // shard (same as link.confirm does).
    if let Ok(value) = &result {
        if value.get("kind").and_then(|k| k.as_str()) == Some("account.ok") {
            if let (Some(account), Some(web_id)) = (
                value.get("account").and_then(|a| a.as_str()),
                value.get("websiteUserId").and_then(|w| w.as_str()),
            ) {
                let t = value.get("t").and_then(|v| v.as_i64()).unwrap_or(0);
                let _ = st.store.record_link(account, web_id, t).await;
            }
        }
    }

    respond_account(result)
}

/// Unlinks a game account from its website user. Body: {"actor"}. Correlated on reqId; a success
/// also clears the sidecar's mirrored link row so attribution stops immediately.
async fn link_delete(
    State(st): State<AppState>,
    Path(account): Path<String>,
    body: Option<Json<Value>>,
) -> impl IntoResponse {
    let actor = body
        .as_ref()
        .and_then(|Json(b)| b.get("actor").and_then(|a| a.as_str()))
        .unwrap_or_default()
        .trim()
        .to_string();

    if actor.is_empty() {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({"error": "actor is required"})),
        );
    }

    let req_id = st.rpc.next_req_id();
    let cmd = json!({
        "kind": "account.unlink", "reqId": req_id, "actor": actor, "account": account
    });
    let result = st.rpc.call(&st.shard, cmd, &req_id).await;

    if let Ok(value) = &result {
        if value.get("kind").and_then(|k| k.as_str()) == Some("account.ok") {
            let _ = st.store.record_unlink(&account).await;
        }
    }

    respond_account(result)
}

// ---- admin write-plane handlers ----

/// Forwards a staff moderation command to the shard, correlated on a fresh reqId. Injects `kind`
/// and `reqId`, requiring the caller-supplied `actor` up front (the shard enforces it too). The
/// body's remaining fields (account/serial/durationSec/reason/text/hue) pass straight through.
async fn admin_call(st: &AppState, kind: &str, body: Value) -> (StatusCode, Json<Value>) {
    let mut obj = match body {
        Value::Object(m) => m,
        _ => {
            return (
                StatusCode::BAD_REQUEST,
                Json(json!({"error": "body must be a JSON object"})),
            )
        }
    };

    let actor_ok = obj
        .get("actor")
        .and_then(|a| a.as_str())
        .map(|s| !s.trim().is_empty())
        .unwrap_or(false);
    if !actor_ok {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({"error": "actor is required"})),
        );
    }

    let req_id = st.rpc.next_req_id();
    obj.insert("kind".to_string(), json!(kind));
    obj.insert("reqId".to_string(), json!(req_id));

    respond_admin(st.rpc.call(&st.shard, Value::Object(obj), &req_id).await)
}

/// Body: {"actor":"...","account":"..."|"serial":"0x.."}. Disconnects the target's live sessions.
async fn admin_kick(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
    admin_call(&st, "admin.kick", body).await
}

/// Body: {"actor":"...","account":"...","durationSec":<opt>,"reason":<opt>}. 0/absent = indefinite.
async fn admin_ban(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
    admin_call(&st, "admin.ban", body).await
}

/// Body: {"actor":"...","account":"..."}.
async fn admin_unban(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
    admin_call(&st, "admin.unban", body).await
}

/// Body: {"actor":"...","text":"...","hue":<opt>}. Announces a system message to everyone online.
async fn admin_broadcast(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
    admin_call(&st, "admin.broadcast", body).await
}

// ---- help-page queue handlers ----

/// The open help-page queue, correlated on reqId. Returns a pages.list.
async fn pages_list(State(st): State<AppState>) -> impl IntoResponse {
    let req_id = st.rpc.next_req_id();
    let cmd = json!({"kind": "pages.snapshot", "reqId": req_id});
    respond(st.rpc.call(&st.shard, cmd, &req_id).await)
}

/// Body: {"message":"...","close":<bool, optional>}. Delivers a staff response to the player.
async fn page_respond(
    State(st): State<AppState>,
    Path(id): Path<String>,
    Json(body): Json<Value>,
) -> impl IntoResponse {
    let message = body
        .get("message")
        .and_then(|m| m.as_str())
        .unwrap_or_default();
    if message.trim().is_empty() {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({"error": "message is required"})),
        );
    }
    let close = body.get("close").and_then(|c| c.as_bool()).unwrap_or(false);

    let req_id = st.rpc.next_req_id();
    let cmd = json!({
        "kind": "page.respond", "reqId": req_id,
        "pageId": id, "message": message, "close": close
    });
    respond(st.rpc.call(&st.shard, cmd, &req_id).await)
}

/// Removes a page from the queue.
async fn page_close(State(st): State<AppState>, Path(id): Path<String>) -> impl IntoResponse {
    let req_id = st.rpc.next_req_id();
    let cmd = json!({"kind": "page.close", "reqId": req_id, "pageId": id});
    respond(st.rpc.call(&st.shard, cmd, &req_id).await)
}

// ---- query handlers ----

async fn char_by_slot(
    State(st): State<AppState>,
    Path((account, slot)): Path<(String, i64)>,
) -> impl IntoResponse {
    let req_id = st.rpc.next_req_id();
    let cmd = json!({"kind":"char.request","reqId":req_id,"account":account,"slot":slot});
    char_reply(&st, st.rpc.call(&st.shard, cmd, &req_id).await).await
}

async fn char_by_serial(
    State(st): State<AppState>,
    Path(serial): Path<String>,
) -> impl IntoResponse {
    let req_id = st.rpc.next_req_id();
    let cmd = json!({"kind":"char.request","reqId":req_id,"serial":serial.clone()});
    let result = st.rpc.call(&st.shard, cmd, &req_id).await;

    // Resilience: if the shard is unreachable, serve the last cached profile if we have one.
    if matches!(result, Err(RpcError::NoShard) | Err(RpcError::Timeout)) {
        if let Ok(Some(cached)) = st.store.get_cached_profile(&serial).await {
            return (StatusCode::OK, Json(cached));
        }
    }
    char_reply(&st, result).await
}

/// Caches a successful profile before responding, so a later shard outage can still serve it.
async fn char_reply(st: &AppState, result: Result<Value, RpcError>) -> (StatusCode, Json<Value>) {
    if let Ok(value) = &result {
        if value.get("kind").and_then(|k| k.as_str()) == Some("char.profile") {
            if let Some(serial) = value.get("serial").and_then(|s| s.as_str()) {
                let t = value.get("t").and_then(|v| v.as_i64()).unwrap_or(0);
                let _ = st
                    .store
                    .cache_profile(
                        serial,
                        value.get("acct").and_then(|a| a.as_str()),
                        value.get("name").and_then(|n| n.as_str()),
                        &value.to_string(),
                        t,
                    )
                    .await;
            }
        }
    }
    respond(result)
}

async fn roster(State(st): State<AppState>, Path(account): Path<String>) -> impl IntoResponse {
    let req_id = st.rpc.next_req_id();
    let cmd = json!({"kind":"account.roster","reqId":req_id,"account":account});
    respond(st.rpc.call(&st.shard, cmd, &req_id).await)
}

async fn vendors(State(st): State<AppState>, Path(account): Path<String>) -> impl IntoResponse {
    let req_id = st.rpc.next_req_id();
    let cmd = json!({"kind":"vendor.snapshot","reqId":req_id,"account":account});
    respond(st.rpc.call(&st.shard, cmd, &req_id).await)
}

// ---- inbound-command handlers ----

/// Body: {"code":"AB12CD","websiteUserId":"9931"}. Correlated on `code`.
async fn link_confirm(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
    let code = body
        .get("code")
        .and_then(|c| c.as_str())
        .unwrap_or_default();
    let web_id = body
        .get("websiteUserId")
        .and_then(|w| w.as_str())
        .unwrap_or_default();

    if code.is_empty() || web_id.is_empty() {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({"error": "code and websiteUserId are required"})),
        );
    }

    let cmd = json!({"kind":"link.confirm","code":code,"websiteUserId":web_id});
    let code = code.to_string();
    let result = st.rpc.call(&st.shard, cmd, &code).await;

    // Mirror a successful link into the store so events can be attributed without the shard.
    if let Ok(value) = &result {
        if value.get("kind").and_then(|k| k.as_str()) == Some("link.ok") {
            if let Some(account) = value.get("account").and_then(|a| a.as_str()) {
                let t = value.get("t").and_then(|v| v.as_i64()).unwrap_or(0);
                let _ = st.store.record_link(account, web_id, t).await;
            }
        }
    }
    respond(result)
}

async fn link_lookup(State(st): State<AppState>, Path(account): Path<String>) -> impl IntoResponse {
    match st.store.get_link(&account).await {
        Ok(Some(web_id)) => (
            StatusCode::OK,
            Json(json!({"account": account, "websiteUserId": web_id})),
        ),
        Ok(None) => (
            StatusCode::NOT_FOUND,
            Json(json!({"account": account, "linked": false})),
        ),
        Err(e) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(json!({"error": e.to_string()})),
        ),
    }
}

/// Body: {"id":"n123","lines":["..."],"durationSec":3600}. Correlated on `id`.
async fn towncrier_add(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
    let id = body.get("id").and_then(|i| i.as_str()).unwrap_or_default();
    if id.is_empty() {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({"error": "id is required"})),
        );
    }

    let mut cmd = body.clone();
    cmd["kind"] = json!("towncrier.add");
    let id = id.to_string();
    respond(st.rpc.call(&st.shard, cmd, &id).await)
}

async fn towncrier_remove(State(st): State<AppState>, Path(id): Path<String>) -> impl IntoResponse {
    let cmd = json!({"kind":"towncrier.remove","id":id});
    respond(st.rpc.call(&st.shard, cmd, &id).await)
}

// ---- history (from SQLite) ----

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

/// Recent events from the store, newest first. `?kind=vendor.sale&limit=50` to filter.
async fn history(State(st): State<AppState>, Query(q): Query<HistoryQuery>) -> impl IntoResponse {
    let limit = q.limit.unwrap_or(100);
    match st.store.recent(q.kind.as_deref(), limit).await {
        Ok(events) => (StatusCode::OK, Json(json!({"events": events}))),
        Err(e) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(json!({"error": e.to_string()})),
        ),
    }
}

/// The money-supply series (economy.supply snapshots), newest first.
async fn economy(State(st): State<AppState>, Query(q): Query<HistoryQuery>) -> impl IntoResponse {
    let limit = q.limit.unwrap_or(100);
    match st.store.economy(limit).await {
        Ok(series) => (StatusCode::OK, Json(json!({"series": series}))),
        Err(e) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(json!({"error": e.to_string()})),
        ),
    }
}

/// The champion-spawn board: every spawn's latest state (status/level/kills/boss/location and, when
/// relevant, the cooldown ETA). Served from the local board table, so it answers without touching
/// the shard and survives a shard outage with the last-known snapshot.
async fn champs(State(st): State<AppState>) -> impl IntoResponse {
    match st.store.champs_all().await {
        Ok(spawns) => (StatusCode::OK, Json(json!({"spawns": spawns}))),
        Err(e) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(json!({"error": e.to_string()})),
        ),
    }
}

/// The guild board: every guild's latest roster snapshot (id/name/abbr/leader/members/alliance).
/// Served from the local board table, so it hydrates a fresh page or a restarted sidecar without a
/// shard round-trip. The live `guild.*` feed then keeps it current.
async fn guilds(State(st): State<AppState>) -> impl IntoResponse {
    match st.store.guilds_all().await {
        Ok(guilds) => (StatusCode::OK, Json(json!({"guilds": guilds}))),
        Err(e) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(json!({"error": e.to_string()})),
        ),
    }
}

/// The governor board: each city's latest governance snapshot (governor/elect/election phase).
/// Served from the local board table for the same reason as `/guilds`.
async fn governors(State(st): State<AppState>) -> impl IntoResponse {
    match st.store.governors_all().await {
        Ok(cities) => (StatusCode::OK, Json(json!({"cities": cities}))),
        Err(e) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(json!({"error": e.to_string()})),
        ),
    }
}

/// The current online population: total plus per-facet and per-region counts. This is the most
/// recent `presence.online` snapshot from the event store (so it survives a sidecar restart); the
/// live `presence.online` stream keeps it current, and `GET /history?kind=presence.online` gives the
/// population time series. Returns `count: 0` if the shard has not reported one yet.
async fn online(State(st): State<AppState>) -> impl IntoResponse {
    match st.store.recent(Some("presence.online"), 1).await {
        Ok(mut events) => match events.pop() {
            Some(latest) => (StatusCode::OK, Json(latest)),
            None => (
                StatusCode::OK,
                Json(json!({"kind": "presence.online", "count": 0, "byFacet": {}, "byRegion": {}})),
            ),
        },
        Err(e) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(json!({"error": e.to_string()})),
        ),
    }
}

// ---- websocket ----

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");
}