//! Radar (spec §17): the human's pages, rendered with askama, refreshed

//! from the event stream, working without JavaScript. Linked by `spectra`

//! and, later, mounted by SpectraHub.

//!

//! The owner signs in with a one-time link their CLI signed (`spectra

//! open`); everything they do here is recorded with that signed request as

//! its proof (spec §8.3). Every unsafe request carries the session's CSRF

//! token and a same-origin `Origin`.

pub mod session;

use askama::Template;

use axum::body::Bytes;

use axum::extract::connect_info::ConnectInfo;

use axum::extract::{Form, Path, Query, State};

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

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

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

use axum::routing::{get, post};

use axum::Router;

use futures::StreamExt;

use serde::Deserialize;

use session::{Session, Sessions};

use spectra_core::conversation::{self as conv, Author, Post, Signer};

use spectra_core::event::{Auth, Method};

use spectra_core::ids::{format_time, now_ms};

use spectra_core::principals::{Principal, Principals};

use spectra_mcp::http::{AppState, PeerInfo};

use std::sync::Arc;

use std::time::Duration;

const MARK: &str = include_str!("../assets/spectra.svg");

const STYLE: &str = include_str!("../assets/style.css");

const SCRIPT: &str = include_str!("../assets/radar.js");

const COOKIE: &str = "spectra_session";

#[derive(Clone)]

pub struct RadarState {

    pub app: AppState,

    pub sessions: Arc<Sessions>,

}

/// Radar's routes.

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

    let state = RadarState { app, sessions: Arc::new(Sessions::default()) };

    Router::new()

        .route("/", get(live))

        .route("/channels", get(channels))

        .route("/c/{name}", get(channel).post(post_channel))

        .route("/c/{name}/t/{id}", get(thread).post(post_thread))

        .route("/log", get(log))

        .route("/log/verify", post(verify))

        .route("/radar/link", post(link))

        .route("/signin", get(signin))

        .route("/signout", post(signout))

        .route("/radar/events", get(events))

        .route("/radar/style.css", get(|| async { asset("text/css; charset=utf-8", STYLE) }))

        .route("/radar/radar.js", get(|| async { asset("text/javascript; charset=utf-8", SCRIPT) }))

        .route("/radar/spectra.svg", get(|| async { asset("image/svg+xml", MARK) }))

        .layer(axum::middleware::map_response(security_headers))

        .with_state(state)

}

fn asset(kind: &'static str, body: &'static str) -> Response {

    ([(header::CONTENT_TYPE, kind), (header::CACHE_CONTROL, "max-age=300")], body).into_response()

}

/// CSP and friends on every response (spec §21.2).

async fn security_headers(mut r: Response) -> Response {

    let h = r.headers_mut();

    h.insert("content-security-policy", HeaderValue::from_static("default-src 'self'; script-src 'self'; style-src 'self'; img-src 'self' data:; frame-ancestors 'none'; form-action 'self'; base-uri 'none'"));

    h.insert("x-content-type-options", HeaderValue::from_static("nosniff"));

    // `same-origin`: a form post still carries our Origin (no-referrer makes it `null`).

    h.insert("referrer-policy", HeaderValue::from_static("same-origin"));

    h.insert("x-frame-options", HeaderValue::from_static("DENY"));

    r

}

/// What every page's frame shows.

pub struct Shell {

    pub page: &'static str,

    pub owner: String,

    pub head: u64,

    pub version: &'static str,

    pub mark: &'static str,

}

/// The signed-in owner, or the page that says how to sign in.

struct Owner {

    session: Session,

    principal: Principal,

    shell: Shell,

}

impl Owner {

    fn author(&self) -> Author<'static> {

        Author {

            principal: self.principal.clone(),

            signer: Signer::Request(Auth {

                key: Some(self.session.key),

                method: Method::Session,

                assurance: 2,

                proof: Some(self.session.proof.clone()),

            }),

        }

    }

}

fn cookie(headers: &HeaderMap) -> Option<String> {

    let c = headers.get(header::COOKIE)?.to_str().ok()?;

    c.split(';').map(str::trim).find_map(|kv| kv.strip_prefix(&format!("{COOKIE}=")).map(String::from))

}

