//! The JSON API (spec §18): what spectra-vscode, scripts and SpectraHub

//! read and write. Every request is a signed request (spec §8.5); a read

//! may instead carry the owner's Radar session. Errors are RFC 7807

//! problems; lists page with `?after=`; reads carry an ETag.

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

use axum::body::Bytes;

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

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

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

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

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

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

use axum::Router;

use futures::StreamExt;

use serde::Deserialize;

use spectra_api as api;

use spectra_core::authn::Caller;

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

use spectra_core::ids::now_ms;

use spectra_core::principals::Principal;

use std::time::Duration;

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

    Router::new()

        .route("/api/v1/health", get(health))

        .route("/api/v1/version", get(version))

        .route("/api/v1/brief", get(brief))

        .route("/api/v1/agents", get(agents))

        .route("/api/v1/projects", get(projects))

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

        .route("/api/v1/channels/{name}/messages", get(messages).post(post_message))

        .route("/api/v1/threads/{id}", get(thread))

        .route("/api/v1/inbox", get(inbox))

        .route("/api/v1/inbox/ack", post(ack))

        .route("/api/v1/search", get(search))

        .route("/api/v1/events", get(events))

        .route("/api/v1/admin/verify", get(admin_verify))

        .route("/api/v1/briefs", get(briefs).put(set_brief))

        .with_state(state)

}

/// An RFC 7807 problem.

pub fn problem(status: StatusCode, title: &str, detail: Option<String>) -> Response {

    let p = api::Problem { kind: "about:blank".into(), title: title.into(), status: status.as_u16(), detail };

    let mut r = (status, axum::Json(p)).into_response();

    r.headers_mut().insert(header::CONTENT_TYPE, "application/problem+json".parse().unwrap());

    r

}

fn bad(e: anyhow::Error) -> Response {

    problem(StatusCode::BAD_REQUEST, "the request cannot be done", Some(format!("{e:#}")))

}

/// JSON with an ETag; 304 when the client has it already.

fn json_etag<T: serde::Serialize>(headers: &HeaderMap, v: &T) -> Response {

    let body = serde_json::to_vec(v).unwrap_or_default();

    let tag = format!("\"{}\"", &blake3_hex(&body)[..32]);

    if headers.get(header::IF_NONE_MATCH).and_then(|h| h.to_str().ok()) == Some(tag.as_str()) {

        return StatusCode::NOT_MODIFIED.into_response();

    }

    ([(header::CONTENT_TYPE, "application/json".to_string()), (header::ETAG, tag)], body).into_response()

}

fn blake3_hex(b: &[u8]) -> String {

    hex::encode(*spectra_core::event::hash_body_plain(b))

}

/// Who is asking: every request is signed (spec §8.5).

async fn caller(

    st: &AppState,

    peer: PeerInfo,

    headers: &HeaderMap,

    method_path: &str,

    body: &[u8],

    read: bool,

) -> Result<Caller, Response> {

    let _ = read;

    let signed = headers.get("spectra-key").is_some();

    crate::http::authenticate_why(st, peer, headers, method_path, body).await.map_err(|(status, why)| {

        if status == StatusCode::FORBIDDEN {

            problem(status, "this origin may not call Spectra", None)

        } else if signed {

            // A signed request is told why its signature was not enough.

            problem(StatusCode::UNAUTHORIZED, "not authenticated", Some(why))

        } else {

            problem(

                StatusCode::UNAUTHORIZED,

                "not authenticated",

                Some("sign the request with your key: spectra's own client, its shims and `spectra bridge` do".into()),

            )

        }

    })

}

fn path_of(uri: &axum::http::Uri) -> String {

    uri.path_and_query().map(|p| p.as_str().to_string()).unwrap_or_else(|| uri.path().to_string())

}

macro_rules! who {

    ($st:expr, $peer:expr, $headers:expr, $uri:expr, $method:literal, $body:expr, $read:expr) => {

        match caller(&$st, $peer, &$headers, &format!("{} {}", $method, path_of(&$uri)), $body, $read).await {

            Ok(c) => c,

            Err(r) => return r,

        }

    };

}

/// `GET /api/v1/admin/verify`: the owner asks the daemon to verify its

/// ledger (spec §8.6): what `spectra admin verify` prints, from afar.

