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
This commit is contained in:
Michel de Graaf
2026-08-28 13:22:50 +00:00
parent b505718db9
commit a91d35b672
13 changed files with 322 additions and 11 deletions
+67
View File
@@ -1216,6 +1216,21 @@
"shared_folders" "shared_folders"
], ],
"type": "object" "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." "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", "name": "shares.status",
"paramStructure": "by-name", "paramStructure": "by-name",
+7
View File
@@ -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, be `null` when the server has not answered that half — `null` means unknown,
not offline. not offline.
**`watch add <user>`** / **`watch remove <user>`** — 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`, **`whoami`** — one object: `user`, `server`, `listening`, `listen_port`,
`shared_folders`, `shared_files`, `download_dir`, `privilege_seconds`. The `shared_folders`, `shared_files`, `download_dir`, `privilege_seconds`. The
cheapest way to prove the credentials and connection work. cheapest way to prove the credentials and connection work.
+18
View File
@@ -140,6 +140,12 @@ pub trait SessionApi: Send + Sync {
fn request_user_info(&self, username: &str) -> Result<()>; fn request_user_info(&self, username: &str) -> Result<()>;
fn user_info(&self, username: &str) -> Option<UserInfo>; fn user_info(&self, username: &str) -> Option<UserInfo>;
/// 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<String>;
/// The conversation this session already knows about. /// The conversation this session already knows about.
/// ///
/// Empty locally, where the TUI's own snapshot on disk is the record. A /// 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) 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<String> {
Self::watched_users(self)
}
fn shared_counts(&self) -> (u32, u32) { fn shared_counts(&self) -> (u32, u32) {
Self::shared_counts(self) Self::shared_counts(self)
} }
+24
View File
@@ -207,6 +207,10 @@ pub enum Commands {
/// What the server knows about another user /// What the server knows about another user
User(UserArgs), User(UserArgs),
/// Keep a daemon updated about users you care about
#[command(subcommand)]
Watch(WatchCommand),
/// Confirm the credentials and connection work /// Confirm the credentials and connection work
Whoami, Whoami,
@@ -571,6 +575,26 @@ pub struct WishRunArgs {
pub sort: SortKey, 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)] #[derive(Subcommand, Debug)]
pub enum RoomCommand { pub enum RoomCommand {
/// List public rooms and their user counts /// List public rooms and their user counts
+1
View File
@@ -309,6 +309,7 @@ fn dispatch(ctx: &Ctx, command: Commands) -> CliResult {
social::message_read(ctx, listen_span(duration, follow)) social::message_read(ctx, listen_span(duration, follow))
} }
Commands::User(args) => peer::user(ctx, &args), Commands::User(args) => peer::user(ctx, &args),
Commands::Watch(command) => peer::watch(ctx, &command),
Commands::Whoami => peer::whoami(ctx), Commands::Whoami => peer::whoami(ctx),
// main.rs runs these itself: some need no account, the rest need the // main.rs runs these itself: some need no account, the rest need the
// config store in scope. The arms only keep the match exhaustive. // config store in scope. The arms only keep the match exhaustive.
+37 -2
View File
@@ -1,9 +1,10 @@
//! Being a peer on the network: `serve`, `whoami`, and `user`. //! Being a peer on the network: `serve`, `whoami`, and `user`.
use super::{Ctx, Session}; use super::{Ctx, Session};
use crate::cli::{ServeArgs, UserArgs}; use crate::cli::{ServeArgs, UserArgs, WatchCommand};
use crate::output::{ use crate::output::{
CliError, CliResult, UploadRecord, UserRecord, WhoamiRecord, CliError, CliResult, UploadRecord, UserRecord, WatchedUserRecord,
WhoamiRecord,
}; };
use soulseek_rs::types::{UploadInfo, UploadStatus, UserInfo}; use soulseek_rs::types::{UploadInfo, UploadStatus, UserInfo};
use std::collections::HashMap; use std::collections::HashMap;
@@ -94,6 +95,40 @@ pub fn user(ctx: &Ctx, args: &UserArgs) -> CliResult {
Ok(()) 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 /// Stay online sharing files, and — with `--wishlist` — re-run the standing
/// searches on the interval the server sets while doing so. /// searches on the interval the server sets while doing so.
pub fn serve( pub fn serve(
+17 -5
View File
@@ -15,6 +15,7 @@ use super::proto::{
Method, QueryParams, RoomEventDto, RoomRef, SayParams, SearchResults, Method, QueryParams, RoomEventDto, RoomRef, SayParams, SearchResults,
Searches, Seconds, SessionLossEvent, SharesStatus, SlotsParams, Searches, Seconds, SessionLossEvent, SharesStatus, SlotsParams,
TransferRef, UploadInfoDto, Uploads, UserMessageDto, UserRef, UserResult, TransferRef, UploadInfoDto, Uploads, UserMessageDto, UserRef, UserResult,
Watched,
}; };
use schemars::{JsonSchema, SchemaGenerator, generate::SchemaSettings}; use schemars::{JsonSchema, SchemaGenerator, generate::SchemaSettings};
use serde_json::{Value, json}; use serde_json::{Value, json};
@@ -51,9 +52,11 @@ fn params_schema(method: Method, generator: &mut SchemaGenerator) -> Value {
} }
Method::RoomSay => schema_of::<SayParams>(generator), Method::RoomSay => schema_of::<SayParams>(generator),
Method::MessageSend => schema_of::<MessageParams>(generator), Method::MessageSend => schema_of::<MessageParams>(generator),
Method::BrowseUser | Method::UserRequest | Method::UserInfoOf => { Method::BrowseUser
schema_of::<UserRef>(generator) | Method::UserRequest
} | Method::UserInfoOf
| Method::UserWatch
| Method::UserUnwatch => schema_of::<UserRef>(generator),
Method::SharesSet => schema_of::<DirectoriesParams>(generator), Method::SharesSet => schema_of::<DirectoriesParams>(generator),
// Methods that take no parameters. // Methods that take no parameters.
Method::DaemonStatus Method::DaemonStatus
@@ -68,7 +71,8 @@ fn params_schema(method: Method, generator: &mut SchemaGenerator) -> Value {
| Method::MessageHistory | Method::MessageHistory
| Method::SearchList | Method::SearchList
| Method::SharesStatusOf | 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::<Uploads>(generator), Method::UploadList => schema_of::<Uploads>(generator),
Method::PrivilegesOwn => schema_of::<Seconds>(generator), Method::PrivilegesOwn => schema_of::<Seconds>(generator),
Method::RoomMembers => schema_of::<Members>(generator), Method::RoomMembers => schema_of::<Members>(generator),
Method::UserWatched => schema_of::<Watched>(generator),
Method::MessageHistory => schema_of::<Messages>(generator), Method::MessageHistory => schema_of::<Messages>(generator),
Method::UserInfoOf => schema_of::<UserResult>(generator), Method::UserInfoOf => schema_of::<UserResult>(generator),
Method::SharesStatusOf | Method::SharesSet | Method::SharesReindex => { Method::SharesStatusOf | Method::SharesSet | Method::SharesReindex => {
@@ -112,7 +117,9 @@ fn result_schema(method: Method, generator: &mut SchemaGenerator) -> Value {
| Method::MessageSend | Method::MessageSend
| Method::BrowseUser | Method::BrowseUser
| Method::SearchForget | Method::SearchForget
| Method::UserRequest => schema_of::<Ack>(generator), | Method::UserRequest
| Method::UserWatch
| Method::UserUnwatch => schema_of::<Ack>(generator),
} }
} }
@@ -172,6 +179,11 @@ fn summary(method: Method) -> &'static str {
Method::RoomLeave => "Leave a chat room.", Method::RoomLeave => "Leave a chat room.",
Method::RoomSay => "Post a message to a room.", Method::RoomSay => "Post a message to a room.",
Method::RoomMembers => "Who is currently in 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::MessageSend => "Send a private message.",
Method::MessageHistory => { Method::MessageHistory => {
"Private messages the daemon collected, including while nobody was attached." "Private messages the daemon collected, including while nobody was attached."
+7
View File
@@ -139,6 +139,13 @@ pub struct Members {
pub users: Vec<String>, pub users: Vec<String>,
} }
/// 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<String>,
}
/// One line of chat history. Unlike a live [`UserMessageDto`] this carries a /// One line of chat history. Unlike a live [`UserMessageDto`] this carries a
/// direction, because a conversation the daemon collected includes what this /// direction, because a conversation the daemon collected includes what this
/// account sent as well as what it received. /// account sent as well as what it received.
+3
View File
@@ -106,6 +106,9 @@ wire_vocabulary! {
BrowseUser => "browse.user", BrowseUser => "browse.user",
UserRequest => "user.request", UserRequest => "user.request",
UserInfoOf => "user.info", UserInfoOf => "user.info",
UserWatch => "user.watch",
UserUnwatch => "user.unwatch",
UserWatched => "user.watched",
SharesStatusOf => "shares.status", SharesStatusOf => "shares.status",
SharesSet => "shares.set", SharesSet => "shares.set",
SharesReindex => "shares.reindex", SharesReindex => "shares.reindex",
+88 -3
View File
@@ -13,7 +13,7 @@ use super::proto::{
MessageParams, Messages, Method, OPENRPC, PROTOCOL_VERSION, QueryParams, MessageParams, Messages, Method, OPENRPC, PROTOCOL_VERSION, QueryParams,
RoomRef, RpcError, SayParams, SearchResultDto, SearchResults, RoomRef, RpcError, SayParams, SearchResultDto, SearchResults,
SearchSummary, Searches, Seconds, SharesStatus, SlotsParams, TransferRef, SearchSummary, Searches, Seconds, SharesStatus, SlotsParams, TransferRef,
UploadInfoDto, Uploads, UserInfoDto, UserRef, UserResult, UploadInfoDto, Uploads, UserInfoDto, UserRef, UserResult, Watched,
}; };
use crate::api::SessionApi; use crate::api::SessionApi;
use crate::output::Exit; 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::SharesStatusOf => ok(self.shares()),
Method::SharesSet => { Method::SharesSet => {
let directories: DirectoriesParams = parse(params)?; 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] #[test]
fn stop_is_recorded_so_the_daemon_can_wind_down() { fn stop_is_recorded_so_the_daemon_can_wind_down() {
let daemon = daemon_with_token("t"); let daemon = daemon_with_token("t");
@@ -605,11 +666,17 @@ mod tests {
// A session that answers nothing: enough to exercise dispatch, parameter // A session that answers nothing: enough to exercise dispatch, parameter
// validation, and the auth gate without a Soulseek server. // 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<Vec<String>>,
}
fn daemon_with_token(token: &str) -> Daemon { fn daemon_with_token(token: &str) -> Daemon {
Daemon { Daemon {
session: Arc::new(SilentSession), session: Arc::new(SilentSession::default()),
hub: Arc::new(Hub::new()), hub: Arc::new(Hub::new()),
browses: Arc::new(PendingBrowses::default()), browses: Arc::new(PendingBrowses::default()),
downloads: Arc::new(Mutex::new(Vec::new())), downloads: Arc::new(Mutex::new(Vec::new())),
@@ -792,6 +859,24 @@ mod tests {
) -> soulseek_rs::Result<()> { ) -> soulseek_rs::Result<()> {
Ok(()) 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<String> {
self.watched.lock().expect("watch list").clone()
}
fn user_info(&self, _username: &str) -> Option<soulseek_rs::UserInfo> { fn user_info(&self, _username: &str) -> Option<soulseek_rs::UserInfo> {
None None
} }
+19
View File
@@ -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<String>,
}
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. /// One configured share folder.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)] #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ShareRecord { pub struct ShareRecord {
+25 -1
View File
@@ -15,7 +15,7 @@ use crate::daemon::proto::{
Members, MessageParams, Messages, Method, PROTOCOL_VERSION, QueryParams, Members, MessageParams, Messages, Method, PROTOCOL_VERSION, QueryParams,
Request, Response, RoomRef, RpcError, SayParams, SearchResults, Searches, Request, Response, RoomRef, RpcError, SayParams, SearchResults, Searches,
Seconds, SharesStatus, SlotsParams, TransferRef, Uploads, UserRef, Seconds, SharesStatus, SlotsParams, TransferRef, Uploads, UserRef,
UserResult, UserResult, Watched,
}; };
use mirror::Mirror; use mirror::Mirror;
use serde::de::DeserializeOwned; 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<String> {
self.request::<_, Watched>(Method::UserWatched, ())
.map(|watched| watched.users)
.unwrap_or_default()
}
fn user_info(&self, username: &str) -> Option<UserInfo> { fn user_info(&self, username: &str) -> Option<UserInfo> {
self.request::<_, UserResult>( self.request::<_, UserResult>(
Method::UserInfoOf, Method::UserInfoOf,
+9
View File
@@ -489,6 +489,15 @@ mod tests {
) -> soulseek_rs::Result<()> { ) -> soulseek_rs::Result<()> {
Ok(()) 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<String> {
Vec::new()
}
fn user_info(&self, _username: &str) -> Option<soulseek_rs::UserInfo> { fn user_info(&self, _username: &str) -> Option<soulseek_rs::UserInfo> {
None None
} }