//! The HTTP surface. The /api/* contract is fixed by web/src/api.ts:
//! errors are `{message}`, login success is `{redirectUrl}`, session is
//! `{did, handle, defaultCamo}` — change either side and change the other.

use std::sync::Arc;

use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
use axum::extract::{Path, Query, State};
use axum::http::{HeaderValue, Method, StatusCode, header};
use axum::response::{IntoResponse, Redirect, Response};
use axum::routing::{get, post};
use axum::{Json, Router};
use axum_extra::extract::cookie::CookieJar;
use base64::Engine;
use tower_http::cors::CorsLayer;

use crate::atproto::{Atproto, LoginError};
use crate::db::Db;
use crate::session::SessionSigner;

#[derive(Clone)]
pub struct AppState {
    pub atproto: Arc<Atproto>,
    pub db: Db,
    pub signer: Arc<SessionSigner>,
    pub web_origin: String,
    /// Where the public match pages are published, when that is not this
    /// service's own address.
    pub share_origin: Option<String>,
    /// None when this instance cannot launch matches (no MATCH_CLUSTER).
    pub matches: crate::matches::SharedMatches,
    /// The live-lobby registry: DB-backed, in-process broadcast per lobby
    /// id. Unlike `matches`, never None — a lobby never touches AWS, so it
    /// works on every deployment this runs on.
    pub lobbies: Arc<crate::matches::lobby::Lobbies>,
    /// None in development; the built image carries BUILD_REF.
    pub build_ref: Option<String>,
    /// The arena image tag matches launch from; None when infra has not
    /// said (or in development).
    pub arena_version: Option<String>,
    /// The at:// URI of the userinput.app space flares are filed to; None
    /// means /api/flare answers that flares are not enabled.
    pub flare_space: Option<String>,
    /// helm's index of the unit library, for the picture a report draws each
    /// machine with. None where no index was named or one would not load, in
    /// which case a report draws weight-class silhouettes as it always has.
    pub units: Option<Arc<crate::units::UnitIndex>>,
}

pub fn app(state: AppState) -> Router {
    // Exact origin plus credentials: the allow-list is the site and nothing
    // else, and a JSON POST from it forces a preflight — which is currently
    // the whole CSRF defence (plan/session-security.md).
    let cors = CorsLayer::new()
        .allow_origin(
            state
                .web_origin
                .parse::<HeaderValue>()
                .expect("config validated origin"),
        )
        .allow_methods([Method::GET, Method::POST, Method::DELETE])
        .allow_headers([header::CONTENT_TYPE])
        .allow_credentials(true);

    let mut router = Router::new()
        .route("/healthz", get(|| async { "ok" }))
        .route("/api/version", get(version))
        .route("/api/login", post(login))
        .route("/api/session", get(session))
        .route("/api/logout", post(logout))
        .route("/oauth/callback", get(callback))
        // A camo is a few tens of kilobytes. The limit is here so a body
        // that is nothing like one is dropped before it is decoded at all.
        .route(
            "/api/camo",
            post(create_camo).layer(axum::extract::DefaultBodyLimit::max(1024 * 1024)),
        )
        .route(
            "/api/camo/{rkey}",
            post(update_camo)
                .delete(delete_camo)
                .layer(axum::extract::DefaultBodyLimit::max(1024 * 1024)),
        )
        // Under /api/account, not /api/camo: it is a setting on the account,
        // and a sibling of /api/camo/{rkey} would collide with a camo whose
        // key happened to be the word.
        .route(
            "/api/account/camo",
            // A record key and nothing else. The default here is 2MB, which
            // is four thousand times the largest body this can accept.
            post(set_default_camo).layer(axum::extract::DefaultBodyLimit::max(4 * 1024)),
        )
        .route(
            "/api/flare",
            // A screenshot arrives base64-encoded, which puts a 1MB PNG at
            // ~1.4MB on the wire; 2MB covers that and the words around it.
            post(create_flare).layer(axum::extract::DefaultBodyLimit::max(2 * 1024 * 1024)),
        )
        .route("/api/units/art", get(unit_art))
        .route("/api/scenarios", get(list_scenarios))
        .route("/api/daily/{slug}", get(daily_status).post(start_daily))
        .route("/api/opponents", get(list_opponents))
        .route("/api/opponents/suggested", get(list_suggested_opponents))
        .route("/api/matches", post(create_match).get(list_matches))
        // Any player's matches, by handle or DID and without a session.
        // Under the profile prefix because that is whose record it is —
        // /api/matches is the caller's own, and this one names somebody.
        .route("/api/profile/{actor}/matches", get(player_matches))
        .route("/api/matches/lobby", post(open_lobby))
        .route("/api/matches/{id}/live", get(lobby_live))
        .route("/api/matches/{id}/deploy", post(deploy_lobby))
        .route("/api/matches/{id}", get(match_status))
        .route("/api/matches/{id}/result", get(match_result))
        .route("/matches/{id}/events/{token}", post(match_event))
        // The only way an artifact gets in: we sign a PUT for one key and the
        // container sends the bytes to S3. No body, so no limit - nothing
        // larger than an empty POST ever arrives here.
        .route("/upload-urls/{id}/{token}/{*name}", post(mint_upload_url))
        // Public, and the only routes here that are. A result a player
        // posted to Bluesky is fetched by a card service with no session of
        // theirs; see share.rs for what that bounds it to.
        // The public face of a finished match. Served under one prefix so a
        // CDN in front of the site can route the whole of it here with a
        // single behaviour, which is what lets these pages live at
        // lance.blue/reports/... rather than on the API's own host.
        .route("/reports/{id}", get(crate::share::page))
        .route("/reports/{id}/card.png", get(crate::share::card))
        .route("/reports/fonts/{name}", get(crate::share::font))
        // What a link to a player unfurls as. A second prefix rather than
        // /profile/, because that address belongs to the site and is served
        // from the bucket: the distribution sends a card fetcher here and
        // everybody else there. See unfurl.rs.
        .route("/unfurl/profile/{actor}", get(crate::unfurl::player))
        .route(
            "/unfurl/profile/{actor}/card.png",
            get(crate::unfurl::player_card),
        )
        .route("/match/{id}", get(crate::proxy::entry))
        .route(
            "/match/{id}/{*rest}",
            axum::routing::any(crate::proxy::forward),
        );

    if state.atproto.is_confidential() {
        router = router
            .route("/oauth/client-metadata.json", get(client_metadata))
            .route("/.well-known/jwks.json", get(jwks));
    }

    router.layer(cors).with_state(state)
}

/// What is running: the crate version, the commit the image was built from,
/// the arena image tag matches launch from, and which unit library this
/// instance loaded. Unset values are null rather than omitted, so the site
/// can tell "old API" from "not configured".
///
/// `library` is here rather than on a route of its own because every page
/// already asks for this one: `web/src/chrome.ts` fetches it once per load
/// for the footer and shares the promise, so a screen that needs the library
/// pays nothing for it.
///
/// It is what makes the site and this service read the same index. The site
/// used to name a helm release in its own source, so a helm deploy needed a
/// matching site build - and when it did not get one, the force builder and
/// the match reports read different libraries and nothing said so. Now the
/// site asks what this instance loaded, and the answer comes off the index
/// document itself.
async fn version(State(state): State<AppState>) -> Response {
    let library = state.units.as_ref().map(|units| {
        let (megamek, helm) = units.library();
        serde_json::json!({ "megamek": megamek, "helm": helm })
    });
    Json(serde_json::json!({
        "version": env!("CARGO_PKG_VERSION"),
        "build": state.build_ref,
        "arena": state.arena_version,
        "library": library,
    }))
    .into_response()
}

/// The most designs one lookup will answer for.
///
/// A scenario preview asks about the handful a scenario fields; the front
/// page's daily card asks about six. This is far above either and bounds
/// what a stranger can make this service do per request.
const ART_LOOKUP_MAX: usize = 64;

/// Which picture each of these designs is drawn with.
///
/// The fast path for a screen that draws a few units. The whole index is
/// public and a screen can read it - the force builder does, because it
/// needs all of it - but 300KB to resolve six names is not a trade the front
/// page should make. This answers in a few hundred bytes.
///
/// A path relative to `data/images/units/`, the same form the index carries,
/// because the caller turns it into an address with the release it already
/// asked `/api/version` for. Sending a whole URL from here would put a second
/// address-builder in this service, and `web/src/megamek.ts` is the one that
/// exists.
///
/// Unknown names are absent from the answer rather than null: the caller
/// falls back to the index, which is the same question asked of the same
/// data, so an empty answer costs correctness nothing.
async fn unit_art(
    State(state): State<AppState>,
    Query(asked): Query<Vec<(String, String)>>,
) -> Response {
    let Some(units) = state.units.as_ref() else {
        // No index loaded is not an error: it is this deployment having no
        // library, and the caller's fallback finds the same nothing.
        return Json(serde_json::json!({})).into_response();
    };
    let names: Vec<&str> = asked
        .iter()
        .filter(|(key, _)| key == "name")
        .map(|(_, value)| value.as_str())
        .take(ART_LOOKUP_MAX)
        .collect();
    (
        // Long enough that a second page load reuses it, short enough that a
        // deployment moving to a new library is not serving last release's
        // answers all afternoon. The paths are stable across most releases;
        // the ones that are not are why this is not immutable.
        [(header::CACHE_CONTROL, "public, max-age=300")],
        Json(serde_json::json!(units.sprites_for(names))),
    )
        .into_response()
}

fn message(status: StatusCode, text: &str) -> Response {
    (status, Json(serde_json::json!({ "message": text }))).into_response()
}

#[derive(serde::Deserialize)]
struct LoginBody {
    handle: String,
}

async fn login(State(state): State<AppState>, body: Option<Json<LoginBody>>) -> Response {
    let Some(Json(LoginBody { handle })) = body else {
        return message(StatusCode::BAD_REQUEST, "Send a JSON body with a handle.");
    };
    match state.atproto.begin_login(&handle).await {
        Ok(url) => Json(serde_json::json!({ "redirectUrl": url })).into_response(),
        Err(LoginError::InvalidHandle) => {
            message(StatusCode::BAD_REQUEST, "That doesn't look like a handle.")
        }
        Err(LoginError::Upstream) => message(
            StatusCode::BAD_GATEWAY,
            "Your identity provider could not be reached.",
        ),
    }
}

/// Permissive on purpose: the authorization server may come back with
/// `error=` and no code (the user said no), and atrium's own params type
/// requires `code`.
#[derive(serde::Deserialize)]
struct CallbackQuery {
    code: Option<String>,
    state: Option<String>,
    iss: Option<String>,
    error: Option<String>,
}

async fn callback(State(state): State<AppState>, Query(query): Query<CallbackQuery>) -> Response {
    let denied = format!("{}/?error=denied", state.web_origin);
    let failed = format!("{}/?error=login_failed", state.web_origin);

    let Some(code) = query.code else {
        // Not worth logging the error value: it is attacker-controlled query
        // text on an unauthenticated route.
        return Redirect::to(if query.error.is_some() {
            &denied
        } else {
            &failed
        })
        .into_response();
    };

    match state
        .atproto
        .complete_login(code, query.state, query.iss)
        .await
    {
        Ok((did, handle)) => {
            if state
                .db
                .record_handle_check(&did, handle.as_deref())
                .await
                .is_err()
            {
                tracing::error!("callback: account row write failed");
                return Redirect::to(&failed).into_response();
            }
            let jar = CookieJar::new().add(state.signer.issue(&did));
            // SameSite=Lax lets the cookie ride this top-level cross-site
            // redirect — the reason Lax was chosen over Strict.
            (jar, Redirect::to(&format!("{}/", state.web_origin))).into_response()
        }
        Err(crate::atproto::CallbackFailed) => Redirect::to(&failed).into_response(),
    }
}

async fn session(State(state): State<AppState>, jar: CookieJar) -> Response {
    let did = jar
        .get(crate::session::COOKIE_NAME)
        .and_then(|cookie| state.signer.verify(cookie.value()));
    let Some(did) = did else {
        return StatusCode::UNAUTHORIZED.into_response();
    };

    // The cookie alone is not a session: the stored OAuth session is. Gone —
    // signed out elsewhere, database wiped — means signed out, and the stale
    // cookie gets cleared on the way past.
    match state.db.has_oauth_session(&did).await {
        Ok(true) => {}
        Ok(false) => {
            let jar = CookieJar::new().add(state.signer.clear());
            return (StatusCode::UNAUTHORIZED, jar).into_response();
        }
        Err(_) => {
            tracing::error!(did, "session: oauth session lookup failed");
            return StatusCode::INTERNAL_SERVER_ERROR.into_response();
        }
    }

    match state.db.account_handle(&did).await {
        Ok(handle) => {
            refresh_handle_if_stale(&state, &did);
            // Here rather than on a route of its own: it comes from the same
            // row, every screen that needs it already has this response, and
            // one request is cheaper than two on every page load.
            let Ok(default_camo) = state.db.default_camo(&did).await else {
                // The handle read two lines up answers 500 on a database
                // error, and this comes from the same row. Reporting "no
                // default" instead would have the editor draw the toggle
                // unset, which is a lie the player can act on.
                tracing::error!(did, "session: default camo lookup failed");
                return StatusCode::INTERNAL_SERVER_ERROR.into_response();
            };
            Json(serde_json::json!({
                "did": did,
                "handle": handle.flatten(),
                "defaultCamo": default_camo,
            }))
            .into_response()
        }
        Err(_) => {
            tracing::error!(did, "session: handle lookup failed");
            StatusCode::INTERNAL_SERVER_ERROR.into_response()
        }
    }
}

/// How long a handle check is trusted before the handle is looked at again.
/// A rename shows up within the hour, and an account's own server is asked
/// about it at most a couple of dozen times a day per signed-in player.
const HANDLE_MAX_AGE_SECONDS: i64 = 60 * 60;

/// Re-verifies the handle if the last check has aged out, off the request
/// path. The site calls /api/session on every page load, so verification —
/// a DNS lookup plus an HTTPS fetch, both to servers we do not run — must
/// never be something a page load waits on. The caller is answered from the
/// cached row and the next load sees the new name.
///
/// Which request does the work is settled in sqlite by claim_handle_check,
/// so simultaneous requests do not each start their own check.
fn refresh_handle_if_stale(state: &AppState, did: &str) {
    let state = state.clone();
    let did = did.to_owned();
    tokio::spawn(async move {
        let now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .expect("clock before 1970")
            .as_secs() as i64;
        match state
            .db
            .claim_handle_check(&did, now - HANDLE_MAX_AGE_SECONDS)
            .await
        {
            Ok(true) => {}
            // Fresh enough, or another request is already on it.
            Ok(false) => return,
            Err(_) => {
                tracing::warn!("handle refresh: could not claim the check");
                return;
            }
        }
        // None demotes the row to unverified, which is the honest answer
        // when a name has stopped resolving back to this DID.
        let handle = state.atproto.reverify_handle(&did).await;
        if state
            .db
            .record_handle_check(&did, handle.as_deref())
            .await
            .is_err()
        {
            tracing::warn!("handle refresh: result could not be stored");
        }
    });
}

async fn logout(State(state): State<AppState>, jar: CookieJar) -> Response {
    if let Some(did) = jar
        .get(crate::session::COOKIE_NAME)
        .and_then(|cookie| state.signer.verify(cookie.value()))
    {
        // Best-effort revocation at the server, unconditional deletion here:
        // sign-out must work even when the PDS is down.
        state.atproto.logout(&did).await;
        if state.db.delete_oauth_session(&did).await.is_err() {
            tracing::error!("logout: session row delete failed");
        }
    }
    let jar = CookieJar::new().add(state.signer.clear());
    (StatusCode::NO_CONTENT, jar).into_response()
}

fn session_did(state: &AppState, jar: &CookieJar) -> Option<String> {
    jar.get(crate::session::COOKIE_NAME)
        .and_then(|cookie| state.signer.verify(cookie.value()))
}

// --- camo --------------------------------------------------------------
//
// Three routes, and they exist for one reason: the browser holds no token.
// The OAuth session lives here (docs/identity-and-sessions.md), so a write to
// a player's own repository has to be made from here on their behalf.
//
// Reading is not here, and should not be. A camo record and its blob are
// public, so the site lists a player's collection straight from their PDS
// without asking us anything. Only the writes need what we hold.

#[derive(serde::Deserialize)]
struct CamoBody {
    name: Option<String>,
    /// The 84x72 PNG, base64. A JSON body rather than a raw one so the name
    /// travels in the body as well: a name in a query string ends up in every
    /// access log between here and the browser.
    png: Option<String>,
    /// The editor's own settings, passed through untouched. The schema does
    /// not declare this and neither does this service — it is written and
    /// read by the editor, and nothing here has any reason to understand it.
    source: Option<serde_json::Value>,
    /// The CID the caller last read, for an edit. A write is refused if the
    /// record has changed since.
    #[serde(rename = "swapRecord")]
    swap_record: Option<String>,
}

/// The 84x72 PNG a body carries, re-encoded. `None` when there is no image in
/// the body at all, which is a rename. The error is the sentence to show,
/// the way camo.rs's own rejections are.
fn camo_png(body: &CamoBody) -> Result<Option<Vec<u8>>, crate::camo::Rejected> {
    let Some(encoded) = body.png.as_deref() else {
        return Ok(None);
    };
    let Ok(bytes) = base64::engine::general_purpose::STANDARD.decode(encoded.as_bytes()) else {
        return Err("The image could not be read.");
    };
    // Re-encoded rather than passed through. This is not the security
    // boundary it looks like — a player can write any bytes to their own repo
    // with any client — but a camo that arrives here malformed should not
    // become a camo other players' browsers try to decode (camo.rs).
    crate::camo::reencode(&bytes).map(Some)
}

fn write_failed(error: crate::atproto::WriteError) -> Response {
    match error {
        // 403, not 401: the session is real and the cookie is fine. What we
        // could not do is use it, and saying so is the difference between
        // "sign in again" and a puzzle.
        crate::atproto::WriteError::NeedsSignIn => message(
            StatusCode::FORBIDDEN,
            "Your session could not be used to write to your account. Sign out and sign in again.",
        ),
        crate::atproto::WriteError::Refused(refusal) => refused(&refusal),
        crate::atproto::WriteError::NotFound => {
            message(StatusCode::NOT_FOUND, "That camo is not in your account.")
        }
        // A client bug, not a player problem and not the account's server:
        // 400 rather than 502, and loud rather than dropping the settings and
        // saving anyway, because a build that sends these needs to find out.
        crate::atproto::WriteError::Unstorable => message(
            StatusCode::BAD_REQUEST,
            "This camo's editor settings could not be stored.",
        ),
        crate::atproto::WriteError::Conflict => message(
            StatusCode::CONFLICT,
            "That camo changed somewhere else. Reload before saving over it.",
        ),
        crate::atproto::WriteError::Upstream => message(
            StatusCode::BAD_GATEWAY,
            "Your account's server would not accept the change.",
        ),
    }
}

/// The account's own server said no. Its words are what the player is shown.
///
/// Nothing here decides what the refusal meant. A missing permission arrives
/// as `Missing required scope "repo:app.userinput.discussion?action=create"`,
/// which names the fix precisely and which this service cannot reconstruct on
/// its own — the granted scope is the expanded form of what was asked for, so
/// a permission set's members are only ever known over there. Re-consent is
/// added as a second sentence when it could be the fix, and nothing is
/// claimed about which of the two it was.
fn refused(refusal: &crate::atproto::Refusal) -> Response {
    let mut text = match refusal.detail() {
        Some(detail) => format!("Your account's server refused this: {detail}"),
        None => "Your account's server refused this.".to_owned(),
    };
    if refusal.fixable_by_signing_in() {
        text.push_str(" Sign out and sign in again if lance.blue has asked for a new permission since you signed in.");
    }
    // 403 for a refusal about this account's permission — the request reached
    // the right server and was understood. 502 for anything else it refused:
    // that is the far end misbehaving, not the player.
    let status = if refusal.fixable_by_signing_in() {
        StatusCode::FORBIDDEN
    } else {
        StatusCode::BAD_GATEWAY
    };
    message(status, &text)
}

fn written(w: crate::atproto::Written) -> Response {
    // The rkey is the last path segment of the at-uri. Handing it back saves
    // the site parsing a URI to name what it just wrote.
    let rkey = w.uri.rsplit('/').next().unwrap_or_default().to_owned();
    Json(serde_json::json!({ "uri": w.uri, "cid": w.cid, "rkey": rkey })).into_response()
}

/// Saves a new camo to the signed-in player's own repository.
async fn create_camo(
    State(state): State<AppState>,
    jar: CookieJar,
    body: Option<Json<CamoBody>>,
) -> Response {
    let Some(did) = session_did(&state, &jar) else {
        return StatusCode::UNAUTHORIZED.into_response();
    };
    let Some(Json(body)) = body else {
        return message(
            StatusCode::BAD_REQUEST,
            "Send a JSON body with a name and a PNG.",
        );
    };

    let name = match crate::camo::clean_name(body.name.as_deref().unwrap_or_default()) {
        Ok(name) => name,
        Err(why) => return message(StatusCode::BAD_REQUEST, why),
    };
    let png = match camo_png(&body) {
        Ok(Some(png)) => png,
        Ok(None) => return message(StatusCode::BAD_REQUEST, "Send the camo image."),
        Err(why) => return message(StatusCode::BAD_REQUEST, why),
    };

    match state
        .atproto
        .create_camo(&did, &name, png, body.source)
        .await
    {
        Ok(w) => written(w),
        Err(e) => write_failed(e),
    }
}