async fn admin_verify(

    State(st): State<AppState>,

    ConnectInfo(peer): ConnectInfo<PeerInfo>,

    headers: HeaderMap,

    uri: axum::http::Uri,

) -> Response {

    let c = who!(st, peer, headers, uri, "GET", b"", false);

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

        return problem(StatusCode::FORBIDDEN, "only the owner verifies the ledger", None);

    }

    let ledger = match spectra_core::ledger::Ledger::open_read_only(&st.daemon.data.join("spectra.db")).await {

        Ok(l) => l,

        Err(e) => return bad(e),

    };

    match ledger.verify(&mut spectra_core::principals::Principals::default()).await {

        Ok(v) => axum::Json(v).into_response(),

        Err(e) => bad(e),

    }

}

/// `GET /api/v1/briefs`: the overall brief and every project's (spec §12.1).

async fn briefs(

    State(st): State<AppState>,

    ConnectInfo(peer): ConnectInfo<PeerInfo>,

    headers: HeaderMap,

    uri: axum::http::Uri,

) -> Response {

    let _ = who!(st, peer, headers, uri, "GET", b"", true);

    let pool = st.daemon.ledger.pool();

    let overall = spectra_core::rules::layer(pool, None).await.ok().flatten();

    let rows: Vec<(String, String)> = sqlx::query_as("SELECT project, text FROM briefs WHERE layer = 'project' ORDER BY project")

        .fetch_all(pool)

        .await

        .unwrap_or_default();

    json_etag(

        &headers,

        &api::Briefs { overall, projects: rows.into_iter().map(|(project, text)| api::ProjectBrief { project, text }).collect() },

    )

}

/// `PUT /api/v1/briefs`: the owner sets a brief; the signed request is the

/// event's proof.

async fn set_brief(

    State(st): State<AppState>,

    ConnectInfo(peer): ConnectInfo<PeerInfo>,

    headers: HeaderMap,

    uri: axum::http::Uri,

    body: Bytes,

) -> Response {

    let c = who!(st, peer, headers, uri, "PUT", &body, false);

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

        return problem(StatusCode::FORBIDDEN, "only the owner sets a brief", None);

    }

    let b: api::SetBrief = match serde_json::from_slice(&body) {

        Ok(b) => b,

        Err(e) => return problem(StatusCode::BAD_REQUEST, "not a brief", Some(e.to_string())),

    };

    if let Err(e) = spectra_core::rules::check(&b.text) {

        return problem(StatusCode::BAD_REQUEST, "too long", Some(e.to_string()));

    }

    let project = b.project.filter(|p| !p.is_empty());

    if let Some(p) = &project {

        let known: Option<String> = sqlx::query_scalar("SELECT name FROM projects WHERE name = ?")

            .bind(p)

            .fetch_optional(st.daemon.ledger.pool())

            .await

            .unwrap_or(None);

        if known.is_none() {

            return problem(StatusCode::NOT_FOUND, "no such project", Some(p.clone()));

        }

    }

    let mut e = spectra_core::event::NewEvent::system("brief.set", serde_json::json!({ "project": project, "text": b.text }))

        .by(c.principal.actor(), c.auth.clone());

    e.project = project.clone();

    match st.daemon.ledger.append_one(e).await {

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

        Err(e) => bad(e),

    }

}

async fn health() -> Response {

    axum::Json(serde_json::json!({ "ok": true })).into_response()

}

async fn version() -> Response {

    axum::Json(api::Version {

        version: env!("CARGO_PKG_VERSION").into(),

        ledger_format: spectra_core::LEDGER_FORMAT,

        mcp_protocol_versions: crate::PROTOCOL_VERSIONS.iter().map(|v| v.to_string()).collect(),

    })

    .into_response()

}

async fn agent_list(st: &AppState) -> Vec<api::Agent> {

    #[allow(clippy::type_complexity)]

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

        "SELECT a.name, a.owner_name, a.key, a.kind, a.project, p.status_text, a.disabled_ms, a.scopes FROM agents a LEFT JOIN presence p ON p.agent = a.name ORDER BY a.name",

    )

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

    .await

    .unwrap_or_default();

    let presence = st.daemon.presence.lock().await;

    let principals = st.daemon.principals.read().await;

    let now = now_ms();

    rows.into_iter()

        .map(|(name, owner, key, kind, project, status, disabled, scopes)| {

            let channels = serde_json::from_str::<Vec<String>>(&scopes)

                .unwrap_or_default()

                .into_iter()

                .filter_map(|s| s.strip_prefix("channel:").map(String::from))

                .collect();

            let k = spectra_core::identity::KeyId::from_hex(&key);

            let state = match k.map(|k| principals.principal(&k, now)) {

                Some(Ok(_)) => "ok".to_string(),

                Some(Err(why)) => why,

                None => "a malformed key".into(),

            };

            let delegation_expires = k.and_then(|k| principals.owner().and_then(|o| o.delegate(&k)).and_then(|d| d.expires));

            api::Agent {

                display: format!("{name} · {owner}'s agent"),

                presence: if disabled.is_some() { "offline".into() } else { presence.state(&name, now).into() },

                disabled: disabled.is_some(),

                name,

                owner,

                key,

                kind,

                project,

                status,

                state,

                delegation_expires,

                channels,

            }

        })

        .collect()

}