async fn owner(st: &RadarState, headers: &HeaderMap, page: &'static str) -> Result<Owner, Response> {

    let now = now_ms();

    let session = cookie(headers)

        .and_then(|id| st.sessions.get(&id, now))

        .ok_or_else(|| signin_page("Radar shows Spectra to its owner only.", StatusCode::UNAUTHORIZED))?;

    let d = &st.app.daemon;

    let principal = d

        .principals

        .read()

        .await

        .principal(&oxy_core_key(session.key), now)

        .map_err(|why| signin_page(&format!("This session no longer belongs to the owner: {why}."), StatusCode::UNAUTHORIZED))?;

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

        return Err(signin_page("Radar shows Spectra to its owner only.", StatusCode::FORBIDDEN));

    }

    let shell = Shell { page, owner: principal.display(), head: d.ledger.head().await.0, version: env!("CARGO_PKG_VERSION"), mark: MARK };

    Ok(Owner { session, principal, shell })

}

fn oxy_core_key(k: [u8; 32]) -> oxy_core::hash::KeyId {

    oxy_core::hash::ObjectRef(k)

}

#[derive(Template)]

#[template(path = "signin.html")]

struct SigninPage<'a> {

    mark: &'static str,

    message: &'a str,

}

fn signin_page(message: &str, status: StatusCode) -> Response {

    (status, render(&SigninPage { mark: MARK, message })).into_response()

}

fn render<T: Template>(t: &T) -> Response {

    match t.render() {

        Ok(html) => Html(html).into_response(),

        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, format!("could not render the page: {e}")).into_response(),

    }

}

/// Unsafe requests: the session's CSRF token and our own origin.

/// The refusal, if the form may not go through.

fn check_form(st: &RadarState, o: &Owner, headers: &HeaderMap, csrf: &str) -> Option<Response> {

    if !spectra_mcp::http::origin_ok(headers, &st.app.origins) {

        return Some((StatusCode::FORBIDDEN, "this origin may not change Spectra").into_response());

    }

    if csrf != o.session.csrf {

        return Some((StatusCode::FORBIDDEN, "the form is stale: reload the page").into_response());

    }

    None

}

/// Markdown, sanitised: what a message's body becomes on a page.

fn markdown(body: &str) -> String {

    let mut html = String::new();

    pulldown_cmark::html::push_html(

        &mut html,

        pulldown_cmark::Parser::new_ext(body, pulldown_cmark::Options::ENABLE_STRIKETHROUGH | pulldown_cmark::Options::ENABLE_TABLES),

    );

    ammonia::clean(&html)

}

// ---- Live ----

struct AgentCard {

    name: String,

    owner: String,

    kind: String,

    project: String,

    presence: String,

    status: Option<String>,

    last_tool: Option<String>,

    last_at: String,

}

struct EventRow {

    seq: i64,

    at: String,

    kind: String,

    actor: String,

    subject: Option<String>,

}

#[derive(Template)]

#[template(path = "live.html")]

struct LivePage {

    shell: Shell,

    agents: Vec<AgentCard>,

    unread: i64,

    events: Vec<EventRow>,

}

async fn live(State(st): State<RadarState>, headers: HeaderMap) -> Response {

    let o = match owner(&st, &headers, "live").await {

        Ok(o) => o,

        Err(r) => return r,

    };

    let d = &st.app.daemon;

    let pool = d.ledger.pool();

    let rows: Vec<(String, String, String, String, Option<String>)> =

        sqlx::query_as("SELECT a.name, a.owner_name, a.kind, a.project, p.status_text FROM agents a LEFT JOIN presence p ON p.agent = a.name WHERE a.disabled_ms IS NULL ORDER BY a.name")

            .fetch_all(pool)

            .await

            .unwrap_or_default();

    let mut agents = Vec::new();

    {

        let now = now_ms();

        for (name, owner, kind, project, status) in rows {

            let last: Option<(String, i64)> =

                sqlx::query_as("SELECT tool, at_ms FROM tool_calls WHERE agent = ? ORDER BY seq DESC LIMIT 1")

                    .bind(&name)

                    .fetch_optional(pool)

                    .await

                    .ok()

                    .flatten();

            let presence = d.presence.lock().await.state(&name, now).to_string();

            agents.push(AgentCard {

                name,

                owner,

                kind,

                project,

                presence,

                status,

                last_tool: last.as_ref().map(|(t, _)| t.clone()),

                last_at: last.map(|(_, at)| format_time(at as u64)).unwrap_or_default(),

            });

        }

    }

    let unread = conv::unread(&d.ledger, "human").await.unwrap_or(0);

    let events = recent(&st, None, None, 50)

        .await

        .into_iter()

        .map(|e| EventRow { seq: e.seq, at: e.at, kind: e.kind, actor: e.actor, subject: e.subject })

        .collect();

    render(&LivePage { shell: o.shell, agents, unread, events })

}

// ---- Channels ----

struct ChannelRow {

    name: String,

