//! # HTTP routes (spec §4.1, §4.3)
//!
//! Four routes on axum 0.7:
//! - `POST /v1/sessions` → `{ "session_id": "<uuid>" }`.
//! - `POST /v1/sessions/:id/offer` (`Content-Type: application/sdp` req
//!   + response) → core returns the SDP answer.
//! - `DELETE /v1/sessions/:id` → tear down.
//! - `GET /` → serve the static HTML test client.
//!
//! No authn/authz, no TLS, no multi-tenancy — all deferred per spec §1.2.

use axum::extract::{Path, State};
use axum::http::{StatusCode, header};
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use axum::{Json, Router};
use serde::{Deserialize, Serialize};
use uuid::Uuid;

use crate::session_map::AppState;

#[derive(Serialize)]
struct SessionCreated {
    session_id: String,
}

/// Body of `POST /v1/sessions` (spec §7). The body is OPTIONAL: absent
/// body or absent `tap_url` field → fall back to `RUTSTER_TAP_URL` env
/// default stored on AppState. Present `tap_url` → override the default;
/// validated per spec §4.4 (fail-fast at session-create).
#[derive(Deserialize, Default)]
pub struct CreateSessionBody {
    tap_url: Option<String>,
}

/// POST /v1/sessions — mint a fresh RtcSession (spec §4.1; §7 for the
/// slice-2 `tap_url` body field). Validates the URL per §4.4 and returns
/// `400 Bad Request` on non-loopback `ws://`, `wss://`, or unparseable
/// URLs (fail-fast at session-create so operators see the config error
/// immediately, not mid-call).
pub async fn create_session(
    State(state): State<AppState>,
    body: Option<axum::Json<CreateSessionBody>>,
) -> Response {
    let tap_url_str: Option<&str> = body.as_ref().and_then(|b| b.tap_url.as_deref());
    let tap_url = match resolve_tap_url(tap_url_str, &state.default_tap_url) {
        Ok(u) => u,
        Err(e) => return (StatusCode::BAD_REQUEST, e).into_response(),
    };
    match state.create_session(Some(tap_url)).await {
        Ok(id) => {
            let body = Json(SessionCreated {
                session_id: id.0.to_string(),
            });
            (StatusCode::OK, body).into_response()
        }
        Err(e) => {
            // Contract strings from the media thread (see media_thread.rs):
            // capacity/drain rejections are retryable-elsewhere → 503 so an
            // LB retries the next node; everything else stays 500.
            if e.contains("node full") || e.contains("draining") {
                return (StatusCode::SERVICE_UNAVAILABLE, e).into_response();
            }
            tracing::error!(error = ?e, "session create failed");
            StatusCode::INTERNAL_SERVER_ERROR.into_response()
        }
    }
}

/// Validate a tap URL per spec §4.4 (fail-fast at session-create).
///
/// - `ws://` must have host `127.0.0.1` or `localhost` (loopback-only —
///   the slice-2 dev loop posture; lands with authz in step 6).
/// - `wss://` is rejected outright with `400 Bad Request` — the TLS
///   story needs the cert story from ARCHITECTURE.md (Vault/KMS), which
///   lands with the real trust boundary in step 6. Fails fast so
///   operators see the config error immediately, not mid-call.
/// - Other schemes are rejected.
/// - `None` → return a clone of the default (env-provided) URL.
///
/// # Why a free function (not an `AppState` method?)
///
/// Validation is a pure function on URL — no shared state involved — and
/// the test module below exercises the four cases without constructing
/// an `AppState`. Free functions are the Rust idiom for "pure logic,
/// testable in isolation."
fn resolve_tap_url(override_url: Option<&str>, default: &url::Url) -> Result<url::Url, String> {
    let raw = override_url.unwrap_or_else(|| default.as_str());
    let parsed = url::Url::parse(raw).map_err(|e| format!("invalid tap_url: {e}"))?;
    match parsed.scheme() {
        "ws" => {
            let host = parsed.host_str().unwrap_or("");
            if host != "127.0.0.1" && host != "localhost" {
                return Err(format!(
                    "ws:// tap_url must be loopback (127.0.0.1 or localhost); got host={host:?}"
                ));
            }
            Ok(parsed)
        }
        "wss" => Err("wss:// lands in step 6; use ws:// for the slice-2 dev loop".to_string()),
        other => Err(format!("unsupported tap_url scheme: {other:?} (use ws://)")),
    }
}

