| 1 |
|
| 2 |
|
| 3 |
|
| 4 |
|
| 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 |
|
| 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 |
|
| 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 |
|
| 224 |
|
| 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 |
} |