OXYGENHUB

Coverage / src/cli/human.rs 56.2%

113 of 201 lines run, 15 of 40 functions · run #23 on main · run missed

1
//! The human at a terminal (spec §19): `spectra say`, `spectra inbox
2
//! [--follow]`, `spectra search`, and `spectra api` for anything else. Each
3
//! signs with the owner's own key: through the daemon when it is serving,
4
//! else straight into the ledger.
5
6
use super::identity::owner_home;
7
use super::{ui, Ctx};
8
use anyhow::{bail, Context, Result};
9
use serde_json::{json, Value};
10
use spectra_core::conversation::{self as conv, Author, Post, Posted, Signer};
11
use spectra_core::keys::KeyPair;
12
use spectra_core::ledger::Ledger;
13
use spectra_core::principals::Principals;
14
use spectra_mcp::client::Client;
15
use std::path::PathBuf;
16
1710×
async fn owner_client(ctx: &Ctx, home: &std::path::Path) -> Result<Option<Client>> {
1810×
    ctx.owner_client(home).await
1910×
}
20
215×
async fn api(c: &Client, method: &str, path: &str, body: Option<Value>) -> Result<Value> {
225×
    super::call(c, method, path, body).await
235×
}
24
25
use super::encode;
26
273×
pub async fn say(
283×
    ctx: &Ctx,
293×
    channel: Option<String>,
303×
    reply_to: Option<String>,
313×
    body: String,
323×
    owner_home_flag: Option<PathBuf>,
333×
    as_json: bool,
343×
) -> Result<i32> {
353×
    let home = owner_home(ctx, owner_home_flag);
363×
    let (id, where_) = match owner_client(ctx, &home).await? {
372×
        Some(c) => {
382×
            let ch = channel.clone().unwrap_or_else(|| "general".into());
392×
            let v = api(
402×
                &c,
412×
                "POST",
422×
                &format!("/api/v1/channels/{}/messages", encode(ch.trim_start_matches('#'))),
432×
                Some(json!({ "body": body, "reply_to": reply_to })),
442×
            )
452×
            .await?;
462×
            if v["duplicate"] == true {
470×
                bail!("you said exactly this in the last minute ({})", v["id"].as_str().unwrap_or(""));
482×
            }
492×
            (v["id"].as_str().unwrap_or("").to_string(), v["channel"].as_str().unwrap_or("").to_string())
50
        }
51
        None => {
521×
            let ledger = ctx.writer().await?;
531×
            let owner = KeyPair::load_home(&home)?;
541×
            let principal = Principals::load(&ledger)
551×
                .await?
561×
                .principal(&owner.key, spectra_core::ids::now_ms())
571×
                .map_err(anyhow::Error::msg)
581×
                .context("run `spectra identity refresh` first")?;
591×
            let author = Author { principal, signer: Signer::Cli(&owner) };
601×
            let post = Post {
611×
                channel: if reply_to.is_some() { None } else { Some(channel.unwrap_or_else(|| "general".into())) },
621×
                body,
631×
                reply_to,
641×
                ..Default::default()
65
            };
661×
            let posted = conv::post(&ledger, &author, post, spectra_core::ids::now_ms()).await?;
671×
            ledger.checkpoint().await?;
681×
            match posted {
691×
                Posted::Message { id, channel } => (id, channel),
700×
                Posted::Duplicate { id } => bail!("you said exactly this in the last minute ({id})"),
71
            }
72
        }
73
    };
743×
    if as_json {
753×
        println!("{}", json!({ "id": id, "channel": where_ }));
763×
    } else {
770×
        println!("{} {}{}{}", ui::done("Posted"), ui::subject(&where_), ui::sep(), ui::id(&id));
780×
    }
793×
    Ok(0)
803×
}
81
820×
fn print_item(i: &Value) {
830×
    let kind = i["kind"].as_str().unwrap_or("");
840×
    let mark = if kind == "dm" { ui::warn(ui::glyph("✉", "@")) } else { ui::ok(ui::glyph("●", "*")) };
850×
    let from = i["from_display"].as_str().or(i["from"].as_str()).unwrap_or("");
860×
    let at = i["at_ms"].as_u64().map(ui::ago).unwrap_or_default();
870×
    println!(
88
        "{mark} {} {}{}{}{}{}",
890×
        ui::dim(kind),
900×
        ui::author(from),
910×
        ui::sep(),
920×
        i["channel"].as_str().unwrap_or(""),
930×
        ui::sep(),
940×
        ui::dim(at)
95
    );
960×
    println!("  {}", i["summary"].as_str().unwrap_or(""));
970×
    println!("  {}", ui::dim(format!("reply: spectra say --reply-to {} \"…\"", i["id"].as_str().unwrap_or(""))));
980×
}
99
100
/// The owner's unread items: from the daemon when one serves, else the ledger.
1012×
async fn inbox_items(ctx: &Ctx, client: Option<&Client>) -> Result<Vec<Value>> {
1022×
    if let Some(c) = client {
1031×
        return Ok(api(c, "GET", "/api/v1/inbox", None).await?.as_array().cloned().unwrap_or_default());
1041×
    }
1051×
    let ledger = Ledger::open_read_only(&ctx.db()?).await?;
1061×
    let mut items = Vec::new();
1071×
    for i in conv::inbox(&ledger, "human", None).await? {
1080×
        let from_display = conv::display_name(&ledger, &i.from).await;
1090×
        let mut v = serde_json::to_value(&i)?;
1100×
        v["from_display"] = from_display.into();
1110×
        items.push(v);
112
    }
1131×
    Ok(items)
1142×
}
115
1162×
pub async fn inbox(ctx: &Ctx, follow: bool, owner_home_flag: Option<PathBuf>, as_json: bool) -> Result<i32> {
1172×
    let home = owner_home(ctx, owner_home_flag);
1182×
    let client = owner_client(ctx, &home).await?;
1192×
    let items = inbox_items(ctx, client.as_ref()).await?;
1202×
    if as_json && !follow {
1212×
        println!("{}", Value::Array(items));
1222×
        return Ok(0);
1230×
    }
1240×
    if items.is_empty() && !follow {
1250×
        println!("{} {}", ui::note(), ui::dim("nothing unread"));
1260×
    }
1270×
    let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
1280×
    for i in &items {
1290×
        seen.insert(i["id"].as_str().unwrap_or("").to_string());
1300×
        if as_json {
1310×
            println!("{i}");
1320×
        } else {
1330×
            print_item(i);
1340×
        }
135
    }
1360×
    if !follow {
1370×
        return Ok(0);
1380×
    }
1390×
    let client = client.context("--follow needs `spectra serve` running")?;
1400×
    if !as_json {
1410×
        println!("{} {}", ui::note(), ui::dim("following; Ctrl-C stops"));
1420×
    }
143
    // Every new message is a cue to look at the inbox again.
1440×
    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<()>();
1450×
    let stream_client = owner_client(ctx, &home).await?.context("--follow needs `spectra serve` running")?;
1460×
    let watcher = tokio::spawn(async move {
1470×
        stream_client
1480×
            .stream("/api/v1/events?follow=1&kind=message.posted", |l| {
1490×
                if l.starts_with("data:") {
1500×
                    let _ = tx.send(());
1510×
                }
1520×
                true
1530×
            })
1540×
            .await
1550×
    });
1560×
    while rx.recv().await.is_some() {
1570×
        for v in inbox_items(ctx, Some(&client)).await? {
1580×
            if seen.insert(v["id"].as_str().unwrap_or("").to_string()) {
1590×
                if as_json {
1600×
                    println!("{v}");
1610×
                } else {
1620×
                    print_item(&v);
1630×
                }
1640×
            }
165
        }
166
    }
1670×
    watcher.await??;
1680×
    Ok(0)
1692×
}
170
1713×
pub async fn search(
1723×
    ctx: &Ctx,
1733×
    query: &str,
1743×
    channel: Option<String>,
1753×
    author: Option<String>,
1763×
    since: Option<String>,
1773×
    limit: u32,
1783×
    as_json: bool,
1793×
) -> Result<i32> {
1803×
    let hits: Vec<Value> = match owner_client(ctx, &owner_home(ctx, None)).await? {
1812×
        Some(c) => {
1822×
            let mut q = format!("/api/v1/search?q={}&limit={limit}", encode(query));
1836×
            for (k, v) in [("channel", &channel), ("author", &author), ("since", &since)] {
1846×
                if let Some(v) = v {
1850×
                    q.push_str(&format!("&{k}={}", encode(v)));
1866×
                }
187
            }
1882×
            api(&c, "GET", &q, None).await?.as_array().cloned().unwrap_or_default()
189
        }
190
        None => {
1911×
            let ledger = Ledger::open_read_only(&ctx.db()?).await?;
1921×
            let owner = Principals::load(&ledger).await?;
1931×
            let me = owner.owner().context("no owner is recorded")?;
1941×
            let principal = spectra_core::principals::Principal::Owner { key: me.hash(), name: me.name.clone() };
1951×
            let since = since.as_deref().and_then(|s| spectra_mcp::tools::parse_since(s, spectra_core::ids::now_ms()));
1961×
            let mut out = Vec::new();
1971×
            for h in conv::search(&ledger, &principal, query, channel.as_deref(), author.as_deref(), since, limit).await? {
1981×
                let mut v = serde_json::to_value(&h)?;
1991×
                v["author_display"] = conv::display_name(&ledger, &h.author).await.into();
2001×
                out.push(v);
201
            }
2021×
            out
203
        }
204
    };
2053×
    if as_json {
2063×
        println!("{}", Value::Array(hits));
2073×
        return Ok(0);
2080×
    }
2090×
    if hits.is_empty() {
2100×
        println!("{} {}", ui::note(), ui::dim("no matches"));
2110×
    }
2120×
    for h in hits {
2130×
        let id = h["id"].as_str().unwrap_or("");
2140×
        println!(
2150×
            "{} {}{}{}{}{}",
2160×
            ui::id(&id[..12.min(id.len())]),
2170×
            ui::author(h["author_display"].as_str().or(h["author"].as_str()).unwrap_or("")),
2180×
            ui::sep(),
2190×
            h["channel"].as_str().unwrap_or(""),
2200×
            ui::sep(),
2210×
            ui::dim(ui::ago(h["at_ms"].as_u64().unwrap_or(0)))
2220×
        );
2230×
        println!("  {}", h["snippet"].as_str().unwrap_or(""));
2240×
    }
2250×
    Ok(0)
2263×
}
227
228
/// `spectra api METHOD PATH [-d JSON]`: any API call, signed as the owner,
229
/// JSON out (the signing proxy for scripts and the editor extensions).
2302×
pub async fn raw(ctx: &Ctx, method: &str, path: &str, data: Option<String>, owner_home_flag: Option<PathBuf>) -> Result<i32> {
2312×
    let home = owner_home(ctx, owner_home_flag);
2322×
    let c = owner_client(ctx, &home).await?.context("`spectra api` needs `spectra serve` running, or `spectra connect`")?;
2332×
    let r = c.send(&method.to_uppercase(), path, &[], data.map(String::into_bytes).unwrap_or_default()).await?;
2342×
    println!("{}", String::from_utf8_lossy(&r.body));
2352×
    Ok(if r.status < 400 { 0 } else { 2 })
2362×
}