/// POST /v1/sessions/:id/offer — accept browser SDP offer, return answer
/// (spec §4.1). Non-trickle: the offer body carries all browser ICE
/// candidates; the answer carries the core's candidates (filled natively
/// by str0m 0.21's sdp_api().accept_offer).
pub async fn post_offer(
    State(state): State<AppState>,
    Path(id_str): Path<String>,
    body: String,
) -> Response {
    let Ok(id_uuid) = Uuid::parse_str(&id_str) else {
        return (StatusCode::NOT_FOUND, "bad session id").into_response();
    };
    let id = rutster_call_model::ChannelId(id_uuid);
    match state.accept_offer(id, body).await {
        Ok(answer_sdp) => (
            StatusCode::OK,
            [(header::CONTENT_TYPE, "application/sdp")],
            answer_sdp,
        )
            .into_response(),
        Err(e) => {
            tracing::error!(error = %e, "SDP accept failed");
            if e.contains("not found") {
                (StatusCode::NOT_FOUND, e).into_response()
            } else {
                StatusCode::BAD_REQUEST.into_response()
            }
        }
    }
}

/// DELETE /v1/sessions/:id — tear down (spec §4.1, §4.5).
pub async fn delete_session(State(state): State<AppState>, Path(id_str): Path<String>) -> Response {
    let Ok(id_uuid) = Uuid::parse_str(&id_str) else {
        return StatusCode::NOT_FOUND.into_response();
    };
    let id = rutster_call_model::ChannelId(id_uuid);
    state.close(id).await;
    StatusCode::NO_CONTENT.into_response()
}

/// GET / — serve the static browser test client (spec §4.4).
pub async fn index() -> Response {
    (
        StatusCode::OK,
        [(header::CONTENT_TYPE, "text/html; charset=utf-8")],
        include_str!("../static/index.html"),
    )
        .into_response()
}

/// GET /healthz — liveness only: the process and HTTP stack are up.
/// Deliberately does NOT consult the media thread — liveness and
/// readiness are different probes with different restart semantics.
pub async fn healthz() -> Response {
    (StatusCode::OK, "ok").into_response()
}

/// GET /readyz — "can this node accept a NEW call right now?"
/// LB target membership + autoscaler both read this. 503 when draining,
/// at capacity, or when the media thread doesn't answer Stats in 250ms
/// (a wedged/dead media thread previously left GET / answering 200 —
/// the zombie-node failure from the 2026-07-04 review).
pub async fn readyz(State(state): State<AppState>) -> Response {
    let (reply, rx) = tokio::sync::oneshot::channel();
    // One timeout bounds the WHOLE round trip — send AND reply. A wedged
    // media thread can stall the send (full command channel), not just
    // the reply, and the probe's 250ms promise must hold in that case too.
    let stats = match tokio::time::timeout(std::time::Duration::from_millis(250), async {
        state
            .cmd_tx
            .send(crate::media_thread::MediaCmd::Stats { reply })
            .await
            .ok()?;
        rx.await.ok()
    })
    .await
    {
        Ok(Some(s)) => s,
        // Elapsed timeout, send failure (thread gone), or dropped reply —
        // all mean the media thread cannot vouch for itself right now.
        _ => {
            return (StatusCode::SERVICE_UNAVAILABLE, "media thread unresponsive").into_response();
        }
    };
    let ready = !stats.draining && stats.sessions < stats.max_sessions;
    let code = if ready {
        StatusCode::OK
    } else {
        StatusCode::SERVICE_UNAVAILABLE
    };
    (code, Json(stats)).into_response()
}

/// Build the axum router.
pub fn router(state: AppState) -> Router {
    Router::new()
        .route("/", get(index))
        .route("/healthz", get(healthz))
        .route("/readyz", get(readyz))
        // `POST /v1/sessions` creates; `DELETE /v1/sessions/:id` destroys
        // (note the `:id` — deleting the collection root has no meaning and
        // would extract a missing `:id` path parameter, so the two routes
        // live at different paths, not chained via `.delete(...)` on the
        // collection route as axum's method chaining would suggest).
        .route("/v1/sessions", post(create_session))
        .route("/v1/sessions/:id", axum::routing::delete(delete_session))
        .route("/v1/sessions/:id/offer", post(post_offer))
        // slice-5 T5 -- trunk-origin outbound + Twilio inbound-call webhook.
        // Stub handlers (respond 503 / minimal TwiML) until full T5 integration:
        // the operator can POST /v1/trunk/sessions against dev-b's
        // TwilioCallControlClient (live impl behind twilio-live feature flag).
        // The webhook responds with a <Connect><Stream> TwiML instructing
        // Twilio to open a Media Streams fork against the binary's
        // /twilio/media-stream WSS endpoint.
        .route("/v1/trunk/sessions", post(originate_trunk_call))
        .route("/v1/trunk/webhook", post(twilio_inbound_webhook))
        .with_state(state)
}

