//! The match proxy: /match/{id}/... to the task's Suramadu on port 8080,
//! HTTP and WebSocket alike. This is the auth boundary arena assumes exists
//! ("proxy is the auth boundary" - arena's suramadu config): only the match
//! owner's session gets through, and the task's address never reaches a
//! browser.
//!
//! Deliberately dumb: no rewriting, no caching, no buffering beyond what
//! hyper does. Bytes in, bytes out.

use axum::body::Body;
use axum::extract::{Path, State};
use axum::http::{HeaderMap, Request, StatusCode, Uri, header};
use axum::response::{IntoResponse, Redirect, Response};
use axum_extra::extract::cookie::CookieJar;
use hyper_util::rt::TokioIo;
use tokio::net::TcpStream;

use crate::routes::AppState;

/// Must match `container_port` in infra's `match-cluster` module.
pub const ARENA_PORT: u16 = 8080;

/// The match's front door: sends the caller to their own client, which
/// lives at their DID.
///
/// arena renders one Suramadu app per human seat at `/<did>` — one root
/// segment, because Suramadu's WebSocket routing resolves only
/// single-segment app paths (a nested `/play/<did>` served its page and
/// 404'd its socket). The identity in the path is public and unique, and it
/// is what lets the forward gate below be a string comparison. A caller
/// with no fighting seat has no client in the container to be sent to, and
/// is told so.
pub async fn entry(
    State(state): State<AppState>,
    Path(id): Path<String>,
    jar: CookieJar,
) -> Response {
    let (_ip, did) = match authorize(&state, &id, &jar).await {
        Ok(authorized) => authorized,
        Err(response) => return response,
    };
    if !state.db.is_human_player(&id, &did).await.unwrap_or(false) {
        return StatusCode::NOT_FOUND.into_response();
    }
    Redirect::to(&format!("/match/{id}/{did}/")).into_response()
}

pub async fn forward(
    State(state): State<AppState>,
    Path((id, rest)): Path<(String, String)>,
    jar: CookieJar,
    request: Request<Body>,
) -> Response {
    let (ip, did) = match authorize(&state, &id, &jar).await {
        Ok(authorized) => authorized,
        Err(response) => return response,
    };
    if let Err(refused) = allow_client_path(&id, &did, &rest) {
        return refused.into_response();
    }

    match proxy_to(&ip, &rest, request).await {
        Ok(response) => response,
        Err(stage) => {
            tracing::warn!(match_id = id, stage, "match proxy: upstream failure");
            StatusCode::BAD_GATEWAY.into_response()
        }
    }
}

/// Which client paths this DID may walk: exactly one, the one named after
/// them.
///
/// A client path is a first segment that is a DID — arena renders each
/// seat's app at `/<did>` — and the check is the whole of it: public,
/// unique, and equal to the session's or not. No lookup: authorize already
/// proved the caller is in this match, and the path says whose client it
/// is. First segments that are not DIDs pass untouched — shared assets and
/// the spectator app, not another player's client — and gating them would
/// be guessing at Suramadu's URL surface.
fn allow_client_path(id: &str, did: &str, rest: &str) -> Result<(), StatusCode> {
    let path_did = rest.split('/').next().unwrap_or_default();
    if !path_did.starts_with("did:") {
        return Ok(());
    }
    if path_did == did {
        return Ok(());
    }
    // Someone else's client. Refused the same way a stranger's match is.
    tracing::info!(
        match_id = id,
        path_did,
        "match proxy: refused another identity's client"
    );
    Err(StatusCode::FORBIDDEN)
}

/// The whole authorization decision: a valid session cookie naming someone
/// with a seat in the match — the launcher or an invited player — and a
/// match that is actually ready. Returns the task's address and who asked.
async fn authorize(
    state: &AppState,
    id: &str,
    jar: &CookieJar,
) -> Result<(String, String), Response> {
    let did = jar
        .get(crate::session::COOKIE_NAME)
        .and_then(|cookie| state.signer.verify(cookie.value()));
    let Some(did) = did else {
        // Signed out: send them to the site to sign in, not a bare 401 -
        // this URL arrives by top-level navigation.
        return Err(Redirect::to(&state.web_origin).into_response());
    };

    let row = match state.db.get_match(id).await {
        Ok(Some(row)) => row,
        Ok(None) => return Err(StatusCode::NOT_FOUND.into_response()),
        Err(_) => {
            tracing::error!(match_id = id, "match proxy: match lookup failed");
            return Err(StatusCode::INTERNAL_SERVER_ERROR.into_response());
        }
    };
    // Owner first: matches from before the seats table have only their owner
    // row.
    if row.owner_did != did
        && !state
            .db
            .is_match_player(&row.id, &did)
            .await
            .unwrap_or_else(|_| {
                // Fail closed - keep the caller out - but say so: otherwise a
                // real seat-holder hitting a transient database error looks
                // exactly like a stranger at this match's door.
                tracing::warn!(
                    match_id = row.id,
                    did,
                    "match proxy: seat lookup failed, refusing"
                );
                false
            })
    {
        // Not their match. The social identity check the URL promises.
        return Err(StatusCode::FORBIDDEN.into_response());
    }
    match (row.status.as_str(), row.task_ip) {
        ("ready", Some(ip)) => Ok((ip, did)),
        ("over" | "failed", _) => Err(StatusCode::GONE.into_response()),
        _ => Err(StatusCode::CONFLICT.into_response()),
    }
}