/// Renames a camo, changes its picture, or both.
///
/// A body naming only `name` is a rename, and the picture and settings are
/// left exactly as they were — including any field this service has never
/// heard of, because the record is read and put back rather than rebuilt.
async fn update_camo(
    State(state): State<AppState>,
    jar: CookieJar,
    Path(rkey): Path<String>,
    body: Option<Json<CamoBody>>,
) -> Response {
    let Some(did) = session_did(&state, &jar) else {
        return StatusCode::UNAUTHORIZED.into_response();
    };
    let Some(Json(body)) = body else {
        return message(StatusCode::BAD_REQUEST, "Send a JSON body.");
    };

    let name = match body.name.as_deref() {
        Some(raw) => match crate::camo::clean_name(raw) {
            Ok(name) => Some(name),
            Err(why) => return message(StatusCode::BAD_REQUEST, why),
        },
        None => None,
    };
    let png = match camo_png(&body) {
        Ok(png) => png,
        Err(why) => return message(StatusCode::BAD_REQUEST, why),
    };
    if name.is_none() && png.is_none() && body.source.is_none() {
        return message(StatusCode::BAD_REQUEST, "Nothing to change.");
    }
    // Absent means leave the settings alone; present means replace them.
    // There is no way to ask for their removal, because the editor has no
    // reason to: a camo it can no longer describe keeps the description it
    // had.
    let source = body.source.map(Some);

    match state
        .atproto
        .update_camo(
            &did,
            &rkey,
            name.as_deref(),
            png,
            source,
            body.swap_record.as_deref(),
        )
        .await
    {
        Ok(w) => written(w),
        Err(e) => write_failed(e),
    }
}

/// Deletes a camo from the signed-in player's own repository.
async fn delete_camo(
    State(state): State<AppState>,
    jar: CookieJar,
    Path(rkey): Path<String>,
) -> Response {
    let Some(did) = session_did(&state, &jar) else {
        return StatusCode::UNAUTHORIZED.into_response();
    };
    match state.atproto.delete_camo(&did, &rkey).await {
        Ok(()) => StatusCode::NO_CONTENT.into_response(),
        Err(e) => write_failed(e),
    }
}

/// The atproto record-key syntax: 1..=512 of `A-Za-z0-9.-_~:`, and neither
/// `.` nor `..`. Duplicated from the spec rather than borrowed from atrium's
/// `RecordKey`, because that type lives behind the atproto module's wall and
/// this is a route deciding whether to store a string.
fn is_record_key(rkey: &str) -> bool {
    !rkey.is_empty()
        && rkey.len() <= 512
        && rkey != "."
        && rkey != ".."
        && rkey
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || b"-._~:".contains(&b))
}

#[derive(serde::Deserialize)]
struct DefaultCamoBody {
    /// The record key to reach for when a match names none, or null to go
    /// back to having no default.
    rkey: Option<String>,
}

/// Remembers which camo to launch with.
///
/// Whether the camo *exists* is not checked: it is a key this service hands
/// back to itself later, the read that uses it fails softly, and a camo can be
/// deleted a second after this returns — so a round trip here would buy no
/// guarantee. That it is a *record key* is checked, because this value is
/// stored and then echoed on every `/api/session` for as long as the account
/// lasts.
async fn set_default_camo(
    State(state): State<AppState>,
    jar: CookieJar,
    Json(body): Json<DefaultCamoBody>,
) -> Response {
    let Some(did) = session_did(&state, &jar) else {
        return StatusCode::UNAUTHORIZED.into_response();
    };
    let rkey = body.rkey.filter(|r| !r.is_empty());
    if rkey.as_deref().is_some_and(|r| !is_record_key(r)) {
        return message(StatusCode::BAD_REQUEST, "That is not a camo.");
    }
    match state.db.set_default_camo(&did, rkey.as_deref()).await {
        Ok(()) => {
            tracing::info!(did, set = rkey.is_some(), "camo: default changed");
            Json(serde_json::json!({ "defaultCamo": rkey })).into_response()
        }
        Err(_) => {
            tracing::error!(did, "camo: default could not be saved");
            message(
                StatusCode::INTERNAL_SERVER_ERROR,
                "That could not be saved just now.",
            )
        }
    }
}

// --- flare -------------------------------------------------------------
//
// One route, same reason as camo: the browser holds no token, so a post to
// the player's own repository is made from here on their behalf. The
// destination is not our collection at all — the record is userinput.app's
// discussion, aimed at the board space FLARE_SPACE names.

#[derive(serde::Deserialize)]
struct FlareBody {
    title: Option<String>,
    body: Option<String>,
    /// What the player ticked: values from the board's tag list, which both
    /// forms show as checkboxes. Held to the lexicon's shape here; which
    /// values the board actually offers is the board's business.
    tags: Option<Vec<String>>,
    /// The match the flare was sent from. Named so the report can be traced
    /// to a container run; refused unless the sender fights in it.
    #[serde(rename = "matchId")]
    match_id: Option<String>,
    /// The screenshot PNG, base64.
    png: Option<String>,
}

/// Files a flare to the board: an `app.userinput.discussion` written to the
/// signed-in player's own repository.
async fn create_flare(
    State(state): State<AppState>,
    jar: CookieJar,
    body: Option<Json<FlareBody>>,
) -> Response {
    let Some(did) = session_did(&state, &jar) else {
        return StatusCode::UNAUTHORIZED.into_response();
    };
    let Some(space) = state.flare_space.clone() else {
        return message(StatusCode::NOT_FOUND, "Flares are not enabled here.");
    };
    let Some(Json(body)) = body else {
        return message(StatusCode::BAD_REQUEST, "Send a JSON body with a title.");
    };

    let title = match crate::flare::clean_title(body.title.as_deref().unwrap_or_default()) {
        Ok(title) => title,
        Err(why) => return message(StatusCode::BAD_REQUEST, why),
    };
    let mut text = match crate::flare::clean_body(body.body.as_deref().unwrap_or_default()) {
        Ok(text) => text,
        Err(why) => return message(StatusCode::BAD_REQUEST, why),
    };

    // The footer is appended here, not sent by the page: the page could say
    // anything, and the one claim worth carrying — this DID fights in this
    // match — is one only the match table can make.
    if let Some(match_id) = &body.match_id {
        if !state
            .db
            .is_human_player(match_id, &did)
            .await
            .unwrap_or(false)
        {
            return message(StatusCode::FORBIDDEN, "That is not your match.");
        }
        let build = state.build_ref.as_deref().unwrap_or("dev");
        let arena = state.arena_version.as_deref().unwrap_or("unset");
        text.push_str(&format!(
            "\n\n[match {match_id} · arena {arena} · hq {build}]"
        ));
    }

    let tags = match crate::flare::clean_tags(body.tags.as_deref().unwrap_or_default()) {
        Ok(tags) => tags,
        Err(why) => return message(StatusCode::BAD_REQUEST, why),
    };

    let png = match &body.png {
        None => None,
        Some(encoded) => {
            let Ok(bytes) = base64::engine::general_purpose::STANDARD.decode(encoded.as_bytes())
            else {
                return message(StatusCode::BAD_REQUEST, "The screenshot could not be read.");
            };
            match crate::flare::reencode_screenshot(&bytes) {
                Ok(png) => Some(png),
                Err(why) => return message(StatusCode::BAD_REQUEST, why),
            }
        }
    };

    match state
        .atproto
        .create_flare(&did, &title, &text, &tags, png, &space)
        .await
    {
        Ok(w) => {
            tracing::info!(
                did,
                from_match = body.match_id.is_some(),
                tags = tags.len(),
                "flare: filed"
            );
            written(w)
        }
        // write_failed, like every other write: its messages name no
        // collection now, because the server's own refusal names the one that
        // was refused.
        Err(e) => write_failed(e),
    }
}

/// The catalog the challenge screen builds its scenario picker from. Static
/// data out of the arena image, so no session — like /api/version.
async fn list_scenarios() -> Response {
    Json(serde_json::json!({
        "scenarios": crate::matches::scenarios::all(),
    }))
    .into_response()
}

/// Everyone the signed-in player has shared a match with, latest first —
/// the challenge screen's "recent opponents" strip.
async fn list_opponents(State(state): State<AppState>, jar: CookieJar) -> Response {
    let Some(did) = session_did(&state, &jar) else {
        return StatusCode::UNAUTHORIZED.into_response();
    };
    match state.db.recent_opponents(&did, 12).await {
        Ok(opponents) => Json(serde_json::json!({ "opponents": opponents })).into_response(),
        Err(_) => {
            tracing::error!(did, "opponents: recent-opponents lookup failed");
            StatusCode::INTERNAL_SERVER_ERROR.into_response()
        }
    }
}

#[derive(serde::Serialize)]
struct SuggestedOpponent {
    did: String,
    handle: Option<String>,
}

/// Bluesky accounts the signed-in player follows who have played lance.blue
/// before, with anyone already faced left out — new discovery, not a re-sort
/// of GET /api/opponents.
///
/// A separate route, fetched separately by the challenge screen, because
/// this one leaves the process to ask a server we do not control on an
/// account list of unpredictable size. That must never hold up the local
/// list, which the database always answers immediately; an unreachable or
/// slow AppView degrades this to an empty suggestions list, never a failed
/// challenge screen.
async fn list_suggested_opponents(State(state): State<AppState>, jar: CookieJar) -> Response {
    let Some(did) = session_did(&state, &jar) else {
        return StatusCode::UNAUTHORIZED.into_response();
    };

    let follows = match state.atproto.followed(&did).await {
        Ok(follows) => follows,
        Err(_) => return Json(serde_json::json!({ "suggested": [] })).into_response(),
    };
    if follows.is_empty() {
        return Json(serde_json::json!({ "suggested": [] })).into_response();
    }

    let candidates: Vec<String> = follows.iter().map(|f| f.did.clone()).collect();
    let played = match state.db.played_dids(&candidates).await {
        Ok(played) => played,
        Err(_) => {
            tracing::error!(did, "opponents: played-dids lookup failed");
            return StatusCode::INTERNAL_SERVER_ERROR.into_response();
        }
    };
    let faced = match state.db.all_opponent_dids(&did).await {
        Ok(faced) => faced,
        Err(_) => {
            tracing::error!(did, "opponents: all-opponent-dids lookup failed");
            return StatusCode::INTERNAL_SERVER_ERROR.into_response();
        }
    };

    let suggested: Vec<_> = follows
        .into_iter()
        .filter(|f| f.did != did && played.contains(&f.did) && !faced.contains(&f.did))
        .map(|f| SuggestedOpponent {
            did: f.did,
            // "handle.invalid" is the AppView's placeholder for a handle
            // that no longer resolves; treated the same as recent_opponents
            // treats a NULL handle — the chip is shown but stays inert.
            handle: (f.handle != "handle.invalid").then_some(f.handle),
        })
        .take(20)
        .collect();

    Json(serde_json::json!({ "suggested": suggested })).into_response()
}

#[derive(serde::Deserialize)]
struct CreateMatchBody {
    /// The legacy shape: `"bot"` launches the configured default scenario
    /// against Princess. Ignored when `players` is present.
    #[serde(default)]
    mode: Option<String>,
    /// Which of the player's camo to paint their units in, as the record key
    /// of a `blue.lance.camo` in their own repository. Optional: a player who
    /// has never made one, or who picks none, gets the default colours.
    ///
    /// A key rather than an image. The browser holds the picture already and
    /// could send it, but then the thing painted on the units would be
    /// whatever the browser said it was rather than what the player
    /// published, and there would be nothing to check it against.
    #[serde(default)]
    camo: Option<String>,
    /// A bot's camo, as a base64 84x72 PNG the browser rolled for this
    /// match.
    ///
    /// An image and not a key, which is the opposite of the field above and
    /// for the opposite reason: the bot has no account and nothing published,
    /// so there is no record to name and nothing to check the picture
    /// against. It is generated, sanitised, staged and forgotten.
    #[serde(default, rename = "botCamo")]
    bot_camo: Option<String>,
    /// The new shape: a scenario path from the catalog…
    #[serde(default)]
    scenario: Option<String>,
    /// …and one entry per slot of that scenario.
    #[serde(default)]
    players: Option<Vec<PlayerBody>>,
}

#[derive(serde::Deserialize)]
struct PlayerBody {
    /// A faction name from the scenario's `Factions=` line.
    slot: String,
    /// `self`, `bot`, or `player` — who takes the seat.
    control: String,
    /// The invited account, for `player` seats. Resolved and verified here;
    /// the browser's idea of a DID is never trusted.
    #[serde(default)]
    handle: Option<String>,
    /// The caller's own camo choice, for the `self` seat. Same three states
    /// as the top-level field: absent is "my default", empty is "none".
    #[serde(default)]
    camo: Option<String>,
    /// A rolled camo for a `bot` seat, base64.
    #[serde(default, rename = "botCamo")]
    bot_camo: Option<String>,
}

/// Camo the browser rolled for the bot forces of a lobby or a daily.
///
/// The lobby's two launch triggers used to carry no body at all, and bot seats
/// got `camo_png: None` as a documented scope cut: nobody had a browser in
/// front of them to roll one. They do now - the Forces block has a Shuffle
/// beside every bot force - and the rule that made this worth wiring up is
/// that the roll shown must be the roll sent. A preview of a camo the match
/// then ignores is worse than no preview.
///
/// Optional in every sense: no body, an empty map, or a slot nothing rolled
/// for all mean the same thing, which is the behaviour that was here before.
#[derive(serde::Deserialize, Default)]
struct LaunchBody {
    /// Base64 PNG per faction name, for bot seats only.
    #[serde(default, rename = "botCamo")]
    bot_camo: std::collections::HashMap<String, String>,
    /// A base64 PNG the caller rolled for their own seat.
    ///
    /// Bytes rather than a record key, because a rolled camo is not a record:
    /// nothing was written to the player's repository, so there is nothing to
    /// point at. It wins over their stored default for this match and is gone
    /// afterwards.
    #[serde(default)]
    camo: Option<String>,
}

/// The caller's camo choice settled against their stored default. The
/// request wins; the stored default is the fallback. A request that means
/// "no camo at all" says so with an empty string, because omitting the field
/// is how a client that predates this asks for the default.
async fn settle_own_camo(state: &AppState, did: &str, chosen: Option<String>) -> Option<String> {
    match chosen {
        Some(rkey) if rkey.is_empty() => None,
        Some(rkey) => Some(rkey),
        // Soft here, unlike /api/session: the match is worth starting without
        // a camo, and the alternative is refusing to launch over a cosmetic.
        // Logged, because otherwise a database fault looks like a player who
        // never set one.
        None => match state.db.default_camo(did).await {
            Ok(rkey) => rkey,
            Err(_) => {
                tracing::warn!(did, "camo: default could not be read; starting without one");
                None
            }
        },
    }
}

/// Never fatal. Nobody asked for this camo by name and nobody will miss a
/// particular one, so bytes that do not decode are a bot in default colours
/// rather than a refused launch. The re-encode that follows it is in
/// matches::stage_png, where the player's camo meets the same gate.
fn decode_rolled_camo(did: &str, encoded: Option<String>) -> Option<Vec<u8>> {
    encoded.and_then(|encoded| {
        base64::engine::general_purpose::STANDARD
            .decode(encoded.as_bytes())
            .inspect_err(|_| tracing::warn!(did, "camo: the bot's was not base64"))
            .ok()
    })
}

/// Turns the new request shape into launcher seats: scenario checked against
/// the catalog, every slot filled exactly once, exactly one seat taken by the
/// caller, and every invited handle resolved to the account it names.
async fn plan_players(
    state: &AppState,
    did: &str,
    scenario_path: &str,
    bodies: Vec<PlayerBody>,
) -> Result<Vec<crate::matches::NewPlayer>, Response> {
    // The scenario's own sides, by slot. Where a setup starts, and for now the
    // only thing with an opinion about it.
    let team_of = |slot: &str| -> Option<i32> {
        crate::matches::scenarios::find(scenario_path)
            .and_then(|s| s.slots.iter().find(|x| x.name == slot))
            .and_then(|x| x.team)
    };
    let Some(scenario) = crate::matches::scenarios::find(scenario_path) else {
        return Err(message(
            StatusCode::BAD_REQUEST,
            "That scenario is not available.",
        ));
    };

    // The slots of the request must be the slots of the scenario, each once.
    // Strict on purpose: the screen always sends the full mapping, and a gap
    // silently becoming a bot is how a mis-mapped slot hands someone else's
    // units to a player.
    let mut wanted: Vec<&str> = scenario
        .slots
        .iter()
        .map(|slot| slot.name.as_str())
        .collect();
    let mut got: Vec<&str> = bodies.iter().map(|player| player.slot.as_str()).collect();
    wanted.sort_unstable();
    got.sort_unstable();
    if wanted != got {
        return Err(message(
            StatusCode::BAD_REQUEST,
            "Fill every slot of the scenario, each exactly once.",
        ));
    }

    if bodies.iter().filter(|p| p.control == "self").count() != 1 {
        return Err(message(
            StatusCode::BAD_REQUEST,
            "Take exactly one slot yourself.",
        ));
    }

    let mut players = Vec::with_capacity(bodies.len());
    let mut seated_dids = vec![did.to_owned()];
    for body in bodies {
        match body.control.as_str() {
            "self" => {
                let handle = state.db.account_handle(did).await.ok().flatten().flatten();
                let camo = settle_own_camo(state, did, body.camo).await;
                let team = team_of(&body.slot);
                players.push(crate::matches::NewPlayer {
                    team,
                    slot: body.slot,
                    bot: false,
                    did: Some(did.to_owned()),
                    handle,
                    camo_rkey: camo,
                    camo_png: None,
                });
            }
            "bot" => {
                let team = team_of(&body.slot);
                players.push(crate::matches::NewPlayer {
                    team,
                    slot: body.slot,
                    bot: true,
                    did: None,
                    handle: None,
                    camo_rkey: None,
                    camo_png: decode_rolled_camo(did, body.bot_camo),
                });
            }
            "player" => {
                let Some(typed) = body
                    .handle
                    .as_deref()
                    .map(str::trim)
                    .filter(|h| !h.is_empty())
                else {
                    return Err(message(
                        StatusCode::BAD_REQUEST,
                        "Name the player for every invited slot.",
                    ));
                };
                let Some((other_did, handle)) = state.atproto.resolve_player(typed).await else {
                    return Err(message(
                        StatusCode::BAD_REQUEST,
                        &format!("{typed} could not be found."),
                    ));
                };
                if seated_dids.contains(&other_did) {
                    return Err(message(
                        StatusCode::BAD_REQUEST,
                        &format!("{handle} already has a slot in this match."),
                    ));
                }
                seated_dids.push(other_did.clone());
                // Their stored default camo, if they have signed in here and
                // set one. Soft like the caller's: the read that stages it
                // is public, and a missing default is default colours.
                let camo = state.db.default_camo(&other_did).await.unwrap_or(None);
                let team = team_of(&body.slot);
                players.push(crate::matches::NewPlayer {
                    team,
                    slot: body.slot,
                    bot: false,
                    did: Some(other_did),
                    handle: Some(handle),
                    camo_rkey: camo,
                    camo_png: None,
                });
            }
            _ => {
                return Err(message(
                    StatusCode::BAD_REQUEST,
                    "A slot is either yours, a bot's, or another player's.",
                ));
            }
        }
    }
    Ok(players)
}

async fn create_match(
    State(state): State<AppState>,
    jar: CookieJar,
    body: Option<Json<CreateMatchBody>>,
) -> Response {
    let Some(did) = session_did(&state, &jar) else {
        return StatusCode::UNAUTHORIZED.into_response();
    };
    let Some(Json(body)) = body else {
        return message(StatusCode::BAD_REQUEST, "Send a JSON body with a mode.");
    };

    // Shape questions are answered before configuration ones: a request this
    // instance could never launch still deserves "that is not a thing to
    // ask" over 503 when it is malformed.
    let planned = match (body.scenario, body.players) {
        // The new shape: a scenario from the catalog, every slot mapped.
        (Some(scenario), Some(bodies)) => {
            let players = match plan_players(&state, &did, &scenario, bodies).await {
                Ok(players) => players,
                Err(response) => return response,
            };
            Some((scenario, players))
        }
        (Some(_), None) | (None, Some(_)) => {
            return message(
                StatusCode::BAD_REQUEST,
                "Send the scenario and its players together.",
            );
        }
        (None, None) => match body.mode.as_deref() {
            Some("bot") => None,
            Some(_) => return message(StatusCode::BAD_REQUEST, "Not yet available."),
            None => return message(StatusCode::BAD_REQUEST, "Send a JSON body with a mode."),
        },
    };

    let Some(matches) = state.matches.as_ref() else {
        return message(
            StatusCode::SERVICE_UNAVAILABLE,
            "Match launching is not configured here.",
        );
    };

    let (scenario, players) = match planned {
        Some(planned) => planned,
        // The legacy shape: the configured scenario, caller vs Princess.
        None => {
            let handle = state.db.account_handle(&did).await.ok().flatten().flatten();
            let camo = settle_own_camo(&state, &did, body.camo).await;
            let config = matches.match_config();
            (
                config.scenario_name.clone(),
                vec![
                    crate::matches::NewPlayer {
                        slot: config.human_slot.clone(),
                        bot: false,
                        did: Some(did.clone()),
                        handle,
                        camo_rkey: camo,
                        camo_png: None,
                        // The configured fallback scenario is not in the
                        // catalog, so nothing here knows its sides.
                        team: None,
                    },
                    crate::matches::NewPlayer {
                        slot: config.bot_slot.clone(),
                        bot: true,
                        did: None,
                        handle: None,
                        camo_rkey: None,
                        team: None,
                        camo_png: decode_rolled_camo(&did, body.bot_camo),
                    },
                ],
            )
        }
    };

    // Carried when the catalog has it, so arena plays the bytes this build
    // parsed. The configured fallback scenario is not in the catalog and has
    // no bytes here, so that one is still named and comes out of the image.
    let source = match crate::matches::scenarios::find(&scenario) {
        Some(found) => crate::matches::manifest::ScenarioSource::Carried(found.text),
        None => crate::matches::manifest::ScenarioSource::Library(&scenario),
    };
    match matches
        .start(&state.atproto, &did, source, &scenario, players)
        .await
    {
        Ok(id) => Json(serde_json::json!({ "matchId": id })).into_response(),
        Err(_) => message(
            StatusCode::INTERNAL_SERVER_ERROR,
            "The match could not be started.",
        ),
    }
}

