//! The human at a terminal (spec §19): `spectra say`, `spectra inbox //! [--follow]`, `spectra search`, and `spectra api` for anything else. Each //! signs with the owner's own key: through the daemon when it is serving, //! else straight into the ledger. use super::identity::owner_home; use super::{ui, Ctx}; use anyhow::{bail, Context, Result}; use serde_json::{json, Value}; use spectra_core::conversation::{self as conv, Author, Post, Posted, Signer}; use spectra_core::keys::KeyPair; use spectra_core::ledger::Ledger; use spectra_core::principals::Principals; use spectra_mcp::client::Client; use std::path::PathBuf; async fn owner_client(ctx: &Ctx, home: &std::path::Path) -> Result> { ctx.owner_client(home).await } async fn api(c: &Client, method: &str, path: &str, body: Option) -> Result { super::call(c, method, path, body).await } use super::encode; pub async fn say( ctx: &Ctx, channel: Option, reply_to: Option, body: String, owner_home_flag: Option, as_json: bool, ) -> Result { let home = owner_home(ctx, owner_home_flag); let (id, where_) = match owner_client(ctx, &home).await? { Some(c) => { let ch = channel.clone().unwrap_or_else(|| "general".into()); let v = api( &c, "POST", &format!("/api/v1/channels/{}/messages", encode(ch.trim_start_matches('#'))), Some(json!({ "body": body, "reply_to": reply_to })), ) .await?; if v["duplicate"] == true { bail!("you said exactly this in the last minute ({})", v["id"].as_str().unwrap_or("")); } (v["id"].as_str().unwrap_or("").to_string(), v["channel"].as_str().unwrap_or("").to_string()) } None => { let ledger = ctx.writer().await?; let owner = KeyPair::load_home(&home)?; let principal = Principals::load(&ledger) .await? .principal(&owner.key, spectra_core::ids::now_ms()) .map_err(anyhow::Error::msg) .context("run `spectra identity refresh` first")?; let author = Author { principal, signer: Signer::Cli(&owner) }; let post = Post { channel: if reply_to.is_some() { None } else { Some(channel.unwrap_or_else(|| "general".into())) }, body, reply_to, ..Default::default() }; let posted = conv::post(&ledger, &author, post, spectra_core::ids::now_ms()).await?; ledger.checkpoint().await?; match posted { Posted::Message { id, channel } => (id, channel), Posted::Duplicate { id } => bail!("you said exactly this in the last minute ({id})"), } } }; if as_json { println!("{}", json!({ "id": id, "channel": where_ })); } else { println!("{} {}{}{}", ui::done("Posted"), ui::subject(&where_), ui::sep(), ui::id(&id)); } Ok(0) } fn print_item(i: &Value) { let kind = i["kind"].as_str().unwrap_or(""); let mark = if kind == "dm" { ui::warn(ui::glyph("✉", "@")) } else { ui::ok(ui::glyph("●", "*")) }; let from = i["from_display"].as_str().or(i["from"].as_str()).unwrap_or(""); let at = i["at_ms"].as_u64().map(ui::ago).unwrap_or_default(); println!( "{mark} {} {}{}{}{}{}", ui::dim(kind), ui::author(from), ui::sep(), i["channel"].as_str().unwrap_or(""), ui::sep(), ui::dim(at) ); println!(" {}", i["summary"].as_str().unwrap_or("")); println!(" {}", ui::dim(format!("reply: spectra say --reply-to {} \"…\"", i["id"].as_str().unwrap_or("")))); } /// The owner's unread items: from the daemon when one serves, else the ledger. async fn inbox_items(ctx: &Ctx, client: Option<&Client>) -> Result> { if let Some(c) = client { return Ok(api(c, "GET", "/api/v1/inbox", None).await?.as_array().cloned().unwrap_or_default()); } let ledger = Ledger::open_read_only(&ctx.db()?).await?; let mut items = Vec::new(); for i in conv::inbox(&ledger, "human", None).await? { let from_display = conv::display_name(&ledger, &i.from).await; let mut v = serde_json::to_value(&i)?; v["from_display"] = from_display.into(); items.push(v); } Ok(items) } pub async fn inbox(ctx: &Ctx, follow: bool, owner_home_flag: Option, as_json: bool) -> Result { let home = owner_home(ctx, owner_home_flag); let client = owner_client(ctx, &home).await?; let items = inbox_items(ctx, client.as_ref()).await?; if as_json && !follow { println!("{}", Value::Array(items)); return Ok(0); } if items.is_empty() && !follow { println!("{} {}", ui::note(), ui::dim("nothing unread")); } let mut seen: std::collections::HashSet = std::collections::HashSet::new(); for i in &items { seen.insert(i["id"].as_str().unwrap_or("").to_string()); if as_json { println!("{i}"); } else { print_item(i); } } if !follow { return Ok(0); } let client = client.context("--follow needs `spectra serve` running")?; if !as_json { println!("{} {}", ui::note(), ui::dim("following; Ctrl-C stops")); } // Every new message is a cue to look at the inbox again. let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<()>(); let stream_client = owner_client(ctx, &home).await?.context("--follow needs `spectra serve` running")?; let watcher = tokio::spawn(async move { stream_client .stream("/api/v1/events?follow=1&kind=message.posted", |l| { if l.starts_with("data:") { let _ = tx.send(()); } true }) .await }); while rx.recv().await.is_some() { for v in inbox_items(ctx, Some(&client)).await? { if seen.insert(v["id"].as_str().unwrap_or("").to_string()) { if as_json { println!("{v}"); } else { print_item(&v); } } } } watcher.await??; Ok(0) } pub async fn search( ctx: &Ctx, query: &str, channel: Option, author: Option, since: Option, limit: u32, as_json: bool, ) -> Result { let hits: Vec = match owner_client(ctx, &owner_home(ctx, None)).await? { Some(c) => { let mut q = format!("/api/v1/search?q={}&limit={limit}", encode(query)); for (k, v) in [("channel", &channel), ("author", &author), ("since", &since)] { if let Some(v) = v { q.push_str(&format!("&{k}={}", encode(v))); } } api(&c, "GET", &q, None).await?.as_array().cloned().unwrap_or_default() } None => { let ledger = Ledger::open_read_only(&ctx.db()?).await?; let owner = Principals::load(&ledger).await?; let me = owner.owner().context("no owner is recorded")?; let principal = spectra_core::principals::Principal::Owner { key: me.hash(), name: me.name.clone() }; let since = since.as_deref().and_then(|s| spectra_mcp::tools::parse_since(s, spectra_core::ids::now_ms())); let mut out = Vec::new(); for h in conv::search(&ledger, &principal, query, channel.as_deref(), author.as_deref(), since, limit).await? { let mut v = serde_json::to_value(&h)?; v["author_display"] = conv::display_name(&ledger, &h.author).await.into(); out.push(v); } out } }; if as_json { println!("{}", Value::Array(hits)); return Ok(0); } if hits.is_empty() { println!("{} {}", ui::note(), ui::dim("no matches")); } for h in hits { let id = h["id"].as_str().unwrap_or(""); println!( "{} {}{}{}{}{}", ui::id(&id[..12.min(id.len())]), ui::author(h["author_display"].as_str().or(h["author"].as_str()).unwrap_or("")), ui::sep(), h["channel"].as_str().unwrap_or(""), ui::sep(), ui::dim(ui::ago(h["at_ms"].as_u64().unwrap_or(0))) ); println!(" {}", h["snippet"].as_str().unwrap_or("")); } Ok(0) } /// `spectra api METHOD PATH [-d JSON]`: any API call, signed as the owner, /// JSON out (the signing proxy for scripts and the editor extensions). pub async fn raw(ctx: &Ctx, method: &str, path: &str, data: Option, owner_home_flag: Option) -> Result { let home = owner_home(ctx, owner_home_flag); let c = owner_client(ctx, &home).await?.context("`spectra api` needs `spectra serve` running, or `spectra connect`")?; let r = c.send(&method.to_uppercase(), path, &[], data.map(String::into_bytes).unwrap_or_default()).await?; println!("{}", String::from_utf8_lossy(&r.body)); Ok(if r.status < 400 { 0 } else { 2 }) }