async fn agents(

    State(st): State<AppState>,

    ConnectInfo(peer): ConnectInfo<PeerInfo>,

    headers: HeaderMap,

    uri: axum::http::Uri,

) -> Response {

    let _ = who!(st, peer, headers, uri, "GET", b"", true);

    json_etag(&headers, &agent_list(&st).await)

}

async fn projects(

    State(st): State<AppState>,

    ConnectInfo(peer): ConnectInfo<PeerInfo>,

    headers: HeaderMap,

    uri: axum::http::Uri,

) -> Response {

    let _ = who!(st, peer, headers, uri, "GET", b"", true);

    let pool = st.daemon.ledger.pool();

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

        sqlx::query_as("SELECT name, path, description, url FROM projects ORDER BY name").fetch_all(pool).await.unwrap_or_default();

    let mut out = Vec::new();

    for (name, path, description, url) in rows {

        let depends_on = sqlx::query_scalar("SELECT depends_on FROM project_relations WHERE project = ? ORDER BY 1")

            .bind(&name)

            .fetch_all(pool)

            .await

            .unwrap_or_default();

        let depended_on_by = sqlx::query_scalar("SELECT project FROM project_relations WHERE depends_on = ? ORDER BY 1")

            .bind(&name)

            .fetch_all(pool)

            .await

            .unwrap_or_default();

        let agents = sqlx::query_scalar("SELECT name FROM agents WHERE project = ? AND disabled_ms IS NULL ORDER BY name")

            .bind(&name)

            .fetch_all(pool)

            .await

            .unwrap_or_default();

        out.push(api::Project { name, path, description, url, depends_on, depended_on_by, agents });

    }

    json_etag(&headers, &out)

}

async fn channels(

    State(st): State<AppState>,

    ConnectInfo(peer): ConnectInfo<PeerInfo>,

    headers: HeaderMap,

    uri: axum::http::Uri,

) -> Response {

    let c = who!(st, peer, headers, uri, "GET", b"", true);

    let all = match conv::channels(&st.daemon.ledger).await {

        Ok(a) => a,

        Err(e) => return bad(e),

    };

    let mut out = Vec::new();

    for ch in all.into_iter().filter(|ch| conv::can_read(&c.principal, ch)) {

        let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM messages WHERE channel = ?")

            .bind(&ch.id)

            .fetch_one(st.daemon.ledger.pool())

            .await

            .unwrap_or(0);

        out.push(api::Channel { display: ch.display(), id: ch.id, name: ch.name, kind: ch.kind, purpose: ch.purpose, messages: n });

    }

    json_etag(&headers, &out)

}

async fn to_api(st: &AppState, m: conv::Message, ch: &str) -> api::Message {

    api::Message {

        author_display: conv::display_name(&st.daemon.ledger, &m.author).await,

        mentions: serde_json::from_str(&m.mentions).unwrap_or_default(),

        refs: serde_json::from_str(&m.refs).unwrap_or_default(),

        redacted: m.redacted.is_some(),

        id: m.id,

        channel: ch.to_string(),

        reply_to: m.reply_to,

        author: m.author,

        kind: m.kind,

        body: m.body,

        at_ms: m.at_ms,

        replies: m.replies,

    }

}

#[derive(Deserialize)]

struct PageQuery {

    after: Option<String>,

    limit: Option<u32>,

}

async fn messages(

    State(st): State<AppState>,

    ConnectInfo(peer): ConnectInfo<PeerInfo>,

    headers: HeaderMap,

    uri: axum::http::Uri,

    Path(name): Path<String>,

    Query(q): Query<PageQuery>,

) -> Response {

    let c = who!(st, peer, headers, uri, "GET", b"", true);

    let limit = q.limit.unwrap_or(50).clamp(1, 500);

    match conv::read(&st.daemon.ledger, &c.principal, Some(&name), None, q.after.as_deref(), limit).await {

        Ok((ch, msgs)) => {

            let next = (msgs.len() as u32 == limit).then(|| msgs.last().map(|m| m.id.clone())).flatten();

            let mut items = Vec::new();

            for m in msgs {

                items.push(to_api(&st, m, &ch.display()).await);

            }

            json_etag(&headers, &api::Messages { channel: ch.display(), items, next })

        }

        Err(e) => problem(StatusCode::NOT_FOUND, "no such channel, or not yours to read", Some(format!("{e:#}"))),

    }

}

