human.rs 55.2%

116 of 210 lines run, 15 of 40 functions · run #34 on main · run missed

//! 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<Option<Client>> {
    ctx.owner_client(home).await
}

async fn api(c: &Client, method: &str, path: &str, body: Option<Value>) -> Result<Value> {
    super::call(c, method, path, body).await
}

use super::encode;

pub async fn say(
    ctx: &Ctx,
    channel: Option<String>,
    reply_to: Option<String>,
    body: String,
    owner_home_flag: Option<PathBuf>,
    as_json: bool,
) -> Result<i32> {
    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<Vec<Value>> {
    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<PathBuf>, as_json: bool) -> Result<i32> {
    let home = owner_home(ctx, owner_home_flag);
    let client = owner_client(ctx, &home).await?;
    // Following: subscribe first, and read the inbox once the server says
    // the subscription holds, so nothing posted in between is missed.
    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<bool>();
    let watcher = if follow {
        let stream_client = owner_client(ctx, &home).await?.context("--follow needs `spectra serve` running")?;
        let w = tokio::spawn(async move {
            stream_client
                .stream("/api/v1/events?follow=1&kind=message.posted", |l| {
                    if l.starts_with(':') {
                        let _ = tx.send(false);
                    } else if l.starts_with("data:") {
                        let _ = tx.send(true);
                    }
                    true
                })
                .await
        });
        match tokio::time::timeout(std::time::Duration::from_secs(10), rx.recv()).await {
            Ok(Some(_)) => {}
            _ => bail!("the server did not start the stream of new messages"),
        }
        Some(w)
    } else {
        None
    };
    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<String> = 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);
        }
    }
    let Some(watcher) = watcher else { 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.
    while let Some(changed) = rx.recv().await {
        if !changed {
            continue;
        }
        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<String>,
    author: Option<String>,
    since: Option<String>,
    limit: u32,
    as_json: bool,
) -> Result<i32> {
    let hits: Vec<Value> = 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<String>, owner_home_flag: Option<PathBuf>) -> Result<i32> {
    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 })
}

