//! Tuner over HTTP (spec §15.1, §8.5, §21.1): `POST /mcp` on the Unix

//! socket and the loopback port, every request signed and checked before

//! anything else; the peer's uid read from the socket; `Origin` checked

//! against DNS rebinding.

use crate::protocol::{self, Headers, Reply};

use crate::tools::Ctx;

use axum::body::Bytes;

use axum::extract::connect_info::{ConnectInfo, Connected};

use axum::extract::State;

use axum::http::{HeaderMap, StatusCode};

use axum::response::sse::{Event as SseEvent, KeepAlive, Sse};

use axum::response::{IntoResponse, Response};

use axum::routing::post;

use axum::Router;

use futures::stream::{self, StreamExt};

use serde_json::{json, Value as Json};

use spectra_core::authn::Caller;

use spectra_core::daemon::{Daemon, Peer, WireEvent};

use spectra_core::principals::Principal;

use spectra_core::proof::Proof;

use std::convert::Infallible;

use std::sync::Arc;

use std::time::Duration;

/// Who is on the other end of a connection: a Unix peer, or TCP.

#[derive(Clone, Copy, Debug)]

pub struct PeerInfo(pub Option<Peer>);

impl Connected<axum::serve::IncomingStream<'_, tokio::net::UnixListener>> for PeerInfo {

    fn connect_info(stream: axum::serve::IncomingStream<'_, tokio::net::UnixListener>) -> Self {

        let cred = stream.io().peer_cred().ok();

        PeerInfo(cred.map(|c| Peer { uid: c.uid(), pid: c.pid() }))

    }

}

impl Connected<axum::serve::IncomingStream<'_, tokio::net::TcpListener>> for PeerInfo {

    fn connect_info(_: axum::serve::IncomingStream<'_, tokio::net::TcpListener>) -> Self {

        PeerInfo(None)

    }

}

impl Connected<axum::serve::IncomingStream<'_, crate::tls::TlsListener>> for PeerInfo {

    fn connect_info(_: axum::serve::IncomingStream<'_, crate::tls::TlsListener>) -> Self {

        PeerInfo(None)

    }

}

/// The app state: the daemon, and the origins a browser may send from.

#[derive(Clone)]

pub struct AppState {

    pub daemon: Arc<Daemon>,

    pub origins: Arc<Vec<String>>,

    /// The owner's browser sessions (Radar's sign-in link).

    pub sessions: Arc<crate::session::Sessions>,

}

/// Tuner's routes; the binary adds the API and Radar beside them.

pub fn router(state: AppState) -> Router {

    Router::new()

        .route("/mcp", post(mcp).get(not_allowed).delete(not_allowed))

        .route("/api/v1/ledger/events", post(owner_events))

        .route("/api/v1/instance", axum::routing::get(instance))

        .route("/api/v1/claim", post(claim))

        .with_state(state)

}

/// `GET /api/v1/instance`, unauthenticated (spec §8.6): who this Spectra

/// is, whose it is, and the instance key's word for its TLS certificate, so

/// a client can decide whether to trust it before it signs anything.

async fn instance(State(state): State<AppState>) -> Response {

    let (instance, owner) = {

        let p = state.daemon.principals.read().await;

        (p.instance().cloned(), p.owner().map(|o| o.hash().to_hex()))

    };

    let owner_missing = owner.is_none();

    let Some(instance) = instance else {

        return (StatusCode::SERVICE_UNAVAILABLE, axum::Json(json!({ "title": "this Spectra has no instance key" }))).into_response();

    };

    let tls = match crate::instance::latest_tls(&state.daemon.ledger).await {

        Ok(Some(e)) => json!({ "payload": e.json(), "proof": e.auth.proof.as_ref().map(hex::encode) }),

        _ => Json::Null,

    };

    axum::Json(json!({

        "name": instance.name,

        "instance": spectra_core::principals::identity_payload(&instance),

        "owner": owner,

        // Until claimed: the KeyId `spectra init --owner` named.

        "named_owner": if owner_missing { state.daemon.named_owner() } else { None },

        "tls": tls,

    }))

    .into_response()

}

/// `POST /api/v1/claim`: the owner named by `spectra init --owner` records

/// their identity (spec §8.6). The event proves itself: it is signed by

