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