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

/// Path of the live-feed WebSocket. Named because `--print-config` reports it: an installer builds
/// the website's WS URL from `web.bind` plus this, and neither side should be hardcoding it twice.
pub const WS_PATH: &str = "/ws";

pub async fn serve(addr: &str, state: AppState) -> anyhow::Result<()> {
    // Everything except /health is behind the auth check.
    let protected = Router::new()
        .route(WS_PATH, 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))
        // Town Cryer news gump (Protocol 2.1). Add/replace an article; delete one.
        .route("/news", post(news_add))
        .route("/news/:id", axum::routing::delete(news_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))
        // The event plane (protocol 6, EVENTS_PLAN.md Phase 11b). Leases are a live config value
        // the website holds for a bounded time; the shard restores baseline when the deadline
        // passes whether or not anyone asks it to. GET lists the whole catalog with current values,
        // which is the one read both `read()` and `inForce()` on the website's side are served by.
        .route("/lease", get(lease_list).post(lease_apply))
        .route("/lease/release", post(lease_release))
        // The run-scoped participation ledger. `snapshot` is a POST despite being a read: it
        // carries the caller's `idempotencyKey`, and on a well-attended run the shard walks its
        // members across ticks rather than in one inbound call -- so a repeat arriving mid-walk is
        // answered `bridge.busy`, and a read that can be refused as a repeat is not a GET.
        .route("/participation", post(participation_open))
        .route(
            "/participation/:run_id/snapshot",
            post(participation_snapshot),
        )
        .route("/participation/:run_id/close", post(participation_close))
        // The world verbs (protocol 7, EVENTS_PLAN.md Phase 12a). Five things an event author
        // can place -- creatures, an enhanced "boss", an oracle NPC, a temporary gate,
        // decoration -- and ONE command family, because each of them ends in "an object exists
        // and this run owns it". POST places, GET says what the run still owns, POST .../despawn
        // gives it back. Ownership is held on the shard, so despawn cannot be pointed at a serial
        // the run did not create.
        .route("/world", post(world_spawn))
        .route("/world/:run_id", get(world_owned))
        .route("/world/:run_id/despawn", post(world_despawn))
        // The one-shots (protocol 7 part b, EVENTS_PLAN.md Phase 12b). Neither owned nor
        // borrowed: an item put into somebody's hands, and a world save. Both are `done is
        // done`, which is why they are not in the world family -- there is nothing to give
        // back and no ledger row core would come back for.
        //
        // `GET /items` is the shard's own grant allowlist, so the website's dropdown offers
        // what this shard will actually build rather than what a module guessed.
        .route("/items", get(item_catalog))
        .route("/items/grant", post(item_grant))
        .route("/world/save", post(world_save))
        // 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 (https://gitea.whitlocktech.com/RunicGateway/docs/src/branch/main/link/PROTOCOL_2.md §12.2).
        .route("/guilds", get(guilds))
        .route("/governors", get(governors))
        .route("/online", get(online))
        .route("/houses", get(houses))
        // The shard ruleset (Protocol 3.0), likewise store-backed: the shard publishes it once per
        // connect, so serving it from the store is what lets the site's rules page render while the
        // shard is down.
        .route("/ruleset", get(ruleset))
        // Points/loyalty leaderboards (Protocol 3.0), store-backed like the other boards: the
        // whole set, or one system by its PointsType name.
        .route("/points", get(points))
        .route("/points/:system", get(points_system))
        // The player-vendor market index (Protocol 3.0). `/market`, NOT `/vendors`: axum would
        // route the latter fine, but `/vendors/:account` next door is the per-account RPC, and two
        // routes a prefix apart that mean "this player's shops" and "every shop on the shard" is a
        // readability trap nobody wins. The only PAGED read the sidecar serves — a whole-world
        // market does not fit in one response.
        .route("/market", get(market))
        // The Asset Bridge (Protocol 8). Stage 1 of the two-stage import gate: what the shard's
        // UO client files currently are. RPC, never store-backed — unlike the boards above there
        // is nothing here worth serving stale, because the only question this answers is "have
        // the files on that host changed since the last import", and a cached answer to that is
        // worse than no answer.
        .route("/assets/sources", get(assets_sources))
        // The cliloc table (Protocol 8, phase 2): UO's id -> display-string map, read out of the
        // shard's own client and paged. RPC for the same reason as the source gate, and one more:
        // it is five megabytes of somebody else's copyrighted strings, which this process has no
        // business holding a copy of. It forwards them and forgets them.
        .route("/cliloc", get(cliloc_table))
        .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 ----

/// Protocol 6. `bridge.busy` says a command carrying this `idempotencyKey` is already in flight on
/// the shard: nothing was run, and the caller should come back.
///
/// It maps to **425 Too Early**, which is what that status is for — a server unwilling to risk
/// processing a request that might be a replay. The obvious alternative, 409, is already the
/// protocol-version gate's answer, and those two want opposite dispositions from a client: a version
/// mismatch is a deployment fault nobody should retry, and a busy shard is a retry that should
/// succeed on its own. Sharing a status would have made the difference readable only by inspecting
/// the body, which is exactly how a retry loop ends up hiding a mismatched deployment.
///
/// It is checked BEFORE the `.error` suffix test in each responder below, and it is deliberately not
/// spelled `bridge.busy.error`: nothing is wrong. The work is happening.
const BUSY_KIND: &str = "bridge.busy";
const BUSY_STATUS: StatusCode = StatusCode::TOO_EARLY;

/// Turns an RPC result into an HTTP response. A `bridge.error` reply from the shard becomes a 4xx;
/// a `bridge.busy` reply becomes a 425; 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 == BUSY_KIND {
                (BUSY_STATUS, Json(value))
            } else 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 == BUSY_KIND {
                (BUSY_STATUS, Json(value))
            } else 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 event plane: leases and the participation ledger.