/// the identity it records, over its payload.

async fn claim(State(state): State<AppState>, body: Bytes) -> Response {

    let events: Vec<WireEvent> = match serde_json::from_slice(&body) {

        Ok(v) => v,

        Err(e) => return (StatusCode::BAD_REQUEST, axum::Json(json!({ "title": format!("not a list of events: {e}") }))).into_response(),

    };

    match state.daemon.claim(events).await {

        Ok(seqs) => axum::Json(json!({ "seqs": seqs })).into_response(),

        Err(e) => (StatusCode::FORBIDDEN, axum::Json(json!({ "title": format!("{e:#}") }))).into_response(),

    }

}

async fn not_allowed() -> StatusCode {

    StatusCode::METHOD_NOT_ALLOWED

}

/// The proof the signature headers carry, for `method_path` and `body`.

pub fn proof_from(headers: &HeaderMap, method_path: &str, body: &[u8]) -> Option<Proof> {

    let get = |k: &str| headers.get(k).and_then(|v| v.to_str().ok());

    let key = get("spectra-key")?;

    let time_ms = get("spectra-time")?.parse().ok()?;

    let nonce: [u8; 16] = hex::decode(get("spectra-nonce")?).ok()?.try_into().ok()?;

    let sig_bytes = hex::decode(get("spectra-signature")?).ok()?;

    let sig = oxy_core::cbor::decode(&sig_bytes).ok().and_then(|v| oxy_core::identity::Signature::from_value(&v).ok())?;

    if sig.key.to_hex() != key {

        return None;

    }

    Some(Proof { sig, method: method_path.into(), body_hash: *spectra_core::event::hash_body_plain(body), time_ms, nonce })

}

/// Refuse a browser request from a page that is not ours (spec §21.1).

pub fn origin_ok(headers: &HeaderMap, allowed: &[String]) -> bool {

    match headers.get("origin").and_then(|v| v.to_str().ok()) {

        None => true,

        Some(o) => allowed.iter().any(|a| a == o),

    }

}

/// Authenticate a request, or the status and reason that refuse it.

pub async fn authenticate_why(

    state: &AppState,

    peer: PeerInfo,

    headers: &HeaderMap,

    method_path: &str,

    body: &[u8],

) -> Result<Caller, (StatusCode, String)> {

    if !origin_ok(headers, &state.origins) {

        return Err((StatusCode::FORBIDDEN, "this origin may not call Spectra".into()));

    }

    let proof = proof_from(headers, method_path, body);

    state.daemon.authenticate(proof.as_ref(), method_path, body, peer.0).await.map_err(|why| {

        tracing::info!(%why, "refused a request");

        // The reason concerns the caller's own key, so the caller is told it:

        // an agent whose delegation expired must be able to say so.

        (StatusCode::UNAUTHORIZED, why.to_string())

    })

}

/// Authenticate a request, or the JSON-RPC response that refuses it.

pub async fn authenticate(

    state: &AppState,

    peer: PeerInfo,

    headers: &HeaderMap,

    method_path: &str,

    body: &[u8],

) -> Result<Caller, Response> {

    authenticate_why(state, peer, headers, method_path, body).await.map_err(|(status, why)| {

        let (code, msg) = if status == StatusCode::FORBIDDEN {

            (protocol::INVALID_REQUEST, why)

        } else {

            (protocol::UNAUTHENTICATED, format!("not authenticated: {why}"))

        };

        (status, axum::Json(protocol::error(&Json::Null, code, &msg))).into_response()

    })

}