// --- lobby ---------------------------------------------------------------
//
// The pre-match live-sync slice: a `lobby`-status row holding a scenario
// choice and the seats being claimed for it, kept in sync over a WebSocket
// every connected browser shares, which as of this slice also carries chat.
// No seat gating on the socket itself — any signed-in player may connect to
// any lobby's channel, see its scenario and seats, change the scenario, and
// talk. That is deliberately permissive: nothing exposed here is sensitive
// (a scenario path, who has claimed which slot), and tightening it belongs
// to future work, not this slice.
//
// Chat is the first thing on this channel a player *authors* rather than
// selects from a fixed set, and it is deliberately the least durable thing
// here: never stored, never published, gone on restart, capped at the last
// `CHAT_HISTORY` lines per lobby (see `matches::lobby`). What that leaves
// unaddressed is abuse — there is no rate limit, no mute and no report path
// on this channel yet, only a length cap. A lobby is a small room the people
// in it opened, which is why that is tolerable for now rather than
// indefinitely; see plan/user-safety.md.
//
// Claiming a seat is narrower than connecting: `claim_seat` only ever seats
// the caller's own session DID — there is no "claim for someone else" on
// this channel, the same way `plan_players`'s `self` control only ever
// resolves to the caller. Inviting a handle who is not connected is the
// third resolver, `invite_handle` (`matches::lobby`), and it does use the
// same async `Atproto::resolve_player` lookup `plan_players`'s `player`
// control does — which is exactly why `lobby_socket` spawns it rather than
// awaiting it inline: a DNS-over-HTTPS round trip on the connection's own
// `tokio::select!` loop would leave that connection unable to read further
// messages or relay broadcasts until the resolution finished. It seats the
// resolved DID the moment resolution succeeds, same as `plan_players`
// always has — see plan/invitations.md for why neither of those asks first.
//
// A team (`LobbyRequest::SetTeam`, `matches::lobby::Lobbies::set_team`) is
// the first thing on this channel that changes the *fight* rather than
// filling in who is in it, and the first with a two-identity rule: the seat's
// holder may move their own seat, and the lobby's owner may move any. It is
// stored on the seat row rather than in memory, because it is a setup
// decision that outlives a socket, and resolved against the scenario's own
// team exactly once, in `lobby_seats_to_players`.
//
// Readiness (`LobbyRequest::SetReady`, `matches::lobby::Lobbies::
// set_ready`) is the one thing on this channel that gates anything: every
// seated human except whoever presses Deploy has to have pressed Ready
// before `deploy_lobby` will launch, and any seat or scenario change drops
// all of them. Setting one is as permissive as chat — anyone connected may,
// and an unseated caller's readiness simply gates nothing.
//
// Kicking a seat (`LobbyRequest::Kick`, `matches::lobby::Lobbies::
// kick_seat`) is the one request on this channel that is not everybody's to
// send: only the lobby's own opener may remove a seat, checked freshly
// against the row on every attempt the same way `deploy_lobby` below checks
// ownership before a launch. Everything else here needs only a valid
// session; kicking needs to be that specific caller — see `kick_seat`'s own
// doc for why that check lives in `matches::lobby` rather than here beside
// `deploy_lobby`'s copy of the same comparison.

#[derive(serde::Deserialize)]
struct OpenLobbyBody {
    scenario: String,
}

async fn open_lobby(
    State(state): State<AppState>,
    jar: CookieJar,
    body: Option<Json<OpenLobbyBody>>,
) -> Response {
    let Some(did) = session_did(&state, &jar) else {
        return StatusCode::UNAUTHORIZED.into_response();
    };
    let Some(Json(body)) = body else {
        return message(StatusCode::BAD_REQUEST, "Send a JSON body with a scenario.");
    };
    // The same catalog check `plan_players` makes before a launch — reused
    // rather than reinvented, and made here rather than inside `Lobbies` so
    // it stays the one place a scenario path is judged against the catalog.
    if crate::matches::scenarios::find(&body.scenario).is_none() {
        return message(StatusCode::BAD_REQUEST, "That scenario is not available.");
    }
    match state.lobbies.open(&did, &body.scenario).await {
        Some(id) => Json(serde_json::json!({ "matchId": id })).into_response(),
        None => message(
            StatusCode::INTERNAL_SERVER_ERROR,
            "The lobby could not be opened.",
        ),
    }
}

/// Upgrades to the lobby's live channel. Auth is a valid session and nothing
/// more — see the module note above for why that is enough for now.
async fn lobby_live(
    State(state): State<AppState>,
    Path(id): Path<String>,
    jar: CookieJar,
    ws: WebSocketUpgrade,
) -> Response {
    let Some(did) = session_did(&state, &jar) else {
        return StatusCode::UNAUTHORIZED.into_response();
    };
    let row = match state.db.get_match(&id).await {
        Ok(Some(row)) if row.status == "lobby" => row,
        // Once it starts, the lobby channel is not the mechanism anymore —
        // the waiting screen's poll is. No existence oracle either way: a
        // match that is not a lobby answers the same as one that never
        // seated this caller would, on the routes that do check seats.
        Ok(Some(_)) => return StatusCode::CONFLICT.into_response(),
        Ok(None) => return StatusCode::NOT_FOUND.into_response(),
        Err(_) => {
            tracing::error!(did, match_id = id, "lobby: match lookup failed");
            return StatusCode::INTERNAL_SERVER_ERROR.into_response();
        }
    };
    let lobbies = state.lobbies.clone();
    let atproto = state.atproto.clone();
    ws.on_upgrade(move |socket| {
        lobby_socket(
            socket,
            lobbies,
            atproto,
            id,
            row.owner_did,
            row.scenario,
            did,
        )
    })
}

/// What a client may send on a lobby's live channel. Tagged so this
/// distinguishes cleanly from `LobbyEvent`, the shape the server sends back
/// (see `matches::lobby`): a request names an action, a broadcast carries a
/// full snapshot, and conflating them would make "what did I just receive"
/// ambiguous the moment a message shape overlaps by accident.
#[derive(serde::Deserialize)]
#[serde(tag = "type", rename_all = "camelCase")]
enum LobbyRequest {
    Scenario {
        scenario: String,
    },
    /// Claims `slot` for the caller's own session DID — never anyone else's.
    ClaimSeat {
        slot: String,
    },
    /// Assigns `slot` to the bot.
    BotSeat {
        slot: String,
    },
    /// Says something to everyone connected to this lobby, as the caller's
    /// own session DID — a lobby has no more notion of speaking for someone
    /// else than it has of claiming a seat for them.
    Chat {
        text: String,
    },
    /// Invites `handle` — who need not be connected to this lobby, or to
    /// anything, right now — into `slot`. See `matches::lobby::Lobbies::
    /// invite_handle` for the resolution and seating this triggers, and the
    /// note above `lobby_socket` for why it runs off this loop rather than
    /// on it.
    Invite {
        slot: String,
        handle: String,
    },
    /// Removes whoever holds `slot`, freeing it for someone else to claim.
    /// Only the lobby's own opener may do this — see
    /// `matches::lobby::Lobbies::kick_seat` for the authorization check and
    /// why it lives there rather than here.
    Kick {
        slot: String,
    },
    /// Holds `slot` open for somebody, or lets it go. Host only — see
    /// `matches::lobby::Lobbies::set_lock`.
    SetLock {
        slot: String,
        locked: bool,
    },
    /// Hands the lobby to a seated player. Host only, and the host's own to
    /// give: see `matches::lobby::Lobbies::make_host`.
    MakeHost {
        did: String,
    },
    /// Puts `slot` on `team`, or clears the override (`null`) so the
    /// scenario's own team stands again. The seat's holder or the lobby's
    /// owner — see `matches::lobby::Lobbies::set_team`.
    SetTeam {
        slot: String,
        team: Option<i32>,
    },
    /// Marks the caller ready to fight the lineup as it stands, or takes
    /// that back. Nobody may set anyone else's, the same as a seat claim or
    /// a chat line. `deploy_lobby` is what it gates — see
    /// `matches::lobby::Lobbies::set_ready`.
    SetReady {
        ready: bool,
    },
}

/// One connection's whole life: who owns the lobby, its current scenario,
/// seats and readiness, then the chat backlog, then whatever arrives on the
/// channel, forever, while also relaying anything this client sends.
/// `Lobbies`'s methods do the validating, storing and broadcasting; this
/// loop is only wiring.
///
/// The owner goes first and is never repeated: it is the one thing here
/// that cannot change, and a client needs it before the seat map means
/// anything — the owner is the player nobody waits on for a Ready.
///
/// The backlog goes last of the three because it is the only one that is not
/// state: a client can render the lobby from the scenario and seats alone,
/// and what has already been said is scrollback under it. Sent as one `chat`
/// event per line, identical to a line said a second later — see
/// `LobbyEvent::Chat`.
async fn lobby_socket(
    mut socket: WebSocket,
    lobbies: Arc<crate::matches::lobby::Lobbies>,
    atproto: Arc<Atproto>,
    match_id: String,
    owner_did: String,
    initial_scenario: Option<String>,
    did: String,
) {
    // Presence is taken after the subscribe and never before it: `arrive`
    // broadcasts the room it just joined, so this connection is told who is
    // here by the same message everyone else is, rather than by a separate
    // initial send that could disagree with it. The guard lives for the
    // whole of this function, which is what makes every way out of it —
    // clean close, read error, or an early return below — a departure.
    let handle = lobbies.account_handle(&did).await;

    // Subscribing and taking the backlog in one call is what makes each line
    // arrive exactly once on a connection that opens mid-conversation; see
    // `Lobbies::join`.
    let crate::matches::lobby::Joined {
        mut updates,
        locked,
        ready,
        chat: backlog,
    } = lobbies.join(&match_id);
    let _here = lobbies.arrive(&match_id, &did, handle);

    if !send_event(
        &mut socket,
        &crate::matches::lobby::LobbyEvent::Owner { owner: owner_did },
    )
    .await
    {
        return;
    }
    if let Some(scenario) = initial_scenario
        && !send_event(
            &mut socket,
            &crate::matches::lobby::LobbyEvent::Scenario { scenario },
        )
        .await
    {
        return;
    }
    let seats = lobbies.seats(&match_id).await;
    if !send_event(
        &mut socket,
        &crate::matches::lobby::LobbyEvent::Seats { seats },
    )
    .await
    {
        return;
    }
    // With the seats, not after the backlog: who is ready and which slots
    // are held open are both state a client renders the lobby from, the
    // same as the seat map they are about.
    if !send_event(
        &mut socket,
        &crate::matches::lobby::LobbyEvent::Locked { locked },
    )
    .await
    {
        return;
    }
    if !send_event(
        &mut socket,
        &crate::matches::lobby::LobbyEvent::Ready { ready },
    )
    .await
    {
        return;
    }
    for line in backlog {
        if !send_event(&mut socket, &crate::matches::lobby::LobbyEvent::Chat(line)).await {
            return;
        }
    }

    loop {
        tokio::select! {
            incoming = socket.recv() => {
                match incoming {
                    Some(Ok(Message::Text(text))) => {
                        match serde_json::from_str::<LobbyRequest>(&text) {
                            Ok(LobbyRequest::Scenario { scenario }) => {
                                lobbies.set_scenario(&match_id, &scenario).await;
                            }
                            Ok(LobbyRequest::ClaimSeat { slot }) => {
                                lobbies.claim_seat(&match_id, &slot, &did).await;
                            }
                            Ok(LobbyRequest::BotSeat { slot }) => {
                                lobbies.assign_bot(&match_id, &slot, &did).await;
                            }
                            Ok(LobbyRequest::Chat { text }) => {
                                if !lobbies.say(&match_id, &did, &text).await {
                                    // Blank or over the length cap. Debug
                                    // rather than warn: a client that lets
                                    // one through is a client-side
                                    // validation gap, not an incident.
                                    tracing::debug!(
                                        match_id,
                                        did,
                                        chars = text.chars().count(),
                                        "lobby: chat line refused"
                                    );
                                }
                            }
                            Ok(LobbyRequest::SetLock { slot, locked }) => {
                                lobbies.set_lock(&match_id, &slot, locked, &did).await;
                            }
                            Ok(LobbyRequest::MakeHost { did: new_owner }) => {
                                lobbies.make_host(&match_id, &new_owner, &did).await;
                            }
                            Ok(LobbyRequest::SetTeam { slot, team }) => {
                                // Authorization and its logging both live in
                                // `set_team`, the same as `kick_seat`.
                                lobbies.set_team(&match_id, &slot, team, &did).await;
                            }
                            Ok(LobbyRequest::SetReady { ready }) => {
                                let changed = lobbies.set_ready(&match_id, &did, ready);
                                tracing::debug!(
                                    match_id,
                                    did,
                                    ready,
                                    changed,
                                    "lobby: readiness set"
                                );
                            }
                            Ok(LobbyRequest::Kick { slot }) => {
                                // Authorization and logging both happen
                                // inside `kick_seat` itself — a non-owner's
                                // attempt is already logged there at `warn`,
                                // so there is nothing more for this loop to
                                // report.
                                lobbies.kick_seat(&match_id, &slot, &did).await;
                            }
                            Ok(LobbyRequest::Invite { slot, handle }) => {
                                // Spawned rather than awaited: resolving a
                                // handle is a real DNS-over-HTTPS round
                                // trip, and awaiting it inline here would
                                // leave this one connection unable to read
                                // further messages or relay broadcasts
                                // until it finished. The result reaches
                                // every connection the same way a
                                // synchronous claim's does — the next
                                // `seats` snapshot — so there is nothing for
                                // the spawned task to report back to this
                                // socket specifically.
                                let lobbies = lobbies.clone();
                                let atproto = atproto.clone();
                                let match_id = match_id.clone();
                                let inviter = did.clone();
                                tokio::spawn(async move {
                                    let took = lobbies
                                        .invite_handle(
                                            &atproto, &match_id, &slot, &handle, &inviter,
                                        )
                                        .await;
                                    tracing::debug!(
                                        match_id,
                                        inviter,
                                        slot,
                                        handle,
                                        took,
                                        "lobby: invite resolved"
                                    );
                                });
                            }
                            // A bad shape or an unknown scenario/slot is
                            // ignored rather than dropping the socket: a
                            // typo in flight should not end the session.
                            Err(_) => {}
                        }
                    }
                    Some(Ok(Message::Close(_))) | None => break,
                    Some(Ok(_)) => {} // ping/pong/binary: nothing to do
                    Some(Err(_)) => break,
                }
            }
            update = updates.recv() => {
                let event = match update {
                    Ok(event) => event,
                    // A slow reader missed some history; the next change
                    // still arrives, so there is nothing to recover here.
                    Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
                    Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
                };
                if !send_event(&mut socket, &event).await {
                    break;
                }
            }
        }
    }
    tracing::debug!(match_id, did, "lobby: socket closed");
}

/// Serializes and sends one event, reporting whether the socket is still
/// good — the one place `lobby_socket`'s several send sites agree on what a
/// failed send means (stop, do not keep writing to a dead connection).
async fn send_event(socket: &mut WebSocket, event: &crate::matches::lobby::LobbyEvent) -> bool {
    let text = serde_json::to_string(event).expect("LobbyEvent always serializes");
    socket.send(Message::Text(text.into())).await.is_ok()
}

/// Whether `seats` fills every slot of `scenario`, each exactly once — the
/// same "every slot, exactly once" posture `plan_players` takes with a
/// client-submitted slate, applied here to a lobby's already-claimed seats.
/// An incomplete lobby (any slot unclaimed) fails this, and `deploy_lobby`
/// turns that into a 400 rather than launching short-handed.
fn seats_fill_every_slot(
    scenario: &crate::matches::scenarios::Scenario,
    seats: &[crate::db::MatchPlayer],
) -> bool {
    let mut wanted: Vec<&str> = scenario
        .slots
        .iter()
        .map(|slot| slot.name.as_str())
        .collect();
    let mut got: Vec<&str> = seats.iter().map(|seat| seat.slot.as_str()).collect();
    wanted.sort_unstable();
    got.sort_unstable();
    wanted == got
}

/// Turns a lobby's already-claimed seats into launcher seats, the shape
/// `Matches::start` wants. Unlike `plan_players`, there is no more "which
/// slot is mine" or "resolve this handle" ambiguity to settle — a lobby seat
/// is a `MatchPlayer` row with a slot, a control, and (for a human) a DID
/// and handle already verified when it was claimed — so this is a straight
/// conversion, plus each human seat's camo.
///
/// Camo is settled the same soft-fail way `plan_players`'s `player` branch
/// settles an invited seat's: `default_camo`, never `settle_own_camo` — that
/// one reads an explicit choice out of the *current caller's* request body,
/// which has no meaning for a generic "deploy whoever is seated" trigger.
/// Bot seats always get `camo_png: None`: nobody ever had a browser in front
/// of them to roll one for a lobby-assigned bot, unlike the single-page
/// launch form's bot slot. A real, deliberate scope cut — see plan/lobby.md, not
/// an oversight.
async fn lobby_seats_to_players(
    state: &AppState,
    owner: &str,
    scenario_path: &str,
    seats: Vec<crate::db::MatchPlayer>,
    rolled: &std::collections::HashMap<String, String>,
    own_roll: Option<&str>,
) -> Vec<crate::matches::NewPlayer> {
    let mut players = Vec::with_capacity(seats.len());
    for seat in seats {
        let bot = seat.control == "bot";
        let camo_rkey = match (&seat.did, bot) {
            (Some(seat_did), false) => state.db.default_camo(seat_did).await.unwrap_or(None),
            _ => None,
        };
        // Only a bot seat, and only from this request. A human seat's camo is
        // the one their account points at, which is the thing the browser
        // cannot override here and should not be able to.
        // A bot seat takes whatever was rolled for it; the caller's own seat
        // takes the camo they rolled for themselves, which beats the stored
        // default for this match only.
        let camo_png = match bot {
            true => decode_rolled_camo(owner, rolled.get(&seat.slot).cloned()),
            false if seat.did.as_deref() == Some(owner) => {
                decode_rolled_camo(owner, own_roll.map(str::to_owned))
            }
            false => None,
        };
        // The lobby's own answer wins; the scenario's stands when nobody
        // gave one. This is the only place the two are resolved, so a seat
        // row's `team` stays readable as "what was overridden" rather than
        // "what will run".
        let team = seat.team.or_else(|| {
            crate::matches::scenarios::find(scenario_path)
                .and_then(|s| s.slots.iter().find(|x| x.name == seat.slot))
                .and_then(|x| x.team)
        });
        players.push(crate::matches::NewPlayer {
            team,
            slot: seat.slot,
            bot,
            did: seat.did,
            handle: seat.handle,
            camo_rkey,
            camo_png,
        });
    }
    players
}

/// Turns a fully-seated lobby into a real, launched match — reusing
/// `Matches::start` for the manifest/camo/AWS work `create_match` already
/// drives, rather than duplicating any of it.
///
/// Only the lobby's own opener (`owner_did`, set by `open_lobby`) may
/// deploy it, not any seated player and not any signed-in caller. This
/// mirrors `create_match`'s existing assumption that only the launching
/// identity triggers a launch, and keeps the authorization question simple
/// for this first slice; "any seated player may deploy" is deferred, see
/// plan/invitations.md.
///
/// Every *other* seated human must have pressed Ready first — the caller's
/// own press is this request (see `matches::lobby::Lobbies::waiting_on`), and
/// a bot is never asked. A lobby of one human and any number of bots is
/// therefore unchanged by this, which is what the daily and every solo fight
/// against Princess are.
///
/// Validation and the double-deploy guard both run before `Matches::start`
/// is ever called, in an order chosen to match `create_match`'s own "shape
/// questions before configuration questions" precedent: read-only checks
/// that need no AWS (ownership, slot completeness, readiness) come first and are
/// answerable even when this instance cannot launch at all, then the 503
/// configuration check, then the atomic `deploying` claim — the one write
/// that decides a race between two callers — and only then the real launch.
/// A failed launch leaves the row at `deploying` rather than restoring it to
/// `lobby`: see the module doc in `matches/mod.rs`.
async fn deploy_lobby(
    State(state): State<AppState>,
    Path(id): Path<String>,
    jar: CookieJar,
    body: Option<axum::Json<LaunchBody>>,
) -> Response {
    let (rolled, own_roll) = body
        .map(|axum::Json(body)| (body.bot_camo, body.camo))
        .unwrap_or_default();
    let Some(did) = session_did(&state, &jar) else {
        return StatusCode::UNAUTHORIZED.into_response();
    };
    let row = match state.db.get_match(&id).await {
        Ok(Some(row)) => row,
        Ok(None) => return StatusCode::NOT_FOUND.into_response(),
        Err(_) => {
            tracing::error!(did, match_id = id, "lobby: deploy lookup failed");
            return StatusCode::INTERNAL_SERVER_ERROR.into_response();
        }
    };
    if row.owner_did != did {
        return StatusCode::FORBIDDEN.into_response();
    }
    let Some(scenario_path) = row.scenario.as_deref() else {
        return message(StatusCode::BAD_REQUEST, "This lobby has no scenario yet.");
    };
    let Some(scenario) = crate::matches::scenarios::find(scenario_path) else {
        return message(StatusCode::BAD_REQUEST, "That scenario is not available.");
    };
    let seats = match state.db.match_players(&id).await {
        Ok(seats) => seats,
        Err(_) => {
            tracing::error!(did, match_id = id, "lobby: seat read failed for deploy");
            return StatusCode::INTERNAL_SERVER_ERROR.into_response();
        }
    };
    if !seats_fill_every_slot(scenario, &seats) {
        return message(
            StatusCode::BAD_REQUEST,
            "Every slot must be claimed before deploying.",
        );
    }
    // Read from the same `seats` the completeness check just used, so both
    // answers are about one lobby rather than two a moment apart.
    let waiting = state.lobbies.waiting_on(&id, &seats, &did);
    if !waiting.is_empty() {
        return message(
            StatusCode::BAD_REQUEST,
            &format!("Waiting for {} to be ready.", waiting.join(", ")),
        );
    }

    let Some(matches) = state.matches.as_ref() else {
        return message(
            StatusCode::SERVICE_UNAVAILABLE,
            "Match launching is not configured here.",
        );
    };

    // The atomic guard against a double-deploy: a second caller — a second
    // click, a second browser — loses here, before a second match is ever
    // started, rather than after.
    match state.db.claim_deploy(&id).await {
        Ok(true) => {}
        Ok(false) => return StatusCode::CONFLICT.into_response(),
        Err(_) => {
            tracing::error!(did, match_id = id, "lobby: deploy claim failed");
            return StatusCode::INTERNAL_SERVER_ERROR.into_response();
        }
    }

    let scenario_path = scenario_path.to_owned();
    let players = lobby_seats_to_players(
        &state,
        &did,
        &scenario_path,
        seats,
        &rolled,
        own_roll.as_deref(),
    )
    .await;
    match matches
        .start(
            &state.atproto,
            &did,
            crate::matches::manifest::ScenarioSource::Carried(scenario.text),
            &scenario_path,
            players,
        )
        .await
    {
        Ok(new_match_id) => {
            // The only way a socket that did not itself call /deploy learns
            // the new match's id — it is otherwise just sitting on a
            // WebSocket to a lobby about to disappear.
            state.lobbies.broadcast_deployed(&id, new_match_id.clone());
            // The lobby's whole purpose was to become this match; now that
            // it has, there is nothing left to keep it around for.
            if let Err(e) = state.db.delete_lobby(&id).await {
                tracing::error!(match_id = id, ?e, "lobby: cleanup after deploy failed");
            }
            Json(serde_json::json!({ "matchId": new_match_id })).into_response()
        }
        // Left at `deploying` on purpose: a dead end a human can notice and
        // investigate, not seat claims silently erased for someone to redo
        // from scratch. No automatic retry in this slice; see plan/lobby.md.
        Err(_) => message(
            StatusCode::INTERNAL_SERVER_ERROR,
            "The match could not be started.",
        ),
    }
}

