api.rs 91.8%

360 of 392 lines run, 63 of 140 functions · run #16 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::{authenticate, 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))
        .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,
            });
        }
    }
    authenticate(st, peer, headers, method_path, body).await.map_err(|_| {
        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,
        }
    };
}

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 now = now_ms();
    rows.into_iter()
        .map(|(name, owner, key, kind, project, status, disabled)| 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,
        })
        .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();
        out.push(api::Project { name, path, remote, depends_on, depended_on_by });
    }
    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> {
    conv::inbox(&st.daemon.ledger, actor, None)
        .await
        .unwrap_or_default()
        .into_iter()
        .map(|i| api::InboxItem { id: i.id, kind: i.kind, at_ms: i.at_ms, from: i.from, channel: i.channel, summary: i.summary })
        .collect()
}

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

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

Keyboard