///
/// Two mappings are the point of it existing rather than reusing `respond`.
///
/// **`lease.drifted` is a 200.** The shard was asked to compare and set, it compared, and it
/// refused to overwrite somebody's deliberate change -- that is the mechanism working, not a
/// failure, and `cleanup.js` on the website treats `drifted` as a distinct successful outcome
/// rather than an error. It is also why this is not a 409: 409 is the protocol-version gate's, and
/// a version mismatch and a drifted lease want opposite dispositions from a caller. The same
/// argument protocol 6 made for `bridge.busy` being a 425.
///
/// **The event plane being switched off is a 403**, not the 400 the generic responder's
/// reason-sniffing would produce. `Bridge.EventsEnabled` is an operator's deliberate refusal to let
/// the website change the world on a schedule, and telling the website it sent a bad request would
/// send an administrator hunting a bug in a step that is written correctly.
fn respond_event(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 == BUSY_KIND {
                (BUSY_STATUS, Json(value))
            } else if kind.ends_with(".error") {
                let reason = value
                    .get("reason")
                    .and_then(|r| r.as_str())
                    .unwrap_or("request rejected");
                let code = if reason.contains("disabled") {
                    StatusCode::FORBIDDEN
                } else if reason.contains("no lease is offered")
                    || reason.contains("not counting")
                    // Phase 12b. A grant against a run this shard has never been told to count
                    // is the same shape as an unknown lease key: the caller named something that
                    // does not exist here, which is a 404 and never a retry. It is deliberately
                    // NOT the same as a run whose ledger is open and empty -- that is a 200 with
                    // `granted: 0`, because "nobody came" is a result rather than a mistake.
                    || reason.contains("no participation ledger")
                {
                    StatusCode::NOT_FOUND
                } else if reason.contains("saves at most every") {
                    // A save refused because one just happened is the shard's rate limit, and it
                    // is TRANSIENT in a way nothing else on this plane is: the same request will
                    // succeed once the interval passes. 429 says exactly that, and keeps it out of
                    // the module's permanent-status set so a phase boundary is retried rather than
                    // abandoned.
                    StatusCode::TOO_MANY_REQUESTS
                } 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 == BUSY_KIND {
                (BUSY_STATUS, Json(value))
            } else 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.
///
/// **Protocol 6: `idempotencyKey` is one of those remaining fields**, and passing it through is the
/// whole of the sidecar's part in the guarantee. It is worth stating rather than leaving to the
/// word "remaining", because a later refactor that narrowed this to a known field list would quietly
/// turn every retried world write back into a possible duplicate, and nothing here would fail.
///
/// The key belongs to the CALLER's unit of work — the website's event step — so the sidecar neither
/// generates one nor validates it. Note also that `reqId` is regenerated on every call: a retry
/// carries the same idempotency key under a NEW correlation id, which is exactly why the shard
/// re-stamps a replayed reply rather than echoing the id the first attempt used.
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
}

// ---- event plane handlers (protocol 6, Phase 11b) ----

/// Forwards an event-plane command to the shard, correlated on a fresh reqId.
///
/// Deliberately NOT `admin_call`: that one requires an `actor`, because every verb behind it is a
/// staff member pressing a button and the shard's audit trail has to name them. An event verb's
/// author is a RUN, which the body already carries as `runId` -- and demanding an actor here would
/// have the runner inventing a human name for something no human is doing.
///
/// Everything else about it is the same, and the `idempotencyKey` passthrough matters for the same
/// reason it does there: the key is one of the body's remaining fields, and a refactor that
/// narrowed this to a known field list would silently make every retried lease a possible duplicate.
async fn event_call(st: &AppState, kind: &str, body: Value) -> (StatusCode, Json<Value>) {
    let mut obj = match body {
        Value::Object(m) => m,
        Value::Null => serde_json::Map::new(),
        _ => {
            return (
                StatusCode::BAD_REQUEST,
                Json(json!({"error": "body must be a JSON object"})),
            )
        }
    };

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

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

/// Every lease this shard offers, with what each is worth right now and what is holding it.
///
/// One read answers both questions the website asks about a lease: `read()` wants the current value
/// before it applies anything, and `inForce()` wants to know whether the shard still has a record
/// of the hold. Splitting them would be two round trips for one key.
///
/// **`held` means "the shard still has a record of this lease", not "the value is still
/// overridden".** A lease whose deadline has already fired stays listed, with `expired: true`,
/// until teardown collects its verdict -- otherwise a reconcile in that window would report it gone
/// and the website would write off a correctly-working backstop as an orphaned resource.
/// **`?key=` and `?target=` narrow it to one row, and a targeted key needs them** (protocol 7 part
/// b). `Spawner.MaxCount` is one capability over thousands of spawners, so it has no single
/// "current" and the catalog walk cannot fill one in -- while the website's `read()` needs exactly
/// one value for exactly one target before it applies anything. Naming both answers that.
///
/// The frame also carries `holds`: every lease this shard is actually holding, whatever key or
/// target it is on. A catalog walk can enumerate the KEYS but never the holds on a targeted one --
/// there is no list of spawners to walk -- so without it a reconcile after an outage would have no
/// way to ask "what are you still holding?".
async fn lease_list(State(st): State<AppState>, Query(q): Query<LeaseQuery>) -> impl IntoResponse {
    let mut body = serde_json::Map::new();
    if let Some(key) = q.key {
        body.insert("key".to_string(), json!(key));
    }
    if let Some(target) = q.target {
        body.insert("target".to_string(), json!(target));
    }
    let arg = if body.is_empty() {
        Value::Null
    } else {
        Value::Object(body)
    };
    event_call(&st, "lease.list", arg).await
}

/// Narrowing for `GET /lease`. Both optional: absent means the whole catalog, as before.
#[derive(Debug, Deserialize)]
struct LeaseQuery {
    key: Option<String>,
    target: Option<String>,
}

/// Body: {"key":"...","value":"...","holdMs":<ms>,"untilMs":<opt>,"runId":<opt>,"idempotencyKey":<opt>}.
///
/// **`holdMs` is authoritative and `untilMs` is carried for display.** An absolute deadline computed
/// on the website and honoured on the shard is a deadline measured against two clocks, and a shard
/// running ten minutes fast would restore a ten-minute lease the moment it took it. A duration is
/// immune to that; the absolute time is still worth sending so a console can say when the hold ends.
///
/// Values cross as TEXT whatever the lease's declared type, because JSON would otherwise decide for
/// us: `1200` and `1200.0` are one number to a parser and two strings to a compare-and-set.
async fn lease_apply(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
    event_call(&st, "lease.apply", body).await
}

/// Body: {"key":"...","expected":"...","baseline":"...","idempotencyKey":<opt>}.
///
/// `expected` is what the event applied and `baseline` is what to put back, both out of the
/// website's ledger rather than the shard's memory -- so a release still works after a reconnect,
/// and a shard that has forgotten the lease entirely (a restart, which reverts every lease anyway)
/// can answer honestly instead of refusing.
///
/// A mismatch comes back `lease.drifted` with a **200**: see `respond_event`.
async fn lease_release(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
    event_call(&st, "lease.release", body).await
}

/// Body: {"runId":"...","map":"Felucca","x":N,"y":N,"radius":N,"holdMs":<opt>}.
///
/// Declares where a run happens and starts counting who is there. The area is a map, a point and a
/// radius rather than a region name, because protocol 6's own live walk established that the most
/// specific region containing an event is routinely anonymous.
async fn participation_open(
    State(st): State<AppState>,
    Json(body): Json<Value>,
) -> impl IntoResponse {
    event_call(&st, "participation.open", body).await
}

/// Body: {"idempotencyKey":<opt>}. Answers the run's tally, best-effort resolved to accounts.
///
/// A POST for a read, and the reason is worth keeping: on a well-attended run the shard walks its
/// members in chunks across Core ticks rather than handing the whole resolve to one inbound call,
/// so the handler completes after its call returned and a repeat arriving in between is answered
/// `bridge.busy`. A read that can legitimately be refused as a repeat in flight is not a GET.
async fn participation_snapshot(
    State(st): State<AppState>,
    Path(run_id): Path<String>,
    Json(body): Json<Value>,
) -> impl IntoResponse {
    let mut obj = match body {
        Value::Object(m) => m,
        _ => serde_json::Map::new(),
    };
    obj.insert("runId".to_string(), json!(run_id));
    event_call(&st, "participation.snapshot", Value::Object(obj)).await
}

/// Body: {"idempotencyKey":<opt>}. Stops counting; the tally stays readable through the shard's
/// grace window, because closing an event and collecting its results are two steps and either can
/// be retried.
async fn participation_close(
    State(st): State<AppState>,
    Path(run_id): Path<String>,
    Json(body): Json<Value>,
) -> impl IntoResponse {
    let mut obj = match body {
        Value::Object(m) => m,
        _ => serde_json::Map::new(),
    };
    obj.insert("runId".to_string(), json!(run_id));
    event_call(&st, "participation.close", Value::Object(obj)).await
}

/// Body: {"runId":"...","what":"creature|boss|npc|gate|decor","map":"...","x":N,"y":N,...}.
///
/// One route for five author-facing verbs. The `what` discriminator is a wire detail: the
/// differences between them -- a boss's multipliers, an oracle's lines, a gate's destination and
/// `holdMs` -- are fields on one command rather than five commands, so there is one ledger shape,
/// one teardown path and one reconcile instead of five near-identical ones in three repos.
///
/// The shard registers every serial it places against the run and PERSISTS that registry beside
/// the world save, which is what makes `world_despawn` below safe: a spawned creature survives a
/// restart, so an in-memory registry would leave the website holding serials the shard would not
/// vouch for.
async fn world_spawn(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
    event_call(&st, "world.spawn", body).await
}

/// What the run still owns, and the answer the website's `reconcile()` is built on.
///
/// A GET, unlike `participation_snapshot`: it carries no idempotency key and the shard answers it
/// in one pass, pruning rows whose object the world has already lost as it walks. Anything not
/// listed is gone -- which is the shape core wants, because it takes a row out of its ledger only
/// on an explicit reply and this is that reply.
async fn world_owned(State(st): State<AppState>, Path(run_id): Path<String>) -> impl IntoResponse {
    event_call(&st, "world.owned", json!({ "runId": run_id })).await
}

/// Body: {"serials":[...]} -- or no serials at all, which means everything the run owns and is the
/// call teardown actually makes.
///
/// Three answers, and the split is why the shard keeps a registry at all. `removed` was found and
/// deleted; `gone` was owned but already absent, which is what happens when a player kills an event
/// creature and is a SUCCESS; `refused` was never this run's to delete, and is the only answer here
/// that means somebody asked for something they should not have.
async fn world_despawn(
    State(st): State<AppState>,
    Path(run_id): Path<String>,
    Json(body): Json<Value>,
) -> impl IntoResponse {
    let mut obj = match body {
        Value::Object(m) => m,
        _ => serde_json::Map::new(),
    };
    obj.insert("runId".to_string(), json!(run_id));
    event_call(&st, "world.despawn", Value::Object(obj)).await
}

// ---- the one-shots (protocol 7 part b) ----

/// What this shard is willing to grant, and the bounds it will grant within.
///
/// A read, so the website's option source offers what this shard will actually build. The module
/// holds the same list, which is two copies of a short allowlist on purpose and exactly how the
/// lease bounds are already carried: the module's copy is what makes a bad value a refusal on a
/// form, and this one is what is true when the website is wrong.
async fn item_catalog(State(st): State<AppState>) -> impl IntoResponse {
    event_call(&st, "item.catalog", Value::Null).await
}

/// Body: {"runId":"...","item":"gold","amount":N,"hue":<opt>,"name":<opt>,"where":<opt>,"idempotencyKey":<opt>}.
///
/// **The recipients are not in the body, and that is the design.** The shard already holds the
/// run's participation ledger (protocol 6 part b), keyed by the same character serials the
/// website's `member_key` holds, so the grant names a run and the shard resolves who was there.
/// Sending a list would mean the same list crossing the wire twice with a window in which the two
/// disagree -- and it would have needed a core surface handing a module core's own participants.
///
/// A run with no ledger open is a 404, not an empty success: "nobody came" and "you never told me
/// to count" are different facts, and only the first is a result a run should record.
///
/// **Retryable, and protocol 6 is why.** `EVENTS.md` §G called a grant un-retryable because a lost
/// acknowledgement and a grant that never applied looked the same -- exactly the argument that made
/// `uo.broadcast` answer `retry: false` in Phase 9. An `idempotencyKey` closes that: a repeat is
/// answered by the original reply, so a retried grant cannot be one winner receiving two.
async fn item_grant(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
    event_call(&st, "item.grant", body).await
}

/// Body: {"idempotencyKey":<opt>}. Starts a world save, useful as a phase boundary.
///
/// The reply says the save was STARTED and nothing more. What actually happened rides
/// `world.save.before` / `world.save.after`, which have been on the event stream since protocol 2 --
/// so this route asserts nothing it cannot know, and a caller that needs the completion watches the
/// stream it is already connected to.
///
/// **A save too soon after the last one is refused, not queued**, and the shard counts ServUO's own
/// autosave as the last one. A save stops the world; a queued one would land at a moment nobody
/// chose, in the middle of whatever the next step is doing.
async fn world_save(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
    event_call(&st, "world.save", 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)
}

/// Body: {"id":"42","title":"...","body":"<html>","image":1614,"url":"...","announce":true}.
/// Adds/replaces a Town Cryer news article. Correlated on `id`. A success is stored so the sidecar
/// can replay the article to the shard on reconnect (NewsEntries is not persisted across a reboot).
async fn news_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();
    let title_ok = body
        .get("title")
        .and_then(|t| t.as_str())
        .map(|s| !s.trim().is_empty())
        .unwrap_or(false);
    if id.is_empty() || !title_ok {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({"error": "id and title are required"})),
        );
    }

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

    // Persist the article (as its news.add command) so it can be replayed on shard reconnect.
    if let Ok(value) = &result {
        if value.get("kind").and_then(|k| k.as_str()) == Some("news.ok") {
            let t = value.get("t").and_then(|v| v.as_i64()).unwrap_or(0);
            let _ = st.store.upsert_news(&id, &cmd.to_string(), t).await;
        }
    }
    respond(result)
}

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

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