/// Whether this session may see and reach this match: the launcher, or
/// anyone holding a seat in it. Matches that predate the seats table have
/// only their owner column, which is why that check comes first.
async fn may_touch_match(state: &AppState, row: &crate::db::MatchRow, did: &str) -> bool {
    row.owner_did == did
        || state
            .db
            .is_match_player(&row.id, did)
            .await
            .unwrap_or(false)
}

/// One player's matches, newest first — launched or invited into.
///
/// `viewer` is who is asking, when anybody is signed in. It decides two
/// things and nothing else: whether a `ready` row is polled, and whether it
/// carries a link into the running match. A player's record is public; the
/// door into their container is not, and a stranger driving one DescribeTasks
/// per listed row is a cost anybody could run up.
///
/// Stored status, except for the subject's own `ready` rows. Those are the
/// only ones this hands out a link for, and the only ones whose being wrong
/// costs the player anything: a match that died without reporting it stayed
/// `ready` in the row for good, so the screen kept offering an Open link into
/// a task that no longer exists and the proxy answered it with a 502.
///
/// `starting` and `running` are left alone on purpose. They are corrected by
/// the waiting screen, which is where the Resume button on those rows goes, and
/// `advance` reaches for the container's own web server with a three-second
/// timeout — a wait this route must not take once per listed row.
async fn match_summaries(
    state: &AppState,
    subject: &str,
    viewer: Option<&str>,
) -> Result<Vec<serde_json::Value>, ()> {
    let rows = match state.db.list_matches(subject, 20).await {
        Ok(rows) => rows,
        Err(_) => {
            tracing::error!(did = subject, "matches: list lookup failed");
            return Err(());
        }
    };
    let mine = viewer == Some(subject);
    let mut matches = Vec::with_capacity(rows.len());
    for row in rows {
        // One query per listed row, bounded by the list's LIMIT. The seats
        // are what let the screen say who a match was against.
        let players = state.db.match_players(&row.id).await.unwrap_or_default();
        // At most one DescribeTasks per ready row, and the poll's own
        // one-second window holds it there however often the screen is
        // refreshed. It converges too: the first list load after a match ends
        // writes `over`, and the row is never polled again.
        //
        // None in development, which has no cluster to ask; the stored status
        // is all there is and all there ever was.
        let status = match state.matches.as_ref() {
            Some(matches) if mine && row.status == "ready" => matches.poll(&row).await.status,
            _ => row.status.clone(),
        };
        // The caller's own client, at their DID. A viewer with no fighting
        // seat gets no link: there is no client in the container for them.
        let url = if mine
            && status == "ready"
            && state
                .db
                .is_human_player(&row.id, subject)
                .await
                .unwrap_or(false)
        {
            Some(format!("/match/{}/{}/", row.id, subject))
        } else {
            None
        };
        matches.push(serde_json::json!({
            "matchId": row.id,
            "status": status,
            "createdAt": row.created_at,
            "url": url,
            "scenario": row.scenario,
            "players": players,
        }));
    }
    Ok(matches)
}

/// The signed-in player's own matches. Their own record, read as themselves.
async fn list_matches(State(state): State<AppState>, jar: CookieJar) -> Response {
    let Some(did) = session_did(&state, &jar) else {
        return StatusCode::UNAUTHORIZED.into_response();
    };
    match match_summaries(&state, &did, Some(&did)).await {
        Ok(matches) => Json(serde_json::json!({ "matches": matches })).into_response(),
        Err(()) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
    }
}

/// Any player's matches, by handle or DID, for whoever is asking.
///
/// No session required. A match is a thing two accounts did together and it
/// is published as one — the report at `/reports/<id>` has been public since
/// it existed, and a list of them says nothing the reports do not. What is
/// still session-scoped is the way *in*: see `match_summaries`.
///
/// Resolved the same way the unfurl is (`Atproto::resolve_actor`), so
/// `/matches/@alice.example` and `/matches/did:plc:...` are one address, and
/// an actor nobody can resolve is a 404 rather than an empty list.
async fn player_matches(
    State(state): State<AppState>,
    axum::extract::Path(actor): axum::extract::Path<String>,
    jar: CookieJar,
) -> Response {
    let Some(subject) = state.atproto.resolve_actor(&actor).await else {
        return StatusCode::NOT_FOUND.into_response();
    };
    let viewer = session_did(&state, &jar);
    match match_summaries(&state, &subject.did, viewer.as_deref()).await {
        Ok(matches) => Json(serde_json::json!({
            "did": subject.did,
            "handle": subject.handle,
            "matches": matches,
        }))
        .into_response(),
        Err(()) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
    }
}

async fn match_status(
    State(state): State<AppState>,
    axum::extract::Path(id): axum::extract::Path<String>,
    jar: CookieJar,
) -> Response {
    let Some(did) = session_did(&state, &jar) else {
        return StatusCode::UNAUTHORIZED.into_response();
    };
    let row = match state.db.get_match(&id).await {
        Ok(Some(row)) => row,
        // Not yours reads as not found - no existence oracle.
        Ok(None) => return StatusCode::NOT_FOUND.into_response(),
        Err(_) => {
            tracing::error!(did, match_id = id, "match status: match lookup failed");
            return StatusCode::INTERNAL_SERVER_ERROR.into_response();
        }
    };
    // A seat is enough: an invited player polls the same waiting screen the
    // launcher does.
    if !may_touch_match(&state, &row, &did).await {
        return StatusCode::NOT_FOUND.into_response();
    }
    let Some(matches) = state.matches.as_ref() else {
        return StatusCode::SERVICE_UNAVAILABLE.into_response();
    };
    let mut view = matches.poll(&row).await;
    // The caller's own client, at their DID — or nothing, for a viewer with
    // no fighting seat.
    view.url = if view.url.is_some() && state.db.is_human_player(&id, &did).await.unwrap_or(false) {
        Some(format!("/match/{}/{}/", id, did))
    } else {
        None
    };
    Json(view).into_response()
}

/// What a match ended as: the result document arena uploaded, and the status
/// that says whether there is any point waiting for it.
///
/// The match screen polls this. It has to: the page it polls from is
/// Suramadu's, the container reports to headquarters rather than to a
/// browser, and the first thing that exists anywhere once a game is decided
/// is this artifact. `container/watch/results.sh` sends it about five seconds
/// after the host writes its game-over marker - a minute before the container
/// stops - which is what lets the page put up a report while the player is
/// still looking at it.
///
/// `result` is null until then, and stays null for a match that ended without
/// one: reaped, crashed, or stopped before victory. A caller that sees a
/// terminal status and no result has its answer and should stop asking.
///
/// `home` because this API is served on its own origin. The match page is
/// proxied through here, so it cannot construct the site's address itself,
/// and a player whose match is over needs somewhere to go. `share` is the
/// same problem for the other direction: the public page for this result,
/// which the report screen offers to post.
///
/// Seat-gated like `match_status`, and for the same reason: an invited player
/// is owed the result of the match they fought in, not just the launcher.
async fn match_result(
    State(state): State<AppState>,
    axum::extract::Path(id): axum::extract::Path<String>,
    jar: CookieJar,
) -> Response {
    let Some(did) = session_did(&state, &jar) else {
        return StatusCode::UNAUTHORIZED.into_response();
    };
    let row = match state.db.get_match(&id).await {
        Ok(Some(row)) => row,
        Ok(None) => return StatusCode::NOT_FOUND.into_response(),
        Err(_) => {
            tracing::error!(did, match_id = id, "match result: match lookup failed");
            return StatusCode::INTERNAL_SERVER_ERROR.into_response();
        }
    };
    if !may_touch_match(&state, &row, &did).await {
        return StatusCode::NOT_FOUND.into_response();
    }
    let Some(matches) = state.matches.as_ref() else {
        return StatusCode::SERVICE_UNAVAILABLE.into_response();
    };
    // Polled rather than read off the row: a container that died without
    // saying so leaves the row at `ready` for good, and the page waiting for
    // a result that is never coming is exactly the case this has to end.
    let view = matches.poll(&row).await;
    // Whose side to open the report from. The page is public and sessionless,
    // so it cannot work this out; this route is seat-gated and knows exactly
    // who is asking, which is the only place the two facts meet.
    let mine = state
        .db
        .match_players(&id)
        .await
        .unwrap_or_default()
        .into_iter()
        .find(|seat| seat.did.as_deref() == Some(did.as_str()))
        .and_then(|seat| seat.handle);
    // Without the leading "@": that is the form the page reads back and the
    // form its own links carry, so an address a player is handed and one they
    // copy off the page are the same address.
    //
    // `participant` is what tells the report its reader fought in the match,
    // which is what puts the composer on the page. It is set here and nowhere
    // else, and the page keeps it out of every link it builds - a reader who
    // arrives from the resulting post gets the report, not the composer.
    let ask = mine
        .map(|handle| {
            format!(
                "?perspective={}&participant=1",
                crate::share::urlencode(handle.trim_start_matches('@'))
            )
        })
        .unwrap_or_else(|| "?participant=1".to_owned());
    Json(serde_json::json!({
        "status": view.status,
        "result": matches.result(&id).await,
        "home": state.web_origin,
        // Where this result can be shown to someone who was not in it. The
        // page cannot build this: it is a route on this origin, and which
        // origin that is is exactly what a page inside a match container
        // cannot see.
        // The address the match report hands the player on to, which is
        // where these pages are published rather than where they are served.
        "share": format!(
            "{}/reports/{}{}",
            state.share_origin.as_deref().unwrap_or(matches.public_url()),
            id,
            ask
        ),
    }))
    .into_response()
}

#[derive(serde::Deserialize)]
struct EventBody {
    event: String,
    /// Free text from the container - phase timings on `ready`, a reason on
    /// `error`. Logged, never interpreted. Absent from older arena images, and
    /// from any caller that sends only the event.
    #[serde(default)]
    detail: String,
}

async fn match_event(
    State(state): State<AppState>,
    axum::extract::Path((id, token)): axum::extract::Path<(String, String)>,
    body: Option<Json<EventBody>>,
) -> Response {
    let row = match state.db.get_match(&id).await {
        Ok(Some(row)) => row,
        _ => return StatusCode::NOT_FOUND.into_response(),
    };
    if !crate::matches::token_matches(&row, &token) {
        return StatusCode::NOT_FOUND.into_response();
    }
    let Some(Json(EventBody { event, detail })) = body else {
        return StatusCode::NO_CONTENT.into_response();
    };
    crate::matches::record_event(&state.db, &row, &event, &detail).await;
    // A daily challenge's score is read out of the result the match just
    // uploaded, so the moment the match ends is the first moment there is one.
    // The upload can land after this event, which is why the card's own read
    // tries again.
    if let Some(matches) = state.matches.as_ref() {
        matches.settle_daily(&id).await;
    }
    StatusCode::NO_CONTENT.into_response()
}

/// The answer the upload route gives a match that has already ended.
///
/// `410 Gone` rather than the `404` a bad token gets: the caller held the right
/// token, so there is no existence to hide, and a distinct code is the
/// difference between "that token is wrong" and "that match is finished" when
/// someone is reading a container log.
///
/// Warn rather than info because in a healthy match this never fires - arena
/// sends `game-over` after its last upload. One of these means either an
/// unexpected ordering in the container or a token being replayed, and both are
/// worth finding.
fn upload_window_closed(row: &crate::db::MatchRow, name: &str) -> Response {
    tracing::warn!(
        match_id = row.id,
        status = row.status,
        name,
        "match: upload refused, the match has ended"
    );
    StatusCode::GONE.into_response()
}

/// Hand back a presigned PUT for one artifact key so the container uploads
/// straight to S3. The only way in: there is no route that accepts artifact
/// bytes.
///
/// That is deliberate. The route this replaced took the whole body into memory
/// to re-send it, and api-host is a t4g.nano carrying the API and Caddy in
/// 512MB - so anyone holding a match token could exhaust it, and a match's
/// screenshots, GIF summary, profiles and client logs all land in the same few
/// seconds at the end anyway.
///
/// Minted per upload, not presigned into the manifest at launch: see
/// `Aws::presign_artifact_put` for why a launch-time signature cannot be given
/// a predictable life.
///
/// The match token is the whole authority here - checked against the row,
/// bounded by the upload window, and worth one signature for one key. The
/// container never holds an AWS credential.
async fn mint_upload_url(
    State(state): State<AppState>,
    axum::extract::Path((id, token, name)): axum::extract::Path<(String, String, String)>,
) -> Response {
    let row = match state.db.get_match(&id).await {
        Ok(Some(row)) => row,
        _ => return StatusCode::NOT_FOUND.into_response(),
    };
    if !crate::matches::token_matches(&row, &token) {
        return StatusCode::NOT_FOUND.into_response();
    }
    if !crate::matches::uploads_open(&row) {
        return upload_window_closed(&row, &name);
    }
    if !crate::matches::artifact_name_ok(&name) {
        return StatusCode::BAD_REQUEST.into_response();
    }
    let Some(matches) = state.matches.as_ref() else {
        return StatusCode::SERVICE_UNAVAILABLE.into_response();
    };
    match matches.presign_artifact_put(&id, &name).await {
        // The URL is a bearer credential for that one key. It goes in this
        // response and nowhere else - not the log, not the match row.
        Some(url) => Json(serde_json::json!({
            "url": url,
            "method": "PUT",
            "expiresIn": crate::matches::ARTIFACT_PUT_EXPIRY_SECS,
        }))
        .into_response(),
        None => StatusCode::BAD_GATEWAY.into_response(),
    }
}

async fn client_metadata(State(state): State<AppState>) -> Response {
    Json(state.atproto.client_metadata()).into_response()
}

async fn jwks(State(state): State<AppState>) -> Response {
    Json(state.atproto.jwks()).into_response()
}

// --- the daily challenge ---------------------------------------------------

/// What is known about one challenge: how many have taken it on, and whether
/// the caller has used their attempt.
///
/// Open to everyone. The count is on the front page, which a stranger sees,
/// and `attempted` is simply false for one — a signed-out reader is told about
/// the fight, not about a player.
async fn daily_status(
    State(state): State<AppState>,
    Path(slug): Path<String>,
    jar: CookieJar,
) -> Response {
    let Some(challenge) = crate::matches::daily::challenge(&slug) else {
        return StatusCode::NOT_FOUND.into_response();
    };
    let attempts = match state.db.daily_attempt_count(challenge.slug).await {
        Ok(n) => n,
        Err(_) => {
            tracing::error!(slug = challenge.slug, "daily: attempt count failed");
            return StatusCode::INTERNAL_SERVER_ERROR.into_response();
        }
    };
    let mine = match session_did(&state, &jar) {
        None => None,
        Some(did) => {
            let attempt = state
                .db
                .daily_attempt(challenge.slug, &did)
                .await
                .unwrap_or(None);
            // An attempt with no score whose match has ended: the result may
            // have been uploaded after the event that said the match was over,
            // so this is the retry. Only for the player's own attempt, so it
            // is one read for one person who is waiting for exactly this.
            match (&attempt, state.matches.as_ref()) {
                (Some(a), Some(matches)) if a.score.is_none() => {
                    matches.settle_daily(&a.match_id).await;
                    state
                        .db
                        .daily_attempt(challenge.slug, &did)
                        .await
                        .unwrap_or(attempt)
                }
                _ => attempt,
            }
        }
    };
    Json(serde_json::json!({
        "slug": challenge.slug,
        "attempts": attempts,
        "attempted": mine.is_some(),
        "score": mine.and_then(|a| a.score),
    }))
    .into_response()
}

/// Take on a challenge: launch the fight against Princess, and spend the
/// player's one attempt if this is the day's.
///
/// A challenge whose day has passed is playable and is not an attempt. That is
/// the whole difference between the two paths here, and it is decided against
/// the server's clock rather than anything the caller sends.
async fn start_daily(
    State(state): State<AppState>,
    Path(slug): Path<String>,
    jar: CookieJar,
    body: Option<axum::Json<LaunchBody>>,
) -> Response {
    let (rolled, own_roll) = body
        .map(|axum::Json(body)| (body.bot_camo, body.camo))
        .unwrap_or_default();
    let Some(did) = session_did(&state, &jar) else {
        return message(StatusCode::UNAUTHORIZED, "Sign in to play.");
    };
    let Some(challenge) = crate::matches::daily::challenge(&slug) else {
        return StatusCode::NOT_FOUND.into_response();
    };
    let Some(matches) = state.matches.as_ref() else {
        return message(
            StatusCode::SERVICE_UNAVAILABLE,
            "Match launching is not configured here.",
        );
    };

    let day = crate::matches::daily::utc_day(std::time::SystemTime::now());
    let scored = crate::matches::daily::is_current(challenge, &day);

    // The attempt is taken before the match is launched, not after: the
    // insert is the lock, and a launch that happened first would let two
    // clicks start two matches and only then discover the second was not
    // allowed. A launch that fails gives it back below.
    if scored {
        match state
            .db
            .take_daily_attempt(
                challenge.slug,
                &did,
                &pending_id(challenge.slug, &did),
                &day,
            )
            .await
        {
            Ok(true) => {}
            Ok(false) => {
                return message(
                    StatusCode::CONFLICT,
                    "You have already taken on today's challenge.",
                );
            }
            Err(_) => {
                tracing::error!(did, slug = challenge.slug, "daily: attempt claim failed");
                return StatusCode::INTERNAL_SERVER_ERROR.into_response();
            }
        }
    }

    let handle = state.db.account_handle(&did).await.ok().flatten().flatten();
    // One seat per faction the scenario names, read from the file rather than
    // assumed: a daily may field one opposing lance or three, and every
    // faction that is not the player's is Princess.
    let mut factions: Vec<String> = headquarters_mms::factions(challenge.scenario)
        .into_iter()
        .map(|f| f.name)
        .collect();
    if factions.is_empty() {
        factions.push(crate::matches::daily::PLAYER_SLOT.to_owned());
    }
    // The player's own camo, the same stored default a lobby seat is settled
    // against. It was `None` here, which meant the challenge screen's camo
    // choice reached everything except the one match it was chosen for.
    let own_camo = state.db.default_camo(&did).await.unwrap_or(None);
    // A daily's sides are in its own file, which is not in the catalog - the
    // scenario is compiled in and carried in the manifest - so its teams come
    // straight off the parse rather than through `scenarios::find`.
    let teams: std::collections::HashMap<String, Option<i32>> =
        headquarters_mms::factions(challenge.scenario)
            .into_iter()
            .map(|f| (f.name, f.team))
            .collect();
    let players: Vec<crate::matches::NewPlayer> = factions
        .into_iter()
        .map(|slot| {
            let is_player = slot == crate::matches::daily::PLAYER_SLOT;
            let camo_png = match is_player {
                false => decode_rolled_camo(&did, rolled.get(&slot).cloned()),
                true => decode_rolled_camo(&did, own_roll.clone()),
            };
            crate::matches::NewPlayer {
                bot: !is_player,
                did: is_player.then(|| did.clone()),
                handle: if is_player { handle.clone() } else { None },
                camo_rkey: is_player.then(|| own_camo.clone()).flatten(),
                camo_png,
                team: teams.get(&slot).copied().flatten(),
                slot,
            }
        })
        .collect();

    let label = format!("daily:{}", challenge.slug);
    match matches
        .start(
            &state.atproto,
            &did,
            crate::matches::manifest::ScenarioSource::Carried(challenge.scenario),
            &label,
            players,
        )
        .await
    {
        Ok(match_id) => {
            if scored {
                // The row was inserted against "pending" so that the claim
                // could happen before the launch. Now it gets the match it
                // was spent on, which is what lets the result find it again.
                if let Err(e) = state
                    .db
                    .bind_daily_attempt(challenge.slug, &did, &match_id)
                    .await
                {
                    tracing::error!(
                        did,
                        slug = challenge.slug,
                        match_id,
                        ?e,
                        "daily: attempt not bound to its match; it cannot be scored"
                    );
                }
            }
            tracing::info!(
                did,
                slug = challenge.slug,
                match_id,
                scored,
                "daily: started"
            );
            Json(serde_json::json!({ "matchId": match_id, "scored": scored })).into_response()
        }
        Err(e) => {
            // Nothing was played, so nothing is spent.
            if scored {
                let _ = state
                    .db
                    .release_daily_attempt(&pending_id(challenge.slug, &did))
                    .await;
            }
            let _ = e;
            tracing::error!(did, slug = challenge.slug, "daily: launch failed");
            message(
                StatusCode::BAD_GATEWAY,
                "The match could not be started. Try again.",
            )
        }
    }
}