async fn thread(

    State(st): State<AppState>,

    ConnectInfo(peer): ConnectInfo<PeerInfo>,

    headers: HeaderMap,

    uri: axum::http::Uri,

    Path(id): Path<String>,

    Query(q): Query<PageQuery>,

) -> Response {

    let c = who!(st, peer, headers, uri, "GET", b"", true);

    let limit = q.limit.unwrap_or(200).clamp(1, 500);

    match conv::read(&st.daemon.ledger, &c.principal, None, Some(&id), q.after.as_deref(), limit).await {

        Ok((ch, msgs)) => {

            let next = (msgs.len() as u32 == limit).then(|| msgs.last().map(|m| m.id.clone())).flatten();

            let mut items = Vec::new();

            for m in msgs {

                items.push(to_api(&st, m, &ch.display()).await);

            }

            json_etag(&headers, &api::Messages { channel: ch.display(), items, next })

        }

        Err(e) => problem(StatusCode::NOT_FOUND, "no such thread, or not yours to read", Some(format!("{e:#}"))),

    }

}

async fn post_message(

    State(st): State<AppState>,

    ConnectInfo(peer): ConnectInfo<PeerInfo>,

    headers: HeaderMap,

    uri: axum::http::Uri,

    Path(name): Path<String>,

    body: Bytes,

) -> Response {

    let c = who!(st, peer, headers, uri, "POST", &body, false);

    let n: api::NewMessage = match serde_json::from_slice(&body) {

        Ok(n) => n,

        Err(e) => return problem(StatusCode::BAD_REQUEST, "not a message", Some(e.to_string())),

    };

    let author = Author { principal: c.principal.clone(), signer: Signer::Request(c.auth.clone()) };

    let post = Post {

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

        body: n.body,

        reply_to: n.reply_to,

        kind: n.kind,

        ..Default::default()

    };

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

        Ok(Posted::Message { id, channel }) => {

            (StatusCode::CREATED, axum::Json(api::Posted { id, channel, duplicate: false })).into_response()

        }

        Ok(Posted::Duplicate { id }) => axum::Json(api::Posted { id, channel: String::new(), duplicate: true }).into_response(),

        Err(e) => bad(e),

    }

}

async fn inbox_items(st: &AppState, actor: &str) -> Vec<api::InboxItem> {

    let mut out = Vec::new();

    for i in conv::inbox(&st.daemon.ledger, actor, None).await.unwrap_or_default() {

        let from_display = conv::display_name(&st.daemon.ledger, &i.from).await;

        out.push(api::InboxItem {

            id: i.id,

            kind: i.kind,

            at_ms: i.at_ms,

            from: i.from,

            from_display,

            channel: i.channel,

            summary: i.summary,

        });

    }

    out

}

async fn inbox(State(st): State<AppState>, ConnectInfo(peer): ConnectInfo<PeerInfo>, headers: HeaderMap, uri: axum::http::Uri) -> Response {

    let c = who!(st, peer, headers, uri, "GET", b"", true);

    json_etag(&headers, &inbox_items(&st, &c.principal.actor()).await)

}

async fn ack(

    State(st): State<AppState>,

    ConnectInfo(peer): ConnectInfo<PeerInfo>,

    headers: HeaderMap,

    uri: axum::http::Uri,

    body: Bytes,

) -> Response {

    let c = who!(st, peer, headers, uri, "POST", &body, false);

    let a: api::Ack = match serde_json::from_slice(&body) {

        Ok(a) => a,

        Err(e) => return problem(StatusCode::BAD_REQUEST, "not an ack", Some(e.to_string())),

    };

    let author = Author { principal: c.principal, signer: Signer::Request(c.auth) };

    match conv::ack(&st.daemon.ledger, &author, &a.ids, a.reaction.as_deref()).await {

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

        Err(e) => bad(e),

    }

}

async fn brief(State(st): State<AppState>, ConnectInfo(peer): ConnectInfo<PeerInfo>, headers: HeaderMap, uri: axum::http::Uri) -> Response {