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..1499189 --- /dev/null +++ b/soulseek-rs-lib/src/message/server/watch_user.rs @@ -0,0 +1,146 @@ +//! `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(); + // 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 { + 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_int32(0) // obsolete upload number + .write_int32(0) // obsolete unknown + .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 { .. }) + )); + } +} 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()); +} 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 }