//! 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`.

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 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 spectra_mcp::session::{self, Session};
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,
}

impl RadarState {
    fn sessions(&self) -> &session::Sessions {
        &self.app.sessions
    }
}

/// Radar's routes.
pub fn router(app: AppState) -> Router {
    let state = RadarState { app };
    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("/briefs", get(briefs).post(save_brief))
        .route("/constellation", get(constellation))
        .route("/radar/constellation.json", get(constellation_json))
        .route("/radar/constellation.js", get(|| async { asset("text/javascript; charset=utf-8", SKY_SCRIPT) }))
        .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(&key_id(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 key_id(k: [u8; 32]) -> spectra_core::identity::KeyId {
    spectra_core::identity::KeyId(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 })
}

// ---- Constellation (spec §17.2) ----

const SKY_SCRIPT: &str = include_str!("../assets/constellation.js");

/// The picture's data: the owner, each project with its links and agents.
async fn sky(st: &RadarState, owner: &str) -> serde_json::Value {
    let d = &st.app.daemon;
    let pool = d.ledger.pool();
    let projects: Vec<(String, String)> =
        sqlx::query_as("SELECT name, description FROM projects ORDER BY seq").fetch_all(pool).await.unwrap_or_default();
    let links: Vec<(String, String)> =
        sqlx::query_as("SELECT project, depends_on FROM project_relations").fetch_all(pool).await.unwrap_or_default();
    let agents: Vec<(String, String, String)> =
        sqlx::query_as("SELECT name, kind, project FROM agents WHERE disabled_ms IS NULL ORDER BY name")
            .fetch_all(pool)
            .await
            .unwrap_or_default();
    let presence = d.presence.lock().await;
    let now = now_ms();
    let list: Vec<serde_json::Value> = projects
        .iter()
        .map(|(name, description)| {
            serde_json::json!({
                "name": name,
                "description": description,
                "depends_on": links.iter().filter(|(p, _)| p == name).map(|(_, q)| q).collect::<Vec<_>>(),
                "agents": agents.iter().filter(|(_, _, p)| p == name)
                    .map(|(a, kind, _)| serde_json::json!({ "name": a, "kind": kind, "presence": presence.state(a, now) }))
                    .collect::<Vec<_>>(),
            })
        })
        .collect();
    serde_json::json!({ "owner": owner, "projects": list })
}

fn esc(s: &str) -> String {
    s.replace('&', "&amp;").replace('<', "&lt;").replace('>', "&gt;").replace('"', "&quot;")
}

const HUES: [u32; 8] = [0, 30, 55, 120, 190, 230, 275, 320];

fn presence_colour(p: &str) -> &'static str {
    match p {
        "online" => "#3fb950",
        "busy" => "#a371f7",
        "idle" | "paused" => "#d29922",
        "blocked" => "#f0883e",
        _ => "#6e7681",
    }
}

/// The still picture, for a browser without JavaScript and for the first paint.
fn still(sky: &serde_json::Value) -> String {
    let (cx, cy, ring, orbit) = (450.0f64, 320.0f64, 205.0f64, 54.0f64);
    let projects = sky["projects"].as_array().cloned().unwrap_or_default();
    let n = projects.len().max(1) as f64;
    let mut at = std::collections::HashMap::new();
    let mut nodes = String::new();
    let mut lines = String::new();
    for (i, p) in projects.iter().enumerate() {
        let a = -std::f64::consts::FRAC_PI_2 + std::f64::consts::TAU * i as f64 / n;
        let (x, y) = (cx + ring * a.cos(), cy + ring * a.sin());
        let name = p["name"].as_str().unwrap_or("");
        at.insert(name.to_string(), (x, y));
        let hue = HUES[i % HUES.len()];
        lines.push_str(&format!(r#"<line class="spoke" x1="{cx}" y1="{cy}" x2="{x:.1}" y2="{y:.1}"/>"#));
        let s = 24.0;
        nodes.push_str(&format!(
            r#"<circle class="orbit" cx="{x:.1}" cy="{y:.1}" r="{orbit}"/><polygon class="prism" fill="hsl({hue},80%,60%)" points="{:.1},{:.1} {:.1},{:.1} {:.1},{:.1}"/><text class="plabel" x="{x:.1}" y="{:.1}">{}</text>"#,
            x,
            y - s,
            x + s * 0.87,
            y + s * 0.5,
            x - s * 0.87,
            y + s * 0.5,
            y + orbit + 22.0,
            esc(name)
        ));
        let agents = p["agents"].as_array().cloned().unwrap_or_default();
        for (j, ag) in agents.iter().enumerate() {
            let b = std::f64::consts::TAU * j as f64 / agents.len().max(1) as f64;
            let (ax, ay) = (x + orbit * b.cos(), y + orbit * b.sin());
            nodes.push_str(&format!(
                r#"<circle class="agent" cx="{ax:.1}" cy="{ay:.1}" r="7" fill="{}"/><text class="alabel" x="{ax:.1}" y="{:.1}">{}</text>"#,
                presence_colour(ag["presence"].as_str().unwrap_or("")),
                ay + 18.0,
                esc(ag["name"].as_str().unwrap_or(""))
            ));
        }
    }
    for p in &projects {
        let from = at.get(p["name"].as_str().unwrap_or(""));
        for q in p["depends_on"].as_array().cloned().unwrap_or_default() {
            if let (Some((x1, y1)), Some((x2, y2))) = (from, at.get(q.as_str().unwrap_or(""))) {
                lines.push_str(&format!(r#"<path class="link" d="M{x1:.1},{y1:.1} Q{cx},{cy} {x2:.1},{y2:.1}"/>"#));
            }
        }
    }
    format!(
        r#"{lines}<circle class="core" cx="{cx}" cy="{cy}" r="22"/><text class="olabel" x="{cx}" y="{:.1}">{}</text>{nodes}"#,
        cy + 44.0,
        esc(sky["owner"].as_str().unwrap_or(""))
    )
}

#[derive(Template)]
#[template(path = "constellation.html")]
struct ConstellationPage {
    shell: Shell,
    still: String,
    data: String,
    empty: bool,
}

async fn constellation(State(st): State<RadarState>, headers: HeaderMap) -> Response {
    let o = match owner(&st, &headers, "constellation").await {
        Ok(o) => o,
        Err(r) => return r,
    };
    let sky = sky(&st, &o.principal.display()).await;
    let empty = sky["projects"].as_array().is_none_or(|a| a.is_empty());
    // JSON inside a <script> data block: no `<` may close it early.
    let data = sky.to_string().replace('<', "\\u003c");
    render(&ConstellationPage { shell: o.shell, still: still(&sky), data, empty })
}

async fn constellation_json(State(st): State<RadarState>, headers: HeaderMap) -> Response {
    match owner(&st, &headers, "constellation").await {
        Ok(o) => axum::Json(sky(&st, &o.principal.display()).await).into_response(),
        Err(r) => r,
    }
}

// ---- Briefs (spec §12.1) ----

struct ProjectBriefRow {
    name: String,
    description: String,
    text: String,
}

#[derive(Template)]
#[template(path = "briefs.html")]
struct BriefsPage {
    shell: Shell,
    csrf: String,
    overall: String,
    projects: Vec<ProjectBriefRow>,
}

async fn briefs(State(st): State<RadarState>, headers: HeaderMap) -> Response {
    let o = match owner(&st, &headers, "briefs").await {
        Ok(o) => o,
        Err(r) => return r,
    };
    let pool = st.app.daemon.ledger.pool();
    let overall = spectra_core::rules::layer(pool, None).await.ok().flatten().unwrap_or_default();
    let rows: Vec<(String, String, Option<String>)> = sqlx::query_as(
        "SELECT p.name, p.description, b.text FROM projects p LEFT JOIN briefs b ON b.layer = 'project' AND b.project = p.name ORDER BY p.name",
    )
    .fetch_all(pool)
    .await
    .unwrap_or_default();
    let projects =
        rows.into_iter().map(|(name, description, text)| ProjectBriefRow { name, description, text: text.unwrap_or_default() }).collect();
    let csrf = o.session.csrf.clone();
    render(&BriefsPage { shell: o.shell, csrf, overall, projects })
}

#[derive(Deserialize)]
struct BriefForm {
    _csrf: String,
    project: Option<String>,
    text: String,
}

async fn save_brief(State(st): State<RadarState>, headers: HeaderMap, Form(f): Form<BriefForm>) -> Response {
    let o = match owner(&st, &headers, "briefs").await {
        Ok(o) => o,
        Err(r) => return r,
    };
    if let Some(r) = check_form(&st, &o, &headers, &f._csrf) {
        return r;
    }
    // Browsers send CRLF from a textarea; the brief is stored as written.
    let text = f.text.replace("\r\n", "\n");
    if let Err(e) = spectra_core::rules::check(&text) {
        return (StatusCode::BAD_REQUEST, format!("{e:#}")).into_response();
    }
    let project = f.project.filter(|p| !p.is_empty());
    let pool = st.app.daemon.ledger.pool();
    if let Some(p) = &project {
        let known: Option<String> =
            sqlx::query_scalar("SELECT name FROM projects WHERE name = ?").bind(p).fetch_optional(pool).await.unwrap_or(None);
        if known.is_none() {
            return (StatusCode::NOT_FOUND, "no such project").into_response();
        }
    }
    let Signer::Request(auth) = o.author().signer else { return StatusCode::INTERNAL_SERVER_ERROR.into_response() };
    let mut e =
        spectra_core::event::NewEvent::system("brief.set", serde_json::json!({ "project": project, "text": text })).by("human", auth);
    e.project = project.clone();
    match st.app.daemon.ledger.append_one(e).await {
        Ok(_) => {
            Redirect::to(&format!("/briefs#{}", project.map(|p| format!("p-{p}")).unwrap_or_else(|| "overall".into()))).into_response()
        }
        Err(e) => (StatusCode::BAD_REQUEST, format!("{e:#}")).into_response(),
    }
}

// ---- 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>,
    brief: 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();
    let brief = if ch.kind == "project" { spectra_core::rules::layer(d.ledger.pool(), Some(&ch.name)).await.ok().flatten() } else { None };
    let brief = brief.map(|b| markdown(&b));
    render(&ChannelPage {
        shell: o.shell,
        name: ch.name.clone(),
        display: ch.display(),
        purpose: ch.purpose.clone(),
        brief,
        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!(
                "{} at {} over {} events",
                if n == 0 { "clean".to_string() } else { format!("{n} finding(s)") },
                format_time(at as u64),
                v["events"].as_u64().unwrap_or(0)
            )
        }
        None => "not yet".into(),
    };
    let kinds: Vec<String> =
        sqlx::query_scalar("SELECT kind FROM event_counts ORDER BY n DESC LIMIT 12").fetch_all(d.ledger.pool()).await.unwrap_or_default();
    let kind = q.kind.unwrap_or_default();
    let before = q.before.or(q.around.map(|a| a + 25));
    let events = recent(&st, (!kind.is_empty()).then_some(kind.as_str()), before, 100).await;
    let older = (events.len() == 100).then(|| events.last().map(|e| e.seq)).flatten();
    let csrf = o.session.csrf.clone();
    render(&LogPage { shell: o.shell, head_hash: hex::encode(head), verified, csrf, kinds, kind, events, older })
}

#[derive(Deserialize)]
struct CsrfForm {
    _csrf: String,
}

/// Verify the ledger now and record the result (`ledger.verified`).
async fn verify(State(st): State<RadarState>, headers: HeaderMap, Form(f): Form<CsrfForm>) -> Response {
    let o = match owner(&st, &headers, "log").await {
        Ok(o) => o,
        Err(r) => return r,
    };
    if let Some(r) = check_form(&st, &o, &headers, &f._csrf) {
        return r;
    }
    let ledger = &st.app.daemon.ledger;
    match ledger.verify(&mut Principals::default()).await {
        Ok(v) => {
            let e = spectra_core::event::NewEvent::system(
                "ledger.verified",
                serde_json::json!({ "events": v.events, "checkpoints": v.checkpoints, "findings": v.findings.len(), "head": hex::encode(v.head), "first": v.findings.first().map(|f| format!("#{}: {}", f.seq, f.detail)) }),
            );
            let _ = ledger.append_one(e).await;
            Redirect::to("/log").into_response()
        }
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, format!("{e:#}")).into_response(),
    }
}

// ---- Signing in ----

/// `POST /radar/link`, signed by the owner's CLI: a one-time sign-in link.
async fn link(State(st): State<RadarState>, ConnectInfo(peer): ConnectInfo<PeerInfo>, headers: HeaderMap, body: Bytes) -> Response {
    let caller = match spectra_mcp::http::authenticate(&st.app, peer, &headers, "POST /radar/link", &body).await {
        Ok(c) => c,
        Err(r) => return r,
    };
    if !matches!(caller.principal, Principal::Owner { .. }) {
        return (StatusCode::FORBIDDEN, axum::Json(serde_json::json!({ "title": "only the owner signs in to Radar" }))).into_response();
    }
    let (Some(key), Some(proof)) = (caller.auth.key, caller.auth.proof) else {
        return StatusCode::UNAUTHORIZED.into_response();
    };
    let code = st.sessions().link(key, proof, now_ms());
    let base = st.app.origins.first().cloned().unwrap_or_default();
    axum::Json(serde_json::json!({ "url": format!("{base}/signin?code={code}"), "valid_ms": session::LINK_MS })).into_response()
}

#[derive(Deserialize)]
struct SigninQuery {
    code: Option<String>,
}

async fn signin(State(st): State<RadarState>, Query(q): Query<SigninQuery>) -> Response {
    let Some(id) = q.code.and_then(|c| st.sessions().open(&c, now_ms())) else {
        return signin_page("That sign-in link has been used or has expired.", StatusCode::UNAUTHORIZED);
    };
    let cookie = format!("{COOKIE}={id}; HttpOnly; SameSite=Strict; Path=/; Max-Age={}", session::SESSION_MS / 1000);
    let mut r = Redirect::to("/").into_response();
    r.headers_mut().insert(header::SET_COOKIE, HeaderValue::from_str(&cookie).unwrap());
    r
}

async fn signout(State(st): State<RadarState>, headers: HeaderMap, Form(f): Form<CsrfForm>) -> Response {
    if let Ok(o) = owner(&st, &headers, "live").await {
        if check_form(&st, &o, &headers, &f._csrf).is_none() {
            if let Some(id) = cookie(&headers) {
                st.sessions().close(&id);
            }
        }
    }
    signin_page("Signed out.", StatusCode::OK)
}

// ---- The event stream ----

/// `GET /radar/events`: one SSE message per appended event, carrying its
/// sequence number, kind, actor and subject, never its data (spec §17.3).
async fn events(State(st): State<RadarState>, headers: HeaderMap) -> Response {
    if let Err(r) = owner(&st, &headers, "live").await {
        return r;
    }
    let rx = st.app.daemon.ledger.subscribe();
    let stream = tokio_stream::wrappers::BroadcastStream::new(rx).filter_map(|e| async move {
        let e = e.ok()?;
        let data = serde_json::json!({ "seq": e.seq, "kind": e.kind, "actor": e.actor, "subject": e.subject, "project": e.project });
        Some(Ok::<_, std::convert::Infallible>(SseEvent::default().data(data.to_string())))
    });
    Sse::new(stream).keep_alive(KeepAlive::new().interval(Duration::from_secs(15))).into_response()
}