async fn mcp(State(state): State<AppState>, ConnectInfo(peer): ConnectInfo<PeerInfo>, headers: HeaderMap, body: Bytes) -> Response {

    let caller = match authenticate(&state, peer, &headers, "POST /mcp", &body).await {

        Ok(c) => c,

        Err(r) => return r,

    };

    let req: Json = match serde_json::from_slice(&body) {

        Ok(v) => v,

        Err(e) => {

            return (StatusCode::BAD_REQUEST, axum::Json(protocol::error(&Json::Null, protocol::PARSE_ERROR, &format!("not JSON: {e}"))))

                .into_response()

        }

    };

    let new_session = req.pointer("/params/arguments/new_session").and_then(Json::as_bool).unwrap_or(false)

        && req.pointer("/params/name").and_then(Json::as_str) == Some("brief");

    let session = state.daemon.touch(&caller.principal, new_session).await;

    let h = Headers {

        version: headers.get("mcp-protocol-version").and_then(|v| v.to_str().ok()),

        method: headers.get("mcp-method").and_then(|v| v.to_str().ok()),

        name: headers.get("mcp-name").and_then(|v| v.to_str().ok()),

    };

    let version = h.version.unwrap_or("2025-03-26").to_string();

    let ctx = Ctx { daemon: state.daemon.clone(), caller, session };

    let meta = meta(&ctx).await;

    match protocol::handle(&ctx, &h, req, meta.clone()).await {

        Reply::Json(status, v) => (StatusCode::from_u16(status).unwrap_or(StatusCode::OK), axum::Json(v)).into_response(),

        Reply::Accepted => StatusCode::ACCEPTED.into_response(),

        Reply::Stream(id, tool, args) => {

            let fut = async move {

                let out = crate::tools::invoke(&ctx, tool, args).await;

                let meta = crate::http::meta(&ctx).await;

                let msg = protocol::stream_result(&id, &version, &out, meta);

                Ok::<_, Infallible>(SseEvent::default().event("message").data(msg.to_string()))

            };

            sse(stream::once(fut).boxed())

        }

        Reply::Listen(id, sub) => {

            let ack = json!({ "jsonrpc": "2.0", "method": "notifications/subscriptions/acknowledged", "params": { "_meta": { "io.modelcontextprotocol/subscriptionId": sub } } });

            let _ = id;

            let first = stream::once(async move { Ok::<_, Infallible>(SseEvent::default().event("message").data(ack.to_string())) });

            // Spectra's tool list never changes while it runs: the stream

            // stays open, quiet but for keep-alives, until the client closes it.

            sse(first.chain(stream::pending()).boxed())

        }

    }

}

fn sse(s: futures::stream::BoxStream<'static, Result<SseEvent, Infallible>>) -> Response {

    let mut r = Sse::new(s).keep_alive(KeepAlive::new().interval(Duration::from_secs(15)).text("")).into_response();

    r.headers_mut().insert("x-accel-buffering", "no".parse().unwrap());

    r

}

/// `_meta` on every result (spec §15.2): the rules version, the session and

/// the unread count.

pub async fn meta(ctx: &Ctx) -> Json {

    let mut m = serde_json::Map::new();

    if let Ok(r) = spectra_core::rules::compile(&ctx.daemon.data) {

        m.insert("spectra/rules_version".into(), json!(r.version));

    }

    if let Some(s) = &ctx.session {

        m.insert("spectra/session".into(), json!(s));

    }

    if let Ok(n) = spectra_core::conversation::unread(&ctx.daemon.ledger, &ctx.caller.principal.actor()).await {

        m.insert("spectra/inbox_unread".into(), json!(n));

    }

    Json::Object(m)

}

/// `POST /api/v1/ledger/events`: the owner's CLI hands over events it

/// signed over their payload, so `agent add` and `project add` work while

/// the daemon holds the ledger (spec §8.5, "One writer").

async fn owner_events(

    State(state): State<AppState>,

    ConnectInfo(peer): ConnectInfo<PeerInfo>,

    headers: HeaderMap,

    body: Bytes,

) -> Response {

    let caller = match authenticate(&state, peer, &headers, "POST /api/v1/ledger/events", &body).await {

        Ok(c) => c,

        Err(r) => return r,

    };

    if !matches!(caller.principal, Principal::Owner { .. }) {

        return (StatusCode::FORBIDDEN, axum::Json(json!({ "title": "only the owner hands events to the daemon" }))).into_response();

    }

    let events: Vec<WireEvent> = match serde_json::from_slice(&body) {

        Ok(v) => v,

        Err(e) => return (StatusCode::BAD_REQUEST, axum::Json(json!({ "title": format!("not a list of events: {e}") }))).into_response(),

    };

    match state.daemon.accept_owner_events(events).await {

        Ok(seqs) => axum::Json(json!({ "appended": seqs })).into_response(),

        Err(e) => (StatusCode::UNPROCESSABLE_ENTITY, axum::Json(json!({ "title": e.to_string() }))).into_response(),

    }

}