// ---- 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 house registry: every house's latest snapshot (owner/region/location/decay/value). Served
/// from the local board table, so it hydrates without the shard and survives an outage.
async fn houses(State(st): State<AppState>) -> impl IntoResponse {
    match st.store.houses_all().await {
        Ok(houses) => (StatusCode::OK, Json(json!({"houses": houses}))),
        Err(e) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(json!({"error": e.to_string()})),
        ),
    }
}

/// The shard's published ruleset: expansion, which optional systems are on, skill/stat caps,
/// account and house limits, champion scroll rules, the save/restart schedule. Served from the
/// store, so it answers during a shard outage with the last-known ruleset — which is the whole
/// point, since a rules page that goes blank when the shard restarts is worse than a stale one.
///
/// `{"ruleset": null}` means the shard has never published one (an old plugin, or
/// `Bridge.RulesetEnabled=false`), which the website renders differently from a published ruleset.
async fn ruleset(State(st): State<AppState>) -> impl IntoResponse {
    match st.store.ruleset().await {
        Ok(r) => (StatusCode::OK, Json(json!({ "ruleset": r }))),
        Err(e) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(json!({"error": e.to_string()})),
        ),
    }
}

/// Every points/loyalty leaderboard the shard publishes: one entry per point system, each with its
/// display name (literal and/or cliloc), max points, participant count and top N. Store-backed like
/// the other boards, so the site's leaderboards page renders during a shard outage — which matters
/// more here than elsewhere, since these are month-scale standings that a restart must not blank.
async fn points(State(st): State<AppState>) -> impl IntoResponse {
    match st.store.points_boards_all().await {
        Ok(boards) => (StatusCode::OK, Json(json!({"boards": boards}))),
        Err(e) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(json!({"error": e.to_string()})),
        ),
    }
}