`/`	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 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
17	9×	async fn owner_client(ctx: &Ctx, home: &std::path::Path) -> Result<Option<Client>> {
18	9×	ctx.owner_client(home).await
19	9×	}
20
21	4×	async fn api(c: &Client, method: &str, path: &str, body: Option<Value>) -> Result<Value> {
22	4×	super::call(c, method, path, body).await
23	4×	}
24
25		use super::encode;
26
27	3×	pub async fn say(
28	3×	ctx: &Ctx,
29	3×	channel: Option<String>,
30	3×	reply_to: Option<String>,
31	3×	body: String,
32	3×	owner_home_flag: Option<PathBuf>,
33	3×	as_json: bool,
34	3×	) -> Result<i32> {
35	3×	let home = owner_home(ctx, owner_home_flag);
36	3×	let (id, where_) = match owner_client(ctx, &home).await? {
37	2×	Some(c) => {
38	2×	let ch = channel.clone().unwrap_or_else(\|\| "general".into());
39	2×	let v = api(
40	2×	&c,
41	2×	"POST",
42	2×	&format!("/api/v1/channels/{}/messages", encode(ch.trim_start_matches('#'))),
43	2×	Some(json!({ "body": body, "reply_to": reply_to })),
44	2×	)
45	2×	.await?;
46	2×	if v["duplicate"] == true {
47	0×	bail!("you said exactly this in the last minute ({})", v["id"].as_str().unwrap_or(""));
48	2×	}
49	2×	(v["id"].as_str().unwrap_or("").to_string(), v["channel"].as_str().unwrap_or("").to_string())
50		}
51		None => {
52	1×	let ledger = ctx.writer().await?;
53	1×	let owner = KeyPair::load_home(&home)?;
54	1×	let principal = Principals::load(&ledger)
55	1×	.await?
56	1×	.principal(&owner.key, spectra_core::ids::now_ms())
57	1×	.map_err(anyhow::Error::msg)
58	1×	.context("run `spectra identity refresh` first")?;
59	1×	let author = Author { principal, signer: Signer::Cli(&owner) };
60	1×	let post = Post {
61	1×	channel: if reply_to.is_some() { None } else { Some(channel.unwrap_or_else(\|\| "general".into())) },
62	1×	body,
63	1×	reply_to,
64	1×	..Default::default()
65		};
66	1×	let posted = conv::post(&ledger, &author, post, spectra_core::ids::now_ms()).await?;
67	1×	ledger.checkpoint().await?;
68	1×	match posted {
69	1×	Posted::Message { id, channel } => (id, channel),
70	0×	Posted::Duplicate { id } => bail!("you said exactly this in the last minute ({id})"),
71		}
72		}
73		};
74	3×	if as_json {
75	3×	println!("{}", json!({ "id": id, "channel": where_ }));
76	3×	} else {
77	0×	println!("{} {}{}{}", ui::done("Posted"), ui::subject(&where_), ui::sep(), ui::id(&id));
78	0×	}
79	3×	Ok(0)
80	3×	}
81
82	0×	fn print_item(i: &Value) {
83	0×	let kind = i["kind"].as_str().unwrap_or("");
84	0×	let mark = if kind == "dm" { ui::warn(ui::glyph("✉", "@")) } else { ui::ok(ui::glyph("●", "*")) };
85	0×	let from = i["from_display"].as_str().or(i["from"].as_str()).unwrap_or("");
86	0×	let at = i["at_ms"].as_u64().map(ui::ago).unwrap_or_default();
87	0×	println!(
88		"{mark} {} {}{}{}{}{}",
89	0×	ui::dim(kind),
90	0×	ui::author(from),
91	0×	ui::sep(),
92	0×	i["channel"].as_str().unwrap_or(""),
93	0×	ui::sep(),
94	0×	ui::dim(at)
95		);
96	0×	println!(" {}", i["summary"].as_str().unwrap_or(""));
97	0×	println!(" {}", ui::dim(format!("reply: spectra say --reply-to {} \"…\"", i["id"].as_str().unwrap_or(""))));
98	0×	}
99
100		/// The owner's unread items: from the daemon when one serves, else the ledger.
101	2×	async fn inbox_items(ctx: &Ctx, client: Option<&Client>) -> Result<Vec<Value>> {
102	2×	if let Some(c) = client {
103	1×	return Ok(api(c, "GET", "/api/v1/inbox", None).await?.as_array().cloned().unwrap_or_default());
104	1×	}
105	1×	let ledger = Ledger::open_read_only(&ctx.db()?).await?;
106	1×	let mut items = Vec::new();
107	1×	for i in conv::inbox(&ledger, "human", None).await? {
108	0×	let from_display = conv::display_name(&ledger, &i.from).await;
109	0×	let mut v = serde_json::to_value(&i)?;
110	0×	v["from_display"] = from_display.into();
111	0×	items.push(v);
112		}
113	1×	Ok(items)
114	2×	}
115
116	2×	pub async fn inbox(ctx: &Ctx, follow: bool, owner_home_flag: Option<PathBuf>, as_json: bool) -> Result<i32> {
117	2×	let home = owner_home(ctx, owner_home_flag);
118	2×	let client = owner_client(ctx, &home).await?;
119		// Following: subscribe first, and read the inbox once the server says
120		// the subscription holds, so nothing posted in between is missed.
121	2×	let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<bool>();
122	2×	let watcher = if follow {
123	0×	let stream_client = owner_client(ctx, &home).await?.context("--follow needs `spectra serve` running")?;
124	0×	let w = tokio::spawn(async move {
125	0×	stream_client
126	0×	.stream("/api/v1/events?follow=1&kind=message.posted", \|l\| {
127	0×	if l.starts_with(':') {
128	0×	let _ = tx.send(false);
129	0×	} else if l.starts_with("data:") {
130	0×	let _ = tx.send(true);
131	0×	}
132	0×	true
133	0×	})
134	0×	.await
135	0×	});
136	0×	match tokio::time::timeout(std::time::Duration::from_secs(10), rx.recv()).await {
137	0×	Ok(Some(_)) => {}
138	0×	_ => bail!("the server did not start the stream of new messages"),
139		}
140	0×	Some(w)
141		} else {
142	2×	None
143		};
144	2×	let items = inbox_items(ctx, client.as_ref()).await?;
145	2×	if as_json && !follow {
146	2×	println!("{}", Value::Array(items));
147	2×	return Ok(0);
148	0×	}
149	0×	if items.is_empty() && !follow {
150	0×	println!("{} {}", ui::note(), ui::dim("nothing unread"));
151	0×	}
152	0×	let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
153	0×	for i in &items {
154	0×	seen.insert(i["id"].as_str().unwrap_or("").to_string());
155	0×	if as_json {
156	0×	println!("{i}");
157	0×	} else {
158	0×	print_item(i);
159	0×	}
160		}
161	0×	let Some(watcher) = watcher else { return Ok(0) };
162	0×	let client = client.context("--follow needs `spectra serve` running")?;
163	0×	if !as_json {
164	0×	println!("{} {}", ui::note(), ui::dim("following; Ctrl-C stops"));
165	0×	}
166		// Every new message is a cue to look at the inbox again.
167	0×	while let Some(changed) = rx.recv().await {
168	0×	if !changed {
169	0×	continue;
170	0×	}
171	0×	for v in inbox_items(ctx, Some(&client)).await? {
172	0×	if seen.insert(v["id"].as_str().unwrap_or("").to_string()) {
173	0×	if as_json {
174	0×	println!("{v}");
175	0×	} else {
176	0×	print_item(&v);
177	0×	}
178	0×	}
179		}
180		}
181	0×	watcher.await??;
182	0×	Ok(0)
183	2×	}
184
185	2×	pub async fn search(
186	2×	ctx: &Ctx,
187	2×	query: &str,
188	2×	channel: Option<String>,
189	2×	author: Option<String>,
190	2×	since: Option<String>,
191	2×	limit: u32,
192	2×	as_json: bool,
193	2×	) -> Result<i32> {
194	2×	let hits: Vec<Value> = match owner_client(ctx, &owner_home(ctx, None)).await? {
195	1×	Some(c) => {
196	1×	let mut q = format!("/api/v1/search?q={}&limit={limit}", encode(query));
197	3×	for (k, v) in [("channel", &channel), ("author", &author), ("since", &since)] {
198	3×	if let Some(v) = v {
199	0×	q.push_str(&format!("&{k}={}", encode(v)));
200	3×	}
201		}
202	1×	api(&c, "GET", &q, None).await?.as_array().cloned().unwrap_or_default()
203		}
204		None => {
205	1×	let ledger = Ledger::open_read_only(&ctx.db()?).await?;
206	1×	let owner = Principals::load(&ledger).await?;
207	1×	let me = owner.owner().context("no owner is recorded")?;
208	1×	let principal = spectra_core::principals::Principal::Owner { key: me.hash(), name: me.name.clone() };
209	1×	let since = since.as_deref().and_then(\|s\| spectra_mcp::tools::parse_since(s, spectra_core::ids::now_ms()));
210	1×	let mut out = Vec::new();
211	1×	for h in conv::search(&ledger, &principal, query, channel.as_deref(), author.as_deref(), since, limit).await? {
212	1×	let mut v = serde_json::to_value(&h)?;
213	1×	v["author_display"] = conv::display_name(&ledger, &h.author).await.into();
214	1×	out.push(v);
215		}
216	1×	out
217		}
218		};
219	2×	if as_json {
220	2×	println!("{}", Value::Array(hits));
221	2×	return Ok(0);
222	0×	}
223	0×	if hits.is_empty() {
224	0×	println!("{} {}", ui::note(), ui::dim("no matches"));
225	0×	}
226	0×	for h in hits {
227	0×	let id = h["id"].as_str().unwrap_or("");
228	0×	println!(
229	0×	"{} {}{}{}{}{}",
230	0×	ui::id(&id[..12.min(id.len())]),
231	0×	ui::author(h["author_display"].as_str().or(h["author"].as_str()).unwrap_or("")),
232	0×	ui::sep(),
233	0×	h["channel"].as_str().unwrap_or(""),
234	0×	ui::sep(),
235	0×	ui::dim(ui::ago(h["at_ms"].as_u64().unwrap_or(0)))
236	0×	);
237	0×	println!(" {}", h["snippet"].as_str().unwrap_or(""));
238	0×	}
239	0×	Ok(0)
240	2×	}
241
242		/// `spectra api METHOD PATH [-d JSON]`: any API call, signed as the owner,
243		/// JSON out (the signing proxy for scripts and the editor extensions).
244	2×	pub async fn raw(ctx: &Ctx, method: &str, path: &str, data: Option<String>, owner_home_flag: Option<PathBuf>) -> Result<i32> {
245	2×	let home = owner_home(ctx, owner_home_flag);
246	2×	let c = owner_client(ctx, &home).await?.context("`spectra api` needs `spectra serve` running, or `spectra connect`")?;
247	2×	let r = c.send(&method.to_uppercase(), path, &[], data.map(String::into_bytes).unwrap_or_default()).await?;
248	2×	println!("{}", String::from_utf8_lossy(&r.body));
249	2×	Ok(if r.status < 400 { 0 } else { 2 })
250	2×	}

Coverage / src/cli/human.rs 55.2%

Keyboard