    display: String,

    kind: String,

    purpose: String,

    count: i64,

    last: String,

}

/// A channel with its message count and last message time.

type ChannelCount = (String, String, String, Option<String>, i64, Option<i64>);

#[derive(Template)]

#[template(path = "channels.html")]

struct ChannelsPage {

    shell: Shell,

    channels: Vec<ChannelRow>,

}

async fn channels(State(st): State<RadarState>, headers: HeaderMap) -> Response {

    let o = match owner(&st, &headers, "channels").await {

        Ok(o) => o,

        Err(r) => return r,

    };

    let rows: Vec<ChannelCount> = sqlx::query_as(

        "SELECT c.id, c.name, c.kind, c.purpose, (SELECT COUNT(*) FROM messages m WHERE m.channel = c.id), (SELECT MAX(at_ms) FROM messages m WHERE m.channel = c.id) \

         FROM channels c ORDER BY c.kind = 'direct', c.kind = 'system', c.name",

    )

    .fetch_all(st.app.daemon.ledger.pool())

    .await

    .unwrap_or_default();

    let channels = rows

        .into_iter()

        .map(|(_, name, kind, purpose, count, last)| ChannelRow {

            display: match name.strip_prefix("dm:") {

                Some(p) => format!("@{p}"),

                None => format!("#{name}"),

            },

            name,

            kind,

            purpose: purpose.unwrap_or_default(),

            count,

            last: last.map(|t| format_time(t as u64)).unwrap_or_default(),

        })

        .collect();

    render(&ChannelsPage { shell: o.shell, channels })

}

// ---- A channel and a thread ----

struct MessageRow {

    id: String,

    author: String,

    at: String,

    kind: String,

    replies: i64,

    redacted: bool,

    body_html: String,

}

#[derive(Template)]

#[template(path = "channel.html")]

struct ChannelPage {

    shell: Shell,

    name: String,

    display: String,

    purpose: Option<String>,

    thread: Option<String>,

    messages: Vec<MessageRow>,

    can_post: bool,

    action: String,

    csrf: String,

}

async fn show(st: &RadarState, o: Owner, name: &str, thread: Option<&str>) -> Response {

    let d = &st.app.daemon;

    let (ch, msgs) = match conv::read(&d.ledger, &o.principal, if thread.is_some() { None } else { Some(name) }, thread, None, 200).await {

        Ok(r) => r,

        Err(e) => return (StatusCode::NOT_FOUND, format!("{e:#}")).into_response(),

    };

    let mut messages = Vec::new();

    for m in msgs {

        messages.push(MessageRow {

            author: spectra_core::conversation::display_name(&d.ledger, &m.author).await,

            at: format_time(m.at_ms as u64),

            redacted: m.redacted.is_some(),

            body_html: if m.redacted.is_some() { ammonia::clean_text(&m.body) } else { markdown(&m.body) },

            id: m.id,

            kind: m.kind,

            replies: m.replies,

        });

    }

    let can_post = conv::can_post(&o.principal, &ch).is_ok();

    let action = match thread {

        Some(t) => format!("/c/{}/t/{t}", ch.name),

        None => format!("/c/{}", ch.name),

    };

    let csrf = o.session.csrf.clone();

    render(&ChannelPage {

        shell: o.shell,

        name: ch.name.clone(),

        display: ch.display(),

        purpose: ch.purpose.clone(),

        thread: thread.map(String::from),

        messages,

        can_post,

        action,

        csrf,

    })

}

async fn channel(State(st): State<RadarState>, headers: HeaderMap, Path(name): Path<String>) -> Response {

    match owner(&st, &headers, "channels").await {

        Ok(o) => show(&st, o, &name, None).await,

        Err(r) => r,

    }

}

async fn thread(State(st): State<RadarState>, headers: HeaderMap, Path((name, id)): Path<(String, String)>) -> Response {

    match owner(&st, &headers, "channels").await {

        Ok(o) => show(&st, o, &name, Some(&id)).await,

        Err(r) => r,

    }

}

#[derive(Deserialize)]

struct PostForm {

    _csrf: String,

    body: String,

    reply_to: Option<String>,

}

async fn post_to(st: &RadarState, headers: &HeaderMap, name: &str, form: PostForm, back: String) -> Response {

    let o = match owner(st, headers, "channels").await {

        Ok(o) => o,

        Err(r) => return r,

    };

    if let Some(r) = check_form(st, &o, headers, &form._csrf) {

        return r;

    }

    let post = Post {

        channel: if form.reply_to.is_some() { None } else { Some(name.to_string()) },

        body: form.body,

        reply_to: form.reply_to,

        ..Default::default()

    };

    match conv::post(&st.app.daemon.ledger, &o.author(), post, now_ms()).await {

        Ok(_) => Redirect::to(&back).into_response(),

        Err(e) => (StatusCode::BAD_REQUEST, format!("{e:#}")).into_response(),

    }

}

