From b505718db9055a2023b2dfb77913dbeb8d6bfb55 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Fri, 28 Aug 2026 12:42:13 +0000 Subject: [PATCH 1/3] feat(lib): watch and unwatch users (WatchUser 5 / UnwatchUser 6) Adds the server-side watch subscription a buddy list needs: WatchUser (5) asks the server to keep pushing a user's status, UnwatchUser (6) cancels it. The reply to a watch carries the user's current status and share statistics, which merge into the existing per-user snapshot read via Client::user_info. Issue #47 asked for AddUserList (141) / WatchedUserAdded (142) / WatchedUserRemoved (143). Those codes are EnableRoomInvitations, ChangePassword and AddRoomOperator in the Soulseek protocol -- there are no such messages on the wire. Watching is codes 5 and 6, so that is what this implements. - message/server/watch_user.rs: WatchUserHandler (code 5), tolerant of an absent user and of a truncated frame - MessageFactory::build_watch_user / build_unwatch_user - Client::watch_user / unwatch_user / watched_users - The watch reply has no privileged flag, so it preserves whatever an earlier GetUserStatus recorded rather than asserting false --- soulseek-rs-lib/src/actor/server_actor.rs | 30 ++++ soulseek-rs-lib/src/client/mod.rs | 77 ++++++++++ soulseek-rs-lib/src/client/operations.rs | 20 +++ soulseek-rs-lib/src/client/rooms.rs | 40 +++++ soulseek-rs-lib/src/client/tests.rs | 80 ++++++++++ .../src/message/server/message_factory.rs | 28 ++++ soulseek-rs-lib/src/message/server/mod.rs | 2 + .../src/message/server/watch_user.rs | 140 ++++++++++++++++++ 8 files changed, 417 insertions(+) create mode 100644 soulseek-rs-lib/src/message/server/watch_user.rs diff --git a/soulseek-rs-lib/src/actor/server_actor.rs b/soulseek-rs-lib/src/actor/server_actor.rs index 57c1778..5b3f6e2 100644 --- a/soulseek-rs-lib/src/actor/server_actor.rs +++ b/soulseek-rs-lib/src/actor/server_actor.rs @@ -18,6 +18,7 @@ use crate::message::server::ReloggedHandler; use crate::message::server::SayChatroomHandler; use crate::message::server::UserJoinedRoomHandler; use crate::message::server::UserLeftRoomHandler; +use crate::message::server::WatchUserHandler; use crate::message::server::WishListIntervalHandler; use crate::message::server::{ GetUserStatsHandler, GetUserStatusHandler, RoomListHandler, @@ -202,6 +203,17 @@ pub enum ServerMessage { status: u32, privileged: bool, }, + /// A `WatchUser` (code 5) reply: the initial snapshot for a user we just + /// started watching. The stats are absent when the server does not know + /// the username. + WatchedUserReceived { + username: String, + exists: bool, + status: Option, + average_speed: Option, + shared_files: Option, + shared_folders: Option, + }, /// A `GetUserStats` (code 36) reply. UserStatsReceived { username: String, @@ -359,6 +371,7 @@ impl ServerActor { handlers.register_handler(ReloggedHandler); handlers.register_handler(RoomListHandler); handlers.register_handler(GetUserStatusHandler); + handlers.register_handler(WatchUserHandler); handlers.register_handler(GetUserStatsHandler); handlers.register_handler(JoinRoomHandler); handlers.register_handler(LeaveRoomHandler); @@ -464,6 +477,23 @@ impl ServerActor { privileged, }); } + ServerMessage::WatchedUserReceived { + username, + exists, + status, + average_speed, + shared_files, + shared_folders, + } => { + self.forward_to_client(ClientOperation::WatchedUserReceived { + username, + exists, + status, + average_speed, + shared_files, + shared_folders, + }); + } ServerMessage::UserStatsReceived { username, average_speed, diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index 8d11cb4..798e175 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/soulseek-rs-lib/src/client/mod.rs @@ -245,6 +245,15 @@ pub enum ClientOperation { shared_files: u32, shared_folders: u32, }, + /// The server answered `WatchUser` for a user we started watching. + WatchedUserReceived { + username: String, + exists: bool, + status: Option, + average_speed: Option, + shared_files: Option, + shared_folders: Option, + }, PeerConnected(String), /// A search distributed to us by the server; reply if our shares match. IncomingSearch { @@ -328,6 +337,10 @@ pub struct ClientContext { /// What the server has told us about other users, merged across the /// separate status and statistics replies. user_info: HashMap, + /// Users we asked the server to watch (code 5), so it keeps pushing their + /// status changes. Kept so a UI can render the watch list and so an + /// unwatch can be rejected for someone we never watched. + watched_users: HashSet, /// Seconds the server wants between wishlist searches (code 104), once it /// has told us. wishlist_interval: Option, @@ -416,6 +429,7 @@ impl ClientContext { room_events: Vec::new(), room_members: HashMap::new(), user_info: HashMap::new(), + watched_users: HashSet::new(), wishlist_interval: None, privileged_users: HashSet::new(), own_privileges: None, @@ -503,6 +517,69 @@ impl ClientContext { }); } + /// Record that we are now watching `username`. + pub fn add_watched_user(&mut self, username: &str) { + self.watched_users.insert(username.to_string()); + } + + /// Forget a watch, dropping any snapshot we held for that user so a later + /// re-watch reports a fresh answer rather than a stale one. + pub fn remove_watched_user(&mut self, username: &str) { + self.watched_users.remove(username); + self.user_info.remove(username); + } + + /// Everyone we are currently watching, sorted by name. + #[must_use] + pub fn watched_users(&self) -> Vec { + let mut users: Vec = + self.watched_users.iter().cloned().collect(); + users.sort(); + users + } + + /// Record the initial snapshot from a `WatchUser` reply. A username the + /// server does not know carries no stats and is dropped from the watch + /// list, since the server will never push anything for it. + pub fn apply_watched_user( + &mut self, + username: String, + exists: bool, + status: Option, + average_speed: Option, + shared_files: Option, + shared_folders: Option, + ) { + if !exists { + self.watched_users.remove(&username); + return; + } + if let Some(status) = status { + let entry = self + .user_info + .entry(username.clone()) + .or_insert_with(|| UserInfo::pending(username.clone())); + // The watch reply carries no privileged flag, so keep whatever a + // previous GetUserStatus told us rather than asserting `false`. + let privileged = + entry.presence.as_ref().is_some_and(|p| p.privileged); + entry.presence = Some(UserPresence { + status: UserStatus::from_code(status), + privileged, + }); + } + if let (Some(average_speed), Some(shared_files), Some(shared_folders)) = + (average_speed, shared_files, shared_folders) + { + self.apply_user_stats( + username, + average_speed, + shared_files, + shared_folders, + ); + } + } + /// What the server has said about `username` so far. #[must_use] pub fn user_info(&self, username: &str) -> Option { diff --git a/soulseek-rs-lib/src/client/operations.rs b/soulseek-rs-lib/src/client/operations.rs index d80758f..e5baf46 100644 --- a/soulseek-rs-lib/src/client/operations.rs +++ b/soulseek-rs-lib/src/client/operations.rs @@ -473,6 +473,26 @@ impl Client { error!("[client] UserStatsReceived write: {}", e); } }, + ClientOperation::WatchedUserReceived { + username, + exists, + status, + average_speed, + shared_files, + shared_folders, + } => match client_context.write_safe() { + Ok(mut ctx) => ctx.apply_watched_user( + username, + exists, + status, + average_speed, + shared_files, + shared_folders, + ), + Err(e) => { + error!("[client] WatchedUserReceived write: {}", e); + } + }, ClientOperation::RoomEvent(event) => { match client_context.write_safe() { Ok(mut ctx) => ctx.apply_room_event(event), diff --git a/soulseek-rs-lib/src/client/rooms.rs b/soulseek-rs-lib/src/client/rooms.rs index 8afc871..5b5361c 100644 --- a/soulseek-rs-lib/src/client/rooms.rs +++ b/soulseek-rs-lib/src/client/rooms.rs @@ -138,6 +138,46 @@ impl Client { } } + /// Ask the server to watch `username` (code 5). + /// + /// The server replies with that user's current status and share + /// statistics — readable with [`Client::user_info`] — and then keeps + /// pushing their status changes until [`Client::unwatch_user`]. Watching + /// a user the server does not know is answered with a "does not exist" + /// flag, and that user is dropped from [`Client::watched_users`]. + /// + /// # Errors + /// Returns [`SoulseekRs::NotConnected`] if the client is not connected. + pub fn watch_user(&self, username: &str) -> Result<()> { + use crate::message::server::MessageFactory; + self.send_server_message(MessageFactory::build_watch_user(username))?; + self.context.write_safe()?.add_watched_user(username); + Ok(()) + } + + /// Stop watching `username` (code 6) and drop what we knew about them. + /// + /// # Errors + /// Returns [`SoulseekRs::NotConnected`] if the client is not connected. + pub fn unwatch_user(&self, username: &str) -> Result<()> { + use crate::message::server::MessageFactory; + self.send_server_message(MessageFactory::build_unwatch_user(username))?; + self.context.write_safe()?.remove_watched_user(username); + Ok(()) + } + + /// Everyone this client is currently watching, sorted by name. + #[must_use] + pub fn watched_users(&self) -> Vec { + match self.context.read_safe() { + Ok(ctx) => ctx.watched_users(), + Err(e) => { + error!("[client] watched_users: {}", e); + Vec::new() + } + } + } + /// Who is currently in `room`, sorted by name. /// /// The roster is built from the membership the server sends when diff --git a/soulseek-rs-lib/src/client/tests.rs b/soulseek-rs-lib/src/client/tests.rs index fd90b76..13409b1 100644 --- a/soulseek-rs-lib/src/client/tests.rs +++ b/soulseek-rs-lib/src/client/tests.rs @@ -622,3 +622,83 @@ fn a_replayed_transfer_response_does_not_start_a_second_transfer() { "a replayed TransferResponse must not dial the peer again" ); } + +#[test] +fn a_watch_reply_fills_both_halves_of_the_snapshot() { + let mut context = ClientContext::new(); + context.add_watched_user("alice"); + context.apply_watched_user( + "alice".to_string(), + true, + Some(2), + Some(1024), + Some(20), + Some(3), + ); + + let info = context.user_info("alice").expect("a snapshot"); + assert!(info.is_complete(), "a watch reply carries status and stats"); + assert_eq!(info.stats.map(|s| s.average_speed), Some(1024)); + assert_eq!(context.watched_users(), vec!["alice".to_string()]); +} + +#[test] +fn a_watch_reply_keeps_a_privileged_flag_it_cannot_carry() { + // WatchUser has no privileged field, so it must not overwrite what + // GetUserStatus already told us with a fabricated `false`. + let mut context = ClientContext::new(); + context.apply_user_status("bob".to_string(), 1, true); + context.apply_watched_user( + "bob".to_string(), + true, + Some(2), + Some(1), + Some(2), + Some(3), + ); + + let info = context.user_info("bob").expect("a snapshot"); + assert_eq!(info.presence.map(|p| p.privileged), Some(true)); +} + +#[test] +fn watching_an_unknown_user_drops_them_from_the_list() { + // The server will never push status for a name it does not know, so + // keeping it in the watch list would show a permanently blank row. + let mut context = ClientContext::new(); + context.add_watched_user("ghost"); + context.apply_watched_user( + "ghost".to_string(), + false, + None, + None, + None, + None, + ); + + assert!(context.watched_users().is_empty()); + assert!(context.user_info("ghost").is_none()); +} + +#[test] +fn unwatching_forgets_the_users_snapshot() { + let mut context = ClientContext::new(); + context.add_watched_user("alice"); + context.apply_user_status("alice".to_string(), 2, false); + + context.remove_watched_user("alice"); + assert!(context.watched_users().is_empty()); + assert!( + context.user_info("alice").is_none(), + "a later re-watch must report a fresh answer" + ); +} + +#[test] +fn watched_users_are_listed_in_a_stable_order() { + let mut context = ClientContext::new(); + context.add_watched_user("carol"); + context.add_watched_user("alice"); + context.add_watched_user("bob"); + assert_eq!(context.watched_users(), vec!["alice", "bob", "carol"]); +} diff --git a/soulseek-rs-lib/src/message/server/message_factory.rs b/soulseek-rs-lib/src/message/server/message_factory.rs index cb09a27..4ce3256 100644 --- a/soulseek-rs-lib/src/message/server/message_factory.rs +++ b/soulseek-rs-lib/src/message/server/message_factory.rs @@ -152,6 +152,20 @@ impl MessageFactory { .clone() } + /// Watch `username` (server code 5): the server replies with their + /// current status and share statistics, then pushes every later status + /// change as a `GetUserStatus` (code 7) until we unwatch them. + #[must_use] + pub fn build_watch_user(username: &str) -> Message { + Message::new().write_int32(5).write_string(username).clone() + } + + /// Stop watching `username` (server code 6). The server sends no reply. + #[must_use] + pub fn build_unwatch_user(username: &str) -> Message { + Message::new().write_int32(6).write_string(username).clone() + } + /// Ask the server (code 64) for the list of public chat rooms. #[must_use] pub fn build_room_list_request() -> Message { @@ -424,3 +438,17 @@ fn test_build_file_search_message() { .to_vec(); assert_eq!(expect, message.get_data()); } + +#[test] +fn test_build_watch_user() { + let message = MessageFactory::build_watch_user("bob"); + let expect: Vec = [5, 0, 0, 0, 3, 0, 0, 0, b'b', b'o', b'b'].to_vec(); + assert_eq!(expect, message.get_data()); +} + +#[test] +fn test_build_unwatch_user() { + let message = MessageFactory::build_unwatch_user("bob"); + let expect: Vec = [6, 0, 0, 0, 3, 0, 0, 0, b'b', b'o', b'b'].to_vec(); + assert_eq!(expect, message.get_data()); +} diff --git a/soulseek-rs-lib/src/message/server/mod.rs b/soulseek-rs-lib/src/message/server/mod.rs index 104082b..b2d93b7 100644 --- a/soulseek-rs-lib/src/message/server/mod.rs +++ b/soulseek-rs-lib/src/message/server/mod.rs @@ -16,6 +16,7 @@ mod say_chatroom; mod user_info; mod user_joined_room; mod user_left_room; +mod watch_user; mod wish_list_interval; pub use connect_to_peer::ConnectToPeerHandler; @@ -36,4 +37,5 @@ pub use say_chatroom::SayChatroomHandler; pub use user_info::{GetUserStatsHandler, GetUserStatusHandler}; pub use user_joined_room::UserJoinedRoomHandler; pub use user_left_room::UserLeftRoomHandler; +pub use watch_user::WatchUserHandler; pub use wish_list_interval::WishListIntervalHandler; diff --git a/soulseek-rs-lib/src/message/server/watch_user.rs b/soulseek-rs-lib/src/message/server/watch_user.rs new file mode 100644 index 0000000..ecbd50d --- /dev/null +++ b/soulseek-rs-lib/src/message/server/watch_user.rs @@ -0,0 +1,140 @@ +//! `WatchUser` (code 5): subscribe to a user's status and share statistics. +//! +//! Watching is how a client keeps a buddy list current. After a `WatchUser` +//! request the server answers on the same code with the user's current +//! presence and stats, and from then on pushes every later change as an +//! ordinary `GetUserStatus` (code 7). `UnwatchUser` (code 6) cancels the +//! subscription and has no reply. +//! +//! Note on codes: the protocol has no `AddUserList`/`WatchedUserAdded` +//! /`WatchedUserRemoved` messages — 141/142/143 are `EnableRoomInvitations`, +//! `ChangePassword` and `AddRoomOperator`. Watching is 5/6, per the reference +//! protocol documentation. + +use crate::actor::server_actor::ServerMessage; +use crate::message::{Message, MessageHandler}; +use std::sync::mpsc::Sender; + +/// Receives the `WatchUser` reply (code 5): `username`, `exists`, and — when +/// the user exists — status plus the same share statistics `GetUserStats` +/// carries. +pub struct WatchUserHandler; + +impl MessageHandler for WatchUserHandler { + fn get_code(&self) -> u8 { + 5 + } + + fn handle(&self, message: &mut Message, sender: Sender) { + let username = message.read_string(); + let exists = message.read_bool(); + let watched = if exists { + let status = message.read_int32(); + let average_speed = message.read_int32(); + let _upload_number = message.read_int64(); // obsolete + let shared_files = message.read_int32(); + let shared_folders = message.read_int32(); + ServerMessage::WatchedUserReceived { + username, + exists, + status: Some(status), + average_speed: Some(average_speed), + shared_files: Some(shared_files), + shared_folders: Some(shared_folders), + } + } else { + // An unknown username: the server stops after the flag, so there + // is nothing further to read and nothing to record. + ServerMessage::WatchedUserReceived { + username, + exists, + status: None, + average_speed: None, + shared_files: None, + shared_folders: None, + } + }; + let _ = sender.send(watched); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::message::framed; + + #[test] + fn existing_user_carries_status_and_stats() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = framed(|m| { + m.write_string("alice") + .write_int8(1) // exists + .write_int32(2) // status: online + .write_int32(1024) // average speed + .write_int64(0) // obsolete upload number + .write_int32(4321) // shared files + .write_int32(77); // shared folders + }); + + WatchUserHandler.handle(&mut message, tx); + match rx.try_recv() { + Ok(ServerMessage::WatchedUserReceived { + username, + exists, + status, + average_speed, + shared_files, + shared_folders, + }) => { + assert_eq!(username, "alice"); + assert!(exists); + assert_eq!(status, Some(2)); + assert_eq!(average_speed, Some(1024)); + assert_eq!(shared_files, Some(4321)); + assert_eq!(shared_folders, Some(77)); + } + other => panic!("unexpected: {other:?}"), + } + } + + #[test] + fn unknown_user_reports_absence_without_stats() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = framed(|m| { + m.write_string("ghost").write_int8(0); + }); + + WatchUserHandler.handle(&mut message, tx); + match rx.try_recv() { + Ok(ServerMessage::WatchedUserReceived { + username, + exists, + status, + shared_files, + .. + }) => { + assert_eq!(username, "ghost"); + assert!(!exists); + assert_eq!(status, None); + assert_eq!(shared_files, None); + } + other => panic!("unexpected: {other:?}"), + } + } + + #[test] + fn truncated_payload_does_not_panic() { + // exists = 1 but the stats were cut off: the bounds-checked readers + // must yield zeros rather than panicking on a hostile frame. + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = framed(|m| { + m.write_string("alice").write_int8(1).write_int32(2); + }); + + WatchUserHandler.handle(&mut message, tx); + assert!(matches!( + rx.try_recv(), + Ok(ServerMessage::WatchedUserReceived { .. }) + )); + } +} From a91d35b672eadcba96ff445028e988cd29db5586 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Fri, 28 Aug 2026 13:22:50 +0000 Subject: [PATCH 2/3] feat(daemon): expose watching over the RPC surface and the CLI The watch subscription was reachable only from the library. It is most useful against a daemon -- that is the session that stays connected long enough to receive the server's status pushes -- so it needs the wire. - proto: user.watch / user.unwatch / user.watched, with a Watched response - SessionApi gains watch_user / unwatch_user / watched_users, so a local and a remote session are still interchangeable above the seam - CLI: `watch add`, `watch remove`, `watch list`, documented in SKILL.md including the caveat that a one-shot run's watch ends with the run - docs/openrpc.json regenerated from the wire types --- soulseek-rs/docs/openrpc.json | 67 +++++++++++++++++++ soulseek-rs/src/SKILL.md | 7 ++ soulseek-rs/src/api.rs | 18 +++++ soulseek-rs/src/cli.rs | 24 +++++++ soulseek-rs/src/commands/mod.rs | 1 + soulseek-rs/src/commands/peer.rs | 39 ++++++++++- soulseek-rs/src/daemon/openrpc.rs | 22 +++++-- soulseek-rs/src/daemon/proto/method.rs | 7 ++ soulseek-rs/src/daemon/proto/mod.rs | 3 + soulseek-rs/src/daemon/server/mod.rs | 91 +++++++++++++++++++++++++- soulseek-rs/src/output.rs | 19 ++++++ soulseek-rs/src/remote/mod.rs | 26 +++++++- soulseek-rs/src/ui/main_tui/mod.rs | 9 +++ 13 files changed, 322 insertions(+), 11 deletions(-) diff --git a/soulseek-rs/docs/openrpc.json b/soulseek-rs/docs/openrpc.json index 213bfc7..46e5886 100644 --- a/soulseek-rs/docs/openrpc.json +++ b/soulseek-rs/docs/openrpc.json @@ -1216,6 +1216,21 @@ "shared_folders" ], "type": "object" + }, + "Watched": { + "description": "Everyone this session is watching, sorted by name.", + "properties": { + "users": { + "items": { + "type": "string" + }, + "type": "array" + } + }, + "required": [ + "users" + ], + "type": "object" } } }, @@ -1875,6 +1890,58 @@ }, "summary": "What the server has said about a user so far." }, + { + "name": "user.watch", + "paramStructure": "by-name", + "params": [ + { + "name": "username", + "required": true, + "schema": { + "type": "string" + } + } + ], + "result": { + "name": "result", + "schema": { + "$ref": "#/components/schemas/Ack" + } + }, + "summary": "Watch a user, so the server keeps pushing their status." + }, + { + "name": "user.unwatch", + "paramStructure": "by-name", + "params": [ + { + "name": "username", + "required": true, + "schema": { + "type": "string" + } + } + ], + "result": { + "name": "result", + "schema": { + "$ref": "#/components/schemas/Ack" + } + }, + "summary": "Stop watching a user." + }, + { + "name": "user.watched", + "paramStructure": "by-name", + "params": [], + "result": { + "name": "result", + "schema": { + "$ref": "#/components/schemas/Watched" + } + }, + "summary": "Everyone this session is watching." + }, { "name": "shares.status", "paramStructure": "by-name", diff --git a/soulseek-rs/src/SKILL.md b/soulseek-rs/src/SKILL.md index 44594b3..2ec67ae 100644 --- a/soulseek-rs/src/SKILL.md +++ b/soulseek-rs/src/SKILL.md @@ -133,6 +133,13 @@ wishlist later rather than repeating the same search. be `null` when the server has not answered that half — `null` means unknown, not offline. +**`watch add `** / **`watch remove `** — prints nothing; exit 0 is +the confirmation. **`watch list`** — one object per watched user: `user`, +`status`. A watch lasts only as long as the session holding it, so these are +worth using against a running daemon (`--daemon`), where the connection stays +up and keeps receiving the server's status pushes. A one-shot run's watch ends +with the run. + **`whoami`** — one object: `user`, `server`, `listening`, `listen_port`, `shared_folders`, `shared_files`, `download_dir`, `privilege_seconds`. The cheapest way to prove the credentials and connection work. diff --git a/soulseek-rs/src/api.rs b/soulseek-rs/src/api.rs index 4006a84..bd39684 100644 --- a/soulseek-rs/src/api.rs +++ b/soulseek-rs/src/api.rs @@ -140,6 +140,12 @@ pub trait SessionApi: Send + Sync { fn request_user_info(&self, username: &str) -> Result<()>; fn user_info(&self, username: &str) -> Option; + /// Ask the server to keep pushing `username`'s status until unwatched. + fn watch_user(&self, username: &str) -> Result<()>; + fn unwatch_user(&self, username: &str) -> Result<()>; + /// Everyone this session is watching, sorted by name. + fn watched_users(&self) -> Vec; + /// The conversation this session already knows about. /// /// Empty locally, where the TUI's own snapshot on disk is the record. A @@ -364,6 +370,18 @@ impl SessionApi for Client { Self::user_info(self, username) } + fn watch_user(&self, username: &str) -> Result<()> { + Self::watch_user(self, username) + } + + fn unwatch_user(&self, username: &str) -> Result<()> { + Self::unwatch_user(self, username) + } + + fn watched_users(&self) -> Vec { + Self::watched_users(self) + } + fn shared_counts(&self) -> (u32, u32) { Self::shared_counts(self) } diff --git a/soulseek-rs/src/cli.rs b/soulseek-rs/src/cli.rs index 7b3f90c..e7ae4d6 100644 --- a/soulseek-rs/src/cli.rs +++ b/soulseek-rs/src/cli.rs @@ -207,6 +207,10 @@ pub enum Commands { /// What the server knows about another user User(UserArgs), + /// Keep a daemon updated about users you care about + #[command(subcommand)] + Watch(WatchCommand), + /// Confirm the credentials and connection work Whoami, @@ -571,6 +575,26 @@ pub struct WishRunArgs { pub sort: SortKey, } +/// Watching only lasts as long as the session doing it, so these are aimed at +/// a running daemon: it stays connected and keeps receiving the status pushes. +#[derive(Subcommand, Debug)] +pub enum WatchCommand { + /// Start watching a user + Add { + /// The user to watch + user: String, + }, + + /// Stop watching a user + Remove { + /// The user to stop watching + user: String, + }, + + /// List the users being watched + List, +} + #[derive(Subcommand, Debug)] pub enum RoomCommand { /// List public rooms and their user counts diff --git a/soulseek-rs/src/commands/mod.rs b/soulseek-rs/src/commands/mod.rs index 28161f7..f678aae 100644 --- a/soulseek-rs/src/commands/mod.rs +++ b/soulseek-rs/src/commands/mod.rs @@ -309,6 +309,7 @@ fn dispatch(ctx: &Ctx, command: Commands) -> CliResult { social::message_read(ctx, listen_span(duration, follow)) } Commands::User(args) => peer::user(ctx, &args), + Commands::Watch(command) => peer::watch(ctx, &command), Commands::Whoami => peer::whoami(ctx), // main.rs runs these itself: some need no account, the rest need the // config store in scope. The arms only keep the match exhaustive. diff --git a/soulseek-rs/src/commands/peer.rs b/soulseek-rs/src/commands/peer.rs index 4101175..ba6218c 100644 --- a/soulseek-rs/src/commands/peer.rs +++ b/soulseek-rs/src/commands/peer.rs @@ -1,9 +1,10 @@ //! Being a peer on the network: `serve`, `whoami`, and `user`. use super::{Ctx, Session}; -use crate::cli::{ServeArgs, UserArgs}; +use crate::cli::{ServeArgs, UserArgs, WatchCommand}; use crate::output::{ - CliError, CliResult, UploadRecord, UserRecord, WhoamiRecord, + CliError, CliResult, UploadRecord, UserRecord, WatchedUserRecord, + WhoamiRecord, }; use soulseek_rs::types::{UploadInfo, UploadStatus, UserInfo}; use std::collections::HashMap; @@ -94,6 +95,40 @@ pub fn user(ctx: &Ctx, args: &UserArgs) -> CliResult { Ok(()) } +/// Start or stop watching a user, or list who is being watched. +/// +/// A watch lives in the session that made it, so `add` against a one-shot CLI +/// run ends with that run. Pointed at a daemon it persists for that daemon's +/// lifetime, which is the point of the command. +pub fn watch(ctx: &Ctx, command: &WatchCommand) -> CliResult { + let session = Session::open(ctx)?; + match command { + WatchCommand::Add { user } => { + session.client.watch_user(user).map_err(|e| { + CliError::connection(format!("cannot watch {user}: {e}")) + })?; + ctx.out.status(&format!("watching {user}")); + } + WatchCommand::Remove { user } => { + session.client.unwatch_user(user).map_err(|e| { + CliError::connection(format!("cannot unwatch {user}: {e}")) + })?; + ctx.out.status(&format!("no longer watching {user}")); + } + WatchCommand::List => { + for user in session.client.watched_users() { + let status = session + .client + .user_info(&user) + .and_then(|info| info.presence) + .map(|p| p.status.to_string()); + ctx.out.emit(&WatchedUserRecord { user, status }); + } + } + } + Ok(()) +} + /// Stay online sharing files, and — with `--wishlist` — re-run the standing /// searches on the interval the server sets while doing so. pub fn serve( diff --git a/soulseek-rs/src/daemon/openrpc.rs b/soulseek-rs/src/daemon/openrpc.rs index e58b9c4..a489826 100644 --- a/soulseek-rs/src/daemon/openrpc.rs +++ b/soulseek-rs/src/daemon/openrpc.rs @@ -15,6 +15,7 @@ use super::proto::{ Method, QueryParams, RoomEventDto, RoomRef, SayParams, SearchResults, Searches, Seconds, SessionLossEvent, SharesStatus, SlotsParams, TransferRef, UploadInfoDto, Uploads, UserMessageDto, UserRef, UserResult, + Watched, }; use schemars::{JsonSchema, SchemaGenerator, generate::SchemaSettings}; use serde_json::{Value, json}; @@ -51,9 +52,11 @@ fn params_schema(method: Method, generator: &mut SchemaGenerator) -> Value { } Method::RoomSay => schema_of::(generator), Method::MessageSend => schema_of::(generator), - Method::BrowseUser | Method::UserRequest | Method::UserInfoOf => { - schema_of::(generator) - } + Method::BrowseUser + | Method::UserRequest + | Method::UserInfoOf + | Method::UserWatch + | Method::UserUnwatch => schema_of::(generator), Method::SharesSet => schema_of::(generator), // Methods that take no parameters. Method::DaemonStatus @@ -68,7 +71,8 @@ fn params_schema(method: Method, generator: &mut SchemaGenerator) -> Value { | Method::MessageHistory | Method::SearchList | Method::SharesStatusOf - | Method::SharesReindex => json!({ "type": "null" }), + | Method::SharesReindex + | Method::UserWatched => json!({ "type": "null" }), } } @@ -88,6 +92,7 @@ fn result_schema(method: Method, generator: &mut SchemaGenerator) -> Value { Method::UploadList => schema_of::(generator), Method::PrivilegesOwn => schema_of::(generator), Method::RoomMembers => schema_of::(generator), + Method::UserWatched => schema_of::(generator), Method::MessageHistory => schema_of::(generator), Method::UserInfoOf => schema_of::(generator), Method::SharesStatusOf | Method::SharesSet | Method::SharesReindex => { @@ -112,7 +117,9 @@ fn result_schema(method: Method, generator: &mut SchemaGenerator) -> Value { | Method::MessageSend | Method::BrowseUser | Method::SearchForget - | Method::UserRequest => schema_of::(generator), + | Method::UserRequest + | Method::UserWatch + | Method::UserUnwatch => schema_of::(generator), } } @@ -172,6 +179,11 @@ fn summary(method: Method) -> &'static str { Method::RoomLeave => "Leave a chat room.", Method::RoomSay => "Post a message to a room.", Method::RoomMembers => "Who is currently in a room.", + Method::UserWatch => { + "Watch a user, so the server keeps pushing their status." + } + Method::UserUnwatch => "Stop watching a user.", + Method::UserWatched => "Everyone this session is watching.", Method::MessageSend => "Send a private message.", Method::MessageHistory => { "Private messages the daemon collected, including while nobody was attached." diff --git a/soulseek-rs/src/daemon/proto/method.rs b/soulseek-rs/src/daemon/proto/method.rs index 96a69c2..3b71247 100644 --- a/soulseek-rs/src/daemon/proto/method.rs +++ b/soulseek-rs/src/daemon/proto/method.rs @@ -139,6 +139,13 @@ pub struct Members { pub users: Vec, } +/// Everyone this session is watching, sorted by name. +#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)] +#[cfg_attr(test, derive(schemars::JsonSchema))] +pub struct Watched { + pub users: Vec, +} + /// One line of chat history. Unlike a live [`UserMessageDto`] this carries a /// direction, because a conversation the daemon collected includes what this /// account sent as well as what it received. diff --git a/soulseek-rs/src/daemon/proto/mod.rs b/soulseek-rs/src/daemon/proto/mod.rs index f91b83c..7bd337a 100644 --- a/soulseek-rs/src/daemon/proto/mod.rs +++ b/soulseek-rs/src/daemon/proto/mod.rs @@ -106,6 +106,9 @@ wire_vocabulary! { BrowseUser => "browse.user", UserRequest => "user.request", UserInfoOf => "user.info", + UserWatch => "user.watch", + UserUnwatch => "user.unwatch", + UserWatched => "user.watched", SharesStatusOf => "shares.status", SharesSet => "shares.set", SharesReindex => "shares.reindex", diff --git a/soulseek-rs/src/daemon/server/mod.rs b/soulseek-rs/src/daemon/server/mod.rs index 718a23a..0bd8470 100644 --- a/soulseek-rs/src/daemon/server/mod.rs +++ b/soulseek-rs/src/daemon/server/mod.rs @@ -13,7 +13,7 @@ use super::proto::{ MessageParams, Messages, Method, OPENRPC, PROTOCOL_VERSION, QueryParams, RoomRef, RpcError, SayParams, SearchResultDto, SearchResults, SearchSummary, Searches, Seconds, SharesStatus, SlotsParams, TransferRef, - UploadInfoDto, Uploads, UserInfoDto, UserRef, UserResult, + UploadInfoDto, Uploads, UserInfoDto, UserRef, UserResult, Watched, }; use crate::api::SessionApi; use crate::output::Exit; @@ -306,6 +306,24 @@ impl Daemon { }) } + Method::UserWatch => { + let user: UserRef = parse(params)?; + self.session.watch_user(&user.username).map_err(|e| { + RpcError::application(Exit::Connection, e.to_string()) + })?; + ok(Ack::OK) + } + Method::UserUnwatch => { + let user: UserRef = parse(params)?; + self.session.unwatch_user(&user.username).map_err(|e| { + RpcError::application(Exit::Connection, e.to_string()) + })?; + ok(Ack::OK) + } + Method::UserWatched => ok(Watched { + users: self.session.watched_users(), + }), + Method::SharesStatusOf => ok(self.shares()), Method::SharesSet => { let directories: DirectoriesParams = parse(params)?; @@ -593,6 +611,49 @@ mod tests { ); } + #[test] + fn a_watch_is_readable_back_and_survives_until_unwatched() { + // The three methods are only useful together: what `user.watch` + // records, `user.watched` must report, and `user.unwatch` must undo. + let daemon = daemon_with_token("t"); + let user = |name: &str| json!({ "username": name }); + + daemon + .call(Method::UserWatch, user("alice")) + .expect("watch"); + daemon.call(Method::UserWatch, user("bob")).expect("watch"); + let watched = daemon + .call(Method::UserWatched, Value::Null) + .expect("watched"); + assert_eq!(watched["users"], json!(["alice", "bob"])); + + daemon + .call(Method::UserUnwatch, user("alice")) + .expect("unwatch"); + let watched = daemon + .call(Method::UserWatched, Value::Null) + .expect("watched"); + assert_eq!( + watched["users"], + json!(["bob"]), + "unwatching one must not clear the rest" + ); + } + + #[test] + fn watching_the_same_user_twice_lists_them_once() { + let daemon = daemon_with_token("t"); + let alice = json!({ "username": "alice" }); + daemon + .call(Method::UserWatch, alice.clone()) + .expect("watch"); + daemon.call(Method::UserWatch, alice).expect("watch"); + let watched = daemon + .call(Method::UserWatched, Value::Null) + .expect("watched"); + assert_eq!(watched["users"], json!(["alice"])); + } + #[test] fn stop_is_recorded_so_the_daemon_can_wind_down() { let daemon = daemon_with_token("t"); @@ -605,11 +666,17 @@ mod tests { // A session that answers nothing: enough to exercise dispatch, parameter // validation, and the auth gate without a Soulseek server. - struct SilentSession; + /// A session that answers nothing, except that it remembers watches: + /// the watch methods are only meaningful if what one call records the + /// next call can read back. + #[derive(Default)] + struct SilentSession { + watched: Mutex>, + } fn daemon_with_token(token: &str) -> Daemon { Daemon { - session: Arc::new(SilentSession), + session: Arc::new(SilentSession::default()), hub: Arc::new(Hub::new()), browses: Arc::new(PendingBrowses::default()), downloads: Arc::new(Mutex::new(Vec::new())), @@ -792,6 +859,24 @@ mod tests { ) -> soulseek_rs::Result<()> { Ok(()) } + fn watch_user(&self, username: &str) -> soulseek_rs::Result<()> { + let mut watched = self.watched.lock().expect("watch list"); + if !watched.iter().any(|u| u == username) { + watched.push(username.to_string()); + watched.sort(); + } + Ok(()) + } + fn unwatch_user(&self, username: &str) -> soulseek_rs::Result<()> { + self.watched + .lock() + .expect("watch list") + .retain(|u| u != username); + Ok(()) + } + fn watched_users(&self) -> Vec { + self.watched.lock().expect("watch list").clone() + } fn user_info(&self, _username: &str) -> Option { None } diff --git a/soulseek-rs/src/output.rs b/soulseek-rs/src/output.rs index d50a853..d8fe980 100644 --- a/soulseek-rs/src/output.rs +++ b/soulseek-rs/src/output.rs @@ -383,6 +383,25 @@ impl Record for RoomMemberRecord { } } +/// One user this session is watching, with whatever the server has already +/// said about them. The status is `None` until the watch reply lands. +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct WatchedUserRecord { + pub user: String, + /// `online`, `away`, or `offline`. + pub status: Option, +} + +impl Record for WatchedUserRecord { + fn text(&self) -> String { + format!( + "{}\t{}", + sanitize(&self.user), + self.status.clone().unwrap_or_else(|| "-".to_string()) + ) + } +} + /// One configured share folder. #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub struct ShareRecord { diff --git a/soulseek-rs/src/remote/mod.rs b/soulseek-rs/src/remote/mod.rs index e236d05..24277f8 100644 --- a/soulseek-rs/src/remote/mod.rs +++ b/soulseek-rs/src/remote/mod.rs @@ -15,7 +15,7 @@ use crate::daemon::proto::{ Members, MessageParams, Messages, Method, PROTOCOL_VERSION, QueryParams, Request, Response, RoomRef, RpcError, SayParams, SearchResults, Searches, Seconds, SharesStatus, SlotsParams, TransferRef, Uploads, UserRef, - UserResult, + UserResult, Watched, }; use mirror::Mirror; use serde::de::DeserializeOwned; @@ -757,6 +757,30 @@ impl SessionApi for RemoteSession { ) } + fn watch_user(&self, username: &str) -> Result<()> { + self.tell( + Method::UserWatch, + UserRef { + username: username.to_string(), + }, + ) + } + + fn unwatch_user(&self, username: &str) -> Result<()> { + self.tell( + Method::UserUnwatch, + UserRef { + username: username.to_string(), + }, + ) + } + + fn watched_users(&self) -> Vec { + self.request::<_, Watched>(Method::UserWatched, ()) + .map(|watched| watched.users) + .unwrap_or_default() + } + fn user_info(&self, username: &str) -> Option { self.request::<_, UserResult>( Method::UserInfoOf, diff --git a/soulseek-rs/src/ui/main_tui/mod.rs b/soulseek-rs/src/ui/main_tui/mod.rs index 0b80188..1fe8624 100644 --- a/soulseek-rs/src/ui/main_tui/mod.rs +++ b/soulseek-rs/src/ui/main_tui/mod.rs @@ -489,6 +489,15 @@ mod tests { ) -> soulseek_rs::Result<()> { Ok(()) } + fn watch_user(&self, _username: &str) -> soulseek_rs::Result<()> { + Ok(()) + } + fn unwatch_user(&self, _username: &str) -> soulseek_rs::Result<()> { + Ok(()) + } + fn watched_users(&self) -> Vec { + Vec::new() + } fn user_info(&self, _username: &str) -> Option { None } From 9c1a6d317e0cf256defb5a83cd0b1e983ca55206 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Fri, 28 Aug 2026 14:42:44 +0000 Subject: [PATCH 3/3] test(e2e): cover watching against soulfind, and read the reply as written The e2e job proved nothing about watching: it passed because the suite has no test that exercises it. Two now run against a real soulfind. - a watch reply carries status and share statistics in one message, checked against a subject sharing a known two files, and unwatching drops both the watch and the snapshot - a username the server does not know is answered with an absence flag and must not stay in the watch list Also reads the two obsolete fields of the reply as the two 32-bit values the server writes, matching soulfind and the sibling GetUserStats (36) handler. The byte count is identical, so this is not a behaviour change -- but the handler now says what the wire actually carries. --- .../src/message/server/watch_user.rs | 10 +- soulseek-rs-lib/tests/e2e.rs | 98 +++++++++++++++++++ 2 files changed, 106 insertions(+), 2 deletions(-) diff --git a/soulseek-rs-lib/src/message/server/watch_user.rs b/soulseek-rs-lib/src/message/server/watch_user.rs index ecbd50d..1499189 100644 --- a/soulseek-rs-lib/src/message/server/watch_user.rs +++ b/soulseek-rs-lib/src/message/server/watch_user.rs @@ -31,7 +31,12 @@ impl MessageHandler for WatchUserHandler { let watched = if exists { let status = message.read_int32(); let average_speed = message.read_int32(); - let _upload_number = message.read_int64(); // obsolete + // Two obsolete 32-bit fields, exactly as GetUserStats (36) + // carries them -- not one 64-bit value. The byte count is the + // same either way, but reading them as the server writes them + // keeps this handler and its sibling honest. + let _upload_number = message.read_int32(); // obsolete, always 0 + let _unknown = message.read_int32(); // obsolete, always 0 let shared_files = message.read_int32(); let shared_folders = message.read_int32(); ServerMessage::WatchedUserReceived { @@ -71,7 +76,8 @@ mod tests { .write_int8(1) // exists .write_int32(2) // status: online .write_int32(1024) // average speed - .write_int64(0) // obsolete upload number + .write_int32(0) // obsolete upload number + .write_int32(0) // obsolete unknown .write_int32(4321) // shared files .write_int32(77); // shared folders }); diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index 072bc21..430a4ce 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -2622,3 +2622,101 @@ fn a_peer_that_hangs_up_its_control_connection_still_delivers_the_file() { let _ = std::fs::remove_dir_all(&download_dir); } + +#[test] +fn watching_a_user_returns_their_status_and_share_counts() { + let server = server_or_skip!(); + + // A subject with a known share, so the watch reply's statistics are + // checkable rather than merely present. + let share_dir = unique_download_dir().join("watch_share"); + std::fs::create_dir_all(share_dir.join("album")).unwrap(); + for name in ["one.flac", "two.flac"] { + std::fs::write(share_dir.join("album").join(name), b"xxxx").unwrap(); + } + let mut subject = Client::with_settings(ClientSettings { + shared_directories: vec![share_dir.display().to_string()], + ..server.settings("e2e_watch_subject", "pw") + }); + subject.connect().expect("subject connect"); + assert!(subject.login().expect("subject login")); + + let mut watcher = + Client::with_settings(server.settings("e2e_watch_watcher", "pw")); + watcher.connect().expect("watcher connect"); + assert!(watcher.login().expect("watcher login")); + + // Let the subject's share counts reach the server before watching, so + // the reply describes a user it already knows the statistics for. + std::thread::sleep(Duration::from_millis(750)); + + watcher + .watch_user("e2e_watch_subject") + .expect("watch the subject"); + assert_eq!( + watcher.watched_users(), + vec!["e2e_watch_subject".to_string()] + ); + + // One reply carries both halves, unlike the two GetUserStatus and + // GetUserStats answers request_user_info waits on. + let deadline = Instant::now() + Duration::from_secs(15); + let mut info = None; + while Instant::now() < deadline { + match watcher.user_info("e2e_watch_subject") { + Some(found) if found.is_complete() => { + info = Some(found); + break; + } + _ => std::thread::sleep(Duration::from_millis(100)), + } + } + let info = info.expect("the watch reply should carry status and stats"); + + let presence = info.presence.expect("presence should have arrived"); + assert!( + presence.status.is_reachable(), + "a logged-in user should not read as offline, got {}", + presence.status + ); + let stats = info.stats.expect("stats should have arrived"); + assert_eq!( + stats.shared_files, 2, + "the file count must survive the obsolete fields ahead of it" + ); + assert_eq!(stats.shared_folders, 1); + + // Unwatching drops the user and what we knew about them, so a later + // re-watch reports a fresh answer rather than this stale one. + watcher + .unwatch_user("e2e_watch_subject") + .expect("unwatch the subject"); + assert!(watcher.watched_users().is_empty()); + assert!(watcher.user_info("e2e_watch_subject").is_none()); +} + +#[test] +fn watching_a_user_the_server_does_not_know_reports_their_absence() { + let server = server_or_skip!(); + + let mut watcher = + Client::with_settings(server.settings("e2e_watch_ghost_asker", "pw")); + watcher.connect().expect("watcher connect"); + assert!(watcher.login().expect("watcher login")); + + watcher + .watch_user("e2e_watch_nobody_here") + .expect("watch an unknown user"); + + // The server answers that it has no such user; the watch is dropped + // rather than left showing a permanently blank row. + let deadline = Instant::now() + Duration::from_secs(10); + while Instant::now() < deadline && !watcher.watched_users().is_empty() { + std::thread::sleep(Duration::from_millis(100)); + } + assert!( + watcher.watched_users().is_empty(), + "a username the server does not know must not stay watched" + ); + assert!(watcher.user_info("e2e_watch_nobody_here").is_none()); +}