/// The `match_id` an attempt holds between being claimed and being launched.
///
/// The claim has to happen before the launch — it is the lock that stops two
/// clicks starting two matches — but the match has no id until it starts. This
/// is the placeholder that stands in, and it is per challenge and per player
/// because `daily_attempt_by_match` is a unique index: a shared placeholder
/// would have two players collide on it.
fn pending_id(slug: &str, did: &str) -> String {
    format!("pending:{slug}:{did}")
}

#[cfg(test)]
mod tests {
    use super::*;
    use axum::body::Body;
    use axum::http::Request;
    use http_body_util::BodyExt;
    use tower::ServiceExt;

    const SECRET: &[u8] = b"0123456789abcdef0123456789abcdef";

    async fn loopback_state() -> (tempfile::TempDir, AppState) {
        let dir = tempfile::tempdir().unwrap();
        let db = crate::db::Db::open(&dir.path().join("test.sqlite")).unwrap();
        let config = crate::config::Config::from_lookup(|_| None).unwrap();
        let state = AppState {
            units: None,
            atproto: Arc::new(Atproto::new(&config, db.clone()).unwrap()),
            lobbies: Arc::new(crate::matches::lobby::Lobbies::new(db.clone())),
            db,
            signer: Arc::new(SessionSigner::new(SECRET.to_vec(), false, None)),
            web_origin: config.web_origin,
            share_origin: config.share_origin,
            matches: None,
            build_ref: config.build_ref,
            arena_version: config.arena_version,
            // Enabled in tests: the write itself never runs (no session gets
            // that far), and the routes below assert everything before it.
            flare_space: Some("at://did:plc:board/app.userinput.space/3kspace".into()),
        };
        (dir, state)
    }

    async fn body_json(response: Response) -> serde_json::Value {
        let bytes = response.into_body().collect().await.unwrap().to_bytes();
        serde_json::from_slice(&bytes).unwrap()
    }

    #[tokio::test]
    async fn healthz_still_responds() {
        let (_dir, state) = loopback_state().await;
        let response = app(state)
            .oneshot(Request::get("/healthz").body(Body::empty()).unwrap())
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn version_reports_build_and_arena() {
        let (_dir, mut state) = loopback_state().await;
        state.build_ref = Some("abc123def456".into());
        state.arena_version = Some("mm0.51.0-ws26.4.5-abc123def456".into());
        let response = app(state)
            .oneshot(Request::get("/api/version").body(Body::empty()).unwrap())
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::OK);
        assert_eq!(
            body_json(response).await,
            serde_json::json!({
                "version": env!("CARGO_PKG_VERSION"),
                "build": "abc123def456",
                "arena": "mm0.51.0-ws26.4.5-abc123def456",
                "library": serde_json::Value::Null,
            })
        );
    }

    /// The site reads the helm release off this rather than compiling one in,
    /// so a helm deploy reaches the force builder without a site build. Both
    /// halves come off the index document itself, which is what stops this
    /// service and that screen from naming different libraries.
    #[tokio::test]
    async fn version_names_the_library_it_loaded() {
        let (_dir, mut state) = loopback_state().await;
        state.units = Some(std::sync::Arc::new(crate::units::for_test(&[(
            "Atlas AS7-D",
            "meks/atlas.png",
        )])));
        let response = app(state)
            .oneshot(Request::get("/api/version").body(Body::empty()).unwrap())
            .await
            .unwrap();
        let body = body_json(response).await;
        assert_eq!(body["library"]["megamek"], "0.51.0");
        assert_eq!(body["library"]["helm"], "main-0bb2bf14a1a3");
    }

    /// The fast path a screen drawing a few units takes, so the front page
    /// does not fetch 300KB of index to resolve six names.
    #[tokio::test]
    async fn unit_art_answers_for_the_names_asked() {
        let (_dir, mut state) = loopback_state().await;
        state.units = Some(std::sync::Arc::new(crate::units::for_test(&[
            ("Atlas AS7-D", "meks/atlas.png"),
            ("Locust LCT-1V", "meks/locust.png"),
        ])));
        let response = app(state)
            .oneshot(
                Request::get("/api/units/art?name=Atlas+AS7-D&name=Nothing+Here")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::OK);
        assert_eq!(
            body_json(response).await,
            // The unknown name is absent, not null: the caller is asking what
            // this library knows, and it does not know that one.
            serde_json::json!({ "Atlas AS7-D": "meks/atlas.png" })
        );
    }

    /// A deployment with no index answers an empty object rather than an
    /// error: the caller's fallback reads the same library and finds the same
    /// nothing, so a failure here would only make it retry for no reason.
    #[tokio::test]
    async fn unit_art_without_an_index_is_empty() {
        let (_dir, state) = loopback_state().await;
        let response = app(state)
            .oneshot(
                Request::get("/api/units/art?name=Atlas+AS7-D")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::OK);
        assert_eq!(body_json(response).await, serde_json::json!({}));
    }