/// One system's board by its `PointsType` name (`QueensLoyalty`, `CleanUpBritannia`, …).
///
/// 404 rather than an empty board when the system is unknown: the shard publishes only the systems
/// it shows on the loyalty gump (or the explicit `Bridge.cfg PointsSystems` list), so "no such
/// board" and "a board with nobody on it" are different answers and the website renders them
/// differently.
async fn points_system(
    State(st): State<AppState>,
    Path(system): Path<String>,
) -> impl IntoResponse {
    match st.store.points_board(&system).await {
        Ok(Some(board)) => (StatusCode::OK, Json(board)),
        Ok(None) => (
            StatusCode::NOT_FOUND,
            Json(json!({"error": "unknown points system", "system": system})),
        ),
        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()})),
        ),
    }
}

#[derive(Deserialize)]
struct PageQuery {
    limit: Option<i64>,
    offset: Option<i64>,
}

/// The player-vendor market index: every vendor's shop name, owner, location and priced inventory,
/// as the shard last published it. Store-backed like the other boards, which is what lets the
/// website's market page render (labelled stale) while the shard is down.
///
/// Paged — `?limit=&offset=`, limit clamped to 1..1000, default 200 — because this is the one board
/// that can be a whole world's inventory. `total` is returned alongside so the caller knows when to
/// stop rather than paging until it sees a short page, which would race a concurrent sweep.
///
/// The frames are served VERBATIM, including owner names and coordinates. That is not an oversight:
/// the sidecar defines no audiences (docs/link/v3.md §3.2). Deciding who may see a vendor's owner
/// or whereabouts is the website's job and is admin-configurable there.
async fn market(State(st): State<AppState>, Query(q): Query<PageQuery>) -> impl IntoResponse {
    let limit = q.limit.unwrap_or(200);
    let offset = q.offset.unwrap_or(0);

    let total = match st.store.vendors_count().await {
        Ok(n) => n,
        Err(e) => {
            return (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(json!({"error": e.to_string()})),
            )
        }
    };

    match st.store.vendors_page(limit, offset).await {
        Ok(vendors) => (
            StatusCode::OK,
            Json(json!({
                "vendors": vendors,
                "total": total,
                "limit": limit.clamp(1, 1000),
                "offset": offset.max(0),
            })),
        ),
        Err(e) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(json!({"error": e.to_string()})),
        ),
    }
}

