diff --git a/crates/buzz-cli/README.md b/crates/buzz-cli/README.md index 625a7619e3..c279008a4e 100644 --- a/crates/buzz-cli/README.md +++ b/crates/buzz-cli/README.md @@ -47,6 +47,7 @@ buzz messages send-diff --channel --diff - --repo https://github.com/org/ # Channels buzz channels list +buzz channels unread # unread counts from your synced read state (read-only) buzz channels create --name "my-channel" --type stream --visibility open buzz channels join --channel buzz channels topic --channel --topic "New topic" @@ -137,6 +138,7 @@ The table below mirrors that tree for readers who are not at a terminal. | | `vote` | Vote on a forum post | | `channels` | `list` | List channels | | | `get` | Get channel details | +| | `unread` | Unread counts per channel (read-only) | | | `create` | Create a channel | | | `update` | Update channel name/description | | | `topic` | Set channel topic | diff --git a/crates/buzz-cli/src/commands/channel_unread.rs b/crates/buzz-cli/src/commands/channel_unread.rs new file mode 100644 index 0000000000..5d9ec21489 --- /dev/null +++ b/crates/buzz-cli/src/commands/channel_unread.rs @@ -0,0 +1,716 @@ +//! `buzz channels unread` — per-channel unread counts from the synced NIP-RS +//! read state. Read-only: it never publishes. +//! +//! Counting follows desktop's `communityUnreadObserver.ts` +//! (`isUnreadExternalEvent`): an event is unread when it is not yours and is +//! newer than the greatest of the channel frontier, its `msg:` marker and, +//! for a non-broadcast reply, its `thread:` marker. Thread replies count. +//! Desktop's notification gate (participated/followed threads, muted channels, +//! hidden DMs) lives in desktop-local storage and is not applied here. + +use std::collections::HashMap; + +use buzz_core::kind::{ + KIND_FORUM_COMMENT, KIND_FORUM_POST, KIND_HUDDLE_STARTED, KIND_STREAM_MESSAGE, + KIND_STREAM_MESSAGE_V2, +}; +use serde_json::{json, Value}; + +use super::read_state::{self, ReadState}; +use crate::client::{extract_d_tag, extract_tag_value, BuzzClient}; +use crate::error::CliError; +use crate::validate::validate_uuid; + +const KIND_GROUP_METADATA: u32 = 39000; +const KIND_GROUP_MEMBERS: u32 = 39002; +const MEMBER_CHANNEL_LIMIT: u32 = 1000; + +/// Resource bounds of one run. +pub(crate) struct Bounds { + /// Most channels counted in one run. + pub max_channels: usize, + /// Most events read per channel; a channel that reaches it reports `capped`. + pub events_per_channel: u32, +} + +pub(crate) const BOUNDS: Bounds = Bounds { + max_channels: 200, + events_per_channel: 1000, +}; + +#[derive(Debug, PartialEq, Eq, serde::Serialize)] +pub(crate) struct ChannelUnread { + pub channel_id: String, + pub name: String, + pub last_read_at: Option, + pub unread_count: u32, + /// `unread_count` is a lower bound: the per-channel event cap was reached. + pub capped: bool, +} + +#[derive(Debug)] +pub(crate) struct Outcome { + pub channels: Vec, + /// Channels found but not counted because of `--limit`. + pub dropped: usize, + pub skipped_slots: Vec, +} + +struct Channel { + id: String, + name: String, + is_dm: bool, +} + +fn message_kinds(is_dm: bool) -> Vec { + let mut kinds = vec![ + KIND_STREAM_MESSAGE, + KIND_STREAM_MESSAGE_V2, + KIND_FORUM_POST, + KIND_FORUM_COMMENT, + ]; + if is_dm { + kinds.push(KIND_HUDDLE_STARTED); + } + kinds +} + +fn tags(event: &Value) -> impl Iterator> { + event + .get("tags") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter_map(|tag| tag.as_array()) + .map(|tag| tag.iter().map(|v| v.as_str().unwrap_or("")).collect()) +} + +/// The thread root a reply counts against, as desktop's `getThreadReference` +/// and `isBroadcastReply` resolve it. A broadcast reply has none. +fn thread_root(event: &Value) -> Option { + let mut broadcast = false; + let mut root = None; + let mut reply = None; + for tag in tags(event) { + match tag.as_slice() { + ["broadcast", "1", ..] => broadcast = true, + ["e", id, _, "root", ..] => { + root.get_or_insert(*id); + } + ["e", id, _, "reply", ..] => reply = Some(*id), + _ => {} + } + } + if broadcast { + return None; + } + let parent = reply?; + Some(root.unwrap_or(parent).to_string()) +} + +fn is_unread(event: &Value, me: &str, channel_read_at: Option, state: &ReadState) -> bool { + if event + .get("pubkey") + .and_then(Value::as_str) + .is_some_and(|p| p.eq_ignore_ascii_case(me)) + { + return false; + } + let id = event.get("id").and_then(Value::as_str).unwrap_or(""); + let created_at = event.get("created_at").and_then(Value::as_u64).unwrap_or(0); + let read_at = [ + channel_read_at, + state.frontier(&format!("msg:{id}")), + thread_root(event).and_then(|root| state.frontier(&format!("thread:{root}"))), + ] + .into_iter() + .flatten() + .max(); + read_at.is_none_or(|read_at| created_at > read_at) +} + +fn newest_by_d(events: Vec) -> HashMap { + let mut newest: HashMap = HashMap::new(); + for event in events { + let d = extract_d_tag(&event); + if d.is_empty() { + continue; + } + let at = |e: &Value| e.get("created_at").and_then(Value::as_u64).unwrap_or(0); + if newest.get(&d).is_none_or(|head| at(&event) > at(head)) { + newest.insert(d, event); + } + } + newest +} + +async fn member_channel_ids(client: &BuzzClient) -> Result, CliError> { + let filter = json!({ + "kinds": [KIND_GROUP_MEMBERS], + "#p": [client.keys().public_key().to_hex()], + }); + let events = client + .query_all_bounded(filter, MEMBER_CHANNEL_LIMIT) + .await + .map_err(|e| match e { + CliError::Other(msg) => { + CliError::Other(format!("{msg}; name the channels with --channel instead")) + } + other => other, + })?; + let mut ids: Vec = events + .iter() + .map(extract_d_tag) + .filter(|id| !id.is_empty()) + .collect(); + ids.sort(); + ids.dedup(); + Ok(ids) +} + +async fn channel_metadata( + client: &BuzzClient, + ids: &[String], +) -> Result, CliError> { + if ids.is_empty() { + return Ok(HashMap::new()); + } + let filter = json!({"kinds": [KIND_GROUP_METADATA], "#d": ids}); + let events = client + .query_all_bounded(filter, MEMBER_CHANNEL_LIMIT) + .await?; + Ok(newest_by_d(events)) +} + +fn is_archived(metadata: &Value) -> bool { + tags(metadata).any(|t| t.as_slice() == ["archived", "true"]) +} + +async fn resolve_channels( + client: &BuzzClient, + requested: &[String], + limit: usize, +) -> Result<(Vec, usize), CliError> { + let explicit = !requested.is_empty(); + let ids = if explicit { + requested.to_vec() + } else { + member_channel_ids(client).await? + }; + let metadata = channel_metadata(client, &ids).await?; + let mut channels = Vec::new(); + for id in ids { + match metadata.get(&id) { + Some(meta) if !explicit && is_archived(meta) => {} + Some(meta) => channels.push(Channel { + name: extract_tag_value(meta, "name"), + is_dm: extract_tag_value(meta, "t") == "dm", + id, + }), + None if explicit => { + return Err(CliError::NotFound(format!("channel {id} not found"))); + } + None => channels.push(Channel { + name: String::new(), + is_dm: false, + id, + }), + } + } + channels.sort_by(|a, b| a.id.cmp(&b.id)); + let dropped = channels.len().saturating_sub(limit); + channels.truncate(limit); + Ok((channels, dropped)) +} + +async fn count_channel( + client: &BuzzClient, + channel: &Channel, + state: &ReadState, + bounds: &Bounds, +) -> Result { + let last_read_at = state.frontier(&channel.id); + let mut filter = json!({ + "kinds": message_kinds(channel.is_dm), + "#h": [channel.id], + }); + if let Some(read_at) = last_read_at { + filter["since"] = json!(read_at + 1); + } + let events = client + .query_paginated(filter, bounds.events_per_channel) + .await?; + let capped = events.len() >= bounds.events_per_channel as usize; + let me = client.keys().public_key().to_hex(); + let unread = events + .iter() + .filter(|e| is_unread(e, &me, last_read_at, state)) + .count(); + Ok(ChannelUnread { + channel_id: channel.id.clone(), + name: channel.name.clone(), + last_read_at, + unread_count: u32::try_from(unread).unwrap_or(u32::MAX), + capped, + }) +} + +/// Compute unread counts for the user's channels, or for `requested` ones. +pub(crate) async fn unread( + client: &BuzzClient, + requested: &[String], + limit: Option, + bounds: &Bounds, +) -> Result { + let limit = match limit { + None => bounds.max_channels, + Some(0) => return Err(CliError::Usage("--limit must be at least 1".into())), + Some(n) if n as usize > bounds.max_channels => { + return Err(CliError::Usage(format!( + "--limit must be at most {}", + bounds.max_channels + ))); + } + Some(n) => n as usize, + }; + let mut requested: Vec = requested + .iter() + .map(|c| validate_uuid(c).map(|()| c.to_lowercase())) + .collect::>()?; + requested.sort(); + requested.dedup(); + if requested.len() > limit { + return Err(CliError::Usage(format!( + "{} channels given but at most {limit} can be counted per run", + requested.len() + ))); + } + + let state = read_state::fetch(client).await?; + let (channels, dropped) = resolve_channels(client, &requested, limit).await?; + let mut counted = Vec::with_capacity(channels.len()); + for channel in &channels { + counted.push(count_channel(client, channel, &state, bounds).await?); + } + counted.sort_by(|a, b| { + b.unread_count + .cmp(&a.unread_count) + .then_with(|| a.name.cmp(&b.name)) + .then_with(|| a.channel_id.cmp(&b.channel_id)) + }); + Ok(Outcome { + channels: counted, + dropped, + skipped_slots: state.skipped, + }) +} + +pub async fn cmd_unread( + client: &BuzzClient, + requested: &[String], + limit: Option, +) -> Result<(), CliError> { + let outcome = unread(client, requested, limit, &BOUNDS).await?; + if outcome.dropped > 0 { + eprintln!( + "{}", + json!({ + "warning": "channels_truncated", + "counted": outcome.channels.len(), + "dropped": outcome.dropped, + "message": "more channels than --limit; rerun with --channel to count the rest", + }) + ); + } + if !outcome.skipped_slots.is_empty() { + let slots: Vec = outcome + .skipped_slots + .iter() + .map(|s| json!({"d": s.d_tag, "v": s.version})) + .collect(); + eprintln!( + "{}", + json!({ + "warning": "read_state_slots_skipped", + "slots": slots, + "message": "written by a newer schema; counts may overstate unread", + }) + ); + } + let out = serde_json::to_string(&outcome.channels) + .map_err(|e| CliError::Other(format!("failed to encode unread counts: {e}")))?; + println!("{out}"); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::commands::read_state::tests::{slot_event, snapshot_descriptor, snapshot_envelope}; + use axum::{body::Bytes, extract::State, routing::get, routing::post, Json, Router}; + use nostr::Keys; + use std::sync::{Arc, Mutex}; + use tokio::net::TcpListener; + + #[derive(Default)] + struct Relay { + events: Vec, + filters: Vec, + } + type Shared = Arc>; + + fn tag_values(filter: &Value, key: &str) -> Vec { + filter[key] + .as_array() + .map(|a| { + a.iter() + .filter_map(|v| v.as_str().map(str::to_string)) + .collect() + }) + .unwrap_or_default() + } + + fn matches(event: &Value, filter: &Value) -> bool { + let kinds: Vec = filter["kinds"] + .as_array() + .unwrap() + .iter() + .filter_map(Value::as_u64) + .collect(); + let has_tag = |name: &str, wanted: Vec| { + wanted.is_empty() + || tags(event).any(|t| { + t.first() == Some(&name) && wanted.iter().any(|w| t.get(1) == Some(&w.as_str())) + }) + }; + kinds.contains(&event["kind"].as_u64().unwrap()) + && filter["authors"] + .as_array() + .is_none_or(|a| a.iter().any(|x| x == &event["pubkey"])) + && event["created_at"].as_u64().unwrap() >= filter["since"].as_u64().unwrap_or(0) + && has_tag("h", tag_values(filter, "#h")) + && has_tag("d", tag_values(filter, "#d")) + && has_tag("p", tag_values(filter, "#p")) + && has_tag("t", tag_values(filter, "#t")) + } + + async fn fake_relay(relay: Shared) -> String { + let app = Router::new() + .route( + "/query", + post(|State(relay): State, body: Bytes| async move { + let filters: Vec = serde_json::from_slice(&body).unwrap(); + let [filter] = filters.as_slice() else { + panic!("one filter expected") + }; + let mut relay = relay.lock().unwrap(); + relay.filters.push(filter.clone()); + if filter.get("read_state_snapshot").is_some() { + let own: Vec = relay + .events + .iter() + .filter(|e| { + e["kind"] == 30078 && filter["authors"] == json!([e["pubkey"]]) + }) + .cloned() + .collect(); + let author = filter["authors"][0].as_str().unwrap(); + return Json(snapshot_envelope(author, own)); + } + let mut hits: Vec = relay + .events + .iter() + .filter(|e| matches(e, filter)) + .cloned() + .collect(); + hits.sort_by_key(|e| std::cmp::Reverse(e["created_at"].as_u64().unwrap())); + hits.truncate(filter["limit"].as_u64().unwrap() as usize); + Json(json!(hits)) + }), + ) + .route( + "/info", + get(|| async { Json(json!({"read_state_snapshot": snapshot_descriptor()})) }), + ) + .with_state(relay); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + format!("http://{addr}") + } + + struct World { + client: BuzzClient, + relay: Shared, + other: Keys, + counter: u64, + } + + const GEN: &str = "11111111-1111-4111-8111-111111111111"; + const DEV: &str = "22222222-2222-4222-8222-222222222222"; + const OLD: &str = "33333333-3333-4333-8333-333333333333"; + const NEW: &str = "44444444-4444-4444-8444-444444444444"; + + async fn world() -> World { + let relay = Shared::default(); + let url = fake_relay(relay.clone()).await; + let client = BuzzClient::new(url, Keys::generate(), None, None).unwrap(); + World { + client, + relay, + other: Keys::generate(), + counter: 0, + } + } + + impl World { + fn push(&self, event: Value) { + self.relay.lock().unwrap().events.push(event); + } + + fn signed(&self, keys: &Keys, kind: u16, tags: Vec, at: u64) -> Value { + let event = nostr::EventBuilder::new(nostr::Kind::Custom(kind), "x") + .tags(tags) + .custom_created_at(nostr::Timestamp::from(at)) + .sign_with_keys(keys) + .unwrap(); + serde_json::to_value(event).unwrap() + } + + fn member(&self, channel: &str) { + let me = self.client.keys().public_key(); + let tags = vec![nostr::Tag::identifier(channel), nostr::Tag::public_key(me)]; + self.push(self.signed(&Keys::generate(), 39002, tags, 1)); + } + + fn channel(&self, id: &str, name: &str, extra: &[(&str, &str)]) { + let mut tags = vec![ + nostr::Tag::identifier(id), + nostr::Tag::parse(["name", name]).unwrap(), + ]; + tags.extend( + extra + .iter() + .map(|(k, v)| nostr::Tag::parse([*k, *v]).unwrap()), + ); + self.push(self.signed(&Keys::generate(), 39000, tags, 1)); + } + + fn say(&mut self, channel: &str, author: &Keys, at: u64, extra: &[Vec<&str>]) -> String { + self.counter += 1; + let mut tags = vec![nostr::Tag::parse(["h", channel]).unwrap()]; + tags.extend(extra.iter().map(|t| nostr::Tag::parse(t.clone()).unwrap())); + tags.push(nostr::Tag::parse(["nonce", &self.counter.to_string()]).unwrap()); + let event = self.signed(author, 9, tags, at); + let id = event["id"].as_str().unwrap().to_string(); + self.push(event); + id + } + + fn read_state(&self, slot: &str, contexts: Value, at: u64) { + let plaintext = json!({"v": 1, "client_id": slot, "contexts": contexts}).to_string(); + self.push(slot_event(self.client.keys(), slot, &plaintext, at)); + } + + fn queries_for(&self, kind: u64) -> Vec { + self.relay + .lock() + .unwrap() + .filters + .iter() + .filter(|f| { + f["kinds"] + .as_array() + .is_some_and(|k| k.contains(&json!(kind))) + }) + .cloned() + .collect() + } + } + + fn by_id<'a>(out: &'a Outcome, id: &str) -> &'a ChannelUnread { + out.channels.iter().find(|c| c.channel_id == id).unwrap() + } + + #[tokio::test] + async fn counts_unread_with_desktop_rules() { + let mut w = world().await; + let me = w.client.keys().clone(); + let other = w.other.clone(); + for (id, name) in [(GEN, "general"), (DEV, "dev"), (OLD, "old")] { + w.member(id); + w.channel(id, name, &[]); + } + w.member(NEW); + w.channel(NEW, "new", &[("archived", "true")]); + + // general: frontier 100 from slot a, raised to 150 by slot b (max-merge). + w.say(GEN, &other, 90, &[]); + w.say(GEN, &other, 120, &[]); + w.say(GEN, &other, 160, &[]); + w.say(GEN, &me, 170, &[]); + let marked = w.say(GEN, &other, 180, &[]); + let root = w.say(GEN, &other, 80, &[]); + let reply = w.say( + GEN, + &other, + 200, + &[vec!["e", &root, "", "root"], vec!["e", &root, "", "reply"]], + ); + let broadcast = w.say( + GEN, + &other, + 203, + &[ + vec!["e", &root, "", "root"], + vec!["e", &root, "", "reply"], + vec!["broadcast", "1"], + ], + ); + w.read_state("a", json!({GEN: 100, DEV: 5}), 1); + w.read_state( + "b", + json!({GEN: 150, format!("msg:{marked}"): 180, format!("thread:{root}"): 205}), + 2, + ); + // dev: never read, two from others, one of mine. + w.say(DEV, &other, 10, &[]); + w.say(DEV, &other, 11, &[]); + w.say(DEV, &me, 12, &[]); + // old: fully read. + w.say(OLD, &other, 50, &[]); + w.read_state("c", json!({OLD: 50}), 3); + let _ = (reply, broadcast); + + let out = unread(&w.client, &[], None, &BOUNDS).await.unwrap(); + + // 160 (> 150); 180 is covered by its msg marker; `reply` 200 is covered by + // thread:root 205; the broadcast reply 203 ignores the thread marker; 90/120 + // predate the frontier; mine is excluded. + assert_eq!(by_id(&out, GEN).unread_count, 2); + assert_eq!(by_id(&out, GEN).last_read_at, Some(150)); + assert_eq!(by_id(&out, DEV).unread_count, 2); + assert_eq!(by_id(&out, DEV).last_read_at, Some(5)); + assert_eq!(by_id(&out, OLD).unread_count, 0); + assert!( + out.channels.iter().all(|c| c.channel_id != NEW), + "archived is skipped" + ); + assert!(out.channels.iter().all(|c| !c.capped)); + assert_eq!(out.dropped, 0); + let order: Vec<_> = out.channels.iter().map(|c| c.name.as_str()).collect(); + assert_eq!( + order, + ["dev", "general", "old"], + "sorted by unread desc, then name" + ); + + // The production queries are bounded and scoped, not a firehose. + let reads = w.queries_for(9); + assert_eq!(reads.len(), 3); + for q in &reads { + assert!(q["limit"].as_u64().unwrap() <= u64::from(BOUNDS.events_per_channel)); + assert!(q["#h"].as_array().is_some_and(|h| h.len() == 1)); + } + let general_q = reads.iter().find(|q| q["#h"] == json!([GEN])).unwrap(); + assert_eq!( + general_q["since"], + json!(151), + "reads start after the frontier" + ); + let dev_q = reads.iter().find(|q| q["#h"] == json!([DEV])).unwrap(); + assert_eq!(dev_q["since"], json!(6)); + let state_qs = w.queries_for(30078); + assert_eq!(state_qs.len(), 1, "read state is one atomic snapshot"); + assert_eq!(state_qs[0]["read_state_snapshot"], json!(1)); + for paging in ["#t", "since", "until", "limit"] { + assert!(state_qs[0].get(paging).is_none(), "{paging}"); + } + } + + #[tokio::test] + async fn bounds_channels_and_events_and_says_so() { + let mut w = world().await; + let other = w.other.clone(); + for id in [GEN, DEV, OLD] { + w.member(id); + w.channel(id, id, &[]); + for at in 1..=4 { + w.say(id, &other, at, &[]); + } + } + let bounds = Bounds { + max_channels: 2, + events_per_channel: 3, + }; + + let out = unread(&w.client, &[], Some(2), &bounds).await.unwrap(); + assert_eq!((out.channels.len(), out.dropped), (2, 1)); + assert!(out.channels.iter().all(|c| c.capped && c.unread_count == 3)); + assert_eq!( + w.queries_for(9).len(), + 2, + "the dropped channel is never queried" + ); + + let err = unread(&w.client, &[], Some(3), &bounds).await.unwrap_err(); + assert!(matches!(err, CliError::Usage(_))); + let err = unread(&w.client, &[], Some(0), &bounds).await.unwrap_err(); + assert!(matches!(err, CliError::Usage(_))); + let all = [GEN, DEV, OLD].map(String::from); + let err = unread(&w.client, &all, None, &bounds).await.unwrap_err(); + assert!( + matches!(err, CliError::Usage(_)), + "explicit channels are never silently dropped" + ); + } + + #[tokio::test] + async fn explicit_channels_are_validated_and_must_exist() { + let mut w = world().await; + let other = w.other.clone(); + w.channel(GEN, "general", &[("archived", "true"), ("t", "dm")]); + w.say(GEN, &other, 5, &[]); + + let out = unread(&w.client, &[GEN.to_uppercase(), GEN.into()], None, &BOUNDS) + .await + .unwrap(); + assert_eq!( + out.channels.len(), + 1, + "case-insensitive dedupe; archived is allowed when named" + ); + assert_eq!(out.channels[0].unread_count, 1); + let queried = w.queries_for(9); + assert!( + queried[0]["kinds"] + .as_array() + .unwrap() + .contains(&json!(48100)), + "dm counts huddle starts" + ); + + let err = unread(&w.client, &["nope".into()], None, &BOUNDS) + .await + .unwrap_err(); + assert!(matches!(err, CliError::Usage(_))); + let err = unread(&w.client, &[DEV.into()], None, &BOUNDS) + .await + .unwrap_err(); + assert!(matches!(err, CliError::NotFound(_))); + } + + #[tokio::test] + async fn an_unreadable_slot_fails_the_command_before_any_count() { + let w = world().await; + w.member(GEN); + w.channel(GEN, "general", &[]); + w.read_state("a", json!({GEN: 1}), 1); + w.push(slot_event(w.client.keys(), "b", "garbage", 2)); + + let err = unread(&w.client, &[], None, &BOUNDS).await.unwrap_err(); + assert!(err.to_string().contains("unreadable"), "{err}"); + assert!(w.queries_for(9).is_empty()); + assert!(w.queries_for(39002).is_empty()); + } +} diff --git a/crates/buzz-cli/src/commands/channels.rs b/crates/buzz-cli/src/commands/channels.rs index 40de86b686..ee92017570 100644 --- a/crates/buzz-cli/src/commands/channels.rs +++ b/crates/buzz-cli/src/commands/channels.rs @@ -1953,6 +1953,9 @@ pub async fn dispatch( let vis_str = visibility.as_ref().map(|v| v.to_string()); cmd_list_channels(client, vis_str.as_deref(), Some(member), limit, format).await } + ChannelsCmd::Unread { channel, limit } => { + crate::commands::channel_unread::cmd_unread(client, &channel, limit).await + } ChannelsCmd::Get { channel } => cmd_get_channel(client, &channel).await, ChannelsCmd::Search { query, diff --git a/crates/buzz-cli/src/commands/mod.rs b/crates/buzz-cli/src/commands/mod.rs index 7e18bdbd25..cd9f94a21c 100644 --- a/crates/buzz-cli/src/commands/mod.rs +++ b/crates/buzz-cli/src/commands/mod.rs @@ -1,5 +1,6 @@ pub mod agents; pub mod channel_templates; +pub mod channel_unread; pub mod channels; pub mod dms; pub mod emoji; @@ -20,6 +21,7 @@ pub mod pr; pub mod project_channel; pub mod projects; pub mod reactions; +pub(crate) mod read_state; mod repo_default_branch; pub mod repos; pub mod social; diff --git a/crates/buzz-cli/src/commands/read_state.rs b/crates/buzz-cli/src/commands/read_state.rs new file mode 100644 index 0000000000..471ed6dc22 --- /dev/null +++ b/crates/buzz-cli/src/commands/read_state.rs @@ -0,0 +1,520 @@ +//! Read-only view of the user's own NIP-RS read state (kind:30078, +//! `d=read-state:`, NIP-44 encrypted to self). +//! +//! Frontiers are decoded and max-merged the way desktop's +//! `readStateFormat.ts` / `readStateSnapshot.ts` do. Unlike desktop, which +//! drops a slot it cannot decode and carries on, a slot this CLI cannot read +//! is an error: an unread count built on a partial merge would overstate +//! unread, never understate it, and the user could not tell. + +use std::collections::HashMap; + +use buzz_core::kind::KIND_READ_STATE; +use nostr::nips::nip44; +use nostr::Event; +use serde_json::{json, Value}; + +use crate::client::BuzzClient; +use crate::error::CliError; + +const D_TAG_PREFIX: &str = "read-state:"; +const T_TAG: &str = "read-state"; +const MAX_CONTEXTS: usize = 10_000; +const MAX_CONTEXT_KEY_BYTES: usize = 256; +const MAX_SLOT_ID_LEN: usize = 64; +const MAX_CLIENT_ID_UTF16_UNITS: usize = 64; +const SNAPSHOT_VERSION: u64 = 1; + +/// A slot that was valid but written by a newer schema, so left out of the merge. +#[derive(Debug, PartialEq, Eq)] +pub(crate) struct SkippedSlot { + pub d_tag: String, + pub version: i64, +} + +/// Max-merged frontiers of every readable slot. +#[derive(Debug, Default)] +pub(crate) struct ReadState { + frontiers: HashMap, + pub skipped: Vec, +} + +impl ReadState { + /// The merged read marker of a context, in unix seconds. + pub(crate) fn frontier(&self, context: &str) -> Option { + self.frontiers.get(context).copied() + } +} + +/// Load and merge every read-state slot of the signed-in identity. +/// +/// Uses the relay's atomic snapshot (NIP-RS, Buzz Atomic Snapshot Extension): +/// paginating by cursor cannot see a slot another client replaces mid-load, so +/// a relay without the extension is an error rather than a partial read. +pub(crate) async fn fetch(client: &BuzzClient) -> Result { + let community_id = discover_snapshot(client).await?; + let me = client.keys().public_key().to_hex(); + let filter = json!({ + "kinds": [KIND_READ_STATE], + "authors": [me], + "read_state_snapshot": SNAPSHOT_VERSION, + }); + let raw = client.query(&filter).await?; + let events = snapshot_events(&raw, &community_id, &me)?; + merge(client.keys(), events) +} + +fn cannot_prove_complete(detail: impl std::fmt::Display) -> CliError { + CliError::Other(format!( + "cannot prove your read state is complete: {detail}; refusing to report unread counts" + )) +} + +/// The community id a relay advertises for its atomic read-state snapshot. +async fn discover_snapshot(client: &BuzzClient) -> Result { + let raw = client.get_public("/info").await?; + let info: Value = serde_json::from_str(&raw) + .map_err(|e| CliError::Other(format!("invalid NIP-11 response: {e}")))?; + let descriptor = info + .get("read_state_snapshot") + .filter(|d| d.is_object()) + .ok_or_else(|| { + cannot_prove_complete("this relay does not offer the atomic read_state_snapshot") + })?; + if descriptor.get("version").and_then(Value::as_u64) != Some(SNAPSHOT_VERSION) { + return Err(cannot_prove_complete( + "this relay offers an unsupported read_state_snapshot version", + )); + } + descriptor + .get("community_id") + .and_then(Value::as_str) + .map(str::to_owned) + .ok_or_else(|| cannot_prove_complete("the relay advertised no community_id")) +} + +/// Validate the snapshot envelope and return its events. Anything but a +/// complete, version-1 envelope for this identity and community is rejected, +/// including the plain event array an older relay returns when it ignores the +/// extension. +fn snapshot_events(raw: &str, community_id: &str, me: &str) -> Result, CliError> { + let mut envelope: Value = serde_json::from_str(raw) + .map_err(|e| CliError::Other(format!("failed to parse query response: {e}")))?; + let field = |name: &str| envelope.get(name); + if field("read_state_snapshot").and_then(Value::as_u64) != Some(SNAPSHOT_VERSION) + || field("complete") != Some(&Value::Bool(true)) + { + return Err(cannot_prove_complete( + "the relay did not return a complete read_state_snapshot", + )); + } + if field("community_id").and_then(Value::as_str) != Some(community_id) + || field("pubkey").and_then(Value::as_str) != Some(me) + { + return Err(cannot_prove_complete( + "the snapshot belongs to a different community or identity", + )); + } + match envelope.get_mut("events").map(Value::take) { + Some(Value::Array(events)) => Ok(events), + _ => Err(cannot_prove_complete("the snapshot has no events array")), + } +} + +fn unreadable(event: &Event, detail: impl std::fmt::Display) -> CliError { + CliError::Other(format!( + "your read-state slot {} is unreadable ({detail}); refusing to report unread counts from a partial read state", + event.id.to_hex() + )) +} + +fn single_d_tag(event: &Event) -> Option<&str> { + let mut d = event + .tags + .iter() + .filter(|t| t.as_slice().first().map(String::as_str) == Some("d")); + let tag = d.next()?; + if d.next().is_some() { + return None; + } + tag.as_slice().get(1).map(String::as_str) +} + +fn is_slot_coordinate(event: &Event) -> Option<&str> { + let d = single_d_tag(event)?; + let slot = d.strip_prefix(D_TAG_PREFIX)?; + if slot.is_empty() || slot.len() > MAX_SLOT_ID_LEN || !slot.is_ascii() { + return None; + } + let t_count = event + .tags + .iter() + .filter(|t| matches!(t.as_slice(), [k, v, ..] if k == "t" && v == T_TAG)) + .count(); + (t_count == 1).then_some(d) +} + +/// JS `Number.isInteger` within the uint32 range. +fn uint32(value: &Value) -> Option { + let n = value.as_u64().or_else(|| { + let f = value.as_f64()?; + (f >= 0.0 && f.fract() == 0.0 && f <= u32::MAX as f64).then_some(f as u64) + })?; + (n <= u64::from(u32::MAX)).then_some(n) +} + +/// Decode one blob. `Ok(Err(version))` is a valid blob of an unknown schema. +fn decode(event: &Event, plaintext: &str) -> Result, i64>, CliError> { + let blob: Value = serde_json::from_str(plaintext).map_err(|e| unreadable(event, e))?; + let Some(blob) = blob.as_object() else { + return Err(unreadable(event, "not a JSON object")); + }; + let version = blob + .get("v") + .and_then(Value::as_f64) + .filter(|v| v.fract() == 0.0); + match version { + Some(1.0) => {} + Some(other) => return Ok(Err(other as i64)), + None => return Err(unreadable(event, "missing or non-integer schema version")), + } + let client_id_ok = blob + .get("client_id") + .and_then(Value::as_str) + .is_some_and(|id| (1..=MAX_CLIENT_ID_UTF16_UNITS).contains(&id.encode_utf16().count())); + if !client_id_ok { + return Err(unreadable(event, "invalid client_id")); + } + let Some(contexts) = blob.get("contexts").and_then(Value::as_object) else { + return Err(unreadable(event, "contexts is not an object")); + }; + if contexts.len() > MAX_CONTEXTS { + return Err(unreadable(event, "too many contexts")); + } + Ok(Ok(contexts + .iter() + .filter(|(key, _)| key.len() <= MAX_CONTEXT_KEY_BYTES) + .filter_map(|(key, value)| Some((key.clone(), uint32(value)?))) + .collect())) +} + +/// Verify, decrypt and max-merge raw relay events. Events that are not read-state +/// coordinates are ignored; a coordinate that cannot be read is an error. +pub(crate) fn merge(keys: &nostr::Keys, raw: Vec) -> Result { + let me = keys.public_key(); + let mut heads: HashMap = HashMap::new(); + for value in raw { + let event: Event = serde_json::from_value(value) + .map_err(|e| CliError::Other(format!("relay returned a malformed event: {e}")))?; + if event.pubkey != me || event.kind.as_u16() as u32 != KIND_READ_STATE { + return Err(CliError::Other(format!( + "relay returned event {} that is not your read state", + event.id.to_hex() + ))); + } + event.verify().map_err(|_| { + CliError::Other(format!( + "relay returned event {} with an invalid signature", + event.id.to_hex() + )) + })?; + let Some(d) = is_slot_coordinate(&event) else { + continue; + }; + let d = d.to_string(); + let newer = heads + .get(&d) + .is_none_or(|head| (event.created_at, head.id) > (head.created_at, event.id)); + if newer { + heads.insert(d, event); + } + } + + let mut state = ReadState::default(); + let mut ordered: Vec<_> = heads.into_iter().collect(); + ordered.sort_by(|a, b| a.0.cmp(&b.0)); + for (d_tag, event) in ordered { + let plaintext = nip44::decrypt(keys.secret_key(), &me, &event.content) + .map_err(|e| unreadable(&event, e))?; + match decode(&event, &plaintext)? { + Ok(contexts) => { + for (key, ts) in contexts { + let slot = state.frontiers.entry(key).or_insert(0); + *slot = (*slot).max(ts); + } + } + Err(version) => state.skipped.push(SkippedSlot { d_tag, version }), + } + } + Ok(state) +} + +#[cfg(test)] +pub(crate) mod tests { + use super::*; + use axum::{body::Bytes, extract::State, routing::get, routing::post, Json, Router}; + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + use std::sync::{Arc, Mutex}; + use tokio::net::TcpListener; + + pub(crate) const COMMUNITY: &str = "5f0c1b9e-2d0b-4f4e-9a7c-0d6f3f3f3f3f"; + + pub(crate) fn snapshot_envelope(pubkey: &str, events: Vec) -> Value { + json!({ + "read_state_snapshot": 1, + "complete": true, + "community_id": COMMUNITY, + "pubkey": pubkey, + "snapshot_id": "0".repeat(64), + "events": events, + }) + } + + pub(crate) fn snapshot_descriptor() -> Value { + json!({"version": 1, "community_id": COMMUNITY, "max_events": 4096}) + } + + pub(crate) fn slot_event(keys: &Keys, slot: &str, plaintext: &str, at: u64) -> Value { + let content = nip44::encrypt( + keys.secret_key(), + &keys.public_key(), + plaintext, + nip44::Version::V2, + ) + .unwrap(); + let event = EventBuilder::new(Kind::Custom(KIND_READ_STATE as u16), content) + .tags([ + Tag::identifier(format!("{D_TAG_PREFIX}{slot}")), + Tag::hashtag(T_TAG), + ]) + .custom_created_at(Timestamp::from(at)) + .sign_with_keys(keys) + .unwrap(); + serde_json::to_value(event).unwrap() + } + + fn blob(contexts: Value) -> String { + json!({"v": 1, "client_id": "dev", "contexts": contexts}).to_string() + } + + #[test] + fn max_merges_every_slot_and_ignores_override_and_foreign_entries() { + let keys = Keys::generate(); + let state = merge( + &keys, + vec![ + slot_event(&keys, "a", &blob(json!({"c1": 10, "c2": 50, "ov_s:c1": 3, "esc:ov_c:x": 9})), 1), + slot_event(&keys, "b", &blob(json!({"c1": 20, "c2": 5, "c3": 1.0e3, "neg": -1, "k".repeat(257): 9, "big": 4294967296u64, "frac": 1.5, "str": "7"})), 2), + ], + ) + .unwrap(); + assert_eq!(state.frontier("c1"), Some(20)); + assert_eq!(state.frontier("c2"), Some(50)); + assert_eq!(state.frontier("c3"), Some(1000)); + assert_eq!(state.frontier(&"k".repeat(257)), None); + for dropped in ["neg", "big", "frac", "str", "missing"] { + assert_eq!(state.frontier(dropped), None, "{dropped}"); + } + assert!(state.skipped.is_empty()); + } + + #[test] + fn newest_version_of_a_coordinate_wins() { + let keys = Keys::generate(); + let old = slot_event(&keys, "a", &blob(json!({"c1": 99})), 1); + let new = slot_event(&keys, "a", &blob(json!({"c1": 5})), 2); + let state = merge(&keys, vec![new, old]).unwrap(); + assert_eq!(state.frontier("c1"), Some(5)); + } + + #[test] + fn an_unreadable_slot_is_an_error_not_an_empty_state() { + let keys = Keys::generate(); + let good = slot_event(&keys, "a", &blob(json!({"c1": 1})), 1); + for bad in [ + "not json", + "[]", + r#"{"client_id":"d","contexts":{}}"#, + r#"{"v":1,"client_id":"","contexts":{}}"#, + r#"{"v":1,"client_id":"d","contexts":[]}"#, + r#"{"v":"1","client_id":"d","contexts":{}}"#, + ] { + let err = + merge(&keys, vec![good.clone(), slot_event(&keys, "b", bad, 2)]).expect_err(bad); + assert!(err.to_string().contains("unreadable"), "{bad}: {err}"); + } + + let other = Keys::generate(); + let foreign = slot_event(&other, "b", &blob(json!({})), 2); + let err = merge(&keys, vec![foreign]).unwrap_err(); + assert!(err.to_string().contains("not your read state"), "{err}"); + + let mut tampered = slot_event(&keys, "b", &blob(json!({"c1": 1})), 2); + tampered["created_at"] = json!(3); + assert!(merge(&keys, vec![tampered]).is_err()); + + let mut undecryptable = slot_event(&keys, "b", &blob(json!({})), 2); + let wrong = slot_event(&other, "b", &blob(json!({})), 2); + undecryptable["content"] = wrong["content"].clone(); + assert!(merge(&keys, vec![undecryptable]).is_err()); + } + + #[test] + fn a_newer_schema_is_reported_not_merged() { + let keys = Keys::generate(); + let future = json!({"v": 2, "client_id": "d", "contexts": {"c1": 99}}).to_string(); + let state = merge(&keys, vec![slot_event(&keys, "a", &future, 1)]).unwrap(); + assert_eq!(state.frontier("c1"), None); + assert_eq!( + state.skipped, + vec![SkippedSlot { + d_tag: "read-state:a".into(), + version: 2 + }] + ); + } + + #[test] + fn non_coordinates_are_ignored() { + let keys = Keys::generate(); + let unrelated = EventBuilder::new(Kind::Custom(KIND_READ_STATE as u16), "x") + .tags([Tag::identifier("channel-sections"), Tag::hashtag(T_TAG)]) + .sign_with_keys(&keys) + .unwrap(); + let no_t_event = EventBuilder::new(Kind::Custom(KIND_READ_STATE as u16), "x") + .tags([Tag::identifier("read-state:a")]) + .sign_with_keys(&keys) + .unwrap(); + let state = merge( + &keys, + vec![ + serde_json::to_value(unrelated).unwrap(), + serde_json::to_value(no_t_event).unwrap(), + ], + ) + .unwrap(); + assert!(state.frontiers.is_empty()); + } + + type Bodies = Arc>>; + + #[derive(Clone)] + struct Canned { + info: Value, + query: String, + bodies: Bodies, + } + + async fn relay(keys: &Keys, info: Value, query: Value) -> (BuzzClient, Bodies) { + let canned = Canned { + info, + query: query.to_string(), + bodies: Bodies::default(), + }; + let bodies = canned.bodies.clone(); + let app = Router::new() + .route( + "/info", + get(|State(c): State| async move { Json(c.info) }), + ) + .route( + "/query", + post(|State(c): State, body: Bytes| async move { + c.bodies + .lock() + .unwrap() + .push(serde_json::from_slice(&body).unwrap()); + c.query + }), + ) + .with_state(canned); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + ( + BuzzClient::new(url, keys.clone(), None, None).unwrap(), + bodies, + ) + } + + fn advertised() -> Value { + json!({"read_state_snapshot": snapshot_descriptor()}) + } + + #[tokio::test] + async fn fetch_loads_one_atomic_snapshot() { + let keys = Keys::generate(); + let slots = vec![ + slot_event(&keys, "a", &blob(json!({"c1": 10})), 1), + slot_event(&keys, "b", &blob(json!({"c1": 20})), 2), + ]; + let (client, bodies) = relay( + &keys, + advertised(), + snapshot_envelope(&keys.public_key().to_hex(), slots), + ) + .await; + + let state = fetch(&client).await.unwrap(); + + assert_eq!(state.frontier("c1"), Some(20)); + assert_eq!( + *bodies.lock().unwrap(), + vec![json!([{ + "kinds": [KIND_READ_STATE], + "authors": [keys.public_key().to_hex()], + "read_state_snapshot": 1, + }])], + "one exact snapshot request: no tag, limit or cursor that could page" + ); + } + + #[tokio::test] + async fn fetch_refuses_a_load_it_cannot_prove_complete() { + let keys = Keys::generate(); + let slot = slot_event(&keys, "a", &blob(json!({"c1": 10})), 1); + let good = snapshot_envelope(&keys.public_key().to_hex(), vec![slot.clone()]); + let with = |field: &str, value: Value| { + let mut envelope = good.clone(); + envelope[field] = value; + envelope + }; + let cases: Vec<(&str, Value, Value)> = vec![ + ("no descriptor", json!({}), good.clone()), + ( + "descriptor version", + json!({"read_state_snapshot": {"version": 2, "community_id": COMMUNITY}}), + good.clone(), + ), + ( + "descriptor community", + json!({"read_state_snapshot": {"version": 1}}), + good.clone(), + ), + ("relay ignored the extension", advertised(), json!([slot])), + ("incomplete", advertised(), with("complete", json!(false))), + ("unmarked", advertised(), with("complete", Value::Null)), + ( + "version", + advertised(), + with("read_state_snapshot", json!(2)), + ), + ( + "community", + advertised(), + with("community_id", json!("other")), + ), + ( + "identity", + advertised(), + with("pubkey", json!("f".repeat(64))), + ), + ("events", advertised(), with("events", Value::Null)), + ]; + for (name, info, response) in cases { + let (client, _) = relay(&keys, info, response).await; + let err = fetch(&client).await.expect_err(name); + assert!(err.to_string().contains("cannot prove"), "{name}: {err}"); + } + } +} diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index a204bd228b..9c343a61fa 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -664,6 +664,19 @@ pub enum ChannelsCmd { #[arg(long)] limit: Option, }, + /// Unread message counts from your synced read state (read-only) + #[command( + after_help = "Examples:\n buzz channels unread\n buzz channels unread --channel --channel \n buzz channels unread --limit 20\n\nOutput: [{channel_id, name, last_read_at, unread_count, capped}]. `capped: true` means unread_count is a lower bound. When more channels exist than --limit (at most 200), the rest are listed in a stderr warning count and not counted. Fails if a read-state slot cannot be decrypted or decoded." + )] + Unread { + /// Count only these channels (repeatable, at most 200). Default: every + /// non-archived channel you are a member of + #[arg(long)] + channel: Vec, + /// Most channels to count [default: 200, max: 200] + #[arg(long)] + limit: Option, + }, /// Get details for a single channel Get { /// Channel UUID @@ -2618,6 +2631,7 @@ mod tests { "set-add-policy", "topic", "unarchive", + "unread", "update" ] ); @@ -2729,7 +2743,7 @@ mod tests { let expected: Vec<(&str, usize)> = vec![ ("agents", 5), ("canvas", 4), - ("channels", 16), + ("channels", 17), ("dms", 4), ("emoji", 5), ("feed", 1),