    /// Bounded, because the query string is a stranger's to write.
    #[tokio::test]
    async fn unit_art_answers_at_most_a_batch() {
        let (_dir, mut state) = loopback_state().await;
        let names: Vec<(String, String)> = (0..200)
            .map(|n| (format!("Design {n}"), format!("meks/{n}.png")))
            .collect();
        let borrowed: Vec<(&str, &str)> = names
            .iter()
            .map(|(n, s)| (n.as_str(), s.as_str()))
            .collect();
        state.units = Some(std::sync::Arc::new(crate::units::for_test(&borrowed)));
        let query = (0..200)
            .map(|n| format!("name=Design+{n}"))
            .collect::<Vec<_>>()
            .join("&");
        let response = app(state)
            .oneshot(
                Request::get(format!("/api/units/art?{query}"))
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        let body = body_json(response).await;
        assert_eq!(
            body.as_object().expect("an object").len(),
            ART_LOOKUP_MAX,
            "the batch cap did not hold"
        );
    }

    /// No index is null rather than absent, so the site can tell an API that
    /// has none from one too old to say - and falls back to its own constants
    /// in both cases.
    #[tokio::test]
    async fn version_without_an_index_says_so() {
        let (_dir, state) = loopback_state().await;
        let response = app(state)
            .oneshot(Request::get("/api/version").body(Body::empty()).unwrap())
            .await
            .unwrap();
        assert!(body_json(response).await["library"].is_null());
    }

    #[tokio::test]
    async fn version_in_development_is_nulls() {
        let (_dir, state) = loopback_state().await;
        let response = app(state)
            .oneshot(Request::get("/api/version").body(Body::empty()).unwrap())
            .await
            .unwrap();
        let body = body_json(response).await;
        assert!(body["version"].is_string());
        assert!(body["build"].is_null());
        assert!(body["arena"].is_null());
    }

    #[tokio::test]
    async fn session_without_cookie_is_401() {
        let (_dir, state) = loopback_state().await;
        let response = app(state)
            .oneshot(Request::get("/api/session").body(Body::empty()).unwrap())
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
    }

    #[tokio::test]
    async fn session_with_forged_cookie_is_401() {
        let (_dir, state) = loopback_state().await;
        let forged = SessionSigner::new(b"wrong-secret-wrong-secret-wrong!".to_vec(), false, None)
            .issue("did:plc:abc");
        let response = app(state)
            .oneshot(
                Request::get("/api/session")
                    .header(
                        header::COOKIE,
                        format!("{}={}", crate::session::COOKIE_NAME, forged.value()),
                    )
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
    }

    #[tokio::test]
    async fn session_happy_path() {
        let (_dir, state) = loopback_state().await;
        state
            .db
            .record_handle_check("did:plc:abc", Some("someone.example"))
            .await
            .unwrap();
        state
            .db
            .call(|conn| {
                conn.execute(
                    "INSERT INTO oauth_session (did, value) VALUES ('did:plc:abc', '{}')",
                    [],
                )
                .map(|_| ())
            })
            .await
            .unwrap();
        let cookie = state.signer.issue("did:plc:abc");

        let response = app(state)
            .oneshot(
                Request::get("/api/session")
                    .header(
                        header::COOKIE,
                        format!("{}={}", crate::session::COOKIE_NAME, cookie.value()),
                    )
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::OK);
        assert_eq!(
            body_json(response).await,
            serde_json::json!({ "did": "did:plc:abc", "handle": "someone.example", "defaultCamo": null })
        );
    }

    /// Nothing resolves a `did:example:` DID, and atrium says so without
    /// asking anyone: the background re-check runs end to end here, and the
    /// answer it lands on is "unverified".
    const UNRESOLVABLE_DID: &str = "did:example:abc";

    async fn sign_in(state: &AppState, did: &str, handle: Option<&str>) {
        state.db.record_handle_check(did, handle).await.unwrap();
        let did = did.to_owned();
        state
            .db
            .call(move |conn| {
                conn.execute(
                    "INSERT INTO oauth_session (did, value) VALUES (?1, '{}')",
                    [did],
                )
                .map(|_| ())
            })
            .await
            .unwrap();
    }

    /// Backdates the last check so the next request finds the row stale.
    async fn age_the_check(state: &AppState, did: &str) {
        let did = did.to_owned();
        state
            .db
            .call(move |conn| {
                conn.execute("UPDATE account SET checked_at = 1 WHERE did = ?1", [did])
                    .map(|_| ())
            })
            .await
            .unwrap();
    }

    async fn get_session(app: &Router, state: &AppState, did: &str) -> Response {
        let cookie = state.signer.issue(did);
        app.clone()
            .oneshot(
                Request::get("/api/session")
                    .header(
                        header::COOKIE,
                        format!("{}={}", crate::session::COOKIE_NAME, cookie.value()),
                    )
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap()
    }

    #[tokio::test]
    async fn session_answers_from_cache_while_the_handle_is_re_checked() {
        let (_dir, state) = loopback_state().await;
        sign_in(&state, UNRESOLVABLE_DID, Some("old.example")).await;
        age_the_check(&state, UNRESOLVABLE_DID).await;

        let response = get_session(&app(state.clone()), &state, UNRESOLVABLE_DID).await;
        assert_eq!(response.status(), StatusCode::OK);
        assert_eq!(
            body_json(response).await["handle"],
            "old.example",
            "the request must not wait on verification"
        );

        // And the check it started lands afterwards. Failing verification
        // demotes the row, which is how a handle someone lost stops being
        // shown - the same write path that promotes one.
        for _ in 0..100 {
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
            if state.db.account_handle(UNRESOLVABLE_DID).await.unwrap() == Some(None) {
                return;
            }
        }
        panic!("the background re-check never stored its result");
    }

    #[tokio::test]
    async fn a_fresh_handle_is_not_re_checked() {
        let (_dir, state) = loopback_state().await;
        // sign_in leaves the check fresh, so this DID never resolving does
        // not matter: nothing should look it up.
        sign_in(&state, UNRESOLVABLE_DID, Some("someone.example")).await;

        let app = app(state.clone());
        for _ in 0..3 {
            let response = get_session(&app, &state, UNRESOLVABLE_DID).await;
            assert_eq!(response.status(), StatusCode::OK);
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        assert_eq!(
            state.db.account_handle(UNRESOLVABLE_DID).await.unwrap(),
            Some(Some("someone.example".to_owned())),
            "a re-check would have demoted this handle"
        );
    }

    #[tokio::test]
    async fn a_changed_handle_does_not_disturb_the_session() {
        let (_dir, state) = loopback_state().await;
        sign_in(&state, "did:plc:abc", Some("old.example")).await;
        let app = app(state.clone());

        // What a completed re-check writes. The cookie carries only the DID,
        // which a rename never touches, so it keeps working.
        state
            .db
            .record_handle_check("did:plc:abc", Some("new.example"))
            .await
            .unwrap();

        let response = get_session(&app, &state, "did:plc:abc").await;
        assert_eq!(response.status(), StatusCode::OK);
        assert!(
            response.headers().get(header::SET_COOKIE).is_none(),
            "the session must survive a rename untouched"
        );
        assert_eq!(
            body_json(response).await,
            serde_json::json!({ "did": "did:plc:abc", "handle": "new.example", "defaultCamo": null })
        );
    }

    #[tokio::test]
    async fn session_with_cookie_but_no_stored_session_clears_it() {
        let (_dir, state) = loopback_state().await;
        let cookie = state.signer.issue("did:plc:abc");

        let response = app(state)
            .oneshot(
                Request::get("/api/session")
                    .header(
                        header::COOKIE,
                        format!("{}={}", crate::session::COOKIE_NAME, cookie.value()),
                    )
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
        let set_cookie = response
            .headers()
            .get(header::SET_COOKIE)
            .unwrap()
            .to_str()
            .unwrap();
        assert!(
            set_cookie.contains("Max-Age=0"),
            "stale cookie must be cleared: {set_cookie}"
        );
    }

    #[tokio::test]
    async fn login_wants_a_handle() {
        let (_dir, state) = loopback_state().await;
        let app = app(state);

        let response = app
            .clone()
            .oneshot(
                Request::post("/api/login")
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(Body::from(r#"{"handle": ""}"#))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
        assert!(body_json(response).await["message"].is_string());

        let response = app
            .oneshot(Request::post("/api/login").body(Body::empty()).unwrap())
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
    }

    #[tokio::test]
    async fn logout_always_clears() {
        let (_dir, state) = loopback_state().await;
        let response = app(state)
            .oneshot(Request::post("/api/logout").body(Body::empty()).unwrap())
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::NO_CONTENT);
        let set_cookie = response
            .headers()
            .get(header::SET_COOKIE)
            .unwrap()
            .to_str()
            .unwrap();
        assert!(set_cookie.contains("Max-Age=0"));
    }

    #[tokio::test]
    async fn callback_without_code_redirects_back() {
        let (_dir, state) = loopback_state().await;
        let web_origin = state.web_origin.clone();
        let response = app(state)
            .oneshot(
                Request::get("/oauth/callback?error=access_denied")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::SEE_OTHER);
        assert_eq!(
            response.headers().get(header::LOCATION).unwrap(),
            &format!("{web_origin}/?error=denied")
        );
    }

    #[tokio::test]
    async fn metadata_routes_absent_in_loopback_mode() {
        let (_dir, state) = loopback_state().await;
        let app = app(state);
        for path in ["/oauth/client-metadata.json", "/.well-known/jwks.json"] {
            let response = app
                .clone()
                .oneshot(Request::get(path).body(Body::empty()).unwrap())
                .await
                .unwrap();
            assert_eq!(response.status(), StatusCode::NOT_FOUND, "{path}");
        }
    }

    async fn seed_match(state: &AppState, owner: &str) {
        state
            .db
            .insert_match(&crate::db::MatchRow {
                id: "m1".into(),
                owner_did: owner.into(),
                status: "starting".into(),
                token: "tok".into(),
                task_arn: None,
                task_ip: None,
                created_at: 0, // the insert uses the column default
                scenario: None,
                started_at: None,
                ended_at: None,
            })
            .await
            .unwrap();
    }

    /// A listed `ready` match is now polled before it is reported — but only
    /// where there is a cluster to poll. Development has none, and this is the
    /// path that would have gone quiet if the fallback were wrong: a player
    /// running the API locally would see every match they ever launched lose
    /// its Open link.
    ///
    /// The polled path itself is not covered. `Matches` holds a concrete
    /// `aws::Aws` with no seam to fake, the same shape `plan_players`'s
    /// resolver has and for the same reason — a test that reached it would need
    /// live AWS egress, which `cargo test` must not.
    #[tokio::test]
    async fn a_ready_match_still_lists_where_there_is_no_cluster() {
        let (_dir, state) = loopback_state().await;
        assert!(state.matches.is_none(), "the loopback state grew a cluster");
        sign_in(&state, "did:plc:abc", None).await;
        seed_match(&state, "did:plc:abc").await;
        state.db.set_match_status("m1", "ready").await.unwrap();
        // A fighting seat: the link is the caller's own client, and a row with
        // no human seat for them is given none.
        state
            .db
            .insert_match_players(
                "m1",
                &[crate::db::MatchPlayer {
                    slot: "Davion".into(),
                    control: "human".into(),
                    did: Some("did:plc:abc".into()),
                    handle: None,
                    team: None,
                }],
            )
            .await
            .unwrap();

        let response = app(state.clone())
            .oneshot(
                with_cookie(Request::get("/api/matches"), &state)
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::OK);

        let body = body_json(response).await;
        let listed = &body["matches"][0];
        assert_eq!(listed["status"], "ready");
        assert_eq!(listed["url"], "/match/m1/did:plc:abc/");
    }

    /// A player's record is public, and the door into their match is not.
    ///
    /// The list itself is one query and covered by `list_matches`' own tests;
    /// what is new here is the split, which is invisible in both directions.
    /// A stranger handed a `url` would be handed a way into somebody else's
    /// container, and a player denied their own would find their Open link
    /// gone from the page the account menu leads to.
    ///
    /// `match_summaries` directly rather than through the route: the route
    /// resolves the address first, and resolution reads a DID document off
    /// the network, which `cargo test` must not.
    #[tokio::test]
    async fn a_strangers_row_carries_no_way_into_the_match() {
        let (_dir, state) = loopback_state().await;
        seed_match(&state, "did:plc:abc").await;
        state.db.set_match_status("m1", "ready").await.unwrap();
        state
            .db
            .insert_match_players(
                "m1",
                &[crate::db::MatchPlayer {
                    slot: "Davion".into(),
                    control: "human".into(),
                    did: Some("did:plc:abc".into()),
                    handle: None,
                    team: None,
                }],
            )
            .await
            .unwrap();

        let mine = match_summaries(&state, "did:plc:abc", Some("did:plc:abc"))
            .await
            .unwrap();
        assert_eq!(mine[0]["url"], "/match/m1/did:plc:abc/");

        for viewer in [Some("did:plc:stranger"), None] {
            let theirs = match_summaries(&state, "did:plc:abc", viewer)
                .await
                .unwrap();
            assert_eq!(theirs.len(), 1, "the record itself is public");
            assert_eq!(theirs[0]["matchId"], "m1");
            assert!(
                theirs[0]["url"].is_null(),
                "a viewer who is not the player was handed a way in",
            );
        }
    }

    /// An address nobody resolves is a 404, not an empty record: a page that
    /// answered with no rows would tell a reader this account has never
    /// fought, when what happened is that there is no account.
    #[tokio::test]
    async fn matches_for_nobody_are_not_an_empty_list() {
        let (_dir, mut state) = loopback_state().await;
        // Resolves nothing, so the handle below is nobody. Offline: the
        // fake resolver answers before anything reaches for the network.
        state.atproto = Arc::new(Atproto::for_test(state.db.clone(), &[]));

        let response = app(state.clone())
            .oneshot(
                Request::get("/api/profile/nobody.example/matches")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();

        assert_eq!(response.status(), StatusCode::NOT_FOUND);
    }

    fn with_cookie(
        request: axum::http::request::Builder,
        state: &AppState,
    ) -> axum::http::request::Builder {
        let cookie = state.signer.issue("did:plc:abc");
        request.header(
            header::COOKIE,
            format!("{}={}", crate::session::COOKIE_NAME, cookie.value()),
        )
    }

    #[tokio::test]
    async fn create_match_wants_session_mode_and_config() {
        let (_dir, state) = loopback_state().await;
        let app = app(state.clone());

        let response = app
            .clone()
            .oneshot(
                Request::post("/api/matches")
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(Body::from(r#"{"mode":"bot"}"#))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::UNAUTHORIZED, "no session");

        let response = app
            .clone()
            .oneshot(
                with_cookie(Request::post("/api/matches"), &state)
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(Body::from(r#"{"mode":"player"}"#))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::BAD_REQUEST, "non-bot mode");
        assert_eq!(body_json(response).await["message"], "Not yet available.");

        let response = app
            .oneshot(
                with_cookie(Request::post("/api/matches"), &state)
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(Body::from(r#"{"mode":"bot"}"#))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(
            response.status(),
            StatusCode::SERVICE_UNAVAILABLE,
            "no match config in loopback tests"
        );
    }

    /// The opponent's camo rides in on the same request. Staging it needs a
    /// match config these tests do not have, so what is checked here is that
    /// the field parses and that bytes nobody can use never fail a launch —
    /// 503 is the request having reached the launcher, which is as far as it
    /// goes without AWS.
    #[tokio::test]
    async fn a_bot_camo_is_accepted_and_never_fatal() {
        let (_dir, state) = loopback_state().await;
        let app = app(state.clone());

        for body in [
            r#"{"mode":"bot","botCamo":"AAAA"}"#,
            r#"{"mode":"bot","botCamo":"not base64!"}"#,
        ] {
            let response = app
                .clone()
                .oneshot(
                    with_cookie(Request::post("/api/matches"), &state)
                        .header(header::CONTENT_TYPE, "application/json")
                        .body(Body::from(body))
                        .unwrap(),
                )
                .await
                .unwrap();
            assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE, "{body}");
        }
    }

    /// The default is a setting, so it round-trips through the session
    /// response the site already asks for on every page load.
    #[tokio::test]
    async fn the_default_camo_is_set_cleared_and_reported() {
        let (_dir, state) = loopback_state().await;
        // /api/session answers 401 without a stored OAuth session, cookie or
        // not: the cookie alone is not a session.
        state
            .db
            .call(|conn| {
                conn.execute(
                    "INSERT INTO oauth_session (did, value) VALUES ('did:plc:abc', '{}')",
                    [],
                )
                .map(|_| ())
            })
            .await
            .unwrap();
        let app = app(state.clone());

        let post = |body: &'static str| {
            app.clone().oneshot(
                with_cookie(Request::post("/api/account/camo"), &state)
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(Body::from(body))
                    .unwrap(),
            )
        };
        let read = || {
            app.clone().oneshot(
                with_cookie(Request::get("/api/session"), &state)
                    .body(Body::empty())
                    .unwrap(),
            )
        };

        assert!(read().await.unwrap().status().is_success());
        assert_eq!(
            body_json(read().await.unwrap()).await["defaultCamo"],
            serde_json::Value::Null,
            "nothing is the default until someone says so"
        );

        let response = post(r#"{"rkey":"3mshoxhvtgt2f"}"#).await.unwrap();
        assert!(response.status().is_success());
        assert_eq!(
            body_json(read().await.unwrap()).await["defaultCamo"],
            "3mshoxhvtgt2f"
        );

        // Null clears it. This is how "start in the plain colours from now on"
        // is said, and it must not need deleting the camo to say it.
        let response = post(r#"{"rkey":null}"#).await.unwrap();
        assert!(response.status().is_success());
        assert_eq!(
            body_json(read().await.unwrap()).await["defaultCamo"],
            serde_json::Value::Null
        );
    }

    /// The value is stored and then echoed on every `/api/session` for as
    /// long as the account lasts, so it has to be a record key and not just a
    /// string that arrived.
    #[tokio::test]
    async fn a_default_camo_has_to_be_a_record_key() {
        let (_dir, state) = loopback_state().await;
        let app = app(state.clone());

        let post = |body: String| {
            app.clone().oneshot(
                with_cookie(Request::post("/api/account/camo"), &state)
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(Body::from(body))
                    .unwrap(),
            )
        };

        for bad in [
            r#"{"rkey":"has a space"}"#.to_owned(),
            r#"{"rkey":"."}"#.to_owned(),
            r#"{"rkey":".."}"#.to_owned(),
            r#"{"rkey":"slash/es"}"#.to_owned(),
            format!(r#"{{"rkey":"{}"}}"#, "a".repeat(513)),
        ] {
            let response = post(bad.clone()).await.unwrap();
            assert_eq!(
                response.status(),
                StatusCode::BAD_REQUEST,
                "should have been refused: {bad}"
            );
        }

        // And the shape a PDS actually mints still goes through.
        let response = post(r#"{"rkey":"3mshoxhvtgt2f"}"#.to_owned())
            .await
            .unwrap();
        assert!(response.status().is_success());
    }

    #[tokio::test]
    async fn setting_a_default_camo_needs_a_session() {
        let (_dir, state) = loopback_state().await;
        let response = app(state)
            .oneshot(
                Request::post("/api/account/camo")
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(Body::from(r#"{"rkey":"x"}"#))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
    }

    /// Everything /api/flare can decide without talking to a PDS, in the
    /// order it decides it: session, then whether flares are on, then the
    /// body, then the match claim.
    #[tokio::test]
    async fn flare_route_checks_session_body_and_match_first() {
        let (_dir, state) = loopback_state().await;
        let app = app(state.clone());

        let no_cookie = Request::post("/api/flare")
            .header(header::CONTENT_TYPE, "application/json")
            .body(Body::from(r#"{"title":"Stuck"}"#))
            .unwrap();
        let response = app.clone().oneshot(no_cookie).await.unwrap();
        assert_eq!(response.status(), StatusCode::UNAUTHORIZED);

        let signed_in = |body: &'static str| {
            with_cookie(
                Request::post("/api/flare").header(header::CONTENT_TYPE, "application/json"),
                &state,
            )
            .body(Body::from(body))
            .unwrap()
        };

        for (body, why) in [
            (r#"{}"#, "no title"),
            (r#"{"title":"   "}"#, "blank title"),
            (r#"{"title":"Stuck","png":"not base64!"}"#, "not base64"),
            (r#"{"title":"Stuck","png":"AAAA"}"#, "not a PNG"),
        ] {
            let response = app.clone().oneshot(signed_in(body)).await.unwrap();
            assert_eq!(response.status(), StatusCode::BAD_REQUEST, "{why}");
            assert!(body_json(response).await["message"].is_string(), "{why}");
        }

        // A match the sender does not fight in is refused before any write.
        let response = app
            .clone()
            .oneshot(signed_in(r#"{"title":"Stuck","matchId":"m1"}"#))
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::FORBIDDEN);
    }

    /// What the account's server said is what the player reads. The sign-in
    /// sentence is added only where signing in again could be the fix, so a
    /// 502 from a broken server does not send anyone round that loop.
    #[tokio::test]
    async fn a_servers_refusal_reaches_the_player_in_its_own_words() {
        let scope_missing = crate::atproto::Refusal {
            status: 403,
            error: Some("ScopeMissingError".into()),
            message: Some(r#"Missing required scope "repo:app.userinput.discussion""#.into()),
        };
        let response = write_failed(crate::atproto::WriteError::Refused(scope_missing));
        assert_eq!(response.status(), StatusCode::FORBIDDEN);
        let text = body_json(response).await["message"]
            .as_str()
            .unwrap()
            .to_owned();
        assert!(
            text.contains(r#"Missing required scope "repo:app.userinput.discussion""#),
            "the server's own words must survive: {text}"
        );
        assert!(text.contains("Sign out and sign in again"), "{text}");

        // A server that refused without saying why, and not with a status
        // fresh consent could change.
        let silent = crate::atproto::Refusal {
            status: 500,
            error: None,
            message: None,
        };
        let response = write_failed(crate::atproto::WriteError::Refused(silent));
        assert_eq!(response.status(), StatusCode::BAD_GATEWAY);
        let text = body_json(response).await["message"]
            .as_str()
            .unwrap()
            .to_owned();
        assert!(!text.contains("Sign out"), "{text}");
    }

    #[tokio::test]
    async fn flares_answer_not_found_when_no_space_is_configured() {
        let (_dir, mut state) = loopback_state().await;
        state.flare_space = None;
        let request = with_cookie(
            Request::post("/api/flare").header(header::CONTENT_TYPE, "application/json"),
            &state,
        )
        .body(Body::from(r#"{"title":"Stuck"}"#))
        .unwrap();
        let response = app(state).oneshot(request).await.unwrap();
        assert_eq!(response.status(), StatusCode::NOT_FOUND);
    }

    /// Everything the camo routes can decide without talking to a PDS. The
    /// writes themselves need a real session against a real server, so they
    /// are not reachable from here.
    #[tokio::test]
    async fn camo_routes_check_session_name_and_image_first() {
        let (_dir, state) = loopback_state().await;
        let app = app(state.clone());

        let signed_in = |method: &str, path: &str, body: &'static str| {
            with_cookie(
                axum::http::Request::builder()
                    .method(method)
                    .uri(path)
                    .header(header::CONTENT_TYPE, "application/json"),
                &state,
            )
            .body(Body::from(body))
            .unwrap()
        };

        // No cookie at all.
        for (method, path) in [
            ("POST", "/api/camo"),
            ("POST", "/api/camo/3kabc"),
            ("DELETE", "/api/camo/3kabc"),
        ] {
            let response = app
                .clone()
                .oneshot(
                    axum::http::Request::builder()
                        .method(method)
                        .uri(path)
                        .header(header::CONTENT_TYPE, "application/json")
                        .body(Body::from("{}"))
                        .unwrap(),
                )
                .await
                .unwrap();
            assert_eq!(
                response.status(),
                StatusCode::UNAUTHORIZED,
                "{method} {path} without a session"
            );
        }

        for (path, body, why) in [
            ("/api/camo", r#"{"name":"  ","png":"AAAA"}"#, "blank name"),
            (
                "/api/camo",
                r#"{"name":"Highland","png":"not base64!"}"#,
                "not base64",
            ),
            (
                "/api/camo",
                r#"{"name":"Highland","png":"AAAA"}"#,
                "not a PNG",
            ),
            ("/api/camo", r#"{"name":"Highland"}"#, "no image at all"),
            ("/api/camo/3kabc", r#"{}"#, "nothing to change"),
            (
                "/api/camo/3kabc",
                r#"{"name":"   "}"#,
                "a rename to nothing",
            ),
        ] {
            let response = app
                .clone()
                .oneshot(signed_in("POST", path, body))
                .await
                .unwrap();
            assert_eq!(response.status(), StatusCode::BAD_REQUEST, "{why}");
            assert!(body_json(response).await["message"].is_string(), "{why}");
        }
    }

    #[tokio::test]
    async fn match_status_is_owner_only() {
        let (_dir, state) = loopback_state().await;
        seed_match(&state, "did:plc:somebody-else").await;
        let app = app(state.clone());

        let response = app
            .clone()
            .oneshot(
                with_cookie(Request::get("/api/matches/m1"), &state)
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::NOT_FOUND, "not the owner");

        let response = app
            .oneshot(
                with_cookie(Request::get("/api/matches/absent"), &state)
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::NOT_FOUND);
    }

    /// The result route is the match screen's only way to learn its match is
    /// over, so the gate on it is the same one the status route has: a seat,
    /// or nothing. What it answers with a seat needs S3 and a cluster, which
    /// `cargo test` has neither of - the 503 below is that absence, and the
    /// three refusals above it are the whole of what this can prove.
    #[tokio::test]
    async fn match_result_needs_a_seat() {
        let (_dir, state) = loopback_state().await;
        seed_match(&state, "did:plc:somebody-else").await;
        let app = app(state.clone());

        let response = app
            .clone()
            .oneshot(
                Request::get("/api/matches/m1/result")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::UNAUTHORIZED, "no session");

        let response = app
            .clone()
            .oneshot(
                with_cookie(Request::get("/api/matches/m1/result"), &state)
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::NOT_FOUND, "not in the match");

        let response = app
            .clone()
            .oneshot(
                with_cookie(Request::get("/api/matches/absent/result"), &state)
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::NOT_FOUND, "no such match");

        state
            .db
            .insert_match(&crate::db::MatchRow {
                id: "m2".into(),
                owner_did: "did:plc:abc".into(),
                status: "ready".into(),
                token: "tok2".into(),
                task_arn: None,
                task_ip: None,
                created_at: 0,
                scenario: None,
                started_at: None,
                ended_at: None,
            })
            .await
            .unwrap();
        let response = app
            .oneshot(
                with_cookie(Request::get("/api/matches/m2/result"), &state)
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(
            response.status(),
            StatusCode::SERVICE_UNAVAILABLE,
            "a seat, and no cluster to answer from"
        );
    }

    #[tokio::test]
    async fn match_list_is_own_matches_only() {
        let (_dir, state) = loopback_state().await;
        seed_match(&state, "did:plc:abc").await;
        state
            .db
            .insert_match(&crate::db::MatchRow {
                id: "m2".into(),
                owner_did: "did:plc:somebody-else".into(),
                status: "ready".into(),
                token: "tok2".into(),
                task_arn: None,
                task_ip: None,
                created_at: 0,
                scenario: None,
                started_at: None,
                ended_at: None,
            })
            .await
            .unwrap();
        let app = app(state.clone());

        let response = app
            .clone()
            .oneshot(Request::get("/api/matches").body(Body::empty()).unwrap())
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::UNAUTHORIZED, "no session");

        let response = app
            .oneshot(
                with_cookie(Request::get("/api/matches"), &state)
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::OK);
        let body = body_json(response).await;
        let matches = body["matches"].as_array().unwrap();
        assert_eq!(matches.len(), 1, "only the caller's matches");
        assert_eq!(matches[0]["matchId"], "m1");
        assert!(matches[0]["url"].is_null(), "no link until ready");
    }

    #[tokio::test]
    async fn match_events_are_token_gated() {
        let (_dir, state) = loopback_state().await;
        seed_match(&state, "did:plc:abc").await;
        let app = app(state.clone());

        let response = app
            .clone()
            .oneshot(
                Request::post("/matches/m1/events/wrong")
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(Body::from(r#"{"event":"running"}"#))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::NOT_FOUND, "wrong token");

        let response = app
            .oneshot(
                Request::post("/matches/m1/events/tok")
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(Body::from(r#"{"event":"running"}"#))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::NO_CONTENT);
        let row = state.db.get_match("m1").await.unwrap().unwrap();
        assert_eq!(row.status, "running", "event advanced the status");
    }

    /// The minting route hands out an S3 write for a key, and it is the only
    /// way an artifact gets in, so every gate has to hold here or nowhere.
    ///
    /// The happy path is not here: it needs a real S3 client to sign with, and
    /// `loopback_state` builds no AWS. Signing itself is the SDK's.
    #[tokio::test]
    async fn minting_checks_the_token_the_match_and_the_name() {
        let (_dir, state) = loopback_state().await;
        seed_match(&state, "did:plc:abc").await;
        let app = app(state.clone());

        let response = app
            .clone()
            .oneshot(
                Request::post("/upload-urls/m1/wrong/turns/x.txt")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::NOT_FOUND, "wrong token");

        let response = app
            .clone()
            .oneshot(
                Request::post("/upload-urls/nosuchmatch/tok/turns/x.txt")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::NOT_FOUND, "unknown match");

        let response = app
            .clone()
            .oneshot(
                Request::post("/upload-urls/m1/tok/turns/../../../etc/passwd")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::BAD_REQUEST, "traversal name");

        // A well-formed name gets as far as needing AWS, which this state does
        // not have. Reaching 503 is the assertion: the request passed every gate
        // and stopped only for want of a client to sign with.
        let response = app
            .oneshot(
                Request::post("/upload-urls/m1/tok/diagnostics/megamek.log")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(
            response.status(),
            StatusCode::SERVICE_UNAVAILABLE,
            "a valid name should pass the gates"
        );
    }

    /// The other traversal test spells `..` literally, which only proves the
    /// check fires on a name that was never encoded. What actually holds the
    /// prefix is that axum's `Path` percent-decodes the `{*name}` capture
    /// before the handler runs, so an encoded `../` arrives as `../` and meets
    /// the same `contains("..")` it would have met unencoded.
    ///
    /// This matters more than it looks: the signature is handed to the caller to
    /// use with us out of the loop, so a key that gets past this check is a
    /// write we cannot take back.
    #[tokio::test]
    async fn encoded_traversal_names_are_rejected() {
        let (_dir, state) = loopback_state().await;
        seed_match(&state, "did:plc:abc").await;
        let app = app(state.clone());

        // Each of these decodes to a name with `../` in it. Routing happens on
        // the raw path, so `%2f` cannot split the capture; it only widens what
        // the decoded name can say.
        for name in [
            // fully encoded
            "%2e%2e%2f%2e%2e%2fmanifest.json",
            // mixed case, real separator
            "%2E%2E/%2E%2E/manifest.json",
            // dots literal, separator encoded
            "..%2f..%2fmanifest.json",
            // encoded separator after a plausible-looking first segment
            "turns%2f..%2f..%2fmanifest.json",
            // encoded leading separator, which would make the key absolute
            "%2fmatches/other/manifest.json",
        ] {
            let response = app
                .clone()
                .oneshot(
                    Request::post(format!("/upload-urls/m1/tok/{name}"))
                        .body(Body::empty())
                        .unwrap(),
                )
                .await
                .unwrap();
            assert_eq!(
                response.status(),
                StatusCode::BAD_REQUEST,
                "minting should reject {name}"
            );
        }
    }

    /// The decode is one pass deep, and that is the whole reason the check
    /// works. A double-encoded name survives it as the literal text
    /// `%2e%2e%2f`, which holds no `..`, so the name check accepts it - it is
    /// a name, not a traversal. The key becomes
    /// `matches/m1/artifacts/%2e%2e%2f...` with the percent signs as
    /// characters, so the object stays in the prefix under an ugly name.
    ///
    /// This is here as a canary. If it ever starts returning 400, something
    /// grew a second decode, and the danger is not this test failing but a
    /// name arriving at the check already decoded once too few times.
    #[tokio::test]
    async fn the_path_decode_is_one_pass_deep() {
        let (_dir, state) = loopback_state().await;
        seed_match(&state, "did:plc:abc").await;
        let app = app(state.clone());

        let response = app
            .clone()
            .oneshot(
                Request::post("/upload-urls/m1/tok/%252e%252e%252fmanifest.json")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(
            response.status(),
            StatusCode::SERVICE_UNAVAILABLE,
            "one decode leaves a literal name, which passes the gates and stops at AWS"
        );
    }

    /// One mint at one status, as the container would ask for it.
    async fn mint_at(status: &str) -> StatusCode {
        let (_dir, state) = loopback_state().await;
        seed_match(&state, "did:plc:abc").await;
        state.db.set_match_status("m1", status).await.unwrap();
        let app = app(state.clone());

        app.oneshot(
            Request::post("/upload-urls/m1/tok/diagnostics/megamek.log")
                .body(Body::empty())
                .unwrap(),
        )
        .await
        .unwrap()
        .status()
    }

    /// The point of the gate: the token is the whole credential here, and the
    /// mint turns it into a 900s S3 write. Once the match is over it buys
    /// nothing.
    #[tokio::test]
    async fn ended_matches_cannot_mint() {
        for status in ["over", "failed"] {
            assert_eq!(mint_at(status).await, StatusCode::GONE, "mint at {status}");
        }
    }

    /// The other half, and the half that would break production silently.
    ///
    /// arena's `container/exit/finalize.sh` sends the result, the screenshots
    /// and the diagnostics from the entrypoint's shutdown trap and only then
    /// posts `game-over`, so every one of those uploads arrives while the row
    /// still says `ready` - or `running`, if the probe never got that far. All of
    /// them must still be signed.
    ///
    /// 503 is the pass: `loopback_state` builds no AWS client, so a request that
    /// reaches the signer has cleared every gate and stopped only for want of
    /// one.
    #[tokio::test]
    async fn a_finalising_match_can_still_mint() {
        for status in ["starting", "running", "ready"] {
            assert_eq!(
                mint_at(status).await,
                StatusCode::SERVICE_UNAVAILABLE,
                "mint at {status}"
            );
        }
    }

    #[tokio::test]
    async fn scenarios_are_served_without_a_session() {
        let (_dir, state) = loopback_state().await;
        let response = app(state)
            .oneshot(Request::get("/api/scenarios").body(Body::empty()).unwrap())
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::OK);
        let body = body_json(response).await;
        let scenarios = body["scenarios"].as_array().unwrap();
        assert!(!scenarios.is_empty());
        // The fields the challenge screen builds its picker from.
        assert!(scenarios[0]["path"].is_string());
        assert!(scenarios[0]["name"].is_string());
        assert!(scenarios[0]["slots"][0]["name"].is_string());
        assert!(scenarios[0]["slots"][0]["units"].is_array());
    }

    /// The plan is checked before anything is staged or launched, so every
    /// wrong shape has to come back 400 with a sentence — and a plan that is
    /// right reaches the launcher, which without AWS is a 503.
    #[tokio::test]
    async fn create_match_validates_the_plan() {
        let (_dir, state) = loopback_state().await;
        let app = app(state.clone());

        let post = |body: String| {
            app.clone().oneshot(
                with_cookie(Request::post("/api/matches"), &state)
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(Body::from(body))
                    .unwrap(),
            )
        };

        for (body, why) in [
            (
                r#"{"scenario":"nope.mms","players":[]}"#.to_owned(),
                "unknown scenario",
            ),
            (
                r#"{"scenario":"TrainingScenarios/1-FirstRun.mms","players":[{"slot":"TraineeA","control":"self"}]}"#.to_owned(),
                "a slot is missing",
            ),
            (
                r#"{"scenario":"TrainingScenarios/1-FirstRun.mms","players":[{"slot":"TraineeA","control":"bot"},{"slot":"TraineeB","control":"bot"}]}"#.to_owned(),
                "nobody took a seat",
            ),
            (
                r#"{"scenario":"TrainingScenarios/1-FirstRun.mms","players":[{"slot":"TraineeA","control":"self"},{"slot":"TraineeB","control":"self"}]}"#.to_owned(),
                "two seats for one caller",
            ),
            (
                r#"{"scenario":"TrainingScenarios/1-FirstRun.mms","players":[{"slot":"TraineeA","control":"self"},{"slot":"TraineeB","control":"player"}]}"#.to_owned(),
                "an invited seat with nobody named",
            ),
            (
                r#"{"scenario":"TrainingScenarios/1-FirstRun.mms","players":[{"slot":"TraineeA","control":"self"},{"slot":"TraineeB","control":"wamble"}]}"#.to_owned(),
                "a control nothing knows",
            ),
            (
                r#"{"scenario":"TrainingScenarios/1-FirstRun.mms"}"#.to_owned(),
                "a scenario with no players",
            ),
        ] {
            let response = post(body).await.unwrap();
            assert_eq!(response.status(), StatusCode::BAD_REQUEST, "{why}");
            assert!(body_json(response).await["message"].is_string(), "{why}");
        }

        // The right shape passes the gates and stops only for want of a
        // launcher, which loopback tests do not build.
        let response = post(
            r#"{"scenario":"TrainingScenarios/1-FirstRun.mms","players":[{"slot":"TraineeA","control":"self"},{"slot":"TraineeB","control":"bot"}]}"#.to_owned(),
        )
        .await
        .unwrap();
        assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
    }

    /// The duplicate-seat guard `plan_players`'s `player` control shares
    /// with `matches::lobby::Lobbies::invite_handle` (see that module's own
    /// tests for its side of this): an invited handle that resolves to a
    /// DID already seated in this request is refused rather than seating
    /// the same person twice. Exercised at the HTTP layer, the same as
    /// `create_match_validates_the_plan`'s other shape checks — reachable
    /// at all only because `Atproto::for_test` gives `resolve_player` a
    /// scripted answer instead of a live DNS-over-HTTPS round trip
    /// (plan/lobby.md).
    #[tokio::test]
    async fn plan_players_refuses_an_invited_handle_already_seated() {
        let (_dir, mut state) = loopback_state().await;
        // "self" seats did:plc:abc (with_cookie's caller); "dupe.example"
        // resolves to that same DID, so the invited seat collides with it.
        state.atproto = Arc::new(Atproto::for_test(
            state.db.clone(),
            &[("dupe.example", "did:plc:abc")],
        ));
        let app = app(state.clone());

        let response = app
            .oneshot(
                with_cookie(Request::post("/api/matches"), &state)
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(Body::from(
                        r#"{"scenario":"TrainingScenarios/1-FirstRun.mms","players":[{"slot":"TraineeA","control":"self"},{"slot":"TraineeB","control":"player","handle":"dupe.example"}]}"#,
                    ))
                    .unwrap(),
            )
            .await
            .unwrap();

        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
        assert!(
            body_json(response).await["message"]
                .as_str()
                .unwrap()
                .contains("already has a slot"),
        );
    }

    /// Seats are what let an invited player see and reach a match someone
    /// else launched: their list carries it, its status answers them, and
    /// the proxy lets them through to it.
    #[tokio::test]
    async fn a_seat_is_as_good_as_ownership() {
        let (_dir, state) = loopback_state().await;
        seed_match(&state, "did:plc:somebody-else").await;
        state
            .db
            .insert_match_players(
                "m1",
                &[
                    crate::db::MatchPlayer {
                        slot: "TraineeA".into(),
                        control: "human".into(),
                        did: Some("did:plc:somebody-else".into()),
                        handle: Some("rival.example".into()),
                        team: None,
                    },
                    crate::db::MatchPlayer {
                        slot: "TraineeB".into(),
                        control: "human".into(),
                        did: Some("did:plc:abc".into()),
                        handle: Some("someone.example".into()),
                        team: None,
                    },
                ],
            )
            .await
            .unwrap();
        let app = app(state.clone());

        // Their list carries the match, with the seats to say who is in it.
        let response = app
            .clone()
            .oneshot(
                with_cookie(Request::get("/api/matches"), &state)
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        let body = body_json(response).await;
        let matches = body["matches"].as_array().unwrap();
        assert_eq!(matches.len(), 1, "the invited player sees the match");
        assert_eq!(matches[0]["players"][0]["handle"], "rival.example");

        // Status answers a seat holder; 503 is past authorization, stopped
        // only for want of a launcher.
        let response = app
            .clone()
            .oneshot(
                with_cookie(Request::get("/api/matches/m1"), &state)
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);

        // The proxy admits the seat too. 409 is past the identity check: the
        // match is simply not ready, where a stranger gets 403.
        let response = app
            .oneshot(
                with_cookie(Request::get("/match/m1/megamek/"), &state)
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::CONFLICT);
    }

    /// The other side of the seat: who you have fought, for the challenge
    /// screen's recent list. Newest first, bots dropped, self dropped.
    #[tokio::test]
    async fn opponents_are_listed_newest_first() {
        let (_dir, state) = loopback_state().await;
        for (id, other, at) in [("m1", "rival.example", 100), ("m2", "other.example", 200)] {
            state
                .db
                .insert_match(&crate::db::MatchRow {
                    id: id.into(),
                    owner_did: "did:plc:abc".into(),
                    status: "over".into(),
                    token: format!("tok-{id}"),
                    task_arn: None,
                    task_ip: None,
                    created_at: 0,
                    scenario: None,
                    started_at: None,
                    ended_at: None,
                })
                .await
                .unwrap();
            let (id2, at2) = (id.to_owned(), at);
            state
                .db
                .call(move |conn| {
                    conn.execute(
                        "UPDATE matches SET created_at = ?2 WHERE id = ?1",
                        (id2, at2),
                    )
                    .map(|_| ())
                })
                .await
                .unwrap();
            state
                .db
                .insert_match_players(
                    id,
                    &[
                        crate::db::MatchPlayer {
                            slot: "TraineeA".into(),
                            control: "human".into(),
                            did: Some("did:plc:abc".into()),
                            handle: Some("someone.example".into()),
                            team: None,
                        },
                        crate::db::MatchPlayer {
                            slot: "TraineeB".into(),
                            control: "human".into(),
                            did: Some(format!("did:plc:{other}")),
                            handle: Some(other.into()),
                            team: None,
                        },
                        crate::db::MatchPlayer {
                            slot: "TraineeC".into(),
                            control: "bot".into(),
                            did: None,
                            handle: Some("lance.blue".into()),
                            team: None,
                        },
                    ],
                )
                .await
                .unwrap();
        }
        let app = app(state.clone());

        let response = app
            .clone()
            .oneshot(Request::get("/api/opponents").body(Body::empty()).unwrap())
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::UNAUTHORIZED, "no session");

        let response = app
            .oneshot(
                with_cookie(Request::get("/api/opponents"), &state)
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::OK);
        let body = body_json(response).await;
        let opponents = body["opponents"].as_array().unwrap();
        assert_eq!(opponents.len(), 2, "two humans fought, no bot, no self");
        assert_eq!(opponents[0]["handle"], "other.example", "newest first");
        assert_eq!(opponents[1]["handle"], "rival.example");
        assert_eq!(opponents[0]["lastMatchAt"], 200);
    }

    /// Session-gated the same way /api/opponents is. The AppView call this
    /// route makes beyond that is exercised manually against a real account,
    /// same as the other AppView-dependent code in this codebase
    /// (avatars.ts, handle-typeahead.ts) — nothing here can stand up a fake
    /// public.api.bsky.app.
    #[tokio::test]
    async fn suggested_opponents_needs_a_session() {
        let (_dir, state) = loopback_state().await;
        let response = app(state)
            .oneshot(
                Request::get("/api/opponents/suggested")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
    }

    /// The client path is the caller's DID: the entry sends a player to
    /// /play/<their did>/, forward refuses them every identity that is not
    /// theirs, and the API hands each caller their own URL.
    #[tokio::test]
    async fn the_proxy_routes_each_seat_to_its_own_client() {
        let (_dir, state) = loopback_state().await;
        seed_match(&state, "did:plc:somebody-else").await;
        state.db.set_match_status("m1", "ready").await.unwrap();
        state.db.set_match_ip("m1", "127.0.0.1").await.unwrap();
        state
            .db
            .insert_match_players(
                "m1",
                &[
                    crate::db::MatchPlayer {
                        slot: "TraineeA".into(),
                        control: "human".into(),
                        did: Some("did:plc:somebody-else".into()),
                        handle: Some("rival.example".into()),
                        team: None,
                    },
                    crate::db::MatchPlayer {
                        slot: "TraineeB".into(),
                        control: "human".into(),
                        did: Some("did:plc:abc".into()),
                        handle: Some("someone.example".into()),
                        team: None,
                    },
                ],
            )
            .await
            .unwrap();
        let app = app(state.clone());
        let get = |path: &str| {
            app.clone().oneshot(
                with_cookie(Request::get(path), &state)
                    .body(Body::empty())
                    .unwrap(),
            )
        };

        // The front door reads the cookie and answers with the caller's own
        // identity in the path.
        let response = get("/match/m1").await.unwrap();
        assert_eq!(response.status(), StatusCode::SEE_OTHER);
        assert_eq!(
            response.headers().get(header::LOCATION).unwrap(),
            "/match/m1/did:plc:abc/"
        );

        // Another identity's client is refused outright.
        let response = get("/match/m1/did:plc:somebody-else/index.html")
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::FORBIDDEN);

        // Their own path clears the gate. Whatever the upstream answers —
        // nothing is listening in a test — it must not be the gate's 403.
        let response = get("/match/m1/did:plc:abc/").await.unwrap();
        assert_ne!(response.status(), StatusCode::FORBIDDEN);

        // And the list hands this caller their own URL, nobody else's.
        let response = get("/api/matches").await.unwrap();
        let body = body_json(response).await;
        assert_eq!(body["matches"][0]["url"], "/match/m1/did:plc:abc/");
    }

    /// A caller with no fighting seat — a match from before the seats table,
    /// or any future spectator — has no client in the container, and is told
    /// so rather than redirected to a path that cannot exist.
    #[tokio::test]
    async fn a_caller_without_a_seat_has_no_client() {
        let (_dir, state) = loopback_state().await;
        seed_match(&state, "did:plc:abc").await;
        state.db.set_match_status("m1", "ready").await.unwrap();
        state.db.set_match_ip("m1", "127.0.0.1").await.unwrap();
        let app = app(state.clone());

        let response = app
            .clone()
            .oneshot(
                with_cookie(Request::get("/match/m1"), &state)
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::NOT_FOUND);

        let response = app
            .oneshot(
                with_cookie(Request::get("/api/matches"), &state)
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        let body = body_json(response).await;
        assert!(body["matches"][0]["url"].is_null(), "no client, no link");
    }

    #[tokio::test]
    async fn match_proxy_gates_before_proxying() {
        let (_dir, state) = loopback_state().await;
        seed_match(&state, "did:plc:somebody-else").await;
        let app = app(state.clone());

        // Signed out: bounced to the site, not 401.
        let response = app
            .clone()
            .oneshot(Request::get("/match/m1").body(Body::empty()).unwrap())
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::SEE_OTHER);

        // Signed in as someone else: forbidden.
        let response = app
            .oneshot(
                with_cookie(Request::get("/match/m1/megamek/"), &state)
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::FORBIDDEN);
    }

    #[tokio::test]
    async fn cors_preflight_echoes_the_one_origin() {
        let (_dir, state) = loopback_state().await;
        let web_origin = state.web_origin.clone();
        let response = app(state)
            .oneshot(
                Request::builder()
                    .method(Method::OPTIONS)
                    .uri("/api/login")
                    .header(header::ORIGIN, &web_origin)
                    .header(header::ACCESS_CONTROL_REQUEST_METHOD, "POST")
                    .header(header::ACCESS_CONTROL_REQUEST_HEADERS, "content-type")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(
            response
                .headers()
                .get(header::ACCESS_CONTROL_ALLOW_ORIGIN)
                .unwrap(),
            &web_origin
        );
        assert_eq!(
            response
                .headers()
                .get(header::ACCESS_CONTROL_ALLOW_CREDENTIALS)
                .unwrap(),
            "true"
        );
    }

    // --- lobby live-sync -----------------------------------------------
    //
    // The one thing this feature exists to prove: two browsers on the same
    // lobby stay in sync. `tower::ServiceExt::oneshot`, which every test
    // above this uses, answers one request and hands back no upgrade to
    // drive - so this binds a real port and speaks real WebSocket to it,
    // the way the two browsers in production do.

    /// A session cookie for `did`, independent of any particular `AppState`
    /// — every `loopback_state()` in this test module signs with the same
    /// fixed `SECRET`, so a cookie built this way authenticates against any
    /// of them. Lets a lobby test issue as many callers' cookies as it needs
    /// without standing up a state (and its own temp database) per caller.
    fn test_cookie(did: &str) -> String {
        format!(
            "{}={}",
            crate::session::COOKIE_NAME,
            SessionSigner::new(SECRET.to_vec(), false, None)
                .issue(did)
                .value()
        )
    }

    type LobbySocket = tokio_tungstenite::WebSocketStream<
        tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
    >;

    async fn connect_to_lobby(
        addr: std::net::SocketAddr,
        match_id: &str,
        cookie: &str,
    ) -> LobbySocket {
        use tokio_tungstenite::tungstenite::client::ClientRequestBuilder;

        let uri = format!("ws://{addr}/api/matches/{match_id}/live")
            .parse()
            .unwrap();
        let request = ClientRequestBuilder::new(uri).with_header("Cookie", cookie);
        let (socket, response) = tokio_tungstenite::connect_async(request).await.unwrap();
        assert_eq!(response.status(), StatusCode::SWITCHING_PROTOCOLS);
        socket
    }

    /// The next `{"type":"presence", ...}` text frame's dids, skipping
    /// anything else.
    async fn next_presence(socket: &mut LobbySocket) -> Vec<String> {
        use futures_util::StreamExt;
        use tokio_tungstenite::tungstenite::Message as WsMessage;

        loop {
            match socket.next().await.unwrap().unwrap() {
                WsMessage::Text(text) => {
                    let value: serde_json::Value = serde_json::from_str(&text).unwrap();
                    if value["type"] == "presence" {
                        return value["here"]
                            .as_array()
                            .unwrap()
                            .iter()
                            .map(|w| w["did"].as_str().unwrap().to_owned())
                            .collect();
                    }
                }
                _ => continue,
            }
        }
    }

    /// Connecting puts a player in the room, and closing the socket takes
    /// them back out - the `Here` guard's whole job, over a real socket
    /// rather than against `Lobbies` directly.
    #[tokio::test]
    async fn a_socket_opening_and_closing_moves_the_room() {
        let (_dir, addr, match_id) =
            serving_lobby("TrainingScenarios/1-FirstRun.mms", "did:plc:abc").await;

        let mut a = connect_to_lobby(addr, &match_id, &test_cookie("did:plc:abc")).await;
        assert_eq!(next_presence(&mut a).await, vec!["did:plc:abc".to_owned()]);

        let b = connect_to_lobby(addr, &match_id, &test_cookie("did:plc:rival")).await;
        assert_eq!(
            next_presence(&mut a).await,
            vec!["did:plc:abc".to_owned(), "did:plc:rival".to_owned()],
            "the first connection hears the second arrive"
        );

        drop(b);
        assert_eq!(
            next_presence(&mut a).await,
            vec!["did:plc:abc".to_owned()],
            "and hears it leave when the socket closes"
        );
    }

    /// The next `{"type":"owner", ...}` text frame off a lobby socket. Sent
    /// once, first, and never again - see `lobby_socket`.
    async fn next_owner(socket: &mut LobbySocket) -> String {
        use futures_util::StreamExt;
        use tokio_tungstenite::tungstenite::Message as WsMessage;

        loop {
            match socket.next().await.unwrap().unwrap() {
                WsMessage::Text(text) => {
                    let value: serde_json::Value = serde_json::from_str(&text).unwrap();
                    if value["type"] == "owner" {
                        return value["owner"].as_str().unwrap().to_owned();
                    }
                }
                _ => continue,
            }
        }
    }

    /// The next `{"type":"scenario", ...}` text frame off a lobby socket,
    /// skipping anything else (a `seats` event, a ping in principle - the
    /// handler under test never sends binary or close on its own).
    async fn next_scenario(socket: &mut LobbySocket) -> String {
        use futures_util::StreamExt;
        use tokio_tungstenite::tungstenite::Message as WsMessage;

        loop {
            match socket.next().await.unwrap().unwrap() {
                WsMessage::Text(text) => {
                    let value: serde_json::Value = serde_json::from_str(&text).unwrap();
                    if value["type"] == "scenario" {
                        return value["scenario"].as_str().unwrap().to_owned();
                    }
                }
                _ => continue,
            }
        }
    }

    /// The next `{"type":"seats", ...}` text frame's slot->did map, skipping
    /// anything else. A bot seat's did is represented as `None`.
    async fn next_seats(socket: &mut LobbySocket) -> Vec<(String, Option<String>)> {
        use futures_util::StreamExt;
        use tokio_tungstenite::tungstenite::Message as WsMessage;

        loop {
            match socket.next().await.unwrap().unwrap() {
                WsMessage::Text(text) => {
                    let value: serde_json::Value = serde_json::from_str(&text).unwrap();
                    if value["type"] == "seats" {
                        return value["seats"]
                            .as_array()
                            .unwrap()
                            .iter()
                            .map(|s| {
                                (
                                    s["slot"].as_str().unwrap().to_owned(),
                                    s["did"].as_str().map(str::to_owned),
                                )
                            })
                            .collect();
                    }
                }
                _ => continue,
            }
        }
    }

    /// Starts a real bound server for a lobby test and opens one lobby on
    /// it under `opener_did`, returning the temp dir backing its database
    /// (kept alive for as long as the caller holds it - dropping it deletes
    /// the sqlite file out from under the still-running server), the
    /// address, and the new lobby's match id.
    async fn serving_lobby(
        scenario: &str,
        opener_did: &str,
    ) -> (tempfile::TempDir, std::net::SocketAddr, String) {
        serving_lobby_with_resolves(scenario, opener_did, &[]).await
    }

    /// The same real, bound server `serving_lobby` starts, with
    /// `Atproto::for_test` scripting `resolve_player`'s answers instead of
    /// the real one — the seam a happy-path `invite` test over the socket
    /// needs, since resolving for real would mean this test depending on
    /// live DNS-over-HTTPS egress.
    async fn serving_lobby_with_resolves(
        scenario: &str,
        opener_did: &str,
        resolves: &[(&str, &str)],
    ) -> (tempfile::TempDir, std::net::SocketAddr, String) {
        let (dir, mut state) = loopback_state().await;
        state.atproto = Arc::new(Atproto::for_test(state.db.clone(), resolves));
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let app = app(state);
        tokio::spawn(async move {
            axum::serve(listener, app).await.unwrap();
        });

        let client = reqwest::Client::new();
        let opened = client
            .post(format!("http://{addr}/api/matches/lobby"))
            .header("Cookie", test_cookie(opener_did))
            .header("Content-Type", "application/json")
            .body(serde_json::json!({ "scenario": scenario }).to_string())
            .send()
            .await
            .unwrap();
        assert_eq!(opened.status(), reqwest::StatusCode::OK);
        let opened_body: serde_json::Value =
            serde_json::from_str(&opened.text().await.unwrap()).unwrap();
        (
            dir,
            addr,
            opened_body["matchId"].as_str().unwrap().to_owned(),
        )
    }

    /// Opens a lobby, connects two signed-in clients to its channel, and
    /// proves both directions of the sync: a fresh connection sees the
    /// current scenario immediately, and a change one client sends arrives
    /// on the other (and back on itself - `Lobbies::set_scenario` broadcasts
    /// to every subscriber, sender included). An invalid scenario in between
    /// is confirmed inert by checking it never surfaces as the "next" value
    /// anywhere.
    #[tokio::test]
    async fn two_clients_stay_in_sync_over_the_lobby_socket() {
        use tokio_tungstenite::tungstenite::Message as WsMessage;

        let (_dir, addr, match_id) =
            serving_lobby("TrainingScenarios/1-FirstRun.mms", "did:plc:abc").await;
        let cookie_a = test_cookie("did:plc:abc");
        let cookie_b = test_cookie("did:plc:someone-else");

        let mut a = connect_to_lobby(addr, &match_id, &cookie_a).await;
        let mut b = connect_to_lobby(addr, &match_id, &cookie_b).await;

        // Both are told whose lobby this is before anything else, the
        // opener's own connection included - it is the one identity a
        // client cannot work out for itself, and the one player never asked
        // to press Ready.
        assert_eq!(next_owner(&mut a).await, "did:plc:abc");
        assert_eq!(next_owner(&mut b).await, "did:plc:abc");

        // Both start from the scenario the lobby opened with, and an empty
        // seat map - no seat, no invite, nothing but a valid session, per
        // the module's permissive connect-time auth for this slice.
        assert_eq!(
            next_scenario(&mut a).await,
            "TrainingScenarios/1-FirstRun.mms"
        );
        assert_eq!(next_seats(&mut a).await, Vec::new());
        assert_eq!(
            next_scenario(&mut b).await,
            "TrainingScenarios/1-FirstRun.mms"
        );
        assert_eq!(next_seats(&mut b).await, Vec::new());

        // An unknown path is ignored - it must never become the "next"
        // value either client sees.
        futures_util::SinkExt::send(
            &mut a,
            WsMessage::text(r#"{"type":"scenario","scenario":"nope.mms"}"#),
        )
        .await
        .unwrap();
        futures_util::SinkExt::send(
            &mut a,
            WsMessage::text(
                r#"{"type":"scenario","scenario":"TrainingScenarios/2-FinalExam.mms"}"#,
            ),
        )
        .await
        .unwrap();

        assert_eq!(
            next_scenario(&mut b).await,
            "TrainingScenarios/2-FinalExam.mms",
            "the other client hears the change"
        );
        assert_eq!(
            next_scenario(&mut a).await,
            "TrainingScenarios/2-FinalExam.mms",
            "the sender hears its own accepted change too"
        );
    }

    /// A connected player claims a slot as themselves; the other connection
    /// hears the updated seat map, did included.
    #[tokio::test]
    async fn claiming_a_seat_over_the_socket_reaches_the_other_connection() {
        use tokio_tungstenite::tungstenite::Message as WsMessage;

        let (_dir, addr, match_id) =
            serving_lobby("TrainingScenarios/1-FirstRun.mms", "did:plc:abc").await;
        let cookie_a = test_cookie("did:plc:abc");
        let cookie_b = test_cookie("did:plc:someone-else");

        let mut a = connect_to_lobby(addr, &match_id, &cookie_a).await;
        let mut b = connect_to_lobby(addr, &match_id, &cookie_b).await;
        next_scenario(&mut a).await;
        next_seats(&mut a).await;
        next_scenario(&mut b).await;
        next_seats(&mut b).await;

        futures_util::SinkExt::send(
            &mut a,
            WsMessage::text(r#"{"type":"claimSeat","slot":"TraineeA"}"#),
        )
        .await
        .unwrap();

        let seats = next_seats(&mut b).await;
        assert_eq!(
            seats,
            vec![("TraineeA".to_owned(), Some("did:plc:abc".to_owned()))]
        );
        // The claimer hears its own claim too, the same way a scenario
        // change echoes back to its sender.
        assert_eq!(
            next_seats(&mut a).await,
            vec![("TraineeA".to_owned(), Some("did:plc:abc".to_owned()))]
        );
    }

    /// A bot slot has no did in the wire shape.
    #[tokio::test]
    async fn assigning_a_bot_seat_over_the_socket_shows_no_did() {
        use tokio_tungstenite::tungstenite::Message as WsMessage;

        let (_dir, addr, match_id) =
            serving_lobby("TrainingScenarios/1-FirstRun.mms", "did:plc:abc").await;
        let cookie = test_cookie("did:plc:abc");
        let mut socket = connect_to_lobby(addr, &match_id, &cookie).await;
        next_scenario(&mut socket).await;
        next_seats(&mut socket).await;

        futures_util::SinkExt::send(
            &mut socket,
            WsMessage::text(r#"{"type":"botSeat","slot":"TraineeB"}"#),
        )
        .await
        .unwrap();

        assert_eq!(
            next_seats(&mut socket).await,
            vec![("TraineeB".to_owned(), None)]
        );
    }

    /// An invite to a slot the current scenario does not have is a no-op —
    /// the same "cosmetic, not fatal" refusal `claimSeat`/`botSeat` give a
    /// bad slot. Provable without any network: `Lobbies::invite_handle`
    /// checks the slot before it ever calls `Atproto::resolve_player`, so
    /// this fails fast and the socket keeps answering right after — shown
    /// here by a real claim landing on the same connection immediately
    /// after the refused invite.
    #[tokio::test]
    async fn inviting_an_unknown_slot_is_a_silent_no_op() {
        use tokio_tungstenite::tungstenite::Message as WsMessage;

        let (_dir, addr, match_id) =
            serving_lobby("TrainingScenarios/1-FirstRun.mms", "did:plc:abc").await;
        let cookie = test_cookie("did:plc:abc");
        let mut socket = connect_to_lobby(addr, &match_id, &cookie).await;
        next_scenario(&mut socket).await;
        next_seats(&mut socket).await;

        futures_util::SinkExt::send(
            &mut socket,
            WsMessage::text(r#"{"type":"invite","slot":"NotARealSlot","handle":"quire.example"}"#),
        )
        .await
        .unwrap();
        futures_util::SinkExt::send(
            &mut socket,
            WsMessage::text(r#"{"type":"botSeat","slot":"TraineeB"}"#),
        )
        .await
        .unwrap();

        assert_eq!(
            next_seats(&mut socket).await,
            vec![("TraineeB".to_owned(), None)],
            "the socket is still alive and answering after the refused invite"
        );
    }

    /// The happy path over the real wire: a handle that resolves gets
    /// seated, the same event a `claimSeat` or `botSeat` would broadcast.
    /// `matches::lobby`'s own test module covers `invite_handle` directly;
    /// this is the same resolver seam (`Atproto::for_test`, via
    /// `serving_lobby_with_resolves`) proven one layer up, over an actual
    /// socket rather than a direct call.
    #[tokio::test]
    async fn inviting_a_resolvable_handle_seats_it() {
        use tokio_tungstenite::tungstenite::Message as WsMessage;

        let (_dir, addr, match_id) = serving_lobby_with_resolves(
            "TrainingScenarios/1-FirstRun.mms",
            "did:plc:abc",
            &[("quire.example", "did:plc:invited")],
        )
        .await;
        let cookie = test_cookie("did:plc:abc");
        let mut socket = connect_to_lobby(addr, &match_id, &cookie).await;
        next_scenario(&mut socket).await;
        next_seats(&mut socket).await;

        futures_util::SinkExt::send(
            &mut socket,
            WsMessage::text(r#"{"type":"invite","slot":"TraineeB","handle":"quire.example"}"#),
        )
        .await
        .unwrap();

        assert_eq!(
            next_seats(&mut socket).await,
            vec![("TraineeB".to_owned(), Some("did:plc:invited".to_owned()))]
        );
    }

    /// Two real, separately-connected clients racing to claim the same
    /// slot: exactly one claim takes, proven end to end over real sockets
    /// rather than by calling `Lobbies` directly.
    #[tokio::test]
    async fn two_sockets_racing_the_same_slot_seat_exactly_one() {
        use tokio_tungstenite::tungstenite::Message as WsMessage;

        let (_dir, addr, match_id) =
            serving_lobby("TrainingScenarios/1-FirstRun.mms", "did:plc:a").await;
        let cookie_a = test_cookie("did:plc:a");
        let cookie_b = test_cookie("did:plc:b");

        let mut a = connect_to_lobby(addr, &match_id, &cookie_a).await;
        let mut b = connect_to_lobby(addr, &match_id, &cookie_b).await;
        next_scenario(&mut a).await;
        next_seats(&mut a).await;
        next_scenario(&mut b).await;
        next_seats(&mut b).await;

        futures_util::SinkExt::send(
            &mut a,
            WsMessage::text(r#"{"type":"claimSeat","slot":"TraineeA"}"#),
        )
        .await
        .unwrap();
        futures_util::SinkExt::send(
            &mut b,
            WsMessage::text(r#"{"type":"claimSeat","slot":"TraineeA"}"#),
        )
        .await
        .unwrap();

        // Both connections converge on the same one-seat outcome, whichever
        // DID won - `db::tests::claim_match_player_is_first_write_wins_...`
        // already proves the DB layer decides this atomically; what a
        // socket client needs is that both sides of the channel agree.
        let seats_a = next_seats(&mut a).await;
        let seats_b = next_seats(&mut b).await;
        assert_eq!(seats_a, seats_b);
        assert_eq!(seats_a.len(), 1);
        assert_eq!(seats_a[0].0, "TraineeA");
    }

    // --- lobby kick --------------------------------------------------------
    //
    // Real sockets, not just `Lobbies::kick_seat` directly: the check itself
    // is proven at the `matches::lobby` level, but ownership only exists on
    // this channel as "whichever session opened this connection", so what a
    // kick earns a real-socket test over `two_sockets_racing_...`'s own
    // template is proof that the two identities involved genuinely are two
    // separate authenticated connections, not two calls sharing one caller.

    /// The lobby's owner kicks a claimed seat; both connections converge on
    /// the freed seat map, the same "everyone hears it, sender included"
    /// posture every other seat change already has on this channel.
    #[tokio::test]
    async fn kicking_a_seat_over_the_socket_reaches_the_other_connection() {
        use tokio_tungstenite::tungstenite::Message as WsMessage;

        let (_dir, addr, match_id) =
            serving_lobby("TrainingScenarios/1-FirstRun.mms", "did:plc:abc").await;
        let cookie_a = test_cookie("did:plc:abc");
        let cookie_b = test_cookie("did:plc:someone-else");

        let mut a = connect_to_lobby(addr, &match_id, &cookie_a).await;
        let mut b = connect_to_lobby(addr, &match_id, &cookie_b).await;
        next_scenario(&mut a).await;
        next_seats(&mut a).await;
        next_scenario(&mut b).await;
        next_seats(&mut b).await;

        futures_util::SinkExt::send(
            &mut b,
            WsMessage::text(r#"{"type":"claimSeat","slot":"TraineeA"}"#),
        )
        .await
        .unwrap();
        assert_eq!(
            next_seats(&mut a).await,
            vec![(
                "TraineeA".to_owned(),
                Some("did:plc:someone-else".to_owned())
            )]
        );
        next_seats(&mut b).await;

        // `a` is the lobby's opener - it may kick the seat `b` just claimed.
        futures_util::SinkExt::send(
            &mut a,
            WsMessage::text(r#"{"type":"kick","slot":"TraineeA"}"#),
        )
        .await
        .unwrap();

        assert_eq!(next_seats(&mut b).await, Vec::new());
        assert_eq!(next_seats(&mut a).await, Vec::new());
    }

    /// A non-owner's kick request is a silent no-op: the seat it targeted
    /// stays exactly as it was, and the connection that sent the refused
    /// request is still alive and answering right after - shown the same way
    /// `inviting_an_unknown_slot_is_a_silent_no_op` shows a refused invite
    /// does not disturb the socket, by having a real action from the same
    /// connection land immediately afterward.
    #[tokio::test]
    async fn a_non_owners_kick_attempt_over_the_socket_is_a_silent_no_op() {
        use tokio_tungstenite::tungstenite::Message as WsMessage;

        let (_dir, addr, match_id) =
            serving_lobby("TrainingScenarios/1-FirstRun.mms", "did:plc:abc").await;
        let cookie_a = test_cookie("did:plc:abc");
        let cookie_b = test_cookie("did:plc:someone-else");

        let mut a = connect_to_lobby(addr, &match_id, &cookie_a).await;
        let mut b = connect_to_lobby(addr, &match_id, &cookie_b).await;
        next_scenario(&mut a).await;
        next_seats(&mut a).await;
        next_scenario(&mut b).await;
        next_seats(&mut b).await;

        futures_util::SinkExt::send(
            &mut a,
            WsMessage::text(r#"{"type":"claimSeat","slot":"TraineeA"}"#),
        )
        .await
        .unwrap();
        assert_eq!(
            next_seats(&mut a).await,
            vec![("TraineeA".to_owned(), Some("did:plc:abc".to_owned()))]
        );
        next_seats(&mut b).await;

        // `b` is not this lobby's opener (`did:plc:abc` is) - this must not
        // remove `a`'s seat.
        futures_util::SinkExt::send(
            &mut b,
            WsMessage::text(r#"{"type":"kick","slot":"TraineeA"}"#),
        )
        .await
        .unwrap();
        futures_util::SinkExt::send(
            &mut b,
            WsMessage::text(r#"{"type":"botSeat","slot":"TraineeB"}"#),
        )
        .await
        .unwrap();

        assert_eq!(
            next_seats(&mut b).await,
            vec![
                ("TraineeA".to_owned(), Some("did:plc:abc".to_owned())),
                ("TraineeB".to_owned(), None),
            ],
            "the seat a non-owner tried to kick is still there, and the socket is still alive and answering"
        );
    }

    // --- lobby chat ------------------------------------------------------
    //
    // Same shape of proof as the sync tests above, for the one thing on this
    // channel a player types rather than picks: real sockets, and what one
    // says arriving at the other. `matches::lobby`'s own tests cover the
    // backlog cap and what makes a line unsendable; these cover the wire.

    /// The next `{"type":"chat", ...}` text frame off a lobby socket as
    /// `(did, text)`, skipping anything else.
    async fn next_chat(socket: &mut LobbySocket) -> (String, String) {
        use futures_util::StreamExt;
        use tokio_tungstenite::tungstenite::Message as WsMessage;

        loop {
            match socket.next().await.unwrap().unwrap() {
                WsMessage::Text(text) => {
                    let value: serde_json::Value = serde_json::from_str(&text).unwrap();
                    if value["type"] == "chat" {
                        return (
                            value["did"].as_str().unwrap().to_owned(),
                            value["text"].as_str().unwrap().to_owned(),
                        );
                    }
                }
                _ => continue,
            }
        }
    }

    /// A line one connection sends reaches the other, attributed, and echoes
    /// back to the sender — the same "everyone converges, sender included"
    /// posture a scenario change has.
    #[tokio::test]
    async fn chat_reaches_the_other_connection_and_the_sender() {
        use tokio_tungstenite::tungstenite::Message as WsMessage;

        let (_dir, addr, match_id) =
            serving_lobby("TrainingScenarios/1-FirstRun.mms", "did:plc:abc").await;
        let mut a = connect_to_lobby(addr, &match_id, &test_cookie("did:plc:abc")).await;
        let mut b = connect_to_lobby(addr, &match_id, &test_cookie("did:plc:rival")).await;
        next_seats(&mut a).await;
        next_seats(&mut b).await;

        futures_util::SinkExt::send(
            &mut a,
            WsMessage::text(r#"{"type":"chat","text":"garboil"}"#),
        )
        .await
        .unwrap();

        assert_eq!(
            next_chat(&mut b).await,
            ("did:plc:abc".to_owned(), "garboil".to_owned())
        );
        assert_eq!(
            next_chat(&mut a).await,
            ("did:plc:abc".to_owned(), "garboil".to_owned())
        );
    }

    /// A browser that opens the lobby after the talking started is replayed
    /// what it missed, in order, before anything said next.
    #[tokio::test]
    async fn a_connection_that_opens_late_is_replayed_the_backlog() {
        use tokio_tungstenite::tungstenite::Message as WsMessage;

        let (_dir, addr, match_id) =
            serving_lobby("TrainingScenarios/1-FirstRun.mms", "did:plc:abc").await;
        let mut a = connect_to_lobby(addr, &match_id, &test_cookie("did:plc:abc")).await;
        next_seats(&mut a).await;

        for text in ["swive", "thole"] {
            futures_util::SinkExt::send(
                &mut a,
                WsMessage::text(format!(r#"{{"type":"chat","text":"{text}"}}"#)),
            )
            .await
            .unwrap();
            // Waiting for the echo before sending the next one is what makes
            // the replayed order below a real assertion rather than a race
            // between two in-flight requests.
            next_chat(&mut a).await;
        }

        let mut b = connect_to_lobby(addr, &match_id, &test_cookie("did:plc:rival")).await;
        assert_eq!(
            next_scenario(&mut b).await,
            "TrainingScenarios/1-FirstRun.mms"
        );
        assert_eq!(next_seats(&mut b).await, Vec::new());
        assert_eq!(next_chat(&mut b).await.1, "swive");
        assert_eq!(next_chat(&mut b).await.1, "thole");
    }

    /// A line the server refuses produces nothing at all — proven without a
    /// timeout by sending a good line behind a bad one and finding the good
    /// one is what arrives next.
    #[tokio::test]
    async fn a_refused_line_never_reaches_the_channel() {
        use tokio_tungstenite::tungstenite::Message as WsMessage;

        let (_dir, addr, match_id) =
            serving_lobby("TrainingScenarios/1-FirstRun.mms", "did:plc:abc").await;
        let mut socket = connect_to_lobby(addr, &match_id, &test_cookie("did:plc:abc")).await;
        next_seats(&mut socket).await;

        for bad in [
            r#"{"type":"chat","text":"   "}"#,
            r#"{"type":"chat"}"#,          // no text at all: a shape error
            r#"{"type":"chat","text":1}"#, // nor a string
        ] {
            futures_util::SinkExt::send(&mut socket, WsMessage::text(bad))
                .await
                .unwrap();
        }
        futures_util::SinkExt::send(
            &mut socket,
            WsMessage::text(r#"{"type":"chat","text":"yestreen"}"#),
        )
        .await
        .unwrap();

        assert_eq!(next_chat(&mut socket).await.1, "yestreen");
    }

    // --- deploy ----------------------------------------------------------
    //
    // Turning a fully-seated lobby into a real match. `loopback_state`
    // builds no AWS client (same as every other test in this module - see
    // `create_match_wants_session_mode_and_config`), so `Matches::start`
    // itself is never reachable here: the 503 branch is what every test
    // below the ownership/validation checks actually exercises, matching
    // the boundary `create_match`'s own tests already live with. The seat
    // conversion (`lobby_seats_to_players`) and slot-completeness check
    // (`seats_fill_every_slot`) are proven directly, without HTTP or AWS.

    #[test]
    fn seats_fill_every_slot_requires_every_slot_exactly_once() {
        let scenario = crate::matches::scenarios::find("TrainingScenarios/1-FirstRun.mms").unwrap();
        let seat = |slot: &str| crate::db::MatchPlayer {
            slot: slot.into(),
            control: "human".into(),
            did: Some("did:plc:abc".into()),
            handle: None,
            team: None,
        };

        assert!(seats_fill_every_slot(
            scenario,
            &[seat("TraineeA"), seat("TraineeB")]
        ));
        assert!(
            !seats_fill_every_slot(scenario, &[seat("TraineeA")]),
            "a slot left unclaimed must fail"
        );
        assert!(
            !seats_fill_every_slot(
                scenario,
                &[seat("TraineeA"), seat("TraineeB"), seat("TraineeA")]
            ),
            "an extra seat outside the scenario's slots must fail too"
        );
    }

    #[tokio::test]
    async fn lobby_seats_to_players_settles_camo_for_humans_and_none_for_bots() {
        let (_dir, state) = loopback_state().await;
        state
            .db
            .set_default_camo("did:plc:abc", Some("3kabc123"))
            .await
            .unwrap();
        let seats = vec![
            crate::db::MatchPlayer {
                slot: "TraineeA".into(),
                control: "human".into(),
                did: Some("did:plc:abc".into()),
                handle: Some("@alice.example".into()),
                team: None,
            },
            crate::db::MatchPlayer {
                slot: "TraineeB".into(),
                control: "bot".into(),
                did: None,
                handle: Some("Princess".into()),
                team: None,
            },
        ];

        let players = lobby_seats_to_players(
            &state,
            "did:plc:test",
            "TrainingScenarios/1-FirstRun.mms",
            seats,
            &Default::default(),
            None,
        )
        .await;

        assert_eq!(players.len(), 2);
        assert!(!players[0].bot);
        assert_eq!(players[0].did.as_deref(), Some("did:plc:abc"));
        assert_eq!(players[0].camo_rkey.as_deref(), Some("3kabc123"));
        assert_eq!(players[0].camo_png, None);
        assert!(players[1].bot);
        assert_eq!(players[1].did, None);
        assert_eq!(players[1].camo_rkey, None);
        assert_eq!(
            players[1].camo_png, None,
            "no rolled camo exists for a lobby-assigned bot seat"
        );
    }

    #[tokio::test]
    async fn deploy_lobby_requires_a_session() {
        let (_dir, state) = loopback_state().await;
        let response = app(state)
            .oneshot(
                Request::post("/api/matches/nonexistent/deploy")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
    }

    #[tokio::test]
    async fn deploy_lobby_returns_not_found_for_a_missing_id() {
        let (_dir, state) = loopback_state().await;
        let response = app(state.clone())
            .oneshot(
                with_cookie(Request::post("/api/matches/nonexistent/deploy"), &state)
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::NOT_FOUND);
    }

    #[tokio::test]
    async fn deploy_lobby_requires_the_lobby_owner() {
        let (_dir, state) = loopback_state().await;
        let id = state
            .lobbies
            .open("did:plc:owner", "TrainingScenarios/1-FirstRun.mms")
            .await
            .unwrap();
        let app = app(state);

        let response = app
            .oneshot(
                Request::post(format!("/api/matches/{id}/deploy"))
                    .header(header::COOKIE, test_cookie("did:plc:someone-else"))
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::FORBIDDEN);
    }

    /// An incomplete lobby is rejected before any status change: the row is
    /// still `lobby` afterwards, not `deploying` — nothing was claimed,
    /// because nothing was ever going to launch.
    #[tokio::test]
    async fn deploy_lobby_rejects_an_incomplete_lobby() {
        let (_dir, state) = loopback_state().await;
        // `with_cookie` below always signs as "did:plc:abc" (see its own
        // definition), so the lobby's owner has to match for this to reach
        // validation rather than the ownership check.
        let id = state
            .lobbies
            .open("did:plc:abc", "TrainingScenarios/1-FirstRun.mms")
            .await
            .unwrap();
        assert!(
            state
                .lobbies
                .claim_seat(&id, "TraineeA", "did:plc:abc")
                .await
        );
        // TraineeB is left unclaimed.

        let response = app(state.clone())
            .oneshot(
                with_cookie(Request::post(format!("/api/matches/{id}/deploy")), &state)
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
        assert_eq!(
            state.db.get_match(&id).await.unwrap().unwrap().status,
            "lobby",
            "an incomplete lobby must not be claimed for deploy"
        );
    }

    /// A second human in the lobby has to press Ready first, and pressing it
    /// is the whole of what unblocks the launch: the same request that was a
    /// 400 gets as far as the 503 every launch route in this test module
    /// reaches afterwards.
    #[tokio::test]
    async fn deploy_lobby_waits_for_the_other_player_to_be_ready() {
        let (_dir, state) = loopback_state().await;
        let id = state
            .lobbies
            .open("did:plc:abc", "TrainingScenarios/1-FirstRun.mms")
            .await
            .unwrap();
        assert!(
            state
                .lobbies
                .claim_seat(&id, "TraineeA", "did:plc:abc")
                .await
        );
        assert!(
            state
                .lobbies
                .claim_seat(&id, "TraineeB", "did:plc:rival")
                .await
        );

        let response = app(state.clone())
            .oneshot(
                with_cookie(Request::post(format!("/api/matches/{id}/deploy")), &state)
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
        assert_eq!(
            state.db.get_match(&id).await.unwrap().unwrap().status,
            "lobby",
            "a lobby nobody is ready in must not be claimed for deploy"
        );

        // The deploying caller is never waited on for their own press —
        // only the other seat is, so this one press is enough.
        assert!(state.lobbies.set_ready(&id, "did:plc:rival", true));

        let response = app(state.clone())
            .oneshot(
                with_cookie(Request::post(format!("/api/matches/{id}/deploy")), &state)
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
    }

    /// A fully-seated lobby passes validation and reaches the same 503
    /// every launch route in this test module answers with — `loopback_state`
    /// builds no AWS client, so this is as far as a real deploy can go here.
    /// Nobody presses Ready here and nobody has to: the only human seat is
    /// the caller's own, which is what a fight against the bot alone is.
    #[tokio::test]
    async fn deploy_lobby_needs_match_launching_configured() {
        let (_dir, state) = loopback_state().await;
        let id = state
            .lobbies
            .open("did:plc:abc", "TrainingScenarios/1-FirstRun.mms")
            .await
            .unwrap();
        assert!(
            state
                .lobbies
                .claim_seat(&id, "TraineeA", "did:plc:abc")
                .await
        );
        assert!(
            state
                .lobbies
                .assign_bot(&id, "TraineeB", "did:plc:abc")
                .await
        );

        let response = app(state.clone())
            .oneshot(
                with_cookie(Request::post(format!("/api/matches/{id}/deploy")), &state)
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
    }

    /// Every socket still on a lobby's channel, not just the HTTP caller,
    /// learns the new match's id — `Deployed` is what `deploy_lobby`
    /// broadcasts the moment `Matches::start` succeeds. Simulated directly
    /// via `broadcast_deployed` rather than through the real endpoint: a
    /// real launch needs AWS this test suite does not have, the same
    /// limitation `deploy_lobby_needs_match_launching_configured` above
    /// hits; what is under test here is the wire shape and delivery, not
    /// the launch itself.
    #[tokio::test]
    async fn a_connected_socket_receives_the_deployed_event() {
        use futures_util::StreamExt;
        use tokio_tungstenite::tungstenite::Message as WsMessage;

        let (_dir, state) = loopback_state().await;
        // Kept before `app(state)` consumes it: this is the same channel
        // registry the running server's `deploy_lobby` would broadcast on,
        // and the only way this test can act as "the deploy that just
        // succeeded" without a real AWS-backed launch.
        let lobbies = state.lobbies.clone();
        let id = lobbies
            .open("did:plc:abc", "TrainingScenarios/1-FirstRun.mms")
            .await
            .unwrap();

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let app = app(state);
        tokio::spawn(async move {
            axum::serve(listener, app).await.unwrap();
        });

        let cookie = test_cookie("did:plc:abc");
        let mut socket = connect_to_lobby(addr, &id, &cookie).await;
        next_scenario(&mut socket).await;
        next_seats(&mut socket).await;

        lobbies.broadcast_deployed(&id, "m2".into());

        loop {
            match socket.next().await.unwrap().unwrap() {
                WsMessage::Text(text) => {
                    let value: serde_json::Value = serde_json::from_str(&text).unwrap();
                    if value["type"] == "deployed" {
                        assert_eq!(value["matchId"], "m2");
                        break;
                    }
                }
                _ => continue,
            }
        }
    }
}