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