/// POST /v1/trunk/sessions -- originate an outbound call via the
/// provider call-control client (spec §3.6). The body shape: `{ "to": "+...", "from": "+..." }`.
/// The handler routes via `CallControlClient::originate(to, from, None)`
/// (None because spend gate is step 6). The CallControlClient is
/// dev-b's `MockCallControlClient` (CI default) OR the feature-gated
/// `TwilioCallControlClient` (live -- maintainer-triggered only).
///
/// Stub T5: responds 503 until the binary's `AppState` carries a
/// `CallControlClient` trait object (Step B refactor, dev-b handover).
pub async fn originate_trunk_call() -> Response {
    tracing::info!(
        "POST /v1/trunk/sessions -- originates routed via CallControlClient (slice-5 T5 Step B interface)"
    );
    (
        StatusCode::SERVICE_UNAVAILABLE,
        "slice-5 T5 Step B: CallControlClient trait not yet threaded into AppState",
    )
        .into_response()
}

/// Derive the TwiML `<Stream>` URL from the operator's public webhook
/// base (`RUTSTER_TWILIO_WEBHOOK_BASE`): `https://…` → `wss://…`,
/// `http://…` → `ws://…` (dev loop), path fixed to the WS route's mount
/// point.
///
/// Authority-only by design: any path on the base is dropped, because
/// `TwilioMediaStreamsServer::router` mounts the WS route absolutely at
/// `/twilio/media-stream` — a path-prefixing edge must strip its prefix
/// before the FOB, exactly like it must for every other route.
fn stream_url_from_webhook_base(base: &url::Url) -> Result<url::Url, String> {
    let ws_scheme = match base.scheme() {
        "https" => "wss",
        "http" => "ws",
        other => {
            return Err(format!(
                "webhook base must be http(s); got scheme {other:?}"
            ));
        }
    };
    let mut stream_url = base
        .join("/twilio/media-stream")
        .map_err(|e| format!("webhook base rejects a path join: {e}"))?;
    stream_url
        .set_scheme(ws_scheme)
        .map_err(|()| format!("cannot rewrite scheme to {ws_scheme}"))?;
    Ok(stream_url)
}