// ---- the Asset Bridge (Protocol 8) ----

/// Stage 1 of the import gate: the shard's UO client files as they are right now — size, mtime and
/// content hash — plus the version of the extractor that would read them, and whether this host
/// can render an image at all.
///
/// The website diffs this against what it last imported and, in the overwhelmingly common case
/// that nothing changed, stops. That is the whole reason stage 1 exists separately from the asset
/// manifest: the normal case is a restart that changed nothing, and it has to cost nothing.
///
/// Forwarded verbatim, like everything else on this link. The sidecar does not know what a cliloc
/// or an anim file is, does not cache this, and has no opinion about what the website does with
/// the answer — the same dumb-forwarder property that keeps access control on the website where it
/// belongs.
async fn assets_sources(State(st): State<AppState>) -> impl IntoResponse {
    let req_id = st.rpc.next_req_id();
    let cmd = json!({"kind": "assets.sources", "reqId": req_id});

    respond_assets(st.rpc.call(&st.shard, cmd, &req_id).await)
}

/// Maps an asset-plane reply to a status.
///
/// Two of these matter more than the rest and neither is the generic responder's answer:
///
/// **`bridge.busy` is a 425**, as everywhere else. On this plane it is not an idempotency
/// collision, it is flow control: the shard serves one asset request at a time on purpose, because
/// its outbound queue is bounded in *lines* and a queue of large replies is how the shard runs out
/// of memory. So it means "come back", it is entirely expected during an import, and a caller that
/// treated it as an error would abandon a perfectly healthy transfer.
///
/// **The plane being switched off is a 403.** `Bridge.AssetsEnabled` is an operator declining to
/// let the website read their client files off this host — a deliberate refusal, not a malformed
/// request — and answering 400 would send an administrator hunting a bug in a call that is written
/// correctly. Same argument the event plane's gate made in protocol 7.
fn respond_assets(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 == BUSY_KIND {
                (BUSY_STATUS, Json(value))
            } else if kind == "assets.error" {
                (asset_error_status(&value), 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"})),
        ),
    }
}

/// The status behind one `assets.error`.
///
/// Phase 2 gave the frame a `code`, and the reason is worth stating: phase 1 decided between 403
/// and 400 by looking for the word "disabled" **in the operator-facing sentence**. That works
/// until someone improves the wording, at which point a refusal quietly becomes a bad request and
/// an administrator goes hunting for a bug in a correctly-written call. The sentence is for a
/// human; the code is for this function.
///
/// The substring check survives as a fallback because an overlay is deployed independently of the
/// sidecar: a phase-1 shard paired with a phase-2 sidecar still sends the codeless frame, and it
/// must keep getting its 403.
fn asset_error_status(value: &Value) -> StatusCode {
    match value.get("code").and_then(|c| c.as_str()) {
        Some("DISABLED") => return StatusCode::FORBIDDEN,
        // The shard has no such file. Not the caller's mistake and not a broken shard -- a client
        // that does not carry what was asked for, which the website reports to its operator.
        Some("NOT_FOUND") => return StatusCode::NOT_FOUND,
        // It has the file and cannot decode it: truncated, hand-edited, or not what it claims to
        // be. 422 rather than 400, because the request was fine and repeating it will not help.
        Some("UNREADABLE") => return StatusCode::UNPROCESSABLE_ENTITY,
        // The shard cannot do this right now (it could not start its asset worker, say). Same
        // status as "no shard connected", because it means the same thing to a caller: come back.
        Some("UNAVAILABLE") => return StatusCode::SERVICE_UNAVAILABLE,
        Some("BAD_REQUEST") => return StatusCode::BAD_REQUEST,
        _ => {}
    }

    let reason = value
        .get("reason")
        .and_then(|r| r.as_str())
        .unwrap_or("request rejected");

    if reason.contains("disabled") {
        StatusCode::FORBIDDEN
    } else {
        StatusCode::BAD_REQUEST
    }
}

