api.rs 91.7%

364 of 397 lines run, 63 of 140 functions · run #19 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))
        .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,
        }
    };
}

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

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

Keyboard