api.rs 91.7%

410 of 447 lines run, 66 of 146 functions · run #21 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))
        .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 = oxy_core::hash::ObjectRef(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),
    }
}

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 = oxy_core::hash::ObjectRef::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, String, Option<String>)> =
        sqlx::query_as("SELECT name, path, remote FROM projects ORDER BY name").fetch_all(pool).await.unwrap_or_default();
    let mut out = Vec::new();
    for (name, path, remote) 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, remote, 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.data).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));
        return Sse::new(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

1		//! The JSON API (spec §18): what spectra-vscode, scripts and SpectraHub
2		//! read and write. Every request is a signed request (spec §8.5); a read
3		//! may instead carry the owner's Radar session. Errors are RFC 7807
4		//! problems; lists page with `?after=`; reads carry an ETag.
5
6		use crate::http::{AppState, PeerInfo};
7		use axum::body::Bytes;
8		use axum::extract::connect_info::ConnectInfo;
9		use axum::extract::{Path, Query, State};
10		use axum::http::{header, HeaderMap, StatusCode};
11		use axum::response::sse::{Event as SseEvent, KeepAlive, Sse};
12		use axum::response::{IntoResponse, Response};
13		use axum::routing::{get, post};
14		use axum::Router;
15		use futures::StreamExt;
16		use serde::Deserialize;
17		use spectra_api as api;
18		use spectra_core::authn::Caller;
19		use spectra_core::conversation::{self as conv, Author, Post, Posted, Signer};
20		use spectra_core::event::{Auth, Method};
21		use spectra_core::ids::now_ms;
22		use spectra_core::principals::Principal;
23		use std::time::Duration;
24
25	18×	pub fn router(state: AppState) -> Router {
26	18×	Router::new()
27	18×	.route("/api/v1/health", get(health))
28	18×	.route("/api/v1/version", get(version))
29	18×	.route("/api/v1/brief", get(brief))
30	18×	.route("/api/v1/agents", get(agents))
31	18×	.route("/api/v1/projects", get(projects))
32	18×	.route("/api/v1/channels", get(channels))
33	18×	.route("/api/v1/channels/{name}/messages", get(messages).post(post_message))
34	18×	.route("/api/v1/threads/{id}", get(thread))
35	18×	.route("/api/v1/inbox", get(inbox))
36	18×	.route("/api/v1/inbox/ack", post(ack))
37	18×	.route("/api/v1/search", get(search))
38	18×	.route("/api/v1/events", get(events))
39	18×	.route("/api/v1/admin/verify", get(admin_verify))
40	18×	.with_state(state)
41	18×	}
42
43		/// An RFC 7807 problem.
44	4×	pub fn problem(status: StatusCode, title: &str, detail: Option<String>) -> Response {
45	4×	let p = api::Problem { kind: "about:blank".into(), title: title.into(), status: status.as_u16(), detail };
46	4×	let mut r = (status, axum::Json(p)).into_response();
47	4×	r.headers_mut().insert(header::CONTENT_TYPE, "application/problem+json".parse().unwrap());
48	4×	r
49	4×	}
50
51	1×	fn bad(e: anyhow::Error) -> Response {
52	1×	problem(StatusCode::BAD_REQUEST, "the request cannot be done", Some(format!("{e:#}")))
53	1×	}
54
55		/// JSON with an ETag; 304 when the client has it already.
56	34×	fn json_etag<T: serde::Serialize>(headers: &HeaderMap, v: &T) -> Response {
57	34×	let body = serde_json::to_vec(v).unwrap_or_default();
58	34×	let tag = format!("\"{}\"", &blake3_hex(&body)[..32]);
59	34×	if headers.get(header::IF_NONE_MATCH).and_then(\|h\| h.to_str().ok()) == Some(tag.as_str()) {
60	1×	return StatusCode::NOT_MODIFIED.into_response();
61	33×	}
62	33×	([(header::CONTENT_TYPE, "application/json".to_string()), (header::ETAG, tag)], body).into_response()
63	34×	}
64
65	34×	fn blake3_hex(b: &[u8]) -> String {
66	34×	hex::encode(*spectra_core::event::hash_body_plain(b))
67	34×	}
68
69		/// Who is asking: a signed request, or for a read the owner's session.
70	51×	async fn caller(
71	51×	st: &AppState,
72	51×	peer: PeerInfo,
73	51×	headers: &HeaderMap,
74	51×	method_path: &str,
75	51×	body: &[u8],
76	51×	read: bool,
77	51×	) -> Result<Caller, Response> {
78	51×	if read && headers.get("spectra-key").is_none() {
79	1×	let cookie = headers.get(header::COOKIE).and_then(\|c\| c.to_str().ok()).unwrap_or("");
80	1×	let id = cookie.split(';').map(str::trim).find_map(\|kv\| kv.strip_prefix("spectra_session=")).unwrap_or("");
81	1×	if let Some(s) = st.sessions.get(id, now_ms()) {
82	0×	if !crate::http::origin_ok(headers, &st.origins) {
83	0×	return Err(problem(StatusCode::FORBIDDEN, "this origin may not read Spectra", None));
84	0×	}
85	0×	let key = oxy_core::hash::ObjectRef(s.key);
86	0×	let principal = st
87	0×	.daemon
88	0×	.principals
89	0×	.read()
90	0×	.await
91	0×	.principal(&key, now_ms())
92	0×	.map_err(\|why\| problem(StatusCode::UNAUTHORIZED, "not authenticated", Some(why)))?;
93	0×	return Ok(Caller {
94	0×	principal,
95	0×	auth: Auth { key: Some(s.key), method: Method::Session, assurance: 2, proof: Some(s.proof) },
96	0×	learn_uid: None,
97	0×	});
98	1×	}
99	50×	}
100	51×	let signed = headers.get("spectra-key").is_some();
101	51×	crate::http::authenticate_why(st, peer, headers, method_path, body).await.map_err(\|(status, why)\| {
102	2×	if status == StatusCode::FORBIDDEN {
103	0×	problem(status, "this origin may not call Spectra", None)
104	2×	} else if signed {
105		// A signed request is told why its signature was not enough.
106	1×	problem(StatusCode::UNAUTHORIZED, "not authenticated", Some(why))
107		} else {
108	1×	problem(
109		StatusCode::UNAUTHORIZED,
110	1×	"not authenticated",
111	1×	Some("sign the request with your key (spectra's client does), or open Radar with `spectra open`".into()),
112		)
113		}
114	2×	})
115	51×	}
116
117	51×	fn path_of(uri: &axum::http::Uri) -> String {
118	51×	uri.path_and_query().map(\|p\| p.as_str().to_string()).unwrap_or_else(\|\| uri.path().to_string())
119	51×	}
120
121		macro_rules! who {
122		($st:expr, $peer:expr, $headers:expr, $uri:expr, $method:literal, $body:expr, $read:expr) => {
123		match caller(&$st, $peer, &$headers, &format!("{} {}", $method, path_of(&$uri)), $body, $read).await {
124		Ok(c) => c,
125		Err(r) => return r,
126		}
127		};
128		}
129
130		/// `GET /api/v1/admin/verify`: the owner asks the daemon to verify its
131		/// ledger (spec §8.6): what `spectra admin verify` prints, from afar.
132	1×	async fn admin_verify(
133	1×	State(st): State<AppState>,
134	1×	ConnectInfo(peer): ConnectInfo<PeerInfo>,
135	1×	headers: HeaderMap,
136	1×	uri: axum::http::Uri,
137	1×	) -> Response {
138	1×	let c = who!(st, peer, headers, uri, "GET", b"", false);
139	1×	if !matches!(c.principal, Principal::Owner { .. }) {
140	0×	return problem(StatusCode::FORBIDDEN, "only the owner verifies the ledger", None);
141	1×	}
142	1×	let ledger = match spectra_core::ledger::Ledger::open_read_only(&st.daemon.data.join("spectra.db")).await {
143	1×	Ok(l) => l,
144	0×	Err(e) => return bad(e),
145		};
146	1×	match ledger.verify(&mut spectra_core::principals::Principals::default()).await {
147	1×	Ok(v) => axum::Json(v).into_response(),
148	0×	Err(e) => bad(e),
149		}
150	1×	}
151
152	1×	async fn health() -> Response {
153	1×	axum::Json(serde_json::json!({ "ok": true })).into_response()
154	1×	}
155
156	1×	async fn version() -> Response {
157		axum::Json(api::Version {
158	1×	version: env!("CARGO_PKG_VERSION").into(),
159		ledger_format: spectra_core::LEDGER_FORMAT,
160	4×	mcp_protocol_versions: crate::PROTOCOL_VERSIONS.iter().map(\|v\| v.to_string()).collect(),
161		})
162	1×	.into_response()
163	1×	}
164
165	11×	async fn agent_list(st: &AppState) -> Vec<api::Agent> {
166		#[allow(clippy::type_complexity)]
167	11×	let rows: Vec<(String, String, String, String, String, Option<String>, Option<i64>)> = sqlx::query_as(
168	11×	"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",
169	11×	)
170	11×	.fetch_all(st.daemon.ledger.pool())
171	11×	.await
172	11×	.unwrap_or_default();
173	11×	let presence = st.daemon.presence.lock().await;
174	11×	let principals = st.daemon.principals.read().await;
175	11×	let now = now_ms();
176	11×	rows.into_iter()
177	11×	.map(\|(name, owner, key, kind, project, status, disabled)\| {
178	11×	let k = oxy_core::hash::ObjectRef::from_hex(&key);
179	11×	let state = match k.map(\|k\| principals.principal(&k, now)) {
180	10×	Some(Ok(_)) => "ok".to_string(),
181	1×	Some(Err(why)) => why,
182	0×	None => "a malformed key".into(),
183		};
184	11×	let delegation_expires = k.and_then(\|k\| principals.owner().and_then(\|o\| o.delegate(&k)).and_then(\|d\| d.expires));
185		api::Agent {
186	11×	display: format!("{name} · {owner}'s agent"),
187	11×	presence: if disabled.is_some() { "offline".into() } else { presence.state(&name, now).into() },
188	11×	disabled: disabled.is_some(),
189	11×	name,
190	11×	owner,
191	11×	key,
192	11×	kind,
193	11×	project,
194	11×	status,
195	11×	state,
196	11×	delegation_expires,
197		}
198	11×	})
199	11×	.collect()
200	11×	}
201
202	10×	async fn agents(
203	10×	State(st): State<AppState>,
204	10×	ConnectInfo(peer): ConnectInfo<PeerInfo>,
205	10×	headers: HeaderMap,
206	10×	uri: axum::http::Uri,
207	10×	) -> Response {
208	10×	let _ = who!(st, peer, headers, uri, "GET", b"", true);
209	8×	json_etag(&headers, &agent_list(&st).await)
210	10×	}
211
212	11×	async fn projects(
213	11×	State(st): State<AppState>,
214	11×	ConnectInfo(peer): ConnectInfo<PeerInfo>,
215	11×	headers: HeaderMap,
216	11×	uri: axum::http::Uri,
217	11×	) -> Response {
218	11×	let _ = who!(st, peer, headers, uri, "GET", b"", true);
219	11×	let pool = st.daemon.ledger.pool();
220	11×	let rows: Vec<(String, String, Option<String>)> =
221	11×	sqlx::query_as("SELECT name, path, remote FROM projects ORDER BY name").fetch_all(pool).await.unwrap_or_default();
222	11×	let mut out = Vec::new();
223	11×	for (name, path, remote) in rows {
224	11×	let depends_on = sqlx::query_scalar("SELECT depends_on FROM project_relations WHERE project = ? ORDER BY 1")
225	11×	.bind(&name)
226	11×	.fetch_all(pool)
227	11×	.await
228	11×	.unwrap_or_default();
229	11×	let depended_on_by = sqlx::query_scalar("SELECT project FROM project_relations WHERE depends_on = ? ORDER BY 1")
230	11×	.bind(&name)
231	11×	.fetch_all(pool)
232	11×	.await
233	11×	.unwrap_or_default();
234	11×	let agents = sqlx::query_scalar("SELECT name FROM agents WHERE project = ? AND disabled_ms IS NULL ORDER BY name")
235	11×	.bind(&name)
236	11×	.fetch_all(pool)
237	11×	.await
238	11×	.unwrap_or_default();
239	11×	out.push(api::Project { name, path, remote, depends_on, depended_on_by, agents });
240		}
241	11×	json_etag(&headers, &out)
242	11×	}
243
244	1×	async fn channels(
245	1×	State(st): State<AppState>,
246	1×	ConnectInfo(peer): ConnectInfo<PeerInfo>,
247	1×	headers: HeaderMap,
248	1×	uri: axum::http::Uri,
249	1×	) -> Response {
250	1×	let c = who!(st, peer, headers, uri, "GET", b"", true);
251	1×	let all = match conv::channels(&st.daemon.ledger).await {
252	1×	Ok(a) => a,
253	0×	Err(e) => return bad(e),
254		};
255	1×	let mut out = Vec::new();
256	4×	for ch in all.into_iter().filter(\|ch\| conv::can_read(&c.principal, ch)) {
257	4×	let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM messages WHERE channel = ?")
258	4×	.bind(&ch.id)
259	4×	.fetch_one(st.daemon.ledger.pool())
260	4×	.await
261	4×	.unwrap_or(0);
262	4×	out.push(api::Channel { display: ch.display(), id: ch.id, name: ch.name, kind: ch.kind, purpose: ch.purpose, messages: n });
263		}
264	1×	json_etag(&headers, &out)
265	1×	}
266
267	4×	async fn to_api(st: &AppState, m: conv::Message, ch: &str) -> api::Message {
268		api::Message {
269	4×	author_display: conv::display_name(&st.daemon.ledger, &m.author).await,
270	4×	mentions: serde_json::from_str(&m.mentions).unwrap_or_default(),
271	4×	refs: serde_json::from_str(&m.refs).unwrap_or_default(),
272	4×	redacted: m.redacted.is_some(),
273	4×	id: m.id,
274	4×	channel: ch.to_string(),
275	4×	reply_to: m.reply_to,
276	4×	author: m.author,
277	4×	kind: m.kind,
278	4×	body: m.body,
279	4×	at_ms: m.at_ms,
280	4×	replies: m.replies,
281		}
282	4×	}
283
284		#[derive(Deserialize)]
285		struct PageQuery {
286		after: Option<String>,
287		limit: Option<u32>,
288		}
289
290	1×	async fn messages(
291	1×	State(st): State<AppState>,
292	1×	ConnectInfo(peer): ConnectInfo<PeerInfo>,
293	1×	headers: HeaderMap,
294	1×	uri: axum::http::Uri,
295	1×	Path(name): Path<String>,
296	1×	Query(q): Query<PageQuery>,
297	1×	) -> Response {
298	1×	let c = who!(st, peer, headers, uri, "GET", b"", true);
299	1×	let limit = q.limit.unwrap_or(50).clamp(1, 500);
300	1×	match conv::read(&st.daemon.ledger, &c.principal, Some(&name), None, q.after.as_deref(), limit).await {
301	1×	Ok((ch, msgs)) => {
302	1×	let next = (msgs.len() as u32 == limit).then(\|\| msgs.last().map(\|m\| m.id.clone())).flatten();
303	1×	let mut items = Vec::new();
304	2×	for m in msgs {
305	2×	items.push(to_api(&st, m, &ch.display()).await);
306		}
307	1×	json_etag(&headers, &api::Messages { channel: ch.display(), items, next })
308		}
309	0×	Err(e) => problem(StatusCode::NOT_FOUND, "no such channel, or not yours to read", Some(format!("{e:#}"))),
310		}
311	1×	}
312
313	1×	async fn thread(
314	1×	State(st): State<AppState>,
315	1×	ConnectInfo(peer): ConnectInfo<PeerInfo>,
316	1×	headers: HeaderMap,
317	1×	uri: axum::http::Uri,
318	1×	Path(id): Path<String>,
319	1×	Query(q): Query<PageQuery>,
320	1×	) -> Response {
321	1×	let c = who!(st, peer, headers, uri, "GET", b"", true);
322	1×	let limit = q.limit.unwrap_or(200).clamp(1, 500);
323	1×	match conv::read(&st.daemon.ledger, &c.principal, None, Some(&id), q.after.as_deref(), limit).await {
324	1×	Ok((ch, msgs)) => {
325	1×	let next = (msgs.len() as u32 == limit).then(\|\| msgs.last().map(\|m\| m.id.clone())).flatten();
326	1×	let mut items = Vec::new();
327	2×	for m in msgs {
328	2×	items.push(to_api(&st, m, &ch.display()).await);
329		}
330	1×	json_etag(&headers, &api::Messages { channel: ch.display(), items, next })
331		}
332	0×	Err(e) => problem(StatusCode::NOT_FOUND, "no such thread, or not yours to read", Some(format!("{e:#}"))),
333		}
334	1×	}
335
336	10×	async fn post_message(
337	10×	State(st): State<AppState>,
338	10×	ConnectInfo(peer): ConnectInfo<PeerInfo>,
339	10×	headers: HeaderMap,
340	10×	uri: axum::http::Uri,
341	10×	Path(name): Path<String>,
342	10×	body: Bytes,
343	10×	) -> Response {
344	10×	let c = who!(st, peer, headers, uri, "POST", &body, false);
345	10×	let n: api::NewMessage = match serde_json::from_slice(&body) {
346	10×	Ok(n) => n,
347	0×	Err(e) => return problem(StatusCode::BAD_REQUEST, "not a message", Some(e.to_string())),
348		};
349	10×	let author = Author { principal: c.principal.clone(), signer: Signer::Request(c.auth.clone()) };
350	10×	let post = Post {
351	10×	channel: if n.reply_to.is_some() { None } else { Some(name) },
352	10×	body: n.body,
353	10×	reply_to: n.reply_to,
354	10×	kind: n.kind,
355	10×	..Default::default()
356		};
357	10×	match conv::post(&st.daemon.ledger, &author, post, now_ms()).await {
358	9×	Ok(Posted::Message { id, channel }) => {
359	9×	(StatusCode::CREATED, axum::Json(api::Posted { id, channel, duplicate: false })).into_response()
360		}
361	0×	Ok(Posted::Duplicate { id }) => axum::Json(api::Posted { id, channel: String::new(), duplicate: true }).into_response(),
362	1×	Err(e) => bad(e),
363		}
364	10×	}
365
366	8×	async fn inbox_items(st: &AppState, actor: &str) -> Vec<api::InboxItem> {
367	8×	let mut out = Vec::new();
368	8×	for i in conv::inbox(&st.daemon.ledger, actor, None).await.unwrap_or_default() {
369	6×	let from_display = conv::display_name(&st.daemon.ledger, &i.from).await;
370	6×	out.push(api::InboxItem {
371	6×	id: i.id,
372	6×	kind: i.kind,
373	6×	at_ms: i.at_ms,
374	6×	from: i.from,
375	6×	from_display,
376	6×	channel: i.channel,
377	6×	summary: i.summary,
378	6×	});
379		}
380	8×	out
381	8×	}
382
383	5×	async fn inbox(State(st): State<AppState>, ConnectInfo(peer): ConnectInfo<PeerInfo>, headers: HeaderMap, uri: axum::http::Uri) -> Response {
384	5×	let c = who!(st, peer, headers, uri, "GET", b"", true);
385	5×	json_etag(&headers, &inbox_items(&st, &c.principal.actor()).await)
386	5×	}
387
388	1×	async fn ack(
389	1×	State(st): State<AppState>,
390	1×	ConnectInfo(peer): ConnectInfo<PeerInfo>,
391	1×	headers: HeaderMap,
392	1×	uri: axum::http::Uri,
393	1×	body: Bytes,
394	1×	) -> Response {
395	1×	let c = who!(st, peer, headers, uri, "POST", &body, false);
396	1×	let a: api::Ack = match serde_json::from_slice(&body) {
397	1×	Ok(a) => a,
398	0×	Err(e) => return problem(StatusCode::BAD_REQUEST, "not an ack", Some(e.to_string())),
399		};
400	1×	let author = Author { principal: c.principal, signer: Signer::Request(c.auth) };
401	1×	match conv::ack(&st.daemon.ledger, &author, &a.ids, a.reaction.as_deref()).await {
402	1×	Ok(n) => axum::Json(serde_json::json!({ "read": n })).into_response(),
403	0×	Err(e) => bad(e),
404		}
405	1×	}
406
407	3×	async fn brief(State(st): State<AppState>, ConnectInfo(peer): ConnectInfo<PeerInfo>, headers: HeaderMap, uri: axum::http::Uri) -> Response {
408	3×	let c = who!(st, peer, headers, uri, "GET", b"", true);
409	3×	let project = match &c.principal {
410	1×	Principal::Agent { project, .. } => Some(project.clone()),
411	2×	_ => None,
412		};
413	3×	let rules = spectra_core::rules::compile(&st.daemon.data).map(\|r\| r.version).unwrap_or_default();
414	3×	let inbox = inbox_items(&st, &c.principal.actor()).await;
415	3×	let b = api::Brief {
416	3×	you: c.principal.display(),
417	3×	actor: c.principal.actor(),
418	3×	project,
419	3×	rules_version: rules,
420	3×	unread: inbox.len() as i64,
421	3×	inbox,
422	3×	agents: agent_list(&st).await,
423	3×	head: st.daemon.ledger.head().await.0,
424		};
425	3×	json_etag(&headers, &b)
426	3×	}
427
428		#[derive(Deserialize)]
429		struct SearchQuery {
430		q: String,
431		channel: Option<String>,
432		author: Option<String>,
433		since: Option<String>,
434		limit: Option<u32>,
435		}
436
437	2×	async fn search(
438	2×	State(st): State<AppState>,
439	2×	ConnectInfo(peer): ConnectInfo<PeerInfo>,
440	2×	headers: HeaderMap,
441	2×	uri: axum::http::Uri,
442	2×	Query(q): Query<SearchQuery>,
443	2×	) -> Response {
444	2×	let c = who!(st, peer, headers, uri, "GET", b"", true);
445	2×	let since = q.since.as_deref().and_then(\|s\| crate::tools::parse_since(s, now_ms()));
446	2×	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
447		{
448	2×	Ok(hits) => {
449	2×	let mut out: Vec<api::Hit> = Vec::new();
450	4×	for h in hits {
451	4×	let author_display = conv::display_name(&st.daemon.ledger, &h.author).await;
452	4×	out.push(api::Hit {
453	4×	id: h.id,
454	4×	channel: h.channel,
455	4×	author: h.author,
456	4×	at_ms: h.at_ms,
457	4×	snippet: h.snippet,
458	4×	thread: h.thread,
459	4×	author_display,
460	4×	});
461		}
462	2×	json_etag(&headers, &out)
463		}
464	0×	Err(e) => bad(e),
465		}
466	2×	}
467
468		#[derive(Deserialize)]
469		struct EventsQuery {
470		after: Option<u64>,
471		kind: Option<String>,
472		limit: Option<u32>,
473		follow: Option<u8>,
474		/// With `follow`, end the stream after this many events.
475		max: Option<usize>,
476		}
477
478	13×	fn summary(e: &spectra_core::event::Event) -> api::EventSummary {
479	13×	api::EventSummary {
480	13×	seq: e.seq,
481	13×	id: e.id.clone(),
482	13×	at_ms: e.at_ms,
483	13×	kind: e.kind.clone(),
484	13×	actor: e.actor.clone(),
485	13×	subject: e.subject.clone(),
486	13×	method: e.auth.method.name().into(),
487	13×	assurance: e.auth.assurance,
488	13×	hash: hex::encode(e.hash),
489	13×	}
490	13×	}
491
492		/// The log as summaries; `follow=1` streams new ones as SSE instead.
493	5×	async fn events(
494	5×	State(st): State<AppState>,
495	5×	ConnectInfo(peer): ConnectInfo<PeerInfo>,
496	5×	headers: HeaderMap,
497	5×	uri: axum::http::Uri,
498	5×	Query(q): Query<EventsQuery>,
499	5×	) -> Response {
500	5×	let c = who!(st, peer, headers, uri, "GET", b"", true);
501		// The log is the owner's to read; an agent reads what concerns it
502		// through the conversation and its inbox.
503	5×	if !matches!(c.principal, Principal::Owner { .. }) {
504	1×	return problem(StatusCode::FORBIDDEN, "the event log is the owner's", None);
505	4×	}
506	4×	if q.follow == Some(1) {
507	2×	let kind = q.kind.clone();
508	2×	let rx = st.daemon.ledger.subscribe();
509	2×	let stream = tokio_stream::wrappers::BroadcastStream::new(rx)
510	4×	.filter_map(move \|e\| {
511	4×	let kind = kind.clone();
512	4×	async move {
513	4×	let e = e.ok()?;
514	4×	if kind.as_ref().is_some_and(\|k\| *k != e.kind) {
515	2×	return None;
516	2×	}
517	2×	Some(Ok::<_, std::convert::Infallible>(
518	2×	SseEvent::default().data(serde_json::to_string(&summary(&e)).unwrap_or_default()),
519	2×	))
520	4×	}
521	4×	})
522	2×	.take(q.max.unwrap_or(usize::MAX));
523	2×	return Sse::new(stream).keep_alive(KeepAlive::new().interval(Duration::from_secs(15))).into_response();
524	2×	}
525	2×	let limit = q.limit.unwrap_or(100).clamp(1, 1000);
526	2×	let from = q.after.map(\|a\| a + 1).unwrap_or(1);
527	2×	let evs = match &q.kind {
528	1×	Some(k) => st.daemon.ledger.events_of_kinds(from, &[k.as_str()], limit).await,
529	1×	None => st.daemon.ledger.events(from, limit).await,
530		};
531	2×	match evs {
532	2×	Ok(evs) => {
533	2×	let next = (evs.len() as u32 == limit).then(\|\| evs.last().map(\|e\| e.seq)).flatten();
534	2×	json_etag(&headers, &api::Events { items: evs.iter().map(summary).collect(), next })
535		}
536	0×	Err(e) => bad(e),
537		}
538	5×	}

Coverage / crates/spectra-mcp/src/api.rs 91.7%

Keyboard