/// The cliloc table, as the shard's own client holds it (docs/link/v8.md §9).
///
/// Until protocol 8 this table reached the website by hand: the operator installed UOFiddler,
/// built a converter against its `Ultima.dll`, ran it over their client's compressed `Cliloc.enu`
/// and copied the result to the web host. The shard could not help, because ServUO's bundled
/// `Ultima.StringList` cannot read a modern client's file either. Phase 2 put the decompressor in
/// the overlay, so the shard reads its own client and the operator installs nothing.
///
/// **Paged, and the caller drives the paging** -- `?cursor=` echoes back whatever the previous
/// reply's `cursor` was, until a reply says `more: false`. The pages are cut by byte budget on the
/// shard (512 KiB against the 1 MiB inbound line cap), so a stock English table arrives in about
/// eleven of them. The sidecar keeps none of it: it has no opinion about what a cliloc is, and a
/// cached copy of five megabytes of EA's strings is exactly what this process should not hold.
///
/// `?lang=` selects the file; it defaults to `enu` on the shard and the shard refuses anything its
/// `Ultima.Files` cannot resolve, which is a 404 rather than a 400.
async fn cliloc_table(
    State(st): State<AppState>,
    Query(q): Query<ClilocQuery>,
) -> impl IntoResponse {
    let req_id = st.rpc.next_req_id();
    let mut cmd = json!({"kind": "cliloc.table", "reqId": req_id});

    if let Some(lang) = q.lang.as_deref().filter(|s| !s.is_empty()) {
        cmd["lang"] = json!(lang);
    }

    if let Some(cursor) = q.cursor.as_deref().filter(|s| !s.is_empty()) {
        cmd["cursor"] = json!(cursor);
    }

    respond_assets(st.rpc.call(&st.shard, cmd, &req_id).await)
}

