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

@@ -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<u32>,
+        average_speed: Option<u32>,
+        shared_files: Option<u32>,
+        shared_folders: Option<u32>,
+    },
     /// 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,

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<u32>,
+        average_speed: Option<u32>,
+        shared_files: Option<u32>,
+        shared_folders: Option<u32>,
+    },
     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<String, UserInfo>,
+    /// 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<String>,
     /// Seconds the server wants between wishlist searches (code 104), once it
     /// has told us.
     wishlist_interval: Option<u32>,
@@ -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<String> {
+        let mut users: Vec<String> =
+            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<u32>,
+        average_speed: Option<u32>,
+        shared_files: Option<u32>,
+        shared_folders: Option<u32>,
+    ) {
+        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<UserInfo> {

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),

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<String> {
+        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

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"]);
+}

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<u8> = [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<u8> = [6, 0, 0, 0, 3, 0, 0, 0, b'b', b'o', b'b'].to_vec();
+    assert_eq!(expect, message.get_data());
+}

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;

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<ServerMessage> for WatchUserHandler {
+    fn get_code(&self) -> u8 {
+        5
+    }
+
+    fn handle(&self, message: &mut Message, sender: Sender<ServerMessage>) {
+        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 { .. })
+        ));
+    }
+}