diff --git a/crates/buzz-cli/README.md b/crates/buzz-cli/README.md index 4aebfa55bf..5ee8101dd4 100644 --- a/crates/buzz-cli/README.md +++ b/crates/buzz-cli/README.md @@ -145,6 +145,12 @@ buzz sidebar sections create --name "Engineering" --icon "🛠" buzz sidebar sections move --section "Engineering" --channel --channel buzz sidebar sections list +# Sidebar stars, mutes, and project pins +buzz sidebar stars add --channel --channel +buzz sidebar mutes remove --channel +buzz sidebar projects add --project 30621:: +buzz sidebar stars list + # Agent Memory (NIP-AE) buzz mem ls buzz mem get @@ -167,6 +173,15 @@ takes a section id or a unique section name. A repeated command prints `"message": "unchanged"` and publishes nothing. Concurrent edits from another CLI run or another client can overwrite each other, and the last publish wins. +`sidebar stars`, `sidebar mutes`, and `sidebar projects` edit the other synced +sidebar lists the same way: one head read and one publish per command, and +`"message": "unchanged"` when nothing changes. `remove` keeps a `false` entry +(as desktop does) so the change syncs; removing an id that is not set changes +nothing. Stars and mutes keep the newest 500 entries; a command that would +evict its own ids fails. Project addresses are `30621::`. +Pass the same ids in any case; channel UUIDs and project owners are stored in +lowercase. If the relay reports a newer update (exit 5), run the command again. + `protect set` replaces every existing rule for the exact ref pattern. Any constraint omitted from the command is removed. `protect list` reports malformed stored rules in `validation_error` so an owner can remove and repair them. @@ -214,6 +229,9 @@ The table below mirrors that tree for readers who are not at a terminal. | | `sections reorder` | Put sections in order | | | `sections move` | Move one or more channels into a section | | | `sections unassign` | Return channels to the default channel list | +| | `stars list` / `add` / `remove` | List, star, or unstar channels | +| | `mutes list` / `add` / `remove` | List, mute, or unmute channels | +| | `projects list` / `add` / `remove` | List, pin, or unpin sidebar projects | | `reactions` | `add` | React to a message | | | `remove` | Remove a reaction | | | `get` | List reactions | diff --git a/crates/buzz-cli/src/commands/mod.rs b/crates/buzz-cli/src/commands/mod.rs index 24e613cc43..dd63145441 100644 --- a/crates/buzz-cli/src/commands/mod.rs +++ b/crates/buzz-cli/src/commands/mod.rs @@ -27,6 +27,8 @@ pub mod repos; pub(crate) mod sent_from_thread; pub mod sidebar; mod sidebar_doc; +mod sidebar_flags; +mod sidebar_sync; pub mod social; pub mod upload; pub mod users; diff --git a/crates/buzz-cli/src/commands/sidebar.rs b/crates/buzz-cli/src/commands/sidebar.rs index e0081e0f78..4057ad0269 100644 --- a/crates/buzz-cli/src/commands/sidebar.rs +++ b/crates/buzz-cli/src/commands/sidebar.rs @@ -6,90 +6,37 @@ //! multi-channel `move` lands atomically. A head this CLI cannot read is never //! overwritten. -use std::time::SystemTime; - -use buzz_core::kind::KIND_READ_STATE; -use nostr::nips::nip44; -use nostr::{EventBuilder, Kind, Tag, Timestamp}; use serde_json::{json, Value}; -use super::parse_write_response; use super::sidebar_doc::{self, Edit, Projection, Section, Stamp, Tree}; +use super::sidebar_flags; +use super::sidebar_sync::{self, channel_ids, now, Lane}; use crate::client::BuzzClient; use crate::error::CliError; use crate::{SidebarCmd, SidebarSectionsCmd}; -const D_TAG: &str = "channel-sections"; -/// Relay event size ceiling shared with desktop's sidebar blobs. -const MAX_CIPHERTEXT_BYTES: usize = 65_535; +const LANE: Lane = Lane { + d_tag: "channel-sections", + label: "sidebar sections", +}; struct Head { created_at: u64, tree: Tree, } -fn now() -> std::time::Duration { - SystemTime::now() - .duration_since(SystemTime::UNIX_EPOCH) - .unwrap_or_default() -} - -/// The relay's current sections event for this identity, decoded. An absent -/// head is an empty tree; an unreadable one is an error, never an empty list. -/// The read is pinned to the writer (`consistency: strong`): a write derives a -/// full snapshot from it, and a lagging replica would drop the caller's own -/// just-accepted edits. +/// The relay's current sections event for this identity, decoded; an absent +/// head is an empty tree. async fn fetch_head(client: &BuzzClient) -> Result { - let me = client.keys().public_key(); - let filter = json!({ - "kinds": [KIND_READ_STATE], - "authors": [me.to_hex()], - "#d": [D_TAG], - "limit": 1, - "consistency": "strong", - }); - let raw = client.query(&filter).await?; - let events: Vec = serde_json::from_str(&raw) - .map_err(|e| CliError::Other(format!("failed to parse sidebar sections query: {e}")))?; - let returned = events.len(); - let head = events - .into_iter() - .filter_map(|event| serde_json::from_value::(event).ok()) - .filter(|event| { - event.pubkey == me && event.tags.identifier() == Some(D_TAG) && event.verify().is_ok() - }) - .min_by(|a, b| { - b.created_at - .cmp(&a.created_at) - .then_with(|| a.id.to_hex().cmp(&b.id.to_hex())) - }); - let Some(event) = head else { - if returned > 0 { - return Err(CliError::Other( - "the relay returned a sidebar sections event that is not yours or fails verification; refusing to overwrite it".into(), - )); - } - return Ok(Head { - created_at: 0, - tree: Tree::new(), - }); - }; - - let unreadable = |detail: String| { - CliError::Other(format!( - "your sidebar sections event {} is unreadable ({detail}); refusing to read or overwrite it", - event.id.to_hex() - )) - }; - let plaintext = nip44::decrypt(client.keys().secret_key(), &me, &event.content) - .map_err(|e| unreadable(e.to_string()))?; - let json: Value = serde_json::from_str(&plaintext).map_err(|e| unreadable(e.to_string()))?; - let created_at = event.created_at.as_secs(); - let legacy_v = i64::try_from(created_at.saturating_mul(1_000)).unwrap_or(i64::MAX); - let tree = sidebar_doc::decode(&json, legacy_v).ok_or_else(|| { - unreadable("unsupported format; a newer client may have written it".into()) - })?; - Ok(Head { created_at, tree }) + let head = sidebar_sync::fetch_head(client, LANE, |json, created_at| { + let legacy_v = i64::try_from(created_at.saturating_mul(1_000)).unwrap_or(i64::MAX); + sidebar_doc::decode(json, legacy_v) + }) + .await?; + Ok(Head { + created_at: head.created_at, + tree: head.doc.unwrap_or_default(), + }) } /// A live section by exact id, or by exact name when that name is unique. @@ -127,18 +74,6 @@ fn section_icon(icon: Option<&str>) -> Result, CliError> { } } -/// Channel ids in the lowercase hyphenated form desktop keys assignments by. -fn channel_ids(channels: &[String]) -> Result, CliError> { - channels - .iter() - .map(|channel| { - uuid::Uuid::parse_str(channel.trim()) - .map(|id| id.hyphenated().to_string()) - .map_err(|_| CliError::Usage(format!("invalid channel UUID: {channel}"))) - }) - .collect() -} - /// Reads the head, applies `apply`, and publishes the result once. Returns /// the write response, or an `unchanged` response without publishing. async fn edit( @@ -154,7 +89,7 @@ async fn edit( }; apply(&mut Edit::new(&mut tree, &stamp), &view)?; if tree == head.tree { - return Ok(json!({"event_id": null, "accepted": true, "message": "unchanged"})); + return Ok(sidebar_sync::unchanged()); } if !sidebar_doc::project(&tree).within_limits() { return Err(CliError::Usage(format!( @@ -163,37 +98,9 @@ async fn edit( sidebar_doc::MAX_ASSIGNMENTS ))); } - - let keys = client.keys(); let plaintext = serde_json::to_string(&sidebar_doc::encode(&tree)) .map_err(|e| CliError::Other(format!("failed to encode sidebar sections: {e}")))?; - let too_large = - || CliError::Usage("sidebar sections would exceed the relay's event size limit".into()); - // NIP-44 rejects longer plaintexts outright. - if plaintext.len() > u16::MAX as usize { - return Err(too_large()); - } - let content = nip44::encrypt( - keys.secret_key(), - &keys.public_key(), - plaintext, - nip44::Version::V2, - ) - .map_err(|e| CliError::Other(format!("failed to encrypt sidebar sections: {e}")))?; - if content.len() > MAX_CIPHERTEXT_BYTES { - return Err(too_large()); - } - let created_at = now().as_secs().max(head.created_at + 1); - let builder = EventBuilder::new(Kind::Custom(KIND_READ_STATE as u16), content) - .tags([Tag::identifier(D_TAG), Tag::hashtag(D_TAG)]) - .custom_created_at(Timestamp::from(created_at)); - let raw = client.submit_event(client.sign_event(builder)?).await?; - let normalized = parse_write_response( - &raw, - "a newer sidebar sections update won; run the command again", - )?; - serde_json::from_str(&normalized) - .map_err(|e| CliError::Other(format!("relay response is not JSON: {e}"))) + sidebar_sync::publish(client, LANE, plaintext, head.created_at).await } async fn run(cmd: SidebarSectionsCmd, client: &BuzzClient) -> Result { @@ -284,101 +191,34 @@ async fn run(cmd: SidebarSectionsCmd, client: &BuzzClient) -> Result Result<(), CliError> { - let SidebarCmd::Sections(cmd) = cmd; - let output = run(cmd, client).await?; + let output = match cmd { + SidebarCmd::Sections(cmd) => run(cmd, client).await?, + SidebarCmd::Stars(cmd) => { + sidebar_flags::run(&sidebar_flags::STARS, cmd.into(), client).await? + } + SidebarCmd::Mutes(cmd) => { + sidebar_flags::run(&sidebar_flags::MUTES, cmd.into(), client).await? + } + SidebarCmd::Projects(cmd) => { + sidebar_flags::run(&sidebar_flags::PROJECTS, cmd.into(), client).await? + } + }; println!("{output}"); Ok(()) } #[cfg(test)] mod tests { + use super::super::sidebar_sync::test_support::{self, Shared}; use super::*; - use axum::{body::Bytes, extract::State, routing::post, Json, Router}; use nostr::Keys; - use std::sync::{Arc, Mutex}; - use tokio::net::TcpListener; - - #[derive(Default)] - struct Relay { - events: Vec, - duplicate: bool, - } - type Shared = Arc>; - - /// Fake relay: `/query` checks the head filter and returns every stored - /// event matching its author and d-tag (more than the relay's `limit`, so - /// head selection is exercised); `/events` stores one. - 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: {filters:?}") - }; - assert_eq!(filter["kinds"], json!([KIND_READ_STATE])); - assert_eq!(filter["#d"], json!([D_TAG])); - assert_eq!(filter["limit"], json!(1)); - assert_eq!(filter["consistency"], json!("strong")); - let author = &filter["authors"][0]; - let relay = relay.lock().unwrap(); - let matching: Vec<_> = relay - .events - .iter() - .filter(|e| &e["pubkey"] == author && e["tags"][0] == json!(["d", D_TAG])) - .cloned() - .collect(); - Json(json!(matching)) - }), - ) - .route( - "/events", - post(|State(relay): State, body: Bytes| async move { - let event: Value = serde_json::from_slice(&body).unwrap(); - let mut relay = relay.lock().unwrap(); - let message = if relay.duplicate { - "duplicate: newer head" - } else { - "" - }; - if !relay.duplicate { - relay.events.push(event.clone()); - } - Json(json!({"event_id": event["id"], "accepted": true, "message": message})) - }), - ) - .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}") - } async fn setup() -> (BuzzClient, Shared) { - let relay = Shared::default(); - let url = fake_relay(relay.clone()).await; - let client = BuzzClient::new(url, Keys::generate(), None, None).unwrap(); - (client, relay) + test_support::setup().await } fn seed_at(client: &BuzzClient, relay: &Shared, plaintext: &str, created_at: u64) -> Value { - let keys = client.keys(); - 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(D_TAG)]) - .custom_created_at(Timestamp::from(created_at)) - .sign_with_keys(keys) - .unwrap(); - let event = serde_json::to_value(event).unwrap(); - relay.lock().unwrap().events.push(event.clone()); - event + test_support::seed_at(client, relay, LANE, plaintext, created_at) } fn seed(client: &BuzzClient, relay: &Shared, plaintext: &str) { @@ -398,19 +238,7 @@ mod tests { sidebar_doc::encode(&tree).to_string() } - /// The decrypted document of the newest published event. - fn published(client: &BuzzClient, relay: &Shared) -> (nostr::Event, Value) { - let raw = relay.lock().unwrap().events.last().cloned().unwrap(); - let event: nostr::Event = serde_json::from_value(raw).unwrap(); - let keys = client.keys(); - let plaintext = - nip44::decrypt(keys.secret_key(), &keys.public_key(), &event.content).unwrap(); - (event, serde_json::from_str(&plaintext).unwrap()) - } - - fn published_count(relay: &Shared) -> usize { - relay.lock().unwrap().events.len() - } + use test_support::{published, published_count}; const C1: &str = "11111111-1111-4111-8111-111111111111"; const C2: &str = "22222222-2222-4222-8222-222222222222"; @@ -455,9 +283,9 @@ mod tests { assert_eq!(published_count(&relay), before + 1); let (event, doc) = published(&client, &relay); - assert_eq!(event.kind.as_u16() as u32, KIND_READ_STATE); - assert!(event.tags.iter().any(|t| t.as_slice() == ["d", D_TAG])); - assert!(event.tags.iter().any(|t| t.as_slice() == ["t", D_TAG])); + assert_eq!(event.kind.as_u16(), 30078); + assert!(event.tags.iter().any(|t| t.as_slice() == ["d", LANE.d_tag])); + assert!(event.tags.iter().any(|t| t.as_slice() == ["t", LANE.d_tag])); assert_eq!(doc["meta"]["v"], json!(1)); assert_eq!(doc["assignments"][C1], json!(eng_id)); assert_eq!(doc["sections"][0]["icon"], json!("🛠")); diff --git a/crates/buzz-cli/src/commands/sidebar_flags.rs b/crates/buzz-cli/src/commands/sidebar_flags.rs new file mode 100644 index 0000000000..e216b92408 --- /dev/null +++ b/crates/buzz-cli/src/commands/sidebar_flags.rs @@ -0,0 +1,624 @@ +//! `buzz sidebar stars|mutes|projects` — your own synced per-id flags +//! (kind:30078, d=`channel-stars` / `channel-mutes` / +//! `project-sidebar-membership`, NIP-44 encrypted to yourself). +//! +//! All three documents are `{version: 1, : {: {, +//! updatedAt}}}` with a per-entry last-writer-wins merge, wire-compatible with +//! desktop `channelStarsStorage.ts`, `channelMutesStorage.ts` and +//! `projectSidebarMembership.ts`. Every write is one head read and one publish. + +use std::collections::BTreeMap; + +use buzz_core::kind::KIND_PROJECT; +use serde_json::{json, Map, Value}; + +use super::sidebar_sync::{self, channel_ids, now, Lane}; +use crate::client::BuzzClient; +use crate::error::CliError; +use crate::{SidebarChannelFlagCmd, SidebarProjectsCmd}; + +/// Entry cap shared with desktop's star and mute stores. +const MAX_ENTRIES: usize = 500; + +#[derive(Clone, Copy)] +enum Unit { + Seconds, + Millis, +} + +#[derive(Clone, Copy)] +enum Ids { + Channels, + Projects, +} + +/// How one document differs from the others. +pub(crate) struct Spec { + lane: Lane, + collection: &'static str, + flag: &'static str, + ids: Ids, + unit: Unit, + /// Oldest entries are evicted past this many (stars and mutes only). + max_entries: Option, + /// Projects reject a document without a collection and accept the legacy + /// array of addresses; stars and mutes read a missing collection as empty. + strict: bool, +} + +pub(crate) const STARS: Spec = Spec { + lane: Lane { + d_tag: "channel-stars", + label: "channel stars", + }, + collection: "channels", + flag: "starred", + ids: Ids::Channels, + unit: Unit::Seconds, + max_entries: Some(MAX_ENTRIES), + strict: false, +}; + +pub(crate) const MUTES: Spec = Spec { + lane: Lane { + d_tag: "channel-mutes", + label: "channel mutes", + }, + collection: "channels", + flag: "muted", + ids: Ids::Channels, + unit: Unit::Seconds, + max_entries: Some(MAX_ENTRIES), + strict: false, +}; + +pub(crate) const PROJECTS: Spec = Spec { + lane: Lane { + d_tag: "project-sidebar-membership", + label: "project sidebar membership", + }, + collection: "projects", + flag: "selected", + ids: Ids::Projects, + unit: Unit::Millis, + max_entries: None, + strict: true, +}; + +impl From for Action { + fn from(cmd: SidebarChannelFlagCmd) -> Self { + match cmd { + SidebarChannelFlagCmd::List => Action::List, + SidebarChannelFlagCmd::Add { channels } => Action::Set { + ids: channels, + on: true, + }, + SidebarChannelFlagCmd::Remove { channels } => Action::Set { + ids: channels, + on: false, + }, + } + } +} + +impl From for Action { + fn from(cmd: SidebarProjectsCmd) -> Self { + match cmd { + SidebarProjectsCmd::List => Action::List, + SidebarProjectsCmd::Add { projects } => Action::Set { + ids: projects, + on: true, + }, + SidebarProjectsCmd::Remove { projects } => Action::Set { + ids: projects, + on: false, + }, + } + } +} + +/// What a flag command asks for; ids are validated and canonicalised by `run`. +pub(crate) enum Action { + List, + Set { ids: Vec, on: bool }, +} + +#[derive(Debug, Clone, PartialEq)] +struct Entry { + on: bool, + updated_at: f64, +} + +type Store = BTreeMap; + +fn valid_entry(spec: &Spec, key: &str, value: &Value) -> Option { + if spec.strict && key.is_empty() { + return None; + } + let object = value.as_object()?; + let updated_at = object.get("updatedAt")?.as_f64().filter(|n| *n >= 0.0)?; + Some(Entry { + on: object.get(spec.flag)?.as_bool()?, + updated_at, + }) +} + +/// Desktop's `boundStarStore`: keeps the newest `max` entries, oldest first by +/// `(updatedAt, id)`. +fn bound(spec: &Spec, store: &mut Store) { + let Some(max) = spec.max_entries else { return }; + if store.len() <= max { + return; + } + let mut entries: Vec<_> = std::mem::take(store).into_iter().collect(); + entries.sort_by(|(left_id, left), (right_id, right)| { + left.updated_at + .total_cmp(&right.updated_at) + .then_with(|| left_id.cmp(right_id)) + }); + store.extend(entries.split_off(entries.len() - max)); +} + +/// Desktop's `parse*Payload`: entries that fail validation are dropped. +fn decode(spec: &Spec, json: &Value) -> Option { + if let (true, Some(addresses)) = (spec.strict, json.as_array()) { + return Some( + addresses + .iter() + .filter_map(Value::as_str) + .filter(|address| !address.is_empty()) + .map(|address| { + ( + address.to_string(), + Entry { + on: true, + updated_at: 0.0, + }, + ) + }) + .collect(), + ); + } + let object = json.as_object()?; + if object.get("version")?.as_f64()? != 1.0 { + return None; + } + let mut store = match object.get(spec.collection) { + Some(Value::Object(entries)) => entries + .iter() + .filter_map(|(key, value)| Some((key.clone(), valid_entry(spec, key, value)?))) + .collect(), + _ if spec.strict => return None, + _ => Store::new(), + }; + bound(spec, &mut store); + Some(store) +} + +fn number(n: f64) -> Value { + if n.fract() == 0.0 && n < 9_007_199_254_740_992.0 { + json!(n as u64) + } else { + json!(n) + } +} + +fn encode(spec: &Spec, store: &Store) -> Value { + let entries: Map = store + .iter() + .map(|(id, entry)| { + ( + id.clone(), + json!({spec.flag: entry.on, "updatedAt": number(entry.updated_at)}), + ) + }) + .collect(); + json!({"version": 1, spec.collection: entries}) +} + +/// `30621::` with a lowercase owner, as desktop builds project +/// addresses. +fn project_addresses(addresses: &[String]) -> Result, CliError> { + addresses + .iter() + .map(|address| { + let invalid = || CliError::Usage(format!("invalid project address: {address}")); + let mut parts = address.trim().splitn(3, ':'); + let (Some(kind), Some(owner), Some(d)) = (parts.next(), parts.next(), parts.next()) + else { + return Err(invalid()); + }; + let owner_ok = owner.len() == 64 && owner.bytes().all(|b| b.is_ascii_hexdigit()); + if kind != KIND_PROJECT.to_string() || !owner_ok || d.is_empty() { + return Err(invalid()); + } + Ok(format!("{kind}:{}:{d}", owner.to_ascii_lowercase())) + }) + .collect() +} + +fn canonical_ids(spec: &Spec, ids: &[String]) -> Result, CliError> { + match spec.ids { + Ids::Channels => channel_ids(ids), + Ids::Projects => project_addresses(ids), + } +} + +fn clock(spec: &Spec) -> f64 { + let now = now(); + match spec.unit { + Unit::Seconds => now.as_secs() as f64, + Unit::Millis => now.as_millis() as f64, + } +} + +/// Sets `ids` to `on`, stamping each changed entry strictly newer than the one +/// it replaces so the edit wins every client's merge. Returns the changed ids; +/// an id already in the requested state, or absent when switching off, is left +/// alone. +fn set_flags(spec: &Spec, store: &mut Store, ids: &[String], on: bool) -> Vec { + let clock = clock(spec); + let mut changed = Vec::new(); + for id in ids { + let current = store.get(id); + if current.map_or(!on, |entry| entry.on == on) { + continue; + } + let updated_at = current.map_or(clock, |entry| clock.max(entry.updated_at + 1.0)); + store.insert(id.clone(), Entry { on, updated_at }); + changed.push(id.clone()); + } + changed +} + +async fn fetch_store(spec: &Spec, client: &BuzzClient) -> Result<(u64, Store), CliError> { + let head = sidebar_sync::fetch_head(client, spec.lane, |json, _| decode(spec, json)).await?; + Ok((head.created_at, head.doc.unwrap_or_default())) +} + +pub(crate) async fn run( + spec: &Spec, + action: Action, + client: &BuzzClient, +) -> Result { + match action { + Action::List => { + let (_, store) = fetch_store(spec, client).await?; + let on: Vec<&String> = store + .iter() + .filter(|(_, entry)| entry.on) + .map(|(id, _)| id) + .collect(); + Ok(json!({spec.collection: on})) + } + Action::Set { ids, on } => { + let ids = canonical_ids(spec, &ids)?; + let (head_created_at, mut store) = fetch_store(spec, client).await?; + let changed = set_flags(spec, &mut store, &ids, on); + if changed.is_empty() { + return Ok(sidebar_sync::unchanged()); + } + bound(spec, &mut store); + if changed.iter().any(|id| !store.contains_key(id)) { + return Err(CliError::Usage(format!( + "{} keeps at most {MAX_ENTRIES} entries; pass fewer ids", + spec.lane.label + ))); + } + let plaintext = serde_json::to_string(&encode(spec, &store)).map_err(|e| { + CliError::Other(format!("failed to encode {}: {e}", spec.lane.label)) + })?; + sidebar_sync::publish(client, spec.lane, plaintext, head_created_at).await + } + } +} + +#[cfg(test)] +mod tests { + use super::super::sidebar_sync::test_support::{ + published, published_count, seed_at, setup, Shared, + }; + use super::*; + use nostr::Keys; + + const C1: &str = "11111111-1111-4111-8111-111111111111"; + const C2: &str = "22222222-2222-4222-8222-222222222222"; + const C3: &str = "abcdef01-2345-4678-89ab-cdef01234567"; + const OWNER: &str = "ab12ab12ab12ab12ab12ab12ab12ab12ab12ab12ab12ab12ab12ab12ab12ab12"; + + fn set(ids: &[&str], on: bool) -> Action { + Action::Set { + ids: ids.iter().map(|id| id.to_string()).collect(), + on, + } + } + + fn seed( + client: &BuzzClient, + relay: &Shared, + spec: &Spec, + doc: Value, + created_at: u64, + ) -> Value { + seed_at(client, relay, spec.lane, &doc.to_string(), created_at) + } + + fn entries(count: usize, flag: &str, first_updated_at: u64) -> Value { + let map: Map = (0..count) + .map(|i| { + let id = format!("{i:036}"); + ( + id, + json!({flag: true, "updatedAt": first_updated_at + i as u64}), + ) + }) + .collect(); + Value::Object(map) + } + + #[tokio::test] + async fn stars_and_mutes_sync_desktop_format() { + for spec in [&STARS, &MUTES] { + let (client, relay) = setup().await; + let list = |client| run(spec, Action::List, client); + assert_eq!(list(&client).await.unwrap(), json!({"channels": []})); + + // Two channels land in one publish; an uppercase id is canonicalised. + let before = now().as_secs(); + run(spec, set(&[C1, &C3.to_uppercase()], true), &client) + .await + .unwrap(); + assert_eq!(published_count(&relay), 1); + let (event, doc) = published(&client, &relay); + assert_eq!(event.kind.as_u16(), 30078); + assert!(event + .tags + .iter() + .any(|t| t.as_slice() == ["d", spec.lane.d_tag])); + assert!(event + .tags + .iter() + .any(|t| t.as_slice() == ["t", spec.lane.d_tag])); + assert_eq!(doc["version"], json!(1)); + assert_eq!(doc["channels"][C3][spec.flag], json!(true)); + let stamp = doc["channels"][C1]["updatedAt"].as_u64().unwrap(); + assert!(stamp >= before && stamp < before + 60, "seconds: {stamp}"); + assert_eq!(list(&client).await.unwrap(), json!({"channels": [C1, C3]})); + + // Repeats and clearing an unknown channel publish nothing. + for action in [set(&[C1], true), set(&[C2], false)] { + let again = run(spec, action, &client).await.unwrap(); + assert_eq!(again["message"], json!("unchanged")); + } + assert_eq!(published_count(&relay), 1); + + // Clearing keeps a `false` entry, as desktop does, so the clear syncs. + run(spec, set(&[C1], false), &client).await.unwrap(); + let (_, doc) = published(&client, &relay); + assert_eq!(doc["channels"][C1][spec.flag], json!(false)); + assert_eq!(list(&client).await.unwrap(), json!({"channels": [C3]})); + + // Invalid input fails before any publish. + let err = run(spec, set(&[C2, "nope"], true), &client) + .await + .unwrap_err(); + assert!(matches!(err, CliError::Usage(_)), "{err:?}"); + assert_eq!(published_count(&relay), 2); + } + } + + #[tokio::test] + async fn an_edit_beats_the_entry_it_replaces_and_follows_the_head() { + let (client, relay) = setup().await; + let future = now().as_secs() + 1_000; + // An entry from a clock ahead of ours, a mobile-shaped entry, and one + // desktop would drop. + seed( + &client, + &relay, + &STARS, + json!({"version": 1.0, "channels": { + C1: {"starred": false, "updatedAt": future}, + C2: {"starred": true, "updatedAt": 5}, + C3: {"starred": "yes", "updatedAt": 5}, + "no-time": {"starred": true}, + "negative": {"starred": true, "updatedAt": -1}, + }}), + future, + ); + run(&STARS, set(&[C1], true), &client).await.unwrap(); + let (event, doc) = published(&client, &relay); + assert_eq!(event.created_at.as_secs(), future + 1); + assert_eq!(doc["channels"][C1]["updatedAt"], json!(future + 1)); + assert_eq!(doc["channels"][C2]["updatedAt"], json!(5)); + assert_eq!(doc["channels"].as_object().unwrap().len(), 2); + } + + #[tokio::test] + async fn refuses_an_unreadable_head_and_a_dominated_write() { + for doc in [ + json!({"version": 2, "channels": {}}), + json!(["not", "a", "store"]), + json!("text"), + ] { + let (client, relay) = setup().await; + seed(&client, &relay, &STARS, doc, 10); + assert!(run(&STARS, Action::List, &client).await.is_err()); + assert!(run(&STARS, set(&[C1], true), &client).await.is_err()); + assert_eq!(published_count(&relay), 1); + } + + let (client, relay) = setup().await; + let mut head = seed(&client, &relay, &MUTES, json!({"version": 1}), 10); + head["content"] = json!("tampered"); + relay.lock().unwrap().events = vec![head]; + let err = run(&MUTES, set(&[C1], true), &client).await.unwrap_err(); + assert!(matches!(err, CliError::Other(_)), "{err:?}"); + + // A head with a forged signature, or from another author, is refused + // even when the relay returns it. + let (client, relay) = setup().await; + let mut forged = seed(&client, &relay, &MUTES, json!({"version": 1}), 10); + forged["sig"] = json!("0".repeat(128)); + relay.lock().unwrap().events = vec![forged]; + relay.lock().unwrap().leaky = true; + let refused = run(&MUTES, set(&[C1], true), &client).await.unwrap_err(); + assert!( + refused.to_string().contains("fails verification"), + "{refused}" + ); + let (client, relay) = setup().await; + let other = BuzzClient::new(String::new(), Keys::generate(), None, None).unwrap(); + seed(&other, &relay, &MUTES, json!({"version": 1}), 10); + relay.lock().unwrap().leaky = true; + let refused = run(&MUTES, set(&[C1], true), &client).await.unwrap_err(); + assert!( + refused.to_string().contains("fails verification"), + "{refused}" + ); + assert_eq!(published_count(&relay), 1); + + // Another author's head is never read as ours. + let (client, relay) = setup().await; + let other = BuzzClient::new(String::new(), Keys::generate(), None, None).unwrap(); + seed( + &other, + &relay, + &MUTES, + json!({"version": 1, "channels": entries(1, "muted", 1)}), + 10, + ); + let list = run(&MUTES, Action::List, &client).await.unwrap(); + assert_eq!(list, json!({"channels": []})); + + relay.lock().unwrap().duplicate = true; + let err = run(&MUTES, set(&[C1], true), &client).await.unwrap_err(); + assert!(matches!(err, CliError::Conflict(_)), "{err:?}"); + } + + #[tokio::test] + async fn keeps_the_newest_500_entries() { + let (client, relay) = setup().await; + seed( + &client, + &relay, + &STARS, + json!({"version": 1, "channels": entries(500, "starred", 100)}), + 10, + ); + run(&STARS, set(&[C1], true), &client).await.unwrap(); + let (_, doc) = published(&client, &relay); + let channels = doc["channels"].as_object().unwrap(); + assert_eq!(channels.len(), 500); + assert!(channels.contains_key(C1)); + assert!(!channels.contains_key(&format!("{:036}", 0))); + assert!(channels.contains_key(&format!("{:036}", 1))); + + // A head over the cap is trimmed the way desktop trims it on read. + let (client, relay) = setup().await; + seed( + &client, + &relay, + &STARS, + json!({"version": 1, "channels": entries(520, "starred", 100)}), + 10, + ); + let list = run(&STARS, Action::List, &client).await.unwrap(); + assert_eq!(list["channels"].as_array().unwrap().len(), 500); + + // An edit that would evict its own entries is refused. + let ids: Vec = (0..501) + .map(|i| format!("ffffffff-0000-4000-8000-{i:012}")) + .collect(); + let err = run(&STARS, Action::Set { ids, on: true }, &client) + .await + .unwrap_err(); + assert!(matches!(err, CliError::Usage(_)), "{err:?}"); + assert_eq!(published_count(&relay), 1); + } + + #[tokio::test] + async fn projects_use_milliseconds_and_keep_every_entry() { + let (client, relay) = setup().await; + let address = format!("30621:{}:my-project", OWNER.to_uppercase()); + let canonical = format!("30621:{OWNER}:my-project"); + let before = now().as_millis() as u64; + run(&PROJECTS, set(&[&address], true), &client) + .await + .unwrap(); + let (event, doc) = published(&client, &relay); + assert!(event + .tags + .iter() + .any(|t| t.as_slice() == ["d", "project-sidebar-membership"])); + assert!(event + .tags + .iter() + .any(|t| t.as_slice() == ["t", "project-sidebar-membership"])); + let entry = &doc["projects"][&canonical]; + assert_eq!(entry["selected"], json!(true)); + assert!(entry["updatedAt"].as_u64().unwrap() >= before); + assert_eq!( + run(&PROJECTS, Action::List, &client).await.unwrap(), + json!({"projects": [canonical]}) + ); + let again = run(&PROJECTS, set(&[&canonical], true), &client) + .await + .unwrap(); + assert_eq!(again["message"], json!("unchanged")); + run(&PROJECTS, set(&[&canonical], false), &client) + .await + .unwrap(); + let (_, doc) = published(&client, &relay); + assert_eq!(doc["projects"][&canonical]["selected"], json!(false)); + + for bad in [ + "my-project", + &format!("30617:{OWNER}:x"), + &format!("30621:{}:x", "zz".repeat(32)), + &format!("30621:{OWNER}:"), + ] { + let err = run(&PROJECTS, set(&[bad], true), &client) + .await + .unwrap_err(); + assert!(matches!(err, CliError::Usage(_)), "{bad}: {err:?}"); + } + + // The legacy address array reads as selected; a document without a + // collection is unreadable rather than empty. + let (client, relay) = setup().await; + seed(&client, &relay, &PROJECTS, json!(["a:b:c", ""]), 10); + let list = run(&PROJECTS, Action::List, &client).await.unwrap(); + assert_eq!(list, json!({"projects": ["a:b:c"]})); + seed(&client, &relay, &PROJECTS, json!({"version": 1}), 20); + assert!(run(&PROJECTS, Action::List, &client).await.is_err()); + + // No eviction: 520 entries survive a write, and the size cap is an error. + let (client, relay) = setup().await; + seed( + &client, + &relay, + &PROJECTS, + json!({"version": 1, "projects": entries(520, "selected", 1)}), + 10, + ); + run(&PROJECTS, set(&[&canonical], true), &client) + .await + .unwrap(); + let (_, doc) = published(&client, &relay); + assert_eq!(doc["projects"].as_object().unwrap().len(), 521); + let (client, relay) = setup().await; + seed( + &client, + &relay, + &PROJECTS, + json!({"version": 1, "projects": entries(800, "selected", 1)}), + 10, + ); + let err = run(&PROJECTS, set(&[&canonical], true), &client) + .await + .unwrap_err(); + assert!(matches!(err, CliError::Usage(_)), "{err:?}"); + assert_eq!(published_count(&relay), 1); + } +} diff --git a/crates/buzz-cli/src/commands/sidebar_sync.rs b/crates/buzz-cli/src/commands/sidebar_sync.rs new file mode 100644 index 0000000000..9e1896ed69 --- /dev/null +++ b/crates/buzz-cli/src/commands/sidebar_sync.rs @@ -0,0 +1,281 @@ +//! Read and write path shared by every synced sidebar document (kind:30078, +//! NIP-44 encrypted to yourself): one verified head read, one signed publish. + +use std::time::SystemTime; + +use buzz_core::kind::KIND_READ_STATE; +use nostr::nips::nip44; +use nostr::{EventBuilder, Kind, Tag, Timestamp}; +use serde_json::{json, Value}; + +use super::parse_write_response; +use crate::client::BuzzClient; +use crate::error::CliError; + +/// Relay event size ceiling shared with desktop's sidebar blobs. +const MAX_CIPHERTEXT_BYTES: usize = 65_535; + +/// One synced document: its `d` tag and the name used in messages. +#[derive(Clone, Copy)] +pub(crate) struct Lane { + pub d_tag: &'static str, + pub label: &'static str, +} + +/// The decoded head. `doc` is `None` when the relay holds no event yet. +pub(crate) struct Head { + pub created_at: u64, + pub doc: Option, +} + +pub(crate) fn now() -> std::time::Duration { + SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap_or_default() +} + +/// Channel ids in the lowercase hyphenated form desktop keys entries by. +pub(crate) fn channel_ids(channels: &[String]) -> Result, CliError> { + channels + .iter() + .map(|channel| { + uuid::Uuid::parse_str(channel.trim()) + .map(|id| id.hyphenated().to_string()) + .map_err(|_| CliError::Usage(format!("invalid channel UUID: {channel}"))) + }) + .collect() +} + +/// The relay's current event for `lane` and this identity, decrypted and +/// passed to `decode` with the event's `created_at`. An absent head is +/// `doc: None`; an unreadable one is an error, never an empty document. The +/// read is pinned to the writer (`consistency: strong`): a write derives a +/// full snapshot from it, and a lagging replica would drop the caller's own +/// just-accepted edits. +pub(crate) async fn fetch_head( + client: &BuzzClient, + lane: Lane, + decode: impl FnOnce(&Value, u64) -> Option, +) -> Result, CliError> { + let me = client.keys().public_key(); + let filter = json!({ + "kinds": [KIND_READ_STATE], + "authors": [me.to_hex()], + "#d": [lane.d_tag], + "limit": 1, + "consistency": "strong", + }); + let raw = client.query(&filter).await?; + let events: Vec = serde_json::from_str(&raw) + .map_err(|e| CliError::Other(format!("failed to parse {} query: {e}", lane.label)))?; + let returned = events.len(); + let head = events + .into_iter() + .filter_map(|event| serde_json::from_value::(event).ok()) + .filter(|event| { + event.pubkey == me + && event.tags.identifier() == Some(lane.d_tag) + && event.verify().is_ok() + }) + .min_by(|a, b| { + b.created_at + .cmp(&a.created_at) + .then_with(|| a.id.to_hex().cmp(&b.id.to_hex())) + }); + let Some(event) = head else { + if returned > 0 { + return Err(CliError::Other(format!( + "the relay returned a {} event that is not yours or fails verification; refusing to overwrite it", + lane.label + ))); + } + return Ok(Head { + created_at: 0, + doc: None, + }); + }; + + let unreadable = |detail: String| { + CliError::Other(format!( + "your {} event {} is unreadable ({detail}); refusing to read or overwrite it", + lane.label, + event.id.to_hex() + )) + }; + let plaintext = nip44::decrypt(client.keys().secret_key(), &me, &event.content) + .map_err(|e| unreadable(e.to_string()))?; + let json: Value = serde_json::from_str(&plaintext).map_err(|e| unreadable(e.to_string()))?; + let created_at = event.created_at.as_secs(); + let doc = decode(&json, created_at).ok_or_else(|| { + unreadable("unsupported format; a newer client may have written it".into()) + })?; + Ok(Head { + created_at, + doc: Some(doc), + }) +} + +/// Encrypts `plaintext` to yourself and publishes it as the new head, one +/// second after `head_created_at` at the earliest. A relay `duplicate:` reply +/// is a conflict. +pub(crate) async fn publish( + client: &BuzzClient, + lane: Lane, + plaintext: String, + head_created_at: u64, +) -> Result { + let keys = client.keys(); + let too_large = || { + CliError::Usage(format!( + "{} would exceed the relay's event size limit", + lane.label + )) + }; + // NIP-44 rejects longer plaintexts outright. + if plaintext.len() > u16::MAX as usize { + return Err(too_large()); + } + let content = nip44::encrypt( + keys.secret_key(), + &keys.public_key(), + plaintext, + nip44::Version::V2, + ) + .map_err(|e| CliError::Other(format!("failed to encrypt {}: {e}", lane.label)))?; + if content.len() > MAX_CIPHERTEXT_BYTES { + return Err(too_large()); + } + let created_at = now().as_secs().max(head_created_at + 1); + let builder = EventBuilder::new(Kind::Custom(KIND_READ_STATE as u16), content) + .tags([Tag::identifier(lane.d_tag), Tag::hashtag(lane.d_tag)]) + .custom_created_at(Timestamp::from(created_at)); + let raw = client.submit_event(client.sign_event(builder)?).await?; + let normalized = parse_write_response( + &raw, + &format!("a newer {} update won; run the command again", lane.label), + )?; + serde_json::from_str(&normalized) + .map_err(|e| CliError::Other(format!("relay response is not JSON: {e}"))) +} + +/// The response of a command that changed nothing and published nothing. +pub(crate) fn unchanged() -> Value { + json!({"event_id": null, "accepted": true, "message": "unchanged"}) +} + +/// Shared fake relay and signing helpers for the sidebar command tests. +#[cfg(test)] +pub(crate) mod test_support { + use super::*; + use axum::{body::Bytes, extract::State, routing::post, Json, Router}; + use nostr::Keys; + use std::sync::{Arc, Mutex}; + use tokio::net::TcpListener; + + #[derive(Default)] + pub struct Relay { + pub events: Vec, + pub duplicate: bool, + /// Answer every query with all stored events, whoever authored them. + pub leaky: bool, + } + pub type Shared = Arc>; + + /// Fake relay: `/query` checks the head filter and returns every stored + /// event matching its author and requested d-tag (more than the relay's + /// `limit`, so head selection is exercised); `/events` stores one. + 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: {filters:?}") + }; + assert_eq!(filter["kinds"], json!([KIND_READ_STATE])); + assert_eq!(filter["limit"], json!(1)); + assert_eq!(filter["consistency"], json!("strong")); + let author = &filter["authors"][0]; + let d_tag = &filter["#d"][0]; + let relay = relay.lock().unwrap(); + let matching: Vec<_> = relay + .events + .iter() + .filter(|e| { + relay.leaky || (&e["pubkey"] == author && &e["tags"][0][1] == d_tag) + }) + .cloned() + .collect(); + Json(json!(matching)) + }), + ) + .route( + "/events", + post(|State(relay): State, body: Bytes| async move { + let event: Value = serde_json::from_slice(&body).unwrap(); + let mut relay = relay.lock().unwrap(); + let message = if relay.duplicate { + "duplicate: newer head" + } else { + "" + }; + if !relay.duplicate { + relay.events.push(event.clone()); + } + Json(json!({"event_id": event["id"], "accepted": true, "message": message})) + }), + ) + .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}") + } + + pub async fn setup() -> (BuzzClient, Shared) { + let relay = Shared::default(); + let url = fake_relay(relay.clone()).await; + let client = BuzzClient::new(url, Keys::generate(), None, None).unwrap(); + (client, relay) + } + + pub fn seed_at( + client: &BuzzClient, + relay: &Shared, + lane: Lane, + plaintext: &str, + created_at: u64, + ) -> Value { + let keys = client.keys(); + 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(lane.d_tag)]) + .custom_created_at(Timestamp::from(created_at)) + .sign_with_keys(keys) + .unwrap(); + let event = serde_json::to_value(event).unwrap(); + relay.lock().unwrap().events.push(event.clone()); + event + } + + /// The decrypted document of the newest published event. + pub fn published(client: &BuzzClient, relay: &Shared) -> (nostr::Event, Value) { + let raw = relay.lock().unwrap().events.last().cloned().unwrap(); + let event: nostr::Event = serde_json::from_value(raw).unwrap(); + let keys = client.keys(); + let plaintext = + nip44::decrypt(keys.secret_key(), &keys.public_key(), &event.content).unwrap(); + (event, serde_json::from_str(&plaintext).unwrap()) + } + + pub fn published_count(relay: &Shared) -> usize { + relay.lock().unwrap().events.len() + } +} diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index df2924eece..2379262013 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -284,7 +284,7 @@ enum Cmd { /// Get and set channel canvas documents #[command(subcommand)] Canvas(CanvasCmd), - /// Organize your own sidebar into channel sections + /// Organize your own sidebar: sections, stars, mutes, and project pins #[command(subcommand)] Sidebar(SidebarCmd), /// Add, remove, and list emoji reactions @@ -1034,6 +1034,52 @@ pub enum SidebarCmd { /// each other, and the last publish wins. #[command(subcommand)] Sections(SidebarSectionsCmd), + /// List, star, and unstar your channels (synced with Buzz desktop) + #[command(subcommand)] + Stars(SidebarChannelFlagCmd), + /// List, mute, and unmute your channels (synced with Buzz desktop) + #[command(subcommand)] + Mutes(SidebarChannelFlagCmd), + /// List, pin, and unpin your sidebar projects (synced with Buzz desktop) + #[command(subcommand)] + Projects(SidebarProjectsCmd), +} + +/// Shared by `sidebar stars` and `sidebar mutes`. +#[derive(Subcommand)] +pub enum SidebarChannelFlagCmd { + /// List the channels that are set + List, + /// Set one or more channels + Add { + /// Channel UUID (repeatable) + #[arg(long = "channel", required = true)] + channels: Vec, + }, + /// Clear one or more channels + Remove { + /// Channel UUID (repeatable) + #[arg(long = "channel", required = true)] + channels: Vec, + }, +} + +#[derive(Subcommand)] +pub enum SidebarProjectsCmd { + /// List the projects pinned to your sidebar + List, + /// Pin one or more projects + Add { + /// Project address `30621::` (repeatable) + #[arg(long = "project", required = true)] + projects: Vec, + }, + /// Unpin one or more projects + Remove { + /// Project address `30621::` (repeatable) + #[arg(long = "project", required = true)] + projects: Vec, + }, } /// `--section` takes a section id, or a section name that is unique.