#[derive(Deserialize)]
struct ClilocQuery {
    lang: Option<String>,
    cursor: Option<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");
}

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

    fn reply(kind: &str) -> Result<Value, RpcError> {
        Ok(json!({"t": 1, "kind": kind, "reqId": "r-9"}))
    }

    /// Protocol 6. Every responder must recognise `bridge.busy`, because every write plane can be
    /// retried: the staff plane, the account plane and the plain command plane all reach handlers
    /// that a keyed retry can arrive at. A responder that missed it would return 200 with a body
    /// saying nothing happened, which is the worst of the three possible answers.
    #[test]
    fn busy_maps_to_425_on_every_plane() {
        assert_eq!(respond(reply("bridge.busy")).0, StatusCode::TOO_EARLY);
        assert_eq!(respond_admin(reply("bridge.busy")).0, StatusCode::TOO_EARLY);
        assert_eq!(
            respond_account(reply("bridge.busy")).0,
            StatusCode::TOO_EARLY
        );
        assert_eq!(respond_event(reply("bridge.busy")).0, StatusCode::TOO_EARLY);

        // Protocol 8. On the asset plane `bridge.busy` is not a keyed retry colliding with itself
        // -- it is flow control, and it is the ORDINARY answer during an import rather than a rare
        // one. The shard serves one asset request at a time because its outbound queue is bounded
        // in lines, not bytes, so a queue of large replies is how it runs out of memory. A
        // responder that answered 200 here would tell the website an import step succeeded and
        // returned nothing.
        assert_eq!(
            respond_assets(reply("bridge.busy")).0,
            StatusCode::TOO_EARLY
        );
    }

    /// The asset plane's own gate, and it is a refusal rather than a mistake: an operator who has
    /// not enabled `Bridge.AssetsEnabled` has declined to let the website read their UO client
    /// files off the shard host. 403, for the same reason the event plane's switch is a 403.
    #[test]
    fn assets_disabled_is_a_403() {
        let value = json!({
            "kind": "assets.error",
            "reqId": "r-9",
            "reason": "asset extraction is disabled on this shard"
        });

        assert_eq!(respond_assets(Ok(value)).0, StatusCode::FORBIDDEN);
    }

    /// Anything else the shard refuses on this plane is the caller's mistake.
    #[test]
    fn other_asset_errors_are_400() {
        let value = json!({
            "kind": "assets.error",
            "reason": "assets.sources requires a reqId"
        });

        assert_eq!(respond_assets(Ok(value)).0, StatusCode::BAD_REQUEST);
    }

    /// Phase 2's codes, each of which says something a 400 does not. `NOT_FOUND` is a client that
    /// does not carry the file (an operator fact, not a bug); `UNREADABLE` is a file that is there
    /// and cannot be decoded, where repeating the request cannot help; `UNAVAILABLE` is the shard
    /// declining for now.
    #[test]
    fn an_asset_error_code_picks_the_status() {
        let with = |code: &str| {
            respond_assets(Ok(json!({
                "kind": "assets.error",
                "reqId": "r-1",
                "code": code,
                "reason": "…"
            })))
            .0
        };

        assert_eq!(with("DISABLED"), StatusCode::FORBIDDEN);
        assert_eq!(with("NOT_FOUND"), StatusCode::NOT_FOUND);
        assert_eq!(with("UNREADABLE"), StatusCode::UNPROCESSABLE_ENTITY);
        assert_eq!(with("UNAVAILABLE"), StatusCode::SERVICE_UNAVAILABLE);
        assert_eq!(with("BAD_REQUEST"), StatusCode::BAD_REQUEST);
        // An unknown code is not a reason to invent a status.
        assert_eq!(with("SOMETHING_NEW"), StatusCode::BAD_REQUEST);
    }

    /// The overlay and the sidecar ship separately and an operator can run a phase-1 shard against
    /// a phase-2 sidecar, so the codeless refusal must keep its 403. This is the whole reason the
    /// substring check survives rather than being deleted with the code it preceded.
    #[test]
    fn a_codeless_disabled_refusal_is_still_a_403() {
        let value = json!({
            "kind": "assets.error",
            "reqId": "r-9",
            "reason": "asset extraction is disabled on this shard"
        });

        assert_eq!(respond_assets(Ok(value)).0, StatusCode::FORBIDDEN);
    }

    /// One page of the cliloc table. The assertion that matters is the envelope: `more` and
    /// `cursor` reach the caller untouched, because the sidecar does not page this — the shard
    /// cuts the pages and the website drives them, and a sidecar that "helpfully" assembled them
    /// would be holding the whole table in memory to do it.
    #[test]
    fn a_cliloc_page_is_a_200_and_keeps_its_cursor() {
        let value = json!({
            "kind": "cliloc.table.ok",
            "reqId": "r-2",
            "lang": "enu",
            "extractorVersion": 1,
            "total": 67496,
            "rows": [{"n": 1023721, "f": 0, "t": "quarter staff"}],
            "more": true,
            "cursor": "n:1023721",
            "cut": "budget"
        });

        let (status, body) = respond_assets(Ok(value));

        assert_eq!(status, StatusCode::OK);
        assert_eq!(body.0.get("more").and_then(|v| v.as_bool()), Some(true));
        assert_eq!(
            body.0.get("cursor").and_then(|v| v.as_str()),
            Some("n:1023721")
        );
        assert_eq!(body.0.get("total").and_then(|v| v.as_i64()), Some(67496));
    }

    /// A source manifest comes back whole. Worth asserting because `respond_assets` sniffs `kind`
    /// and a family whose success kind ends in `.ok` sits one character away from the `.error`
    /// suffix the generic responder matches on -- which is exactly why this plane has its own
    /// responder and matches `assets.error` exactly rather than by suffix.
    #[test]
    fn a_source_manifest_is_a_200() {
        let value = json!({
            "kind": "assets.sources.ok",
            "reqId": "r-9",
            "extractorVersion": 1,
            "imaging": {"ok": true},
            "files": [],
            "more": false,
            "cut": "end"
        });

        let (status, body) = respond_assets(Ok(value));

        assert_eq!(status, StatusCode::OK);
        assert_eq!(
            body.0.get("extractorVersion").and_then(|v| v.as_i64()),
            Some(1)
        );
    }

    /// A shard that is not connected is a 503 and a shard that did not answer in time is a 504,
    /// and the asset plane needs the second one to stay distinct more than any other plane does:
    /// hashing a 195 MB anim.mul is the one thing on this link that can genuinely outlast the
    /// 10 s reply timeout, and the website's response to that is to poll again rather than to
    /// declare the shard down.
    #[test]
    fn asset_transport_failures_keep_their_own_statuses() {
        assert_eq!(
            respond_assets(Err(RpcError::NoShard)).0,
            StatusCode::SERVICE_UNAVAILABLE
        );
        assert_eq!(
            respond_assets(Err(RpcError::Timeout)).0,
            StatusCode::GATEWAY_TIMEOUT
        );
    }

    /// The event plane is the FIRST place `bridge.busy` is reachable on a live shard rather than
    /// only in a unit test: `participation.snapshot` walks a well-attended run's members across
    /// Core ticks, so it completes after its inbound call returned and a repeat can genuinely land
    /// mid-flight. 11a built the door and had nothing to walk through it.
    #[test]
    fn a_drifted_lease_is_a_200_not_a_409() {
        let value =
            json!({"kind": "lease.drifted", "key": "PlayerCaps.SkillCap", "current": "1300"});
        let (status, body) = respond_event(Ok(value));

        // The shard was asked to compare and set, it compared, and it declined to overwrite
        // somebody's deliberate change. That is the mechanism working; the website records
        // `drifted` as a distinct successful outcome rather than an error.
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body.0.get("current").and_then(|v| v.as_str()), Some("1300"));

        // And explicitly not the version gate's status, for the reason 425 is not either: a
        // mismatched deployment and a moved value want opposite dispositions from a caller.
        assert_ne!(status, StatusCode::CONFLICT);
    }

    /// The event plane being switched off is an operator's refusal, not a malformed request. A 400
    /// would send an administrator hunting a bug in a step that is written correctly.
    #[test]
    fn the_event_gate_being_off_is_a_403() {
        assert_eq!(
            respond_event(Ok(json!({
                "kind": "lease.error",
                "reason": "the event plane is disabled on this shard (Bridge.EventsEnabled)"
            })))
            .0,
            StatusCode::FORBIDDEN
        );
        assert_eq!(
            respond_event(Ok(json!({
                "kind": "participation.error",
                "reason": "the event plane is disabled on this shard (Bridge.EventsEnabled)"
            })))
            .0,
            StatusCode::FORBIDDEN
        );
    }

    /// Protocol 7's world verbs go through the same responder, and this pins the two mappings
    /// they depend on rather than trusting that the reason-sniffing above keeps covering a kind
    /// it was written before.
    ///
    /// A CEILING refusal is a 400 on purpose. It is permanent -- retrying "you asked for 80
    /// creatures and this shard places 30" gets the same answer forever -- and it is the module's
    /// `PERMANENT_STATUSES` that has to see it as such, so classifying it as anything retryable
    /// would put a run in a loop against a limit that will never move.
    #[test]
    fn a_world_refusal_is_a_400_and_the_gate_is_still_a_403() {
        assert_eq!(
            respond_event(Ok(json!({
                "kind": "world.error",
                "action": "spawn",
                "reason": "this shard places 1 to 30 of 'creature' at a time, and 80 was asked for"
            })))
            .0,
            StatusCode::BAD_REQUEST
        );
        assert_eq!(
            respond_event(Ok(json!({
                "kind": "world.error",
                "action": "spawn",
                "reason": "the event plane is disabled on this shard (Bridge.EventsEnabled)"
            })))
            .0,
            StatusCode::FORBIDDEN
        );
    }

    /// A run the shard has no registry rows for answers with an EMPTY hand, not a 404, and the
    /// distinction is load-bearing for reconcile.
    ///
    /// "This run owns nothing" and "I have never heard of this run" are the same fact once the
    /// registry is the only record of ownership, and they stay the same fact across a restart:
    /// the registry is written by `EventSink.WorldSave`, so it and the objects it describes are
    /// saved and lost together. A 404 here would make the website treat a run that legitimately
    /// owns nothing as a shard it could not reach.
    #[test]
    fn a_run_owning_nothing_is_an_empty_list_not_a_404() {
        let (status, body) = respond_event(Ok(json!({
            "kind": "world.owned.ok",
            "runId": "77",
            "owned": [],
            "pruned": 0
        })));

        assert_eq!(status, StatusCode::OK);
        assert_eq!(
            body.0
                .get("owned")
                .and_then(|v| v.as_array())
                .map(|a| a.len()),
            Some(0)
        );
    }

    /// An unknown lease key and an unknown run are not-founds; anything else the shard refuses is a
    /// bad request. The catalog is short and a typo in a step is the likely cause of both.
    #[test]
    fn unknown_lease_and_run_are_404s() {
        assert_eq!(
            respond_event(Ok(json!({
                "kind": "lease.error",
                "reason": "no lease is offered for key 'Loot.MaxProps'"
            })))
            .0,
            StatusCode::NOT_FOUND
        );
        assert_eq!(
            respond_event(Ok(json!({
                "kind": "participation.error",
                "reason": "this shard is not counting run '42'"
            })))
            .0,
            StatusCode::NOT_FOUND
        );
        assert_eq!(
            respond_event(Ok(json!({
                "kind": "lease.error",
                "reason": "a lease needs a positive holdMs"
            })))
            .0,
            StatusCode::BAD_REQUEST
        );
    }

    /// A run this shard was never told to count is a 404; a run that WAS counted and had no
    /// attendees is a 200. Protocol 7 part b.
    #[test]
    fn an_uncounted_run_is_a_404_and_an_empty_one_is_not() {
        assert_eq!(
            respond_event(Ok(json!({
                "kind": "oneshot.error",
                "reason": "run 42 has no participation ledger open on this shard"
            })))
            .0,
            StatusCode::NOT_FOUND
        );
        // The distinction the 404 exists to preserve. "Nobody came" is a RESULT -- an event
        // nobody attended still happened -- and answering it as a failure would have the module
        // retry a grant against a ledger that will be just as empty next time.
        assert_eq!(
            respond_event(Ok(json!({
                "kind": "item.grant.ok",
                "runId": "42",
                "granted": 0,
                "missed": []
            })))
            .0,
            StatusCode::OK
        );
    }

    /// The save rate limit is the one refusal on this plane that the same request will get past
    /// by waiting, so it is a 429 rather than the 400 every other refusal is.
    #[test]
    fn a_save_refused_for_coming_too_soon_is_a_429() {
        assert_eq!(
            respond_event(Ok(json!({
                "kind": "oneshot.error",
                "reason": "this shard saves at most every 300 seconds, and the last save was 12 seconds ago"
            })))
            .0,
            StatusCode::TOO_MANY_REQUESTS
        );
        // And an ordinary refusal on the same plane is still a 400, so the 429 is not swallowing
        // the class it sits beside: a grant this shard does not offer will never succeed, however
        // long the caller waits.
        assert_eq!(
            respond_event(Ok(json!({
                "kind": "oneshot.error",
                "reason": "this shard does not grant 'castle'"
            })))
            .0,
            StatusCode::BAD_REQUEST
        );
        // The event gate being off stays a 403 on this plane too -- it is an operator's deliberate
        // refusal, not a bad request.
        assert_eq!(
            respond_event(Ok(json!({
                "kind": "oneshot.error",
                "reason": "the event plane is disabled on this shard (Bridge.EventsEnabled)"
            })))
            .0,
            StatusCode::FORBIDDEN
        );
    }

    /// A lease taken, a tally answered: an ordinary success carries straight through.
    #[test]
    fn event_successes_are_200s() {
        assert_eq!(respond_event(reply("lease.ok")).0, StatusCode::OK);
        assert_eq!(respond_event(reply("lease.list.ok")).0, StatusCode::OK);
        assert_eq!(respond_event(reply("participation.ok")).0, StatusCode::OK);
        assert_eq!(
            respond_event(reply("participation.snapshot.ok")).0,
            StatusCode::OK
        );
    }

    /// 425 must not collide with the protocol-version gate's 409: a mismatch is a deployment fault
    /// nobody should retry, a busy shard is a retry that will succeed. Same-status would make the
    /// two readable only by inspecting the body.
    #[test]
    fn busy_is_not_the_version_gates_status() {
        assert_ne!(BUSY_STATUS, StatusCode::CONFLICT);
    }

    /// A replayed reply is an ordinary success. The shard marks it `replayed: true` for the log, and
    /// the caller must be able to treat it exactly as it would have treated the answer it lost.
    #[test]
    fn a_replayed_reply_is_still_a_200() {
        let value = json!({"t": 1, "kind": "admin.ok", "reqId": "r-9", "replayed": true});
        let (status, body) = respond_admin(Ok(value));
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body.0.get("replayed").and_then(|v| v.as_bool()), Some(true));
    }

    /// The error mapping the busy arm is threaded in front of must be untouched by it.
    #[test]
    fn errors_still_map_as_before() {
        assert_eq!(
            respond(Ok(
                json!({"kind": "bridge.error", "reason": "unknown account"})
            ))
            .0,
            StatusCode::NOT_FOUND
        );
        assert_eq!(
            respond_admin(Ok(json!({"kind": "admin.error", "reason": "protected"}))).0,
            StatusCode::FORBIDDEN
        );
        assert_eq!(
            respond_account(Ok(
                json!({"kind": "account.error", "reason": "already exists"})
            ))
            .0,
            StatusCode::CONFLICT
        );
    }
}