From b505718db9055a2023b2dfb77913dbeb8d6bfb55 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Fri, 28 Aug 2026 12:42:13 +0000 Subject: [PATCH] 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 { .. }) + )); + } +}