/// POST /v1/trunk/webhook — Twilio's inbound-call signaling webhook
/// receiver (deploy slice A §5.3). Responds with TwiML instructing Twilio
/// to open a Media Streams fork against our `/twilio/media-stream` WS
/// endpoint, at the PUBLIC address derived from
/// `RUTSTER_TWILIO_WEBHOOK_BASE` — replacing the slice-5 placeholder's
/// hardcoded `ws://127.0.0.1:8080`, which no CPaaS could ever dial.
///
/// 503 when the trunk is unconfigured: a webhook arriving without
/// `RUTSTER_TWILIO_*` set is an operator misconfiguration (Twilio was
/// pointed here, the binary wasn't told its public name) — failing the
/// call loudly beats returning a loopback Stream URL that dies silently
/// 10 s later. No XML escaping needed: `config::twilio_credentials`
/// rejects non-http(s) schemes + any XML-special char (`"`, `'`, `<`,
/// `>`, `&`) at boot, so the URL's `as_str()` is interpolatable into
/// the TwiML body without an escape pass.
pub async fn twilio_inbound_webhook(State(state): State<AppState>) -> Response {
    let Some(base) = state.trunk_webhook_base.as_ref() else {
        return (
            StatusCode::SERVICE_UNAVAILABLE,
            "trunk not configured: set the RUTSTER_TWILIO_* env vars \
             (webhook received, but there is no RUTSTER_TWILIO_WEBHOOK_BASE \
             to derive the public Stream URL from)",
        )
            .into_response();
    };
    let stream_url = match stream_url_from_webhook_base(base) {
        Ok(u) => u,
        Err(e) => {
            // config::twilio_credentials rejects non-http(s) schemes at
            // boot, so this arm is unreachable in a correctly-booted
            // binary (the scheme check in stream_url_from_webhook_base
            // would only fail on a non-http(s) scheme, which boot
            // validation already rejected). 500, never a panic.
            tracing::error!(error = %e, "webhook base failed Stream-URL derivation");
            return StatusCode::INTERNAL_SERVER_ERROR.into_response();
        }
    };
    let twiml = format!(
        r#"<?xml version="1.0" encoding="UTF-8"?>
<Response>
  <Connect>
    <Stream url="{stream_url}" />
  </Connect>
</Response>"#
    );
    (
        StatusCode::OK,
        [(header::CONTENT_TYPE, "application/xml")],
        twiml,
    )
        .into_response()
}

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

    #[test]
    fn ws_loopback_accepted() {
        let default = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
        let r = resolve_tap_url(Some("ws://localhost:9001/x"), &default).unwrap();
        assert_eq!(r.host_str(), Some("localhost"));
    }

    #[test]
    fn ws_non_loopback_rejected() {
        let default = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
        let err = resolve_tap_url(Some("ws://example.com:8081/echo"), &default).unwrap_err();
        assert!(err.contains("loopback"));
    }

    #[test]
    fn wss_rejected_fail_fast() {
        let default = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
        let err = resolve_tap_url(Some("wss://127.0.0.1:8081/echo"), &default).unwrap_err();
        assert!(err.contains("step 6"));
    }

    #[test]
    fn default_used_when_override_absent() {
        let default = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
        let r = resolve_tap_url(None, &default).unwrap();
        assert_eq!(r.as_str(), "ws://127.0.0.1:8081/echo");
    }

    #[tokio::test]
    async fn readyz_503_when_media_thread_gone_but_healthz_200() {
        // Default AppState: closed placeholder channel = dead media thread.
        let app = router(AppState::default());
        let ready = app
            .clone()
            .oneshot(
                axum::http::Request::builder()
                    .uri("/readyz")
                    .body(axum::body::Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(ready.status(), StatusCode::SERVICE_UNAVAILABLE);
        let health = app
            .oneshot(
                axum::http::Request::builder()
                    .uri("/healthz")
                    .body(axum::body::Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(health.status(), StatusCode::OK);
    }

    #[test]
    fn stream_url_derivation_maps_https_to_wss_and_keeps_authority() {
        let base = url::Url::parse("https://pbx.example.com:8443").unwrap();
        let u = stream_url_from_webhook_base(&base).unwrap();
        assert_eq!(u.as_str(), "wss://pbx.example.com:8443/twilio/media-stream");
    }

    #[test]
    fn stream_url_derivation_maps_http_to_ws_for_dev() {
        let base = url::Url::parse("http://localhost:8080").unwrap();
        let u = stream_url_from_webhook_base(&base).unwrap();
        assert_eq!(u.as_str(), "ws://localhost:8080/twilio/media-stream");
    }

    #[test]
    fn stream_url_derivation_rejects_non_http_schemes() {
        let base = url::Url::parse("ftp://pbx.example.com").unwrap();
        let err = stream_url_from_webhook_base(&base).unwrap_err();
        assert!(err.contains("http"), "msg: {err}");
    }

    #[tokio::test]
    async fn webhook_derives_stream_url_from_configured_base() {
        let state = AppState::default()
            .with_trunk_webhook_base(Some(url::Url::parse("https://pbx.example.com").unwrap()));
        let app = router(state);
        let resp = app
            .oneshot(
                axum::http::Request::builder()
                    .method("POST")
                    .uri("/v1/trunk/webhook")
                    .body(axum::body::Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let body = axum::body::to_bytes(resp.into_body(), 4096).await.unwrap();
        let twiml = String::from_utf8(body.to_vec()).unwrap();
        assert!(
            twiml.contains(r#"<Stream url="wss://pbx.example.com/twilio/media-stream" />"#),
            "twiml: {twiml}"
        );
        assert!(
            !twiml.contains("127.0.0.1"),
            "hardcoded loopback must be gone: {twiml}"
        );
    }

    #[tokio::test]
    async fn webhook_503_when_trunk_unconfigured() {
        // Default AppState: trunk_webhook_base = None (no RUTSTER_TWILIO_*).
        let app = router(AppState::default());
        let resp = app
            .oneshot(
                axum::http::Request::builder()
                    .method("POST")
                    .uri("/v1/trunk/webhook")
                    .body(axum::body::Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
    }
}