async fn proxy_to(
    ip: &str,
    rest: &str,
    mut request: Request<Body>,
) -> Result<Response, &'static str> {
    // Take the client's upgrade handle before the request is consumed; only
    // present when the client asked for an upgrade.
    let wants_upgrade = request
        .headers()
        .get(header::UPGRADE)
        .is_some_and(|value| value.to_str().is_ok());
    let downstream_upgrade = wants_upgrade.then(|| hyper::upgrade::on(&mut request));

    let stream = TcpStream::connect((ip, ARENA_PORT))
        .await
        .map_err(|_| "connect")?;
    stream.set_nodelay(true).map_err(|_| "nodelay")?;
    let (mut sender, connection) = hyper::client::conn::http1::handshake(TokioIo::new(stream))
        .await
        .map_err(|_| "handshake")?;
    tokio::spawn(connection.with_upgrades());

    let path_and_query = match request.uri().query() {
        Some(query) => format!("/{rest}?{query}"),
        None => format!("/{rest}"),
    };
    let uri: Uri = path_and_query.parse().map_err(|_| "uri")?;

    let (parts, body) = request.into_parts();
    let mut outbound = Request::builder()
        .method(parts.method)
        .uri(uri)
        .body(body)
        .map_err(|_| "request build")?;
    *outbound.headers_mut() = proxied_headers(&parts.headers, ip);

    let upstream_response = sender.send_request(outbound).await.map_err(|_| "send")?;

    if upstream_response.status() == StatusCode::SWITCHING_PROTOCOLS {
        let Some(downstream_upgrade) = downstream_upgrade else {
            return Err("unexpected 101");
        };
        let (parts, upgrade_body) = upstream_response.into_parts();
        let upstream_upgrade =
            hyper::upgrade::on(Response::from_parts(parts.clone(), upgrade_body));

        tokio::spawn(async move {
            let (downstream, upstream) =
                match tokio::try_join!(downstream_upgrade, upstream_upgrade) {
                    Ok(pair) => pair,
                    Err(e) => {
                        tracing::debug!("match proxy: upgrade did not complete: {e}");
                        return;
                    }
                };
            let mut downstream = TokioIo::new(downstream);
            let mut upstream = TokioIo::new(upstream);
            // The socket lives as long as the match screen is open.
            let _ = tokio::io::copy_bidirectional(&mut downstream, &mut upstream).await;
        });

        let mut response = Response::builder()
            .status(StatusCode::SWITCHING_PROTOCOLS)
            .body(Body::empty())
            .expect("static response");
        *response.headers_mut() = parts.headers;
        Ok(response)
    } else {
        let (parts, body) = upstream_response.into_parts();
        Ok(Response::from_parts(parts, Body::new(body)))
    }
}

/// Everything except hop-by-hop headers, with Host rewritten to the task.
/// Connection/Upgrade survive because Suramadu needs the upgrade to happen
/// at the task, not here.
fn proxied_headers(headers: &HeaderMap, ip: &str) -> HeaderMap {
    let mut out = HeaderMap::new();
    for (name, value) in headers {
        match name.as_str() {
            "host"
            | "keep-alive"
            | "proxy-authenticate"
            | "proxy-authorization"
            | "te"
            | "trailers"
            | "transfer-encoding"
            | "cookie" => {}
            _ => {
                out.append(name, value.clone());
            }
        }
    }
    // Suramadu sets its own session cookies and logs the tab out when they
    // stop coming back (rest/refreshToken answers 401). Forward every cookie
    // except ours: the task never sees headquarters_session.
    if let Some(value) = filtered_cookies(headers)
        && let Ok(value) = value.parse()
    {
        out.insert(header::COOKIE, value);
    }
    if let Ok(host) = format!("{ip}:{ARENA_PORT}").parse() {
        out.insert(header::HOST, host);
    }
    out
}

fn filtered_cookies(headers: &HeaderMap) -> Option<String> {
    let ours = format!("{}=", crate::session::COOKIE_NAME);
    let kept: Vec<&str> = headers
        .get_all(header::COOKIE)
        .iter()
        .filter_map(|value| value.to_str().ok())
        .flat_map(|value| value.split(';'))
        .map(str::trim)
        .filter(|pair| !pair.is_empty() && !pair.starts_with(&ours))
        .collect();
    (!kept.is_empty()).then(|| kept.join("; "))
}

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

    fn jar(value: &str) -> HeaderMap {
        let mut headers = HeaderMap::new();
        headers.insert(header::COOKIE, value.parse().unwrap());
        headers
    }

    #[test]
    fn ours_is_withheld_theirs_is_forwarded() {
        // The two the container actually sets: Jetty's session cookie and the
        // one Suramadu names token_cookie. Both spellings are upstream's and
        // survived the rename; the filter does not depend on either.
        let headers = jar("headquarters_session=secret.tag; JSESSIONID=abc; token_cookie=xyz");
        assert_eq!(
            filtered_cookies(&headers).as_deref(),
            Some("JSESSIONID=abc; token_cookie=xyz")
        );
    }

    #[test]
    fn only_ours_means_no_cookie_header_at_all() {
        let headers = jar("headquarters_session=secret.tag");
        assert_eq!(filtered_cookies(&headers), None);
    }

    #[test]
    fn no_cookies_stays_no_cookies() {
        assert_eq!(filtered_cookies(&HeaderMap::new()), None);
    }
}