async fn post_channel(State(st): State<RadarState>, headers: HeaderMap, Path(name): Path<String>, Form(form): Form<PostForm>) -> Response {

    let back = format!("/c/{name}");

    post_to(&st, &headers, &name, form, back).await

}

async fn post_thread(

    State(st): State<RadarState>,

    headers: HeaderMap,

    Path((name, id)): Path<(String, String)>,

    Form(mut form): Form<PostForm>,

) -> Response {

    form.reply_to = Some(id.clone());

    let back = format!("/c/{name}/t/{id}");

    post_to(&st, &headers, &name, form, back).await

}

// ---- The log ----

struct LogRow {

    seq: i64,

    id: String,

    at: String,

    kind: String,

    actor: String,

    subject: Option<String>,

    method: String,

    assurance: i64,

    hash: String,

    prev: String,

    key: String,

    proof: String,

    payload: String,

}

async fn recent(st: &RadarState, kind: Option<&str>, before: Option<i64>, limit: i64) -> Vec<LogRow> {

    #[allow(clippy::type_complexity)]

    let rows: Vec<(

        i64,

        String,

        i64,

        String,

        String,

        Option<String>,

        i64,

        i64,

        Vec<u8>,

        Vec<u8>,

        Option<Vec<u8>>,

        Option<Vec<u8>>,

        Option<Vec<u8>>,

    )> = sqlx::query_as(

        "SELECT seq, id, at_ms, kind, actor, subject, auth_method, auth_assurance, hash, prev, auth_key, auth_proof, payload FROM events \

         WHERE (? IS NULL OR kind = ?) AND (? IS NULL OR seq < ?) ORDER BY seq DESC LIMIT ?",

    )

    .bind(kind)

    .bind(kind)

    .bind(before)

    .bind(before)

    .bind(limit)

    .fetch_all(st.app.daemon.ledger.pool())

    .await

    .unwrap_or_default();

    let mut out = Vec::new();

    for (seq, id, at, kind, actor, subject, method, assurance, hash, prev, key, proof, payload) in rows {

        let payload = payload

            .and_then(|p| serde_json::from_slice::<serde_json::Value>(&p).ok())

            .map(|v| serde_json::to_string_pretty(&v).unwrap_or_default())

            .unwrap_or_else(|| "(pruned to its stub)".into());

        out.push(LogRow {

            seq,

            id,

            at: format_time(at as u64),

            kind,

            actor: spectra_core::conversation::display_name(&st.app.daemon.ledger, &actor).await,

            subject,

            method: Method::from_u8(method as u8).map(|m| m.name()).unwrap_or("?").to_string(),

            assurance,

            hash: hex::encode(hash),

            prev: hex::encode(prev),

            key: key.map(hex::encode).unwrap_or_else(|| "none (Spectra itself; covered by the next checkpoint)".into()),

            proof: proof.map(|p| format!("{} bytes", p.len())).unwrap_or_else(|| "none".into()),

            payload,

        });

    }

    out

}

#[derive(Template)]

#[template(path = "log.html")]

struct LogPage {

    shell: Shell,

    head_hash: String,

    verified: String,

    csrf: String,

    kinds: Vec<String>,

    kind: String,

    events: Vec<LogRow>,

    older: Option<i64>,

}

#[derive(Deserialize)]

struct LogQuery {

    kind: Option<String>,

    before: Option<i64>,

    around: Option<i64>,

}

async fn log(State(st): State<RadarState>, headers: HeaderMap, Query(q): Query<LogQuery>) -> Response {

    let o = match owner(&st, &headers, "log").await {

        Ok(o) => o,

        Err(r) => return r,

    };

    let d = &st.app.daemon;

    let (_, head) = d.ledger.head().await;

    let last: Option<(Vec<u8>, i64)> =

        sqlx::query_as("SELECT payload, at_ms FROM events WHERE kind = 'ledger.verified' ORDER BY seq DESC LIMIT 1")

            .fetch_optional(d.ledger.pool())

            .await

            .ok()

            .flatten();

    let verified = match last {

        Some((p, at)) => {

            let v: serde_json::Value = serde_json::from_slice(&p).unwrap_or_default();

            let n = v["findings"].as_u64().unwrap_or(0);

            format!(