api.rs 92.1%

466 of 506 lines run, 74 of 162 functions · run #29 on main · run missed
//! 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::event::{Auth, Method};
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: a signed request, or for a read the owner's session.
async fn caller(
    st: &AppState,
    peer: PeerInfo,
    headers: &HeaderMap,
    method_path: &str,
    body: &[u8],
    read: bool,
) -> Result<Caller, Response> {
    if read && headers.get("spectra-key").is_none() {
        let cookie = headers.get(header::COOKIE).and_then(|c| c.to_str().ok()).unwrap_or("");
        let id = cookie.split(';').map(str::trim).find_map(|kv| kv.strip_prefix("spectra_session=")).unwrap_or("");
        if let Some(s) = st.sessions.get(id, now_ms()) {
            if !crate::http::origin_ok(headers, &st.origins) {
                return Err(problem(StatusCode::FORBIDDEN, "this origin may not read Spectra", None));
            }
            let key = spectra_core::identity::KeyId(s.key);
            let principal = st
                .daemon
                .principals
                .read()
                .await
                .principal(&key, now_ms())
                .map_err(|why| problem(StatusCode::UNAUTHORIZED, "not authenticated", Some(why)))?;
            return Ok(Caller {
                principal,
                auth: Auth { key: Some(s.key), method: Method::Session, assurance: 2, proof: Some(s.proof) },
                learn_uid: None,
            });
        }
    }
    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 client does), or open Radar with `spectra open`".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>)> = sqlx::query_as(
        "SELECT a.name, a.owner_name, a.key, a.kind, a.project, p.status_text, a.disabled_ms 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)| {
            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,
            }
        })
        .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 {
    let c = who!(st, peer, headers, uri, "GET", b"", true);
    let project = match &c.principal {
        Principal::Agent { project, .. } => Some(project.clone()),
        _ => None,
    };
    let rules = spectra_core::rules::compile(st.daemon.ledger.pool(), project.as_deref()).await.map(|r| r.version).unwrap_or_default();
    let inbox = inbox_items(&st, &c.principal.actor()).await;
    let b = api::Brief {
        you: c.principal.display(),
        actor: c.principal.actor(),
        project,
        rules_version: rules,
        unread: inbox.len() as i64,
        inbox,
        agents: agent_list(&st).await,
        head: st.daemon.ledger.head().await.0,
    };
    json_etag(&headers, &b)
}

#[derive(Deserialize)]
struct SearchQuery {
    q: String,
    channel: Option<String>,
    author: Option<String>,
    since: Option<String>,
    limit: Option<u32>,
}

async fn search(
    State(st): State<AppState>,
    ConnectInfo(peer): ConnectInfo<PeerInfo>,
    headers: HeaderMap,
    uri: axum::http::Uri,
    Query(q): Query<SearchQuery>,
) -> Response {
    let c = who!(st, peer, headers, uri, "GET", b"", true);
    let since = q.since.as_deref().and_then(|s| crate::tools::parse_since(s, now_ms()));
    match conv::search(&st.daemon.ledger, &c.principal, &q.q, q.channel.as_deref(), q.author.as_deref(), since, q.limit.unwrap_or(20)).await
    {
        Ok(hits) => {
            let mut out: Vec<api::Hit> = Vec::new();
            for h in hits {
                let author_display = conv::display_name(&st.daemon.ledger, &h.author).await;
                out.push(api::Hit {
                    id: h.id,
                    channel: h.channel,
                    author: h.author,
                    at_ms: h.at_ms,
                    snippet: h.snippet,
                    thread: h.thread,
                    author_display,
                });
            }
            json_etag(&headers, &out)
        }
        Err(e) => bad(e),
    }
}

#[derive(Deserialize)]
struct EventsQuery {
    after: Option<u64>,
    kind: Option<String>,
    limit: Option<u32>,
    follow: Option<u8>,
    /// With `follow`, end the stream after this many events.
    max: Option<usize>,
}

fn summary(e: &spectra_core::event::Event) -> api::EventSummary {
    api::EventSummary {
        seq: e.seq,
        id: e.id.clone(),
        at_ms: e.at_ms,
        kind: e.kind.clone(),
        actor: e.actor.clone(),
        subject: e.subject.clone(),
        method: e.auth.method.name().into(),
        assurance: e.auth.assurance,
        hash: hex::encode(e.hash),
    }
}

/// The log as summaries; `follow=1` streams new ones as SSE instead.
async fn events(
    State(st): State<AppState>,
    ConnectInfo(peer): ConnectInfo<PeerInfo>,
    headers: HeaderMap,
    uri: axum::http::Uri,
    Query(q): Query<EventsQuery>,
) -> Response {
    let c = who!(st, peer, headers, uri, "GET", b"", true);
    // The log is the owner's to read; an agent reads what concerns it
    // through the conversation and its inbox.
    if !matches!(c.principal, Principal::Owner { .. }) {
        return problem(StatusCode::FORBIDDEN, "the event log is the owner's", None);
    }
    if q.follow == Some(1) {
        let kind = q.kind.clone();
        let rx = st.daemon.ledger.subscribe();
        let stream = tokio_stream::wrappers::BroadcastStream::new(rx)
            .filter_map(move |e| {
                let kind = kind.clone();
                async move {
                    let e = e.ok()?;
                    if kind.as_ref().is_some_and(|k| *k != e.kind) {
                        return None;
                    }
                    Some(Ok::<_, std::convert::Infallible>(
                        SseEvent::default().data(serde_json::to_string(&summary(&e)).unwrap_or_default()),
                    ))
                }
            })
            .take(q.max.unwrap_or(usize::MAX));
        // A comment first: the follower knows it is subscribed, so it can
        // read the state and miss nothing that happens after.
        let hello = futures::stream::once(async { Ok::<_, std::convert::Infallible>(SseEvent::default().comment("subscribed")) });
        return Sse::new(hello.chain(stream)).keep_alive(KeepAlive::new().interval(Duration::from_secs(15))).into_response();
    }
    let limit = q.limit.unwrap_or(100).clamp(1, 1000);
    let from = q.after.map(|a| a + 1).unwrap_or(1);
    let evs = match &q.kind {
        Some(k) => st.daemon.ledger.events_of_kinds(from, &[k.as_str()], limit).await,
        None => st.daemon.ledger.events(from, limit).await,
    };
    match evs {
        Ok(evs) => {
            let next = (evs.len() as u32 == limit).then(|| evs.last().map(|e| e.seq)).flatten();
            json_etag(&headers, &api::Events { items: evs.iter().map(summary).collect(), next })
        }
        Err(e) => bad(e),
    }
}
`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet
Coverage / crates/spectra-mcp/src/api.rs 92.1%

Keyboard
