diff --git a/soulseek-rs/src/SKILL.md b/soulseek-rs/src/SKILL.md index 017e324..cf3c6a6 100644 --- a/soulseek-rs/src/SKILL.md +++ b/soulseek-rs/src/SKILL.md @@ -106,6 +106,18 @@ the interval the server sets and prints `search`-shaped objects for what it finds, so one long-running process covers both halves of being a good participant. +**`daemon`** — stay logged in as a service other commands share. Prints +progress on stderr and nothing on stdout; it does not return until stopped. +While it runs, every other command on this machine uses its session +automatically — no flags, no credentials, no second login. `--bind ADDR` also +accepts connections from other machines, which need the token. +**`daemon status`** — one object: `user`, `server`, `version`, `protocol`, +`listening`, `listen_port`, `shared_folders`, `shared_files`, `download_dir`, +`session_loss`, `clients`, `uptime_secs`. Exit 3 when nothing is running. +**`daemon stop`** — ask it to shut down; prints nothing. +**`daemon token`** — the secret a client on another machine needs. Local +clients never need it. + **`shares list`** — one object per configured folder: `directory`, `usable`. **`shares add `** and **`shares remove `** — change the config file; they print nothing. @@ -152,8 +164,27 @@ is the normal way to be selective. ## Running several at once The server allows one session per account, and the later login wins: two runs -sharing a username silently kill each other's session. Give every concurrent -run its own name. +sharing a username silently kill each other's session. + +The daemon is the answer to this. Start one, and every command afterwards +borrows its session instead of opening one: + +```bash +soulseek-rs daemon & # once +soulseek-rs search 'gary beck fold' --json # no login, no flags +soulseek-rs search 'marcal steady' --json # concurrent, same session +``` + +Commands find it on their own: it listens on a Unix socket only you can open, +so there is nothing to configure and no token to pass. `daemon status` says +whether one is running. This also makes downloads outlive the command that +started them — the transfer belongs to the daemon, so the queue survives. + +Two things change when a daemon is in play: files land in the *daemon's* +download directory, and `--no-daemon` is how one run opts out and logs in for +itself. + +Without a daemon, give every concurrent run its own name instead: ```bash for q in 'gary beck fold' 'marcal steady'; do diff --git a/soulseek-rs/src/cli.rs b/soulseek-rs/src/cli.rs index ee5395f..e4628a2 100644 --- a/soulseek-rs/src/cli.rs +++ b/soulseek-rs/src/cli.rs @@ -25,6 +25,7 @@ Examples: soulseek-rs browse someuser --json list a peer's shares soulseek-rs room listen lobby --follow stream room chat soulseek-rs serve --follow --json share files, stream upload events + soulseek-rs daemon run as a service; later commands use it soulseek-rs user someuser is this peer online and worth it soulseek-rs skills install teach your coding agent this CLI soulseek-rs completions install tab completion for your shell @@ -129,6 +130,23 @@ pub struct Cli { )] pub search_timeout: Option, + /// Control a daemon at this address (host:port) or Unix socket path + #[arg(long, global = true, env = "SOULSEEK_DAEMON", value_name = "ADDR")] + pub daemon: Option, + + /// Token for a remote daemon; a local socket needs none + #[arg( + long, + global = true, + env = "SOULSEEK_DAEMON_TOKEN", + hide_env_values = true + )] + pub daemon_token: Option, + + /// Ignore any running daemon and open this run's own session + #[arg(long, global = true, action = ArgAction::SetTrue)] + pub no_daemon: bool, + /// Config file to read instead of the default location #[arg(long, global = true, env = "SOULSEEK_CONFIG", value_name = "FILE")] pub config: Option, @@ -183,6 +201,9 @@ pub enum Commands { /// Stay online sharing files, reporting each upload as it happens Serve(ServeArgs), + /// Run as a service other clients connect to, or control one that is + Daemon(DaemonArgs), + /// What the server knows about another user User(UserArgs), @@ -209,6 +230,33 @@ pub enum Commands { Completions(CompletionsCommand), } +#[derive(Args, Debug, Default)] +pub struct DaemonArgs { + #[command(subcommand)] + pub command: Option, + + /// Also accept control connections on this address, for clients on other + /// machines. They need the token; the local socket does not. + #[arg(long, value_name = "ADDR")] + pub bind: Option, + + /// How many uploads to serve at once + #[arg(long, value_parser = clap::value_parser!(u16).range(1..=1000))] + pub upload_slots: Option, +} + +#[derive(Subcommand, Debug)] +pub enum DaemonCommand { + /// Print the token remote clients authenticate with + Token, + + /// What a running daemon is doing + Status, + + /// Ask a running daemon to shut down + Stop, +} + #[derive(Subcommand, Debug)] pub enum CompletionsCommand { /// Write the completion script where the shell looks for it diff --git a/soulseek-rs/src/commands/daemon.rs b/soulseek-rs/src/commands/daemon.rs new file mode 100644 index 0000000..984ae51 --- /dev/null +++ b/soulseek-rs/src/commands/daemon.rs @@ -0,0 +1,86 @@ +//! Controlling the service: `daemon token`, `daemon status`, `daemon stop`. +//! +//! Running the daemon itself lives in [`crate::daemon`]; these are the +//! one-shot commands that talk *to* one. None of them needs a Soulseek +//! account — the daemon already has one — so they are dispatched before +//! credentials are demanded. + +use crate::daemon::proto::{DaemonStatus, Method}; +use crate::output::{CliError, CliResult, DaemonStatusRecord, Out, TokenRecord}; +use crate::remote::{Endpoint, RemoteSession}; + +/// Print the token a client on another machine needs. +/// +/// Local clients never need this — the socket's permissions already answer the +/// question — so it exists for the moment you point a second machine at this +/// one. +pub fn token(out: &Out) -> CliResult { + let path = crate::daemon::token_path().ok_or_else(|| { + CliError::usage("no state directory: set SOULSEEK_STATE_DIR") + })?; + let token = crate::daemon::load_or_create_token(&path).map_err(|e| { + CliError::usage(format!("cannot read {}: {e}", path.display())) + })?; + out.emit(&TokenRecord { token }); + Ok(()) +} + +/// Ask a running daemon what it is doing. +pub fn status( + out: &Out, + endpoint: Option<&Endpoint>, + token: Option, +) -> CliResult { + let status: DaemonStatus = ask(endpoint, token, Method::DaemonStatus)?; + out.emit(&DaemonStatusRecord { + user: status.username, + server: status.server, + version: status.daemon_version, + protocol: status.protocol, + listening: status.listen_port.is_some(), + listen_port: status.listen_port.unwrap_or(0), + shared_folders: status.shared_folders, + shared_files: status.shared_files, + download_dir: status.download_dir, + session_loss: status + .session_loss + .map(|loss| format!("{loss:?}").to_lowercase()), + clients: status.clients, + uptime_secs: status.uptime_secs, + }); + Ok(()) +} + +/// Ask a running daemon to shut down. +pub fn stop( + out: &Out, + endpoint: Option<&Endpoint>, + token: Option, +) -> CliResult { + let _: serde_json::Value = ask(endpoint, token, Method::DaemonStop)?; + out.status("the daemon is stopping"); + Ok(()) +} + +/// One request to whichever daemon this run is pointed at. +fn ask( + endpoint: Option<&Endpoint>, + token: Option, + method: Method, +) -> CliResult { + let endpoint = endpoint.ok_or_else(no_daemon)?; + let token = token.or_else(crate::daemon::stored_token); + let session = RemoteSession::connect(endpoint, token).map_err(|e| { + CliError::connection(format!("cannot reach the daemon: {e}")) + })?; + session + .ask_for(method) + .map_err(|e| CliError::connection(e.to_string())) +} + +fn no_daemon() -> CliError { + CliError::connection( + "no daemon is running; start one with `soulseek-rs daemon`, or point \ + at a remote one with --daemon", + ) +} diff --git a/soulseek-rs/src/commands/mod.rs b/soulseek-rs/src/commands/mod.rs index f0ffa83..b7071b1 100644 --- a/soulseek-rs/src/commands/mod.rs +++ b/soulseek-rs/src/commands/mod.rs @@ -6,6 +6,7 @@ //! dispatched by `main::run` before any credentials are demanded. pub mod completions; +pub mod daemon; pub mod peer; pub mod settings; pub mod skills; @@ -16,6 +17,7 @@ pub mod wish; use crate::cli::{Commands, MessageCommand, RoomCommand}; use crate::output::{CliError, CliResult, Exit, Out, PortmapRecord}; use crate::port_mapping::{self, PortMapper}; +use crate::remote::{Endpoint, RemoteSession}; use soulseek_rs::types::{RoomEvent, RoomInfo}; use soulseek_rs::{Client, ClientSettings, SessionLoss}; use crate::api::SessionApi; @@ -30,6 +32,11 @@ pub struct Ctx { pub download_dir: String, pub max_concurrent_downloads: usize, pub search_timeout: Duration, + /// Where to find a daemon, when this run should use one rather than open + /// its own session. Resolved once in `main`, so every command agrees. + pub daemon: Option, + /// Token for a remote daemon; a Unix socket needs none. + pub daemon_token: Option, /// The client of the session this command opened, once it has one, so the /// verdict can be checked against the session that produced it. pub session: OnceLock>, @@ -86,7 +93,42 @@ const REACH_TIMEOUT: Duration = Duration::from_secs(10); const LOGIN_TIMEOUT: Duration = Duration::from_secs(30); impl Session { + /// Get a session: the daemon's, if this run is routed to one, or a fresh + /// login of its own. pub fn open(ctx: &Ctx) -> CliResult { + match &ctx.daemon { + Some(endpoint) => Self::attach(ctx, endpoint), + None => Self::login(ctx), + } + } + + /// Borrow the daemon's session instead of logging in again. + /// + /// The server allows one login per account, so a second login would + /// displace the daemon — attaching is not just cheaper, it is the only way + /// two things can be on the network as the same user at once. + fn attach(ctx: &Ctx, endpoint: &Endpoint) -> CliResult { + ctx.out.status(&format!("using the daemon at {endpoint}")); + let token = ctx + .daemon_token + .clone() + .or_else(crate::daemon::stored_token); + let remote = RemoteSession::connect(endpoint, token).map_err(|e| { + CliError::connection(format!("cannot use the daemon: {e}")) + })?; + + let client: Arc = Arc::new(remote); + ctx.out + .status(&format!("logged in as {}", client.username())); + let _ = ctx.session.set(Arc::clone(&client)); + Ok(Self { + client, + // The daemon holds the listener, so it owns the port mapping too. + _mapper: None, + }) + } + + fn login(ctx: &Ctx) -> CliResult { let address = ctx.settings.server_address.to_string(); ctx.out.status(&format!("connecting to {address}…")); reachable(&address, REACH_TIMEOUT)?; @@ -259,6 +301,7 @@ fn dispatch(ctx: &Ctx, command: Commands) -> CliResult { // These need the config file as well as a session, so main.rs runs // them where the store is in scope. Commands::Serve(_) + | Commands::Daemon(_) | Commands::Wish(_) | Commands::Config(_) | Commands::Shares(_) diff --git a/soulseek-rs/src/commands/peer.rs b/soulseek-rs/src/commands/peer.rs index dd06f56..38e7df6 100644 --- a/soulseek-rs/src/commands/peer.rs +++ b/soulseek-rs/src/commands/peer.rs @@ -24,7 +24,10 @@ pub fn whoami(ctx: &Ctx) -> CliResult { // reachable needs what is true, not what was asked for. let bound = session.client.listen_port(); ctx.out.emit(&WhoamiRecord { - user: ctx.settings.username.clone(), + // The session's own name, not this run's configuration: routed to a + // daemon they are different, and the session is the one that is + // actually on the network. + user: session.client.username(), server: ctx.settings.server_address.to_string(), listening: bound.is_some(), listen_port: bound.unwrap_or(0), diff --git a/soulseek-rs/src/commands/settings.rs b/soulseek-rs/src/commands/settings.rs index 672cdad..7781b43 100644 --- a/soulseek-rs/src/commands/settings.rs +++ b/soulseek-rs/src/commands/settings.rs @@ -96,7 +96,9 @@ pub fn shares_status(ctx: &Ctx) -> CliResult { ctx.out.emit(&ShareStatusRecord { folders, files, - directories: ctx.settings.shared_directories.clone(), + // What the session shares, which locally is what this run configured + // and against a daemon is what the daemon configured. + directories: session.client.shared_directories(), }); Ok(()) } @@ -106,16 +108,20 @@ pub fn shares_status(ctx: &Ctx) -> CliResult { pub fn shares_reindex(ctx: &Ctx) -> CliResult { let session = Session::open(ctx)?; ctx.out.status("re-indexing the shared folders…"); + // Re-scan what the session already shares rather than imposing this run's + // configuration on it: against a daemon that would silently replace the + // daemon's shares with the client machine's. + let directories = session.client.shared_directories(); session .client - .set_shared_directories(ctx.settings.shared_directories.clone()) + .set_shared_directories(directories.clone()) .map_err(|e| CliError::usage(format!("cannot re-index: {e}")))?; let (folders, files) = session.client.shared_counts(); ctx.out.emit(&ShareStatusRecord { folders, files, - directories: ctx.settings.shared_directories.clone(), + directories, }); Ok(()) } diff --git a/soulseek-rs/src/commands/transfer.rs b/soulseek-rs/src/commands/transfer.rs index 67d9b36..d902f40 100644 --- a/soulseek-rs/src/commands/transfer.rs +++ b/soulseek-rs/src/commands/transfer.rs @@ -240,7 +240,7 @@ pub fn download(ctx: &Ctx, args: &DownloadArgs) -> CliResult { return Err(CliError::no_results("no files to download")); } - ensure_download_dir(&ctx.download_dir)?; + ensure_download_dir(ctx)?; let session = Session::open(ctx)?; let requests = resolve_sizes(ctx, session.client.as_ref(), requests)?; download_all( @@ -252,7 +252,7 @@ pub fn download(ctx: &Ctx, args: &DownloadArgs) -> CliResult { } pub fn get(ctx: &Ctx, args: &GetArgs) -> CliResult { - ensure_download_dir(&ctx.download_dir)?; + ensure_download_dir(ctx)?; let session = Session::open(ctx)?; let results = collect(ctx, &session.client, &args.query, args.pick == Pick::First)?; @@ -421,8 +421,17 @@ fn resolve_sizes( Ok(resolved) } -fn ensure_download_dir(download_dir: &str) -> CliResult { - let path = expand_tilde(download_dir); +/// Make sure the bytes have somewhere to land. +/// +/// A daemon writes to its own filesystem and has already done this; creating +/// the same path here would leave a stray directory on the client machine and, +/// for a daemon on another host, could fail outright over a path that is none +/// of this machine's business. +fn ensure_download_dir(ctx: &Ctx) -> CliResult { + if ctx.daemon.is_some() { + return Ok(()); + } + let path = expand_tilde(&ctx.download_dir); std::fs::create_dir_all(&path).map_err(|e| { CliError::usage(format!( "cannot use download directory {}: {e}", diff --git a/soulseek-rs/src/directories.rs b/soulseek-rs/src/directories.rs index e048d19..28375bc 100644 --- a/soulseek-rs/src/directories.rs +++ b/soulseek-rs/src/directories.rs @@ -37,6 +37,7 @@ pub fn resolve_shared_directory( /// Validate several configured share paths; invalid ones are skipped with a /// warning instead of aborting (the client just shares less). +#[must_use] pub fn resolve_shared_directories(dirs: &[String]) -> Vec { let mut resolved = Vec::new(); for dir in dirs { diff --git a/soulseek-rs/src/lib.rs b/soulseek-rs/src/lib.rs index fab2657..11ce9c1 100644 --- a/soulseek-rs/src/lib.rs +++ b/soulseek-rs/src/lib.rs @@ -5,9 +5,9 @@ pub mod api; pub mod cli; pub mod commands; pub mod daemon; -pub mod directories; +pub(crate) mod directories; pub mod models; pub mod output; pub mod persist; -pub mod port_mapping; +pub(crate) mod port_mapping; pub mod remote; diff --git a/soulseek-rs/src/main.rs b/soulseek-rs/src/main.rs index ffc4cfe..88297b1 100644 --- a/soulseek-rs/src/main.rs +++ b/soulseek-rs/src/main.rs @@ -7,11 +7,12 @@ mod models; mod output; mod persist; mod port_mapping; +mod remote; mod ui; use clap::Parser; use cli::{ - Cli, Commands, ConfigCommand, SharesCommand, WishCommand, + Cli, Commands, ConfigCommand, DaemonCommand, SharesCommand, WishCommand, parse_server_address, }; use commands::Ctx; @@ -52,6 +53,12 @@ fn run(mut cli: Cli, out: &Out) -> CliResult { let resolved = persist::config::resolve(&cli, &file_config); let Some(command) = cli.command.take() else { + // A running daemon already holds a session; the TUI attaches to it + // rather than logging in a second time, which the server would answer + // by cutting the daemon off. + if let Some(endpoint) = daemon_endpoint(&cli) { + return run_attached_tui(&cli, &resolved, &endpoint); + } return run_default_tui(&cli, &resolved, config_path, &file_config); }; @@ -107,6 +114,25 @@ fn run(mut cli: Cli, out: &Out) -> CliResult { Commands::Wish(WishCommand::List) => { return commands::wish::list(out, &store); } + // Controlling a daemon needs no account of our own: the daemon is + // already logged in, and demanding credentials a script never had + // would turn "is it running?" into a configuration error. + Commands::Daemon(cli::DaemonArgs { + command: Some(ref command), + .. + }) => { + let endpoint = daemon_endpoint(&cli); + let token = cli.daemon_token.clone(); + return match command { + DaemonCommand::Token => commands::daemon::token(out), + DaemonCommand::Status => { + commands::daemon::status(out, endpoint.as_ref(), token) + } + DaemonCommand::Stop => { + commands::daemon::stop(out, endpoint.as_ref(), token) + } + }; + } _ => {} } @@ -121,6 +147,9 @@ fn run(mut cli: Cli, out: &Out) -> CliResult { Commands::Wish(WishCommand::Run(ref args)) => { commands::wish::run(&ctx, &store, args) } + Commands::Daemon(ref args) => { + daemon::run(&ctx, args.bind.as_deref(), args.upload_slots) + } Commands::Serve(ref args) => { let sweeper = args .wishlist @@ -154,40 +183,104 @@ fn context( resolved: &persist::config::Resolved, out: &Out, ) -> CliResult { - let username = resolved.username.clone().ok_or_else(|| { - CliError::usage( - "no username: pass --username, set SOULSEEK_USERNAME, or put one \ - in config.toml", + let daemon = daemon_endpoint(cli); + // A run routed to a daemon needs no account of its own: the daemon is + // already logged in, and asking for credentials it will not use would + // stop a script that never had any. + let (username, password) = if daemon.is_some() { + (String::new(), String::new()) + } else { + ( + resolved.username.clone().ok_or_else(|| { + CliError::usage( + "no username: pass --username, set SOULSEEK_USERNAME, or \ + put one in config.toml", + ) + })?, + password(cli, resolved)?.ok_or_else(|| { + CliError::usage( + "no password: pass --password/--password-stdin, set \ + SOULSEEK_PASSWORD, store one in the OS keychain, or set \ + password_cmd", + ) + })?, ) - })?; - let password = password(cli, resolved)?.ok_or_else(|| { - CliError::usage( - "no password: pass --password/--password-stdin, set \ - SOULSEEK_PASSWORD, store one in the OS keychain, or set \ - password_cmd", - ) - })?; + }; let (host, port) = parse_server_address(&resolved.server).map_err(CliError::usage)?; + // A daemon has its own server and its own filesystem. Asking it once, up + // front, is what keeps every command that reports a path or a server + // honest — `whoami` naming the right server, `download` naming the file + // where the bytes actually landed. + let described = daemon.as_ref().and_then(|endpoint| { + describe_daemon(endpoint, cli.daemon_token.clone()) + }); + let server_address = described.as_ref().map_or_else( + || PeerAddress::new(host, port), + |status| { + parse_server_address(&status.server).map_or_else( + |_| PeerAddress::new(status.server.clone(), 0), + |(host, port)| PeerAddress::new(host, port), + ) + }, + ); + let download_dir = described.as_ref().map_or_else( + || resolved.download_dir.clone(), + |status| status.download_dir.clone(), + ); + Ok(Ctx { out: out.clone(), settings: ClientSettings { username, password, - server_address: PeerAddress::new(host, port), + server_address, enable_listen: !resolved.disable_listener, listen_port: resolved.listener_port, shared_directories: shared_directories(resolved, out), version: ClientVersion::REFERENCE_CLIENT, }, - download_dir: resolved.download_dir.clone(), + download_dir, max_concurrent_downloads: resolved.max_concurrent_downloads, search_timeout: Duration::from_secs(resolved.search_timeout), + daemon, + daemon_token: cli.daemon_token.clone(), session: std::sync::OnceLock::new(), }) } +/// Ask a daemon what it is, so this run describes the session it will actually +/// use. A daemon that cannot be reached is left to `Session::open` to report, +/// which produces a better message than anything this early probe could. +fn describe_daemon( + endpoint: &remote::Endpoint, + token: Option, +) -> Option { + let token = token.or_else(daemon::stored_token); + remote::RemoteSession::connect(endpoint, token) + .ok()? + .ask_for(daemon::proto::Method::DaemonStatus) + .ok() +} + +/// Which daemon this run should use, if any. +/// +/// An address given explicitly always wins. Otherwise a daemon listening on +/// the local socket is used automatically — the convention Docker and +/// ssh-agent set, and what lets an agent run a command with no flags at all. +/// With nothing listening, the run opens its own session exactly as it always +/// has. +fn daemon_endpoint(cli: &Cli) -> Option { + if cli.no_daemon { + return None; + } + if let Some(address) = &cli.daemon { + return Some(remote::Endpoint::parse(address)); + } + daemon::local_daemon().map(remote::Endpoint::Socket) +} + /// Resolve the password: stdin beats the flag/env, which beats the keychain, /// which beats `password_cmd`. fn password( @@ -365,6 +458,54 @@ fn run_default_tui( .map_err(|e| CliError::new(Exit::Failure, e.to_string())) } +/// Run the TUI against a daemon's session. +/// +/// No login screen: the daemon is already logged in, so there is nothing to +/// ask for and nothing to store. Everything the UI shows comes back over the +/// socket, and what it persists stays daemon-side. +fn run_attached_tui( + cli: &Cli, + resolved: &persist::config::Resolved, + endpoint: &remote::Endpoint, +) -> CliResult { + use ratatui::crossterm::{ + event::EnableMouseCapture, + execute, + terminal::{Clear, ClearType}, + }; + + if !std::io::stdout().is_terminal() { + return Err(CliError::usage( + "the interactive interface needs a terminal; run a subcommand for \ + scripted use (see --help)", + )); + } + + let token = cli + .daemon_token + .clone() + .or_else(daemon::stored_token); + let session = remote::RemoteSession::connect(endpoint, token) + .map_err(|e| CliError::connection(format!("cannot use the daemon at {endpoint}: {e}")))?; + + soulseek_rs::utils::logger::enable_buffering(); + let _ = + execute!(std::io::stdout(), Clear(ClearType::All), EnableMouseCapture); + let terminal = ratatui::init(); + + // No StateStore: the daemon owns what survives a restart, so a second + // writer here would fight it for the same files. + launch_main_tui( + terminal, + Arc::new(session), + resolved.download_dir.clone(), + resolved.max_concurrent_downloads, + Duration::from_secs(resolved.search_timeout), + None, + ) + .map_err(|e| CliError::new(Exit::Failure, e.to_string())) +} + /// After a successful login, remember the username in config.toml and — when /// it was typed into the form — the password in the OS keychain. Both are /// best-effort: failing to persist must not break a working session. diff --git a/soulseek-rs/src/output.rs b/soulseek-rs/src/output.rs index 0bd0d30..c2f8fed 100644 --- a/soulseek-rs/src/output.rs +++ b/soulseek-rs/src/output.rs @@ -489,6 +489,49 @@ impl Record for CompletionRecord { } } +/// The token a remote client authenticates with. +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct TokenRecord { + pub token: String, +} + +impl Record for TokenRecord { + fn text(&self) -> String { + sanitize(&self.token) + } +} + +/// What a running daemon is and what its session is doing. +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct DaemonStatusRecord { + pub user: String, + pub server: String, + pub version: String, + pub protocol: u32, + pub listening: bool, + pub listen_port: u16, + pub shared_folders: u32, + pub shared_files: u32, + pub download_dir: String, + /// `displaced` or `disconnected` when the server session has ended. + pub session_loss: Option, + /// Control connections currently attached. + pub clients: usize, + pub uptime_secs: u64, +} + +impl Record for DaemonStatusRecord { + fn text(&self) -> String { + format!( + "{}\t{}\t{}\t{}", + sanitize(&self.user), + sanitize(&self.server), + self.clients, + self.uptime_secs + ) + } +} + /// The result of probing automatic port mapping. #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub struct PortmapRecord { diff --git a/soulseek-rs/src/remote/mirror.rs b/soulseek-rs/src/remote/mirror.rs new file mode 100644 index 0000000..f5327d6 --- /dev/null +++ b/soulseek-rs/src/remote/mirror.rs @@ -0,0 +1,297 @@ +//! The client-side copy of what the daemon drained. +//! +//! A remote session cannot ask the daemon to `take_*` on its behalf — that is +//! destructive, and with two clients attached the first to ask would steal the +//! other's events. So the daemon pushes, this mirror accumulates, and +//! `take_*` empties the mirror instead. From the caller's side the semantics +//! are identical: each event is delivered once, to each client. + +use crate::daemon::proto::{ + BrowseEvent, DownloadStatusEvent, Event, RoomEventDto, SessionLossEvent, + UploadInfoDto, UserMessageDto, +}; +use soulseek_rs::{ + DownloadStatus, RoomEvent, SessionLoss, SharedDirectory, UploadInfo, + UserMessage, +}; +use std::collections::HashMap; +use std::sync::Mutex; +use std::sync::mpsc::Sender; + +/// Everything a pushed event can land in, behind one lock. +/// +/// One lock rather than several: it is taken for the length of a `Vec::push` +/// on an event thread that has nothing else to do, and a single lock cannot +/// deadlock against itself. +#[derive(Default)] +struct Buffers { + rooms: Vec, + messages: Vec, + uploads: Vec, + browses: HashMap>, + /// Where each in-flight transfer's status goes, keyed the way the daemon + /// names it. Registered by `download`, dropped when the transfer ends. + downloads: HashMap<(String, String), Sender>, + loss: Option, +} + +#[derive(Default)] +pub struct Mirror(Mutex); + +impl Mirror { + /// File one pushed event. Anything unparseable is dropped with a warning + /// rather than killing the connection: a newer daemon may legitimately + /// send a shape this client does not know yet. + pub fn apply(&self, event: Event, params: serde_json::Value) { + let Ok(mut buffers) = self.0.lock() else { + return; + }; + match event { + Event::Room => { + if let Ok(room) = parse::(event, params) { + buffers.rooms.push(room.into()); + } + } + Event::Message => { + if let Ok(message) = parse::(event, params) { + buffers.messages.push(message.into()); + } + } + Event::Upload => { + if let Ok(upload) = parse::(event, params) { + buffers.uploads.push(upload.into()); + } + } + Event::Browse => { + if let Ok(browse) = parse::(event, params) { + buffers.browses.insert( + browse.username, + browse + .directories + .into_iter() + .map(Into::into) + .collect(), + ); + } + } + Event::DownloadStatus => { + if let Ok(update) = parse::(event, params) + { + Self::route(&mut buffers, update); + } + } + Event::SessionLoss => { + if let Ok(loss) = parse::(event, params) { + // The first loss is the one worth keeping: a displaced + // session also drops its socket a moment later. + buffers.loss.get_or_insert_with(|| loss.loss.into()); + } + } + } + } + + /// Deliver a transfer update to whoever is waiting on it, forgetting the + /// transfer once it can produce nothing further. + fn route(buffers: &mut Buffers, update: DownloadStatusEvent) { + let key = (update.username, update.filename); + let status = DownloadStatus::from(update.status); + let terminal = matches!( + status, + DownloadStatus::Completed + | DownloadStatus::Failed(_) + | DownloadStatus::TimedOut + ); + let Some(sender) = buffers.downloads.get(&key) else { + return; + }; + // A caller that dropped its receiver is not an error: it stopped + // caring, which is exactly what `remove_download` does. + let delivered = sender.send(status).is_ok(); + if terminal || !delivered { + buffers.downloads.remove(&key); + } + } + + /// Register where a transfer's updates should go before asking for it, so + /// an update that lands immediately is not dropped on the floor. + pub fn expect_download( + &self, + username: &str, + filename: &str, + sender: Sender, + ) { + if let Ok(mut buffers) = self.0.lock() { + buffers + .downloads + .insert((username.to_string(), filename.to_string()), sender); + } + } + + pub fn forget_download(&self, username: &str, filename: &str) { + if let Ok(mut buffers) = self.0.lock() { + buffers + .downloads + .remove(&(username.to_string(), filename.to_string())); + } + } + + pub fn take_rooms(&self) -> Vec { + self.take(|buffers| std::mem::take(&mut buffers.rooms)) + } + + pub fn take_messages(&self) -> Vec { + self.take(|buffers| std::mem::take(&mut buffers.messages)) + } + + pub fn take_uploads(&self) -> Vec { + self.take(|buffers| std::mem::take(&mut buffers.uploads)) + } + + pub fn take_browse(&self, username: &str) -> Option> { + self.0.lock().ok()?.browses.remove(username) + } + + pub fn loss(&self) -> Option { + self.0.lock().ok()?.loss + } + + fn take(&self, drain: impl FnOnce(&mut Buffers) -> T) -> T { + self.0.lock().map_or_else(|_| T::default(), |mut b| drain(&mut b)) + } +} + +fn parse( + event: Event, + params: serde_json::Value, +) -> Result { + serde_json::from_value(params).map_err(|e| { + soulseek_rs::warn!("ignoring an unreadable {event} event: {e}"); + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::daemon::proto::DownloadStatusDto; + use serde_json::json; + + #[test] + fn a_room_event_is_delivered_once_and_then_gone() { + let mirror = Mirror::default(); + mirror.apply( + Event::Room, + json!({ "event": "left", "room": "lobby" }), + ); + + assert_eq!( + mirror.take_rooms(), + [RoomEvent::Left { + room: "lobby".into() + }] + ); + assert!( + mirror.take_rooms().is_empty(), + "draining twice must not replay what was already handed over" + ); + } + + #[test] + fn a_browse_result_is_keyed_by_peer() { + let mirror = Mirror::default(); + mirror.apply( + Event::Browse, + json!({ + "username": "bob", + "directories": [{ "name": "@@x", "files": [["a.mp3", 42]] }], + }), + ); + + assert!(mirror.take_browse("sue").is_none()); + let listing = mirror.take_browse("bob").expect("bob answered"); + assert_eq!(listing[0].name, "@@x"); + assert_eq!(listing[0].files, [("a.mp3".to_string(), 42)]); + assert!(mirror.take_browse("bob").is_none(), "taken means taken"); + } + + #[test] + fn a_transfer_update_reaches_the_caller_waiting_on_it() { + let mirror = Mirror::default(); + let (sender, receiver) = std::sync::mpsc::channel(); + mirror.expect_download("bob", "a.mp3", sender); + + mirror.apply( + Event::DownloadStatus, + serde_json::to_value(DownloadStatusEvent { + username: "bob".into(), + filename: "a.mp3".into(), + status: DownloadStatusDto::Queued, + }) + .unwrap(), + ); + + assert!(matches!(receiver.try_recv(), Ok(DownloadStatus::Queued))); + } + + #[test] + fn a_finished_transfer_stops_being_tracked() { + let mirror = Mirror::default(); + let (sender, receiver) = std::sync::mpsc::channel(); + mirror.expect_download("bob", "a.mp3", sender); + + for status in [DownloadStatusDto::Completed, DownloadStatusDto::Queued] + { + mirror.apply( + Event::DownloadStatus, + serde_json::to_value(DownloadStatusEvent { + username: "bob".into(), + filename: "a.mp3".into(), + status, + }) + .unwrap(), + ); + } + + assert!(matches!(receiver.try_recv(), Ok(DownloadStatus::Completed))); + assert!( + receiver.try_recv().is_err(), + "a completed transfer must not keep receiving updates" + ); + } + + #[test] + fn an_update_for_an_unknown_transfer_is_harmless() { + let mirror = Mirror::default(); + mirror.apply( + Event::DownloadStatus, + serde_json::to_value(DownloadStatusEvent { + username: "nobody".into(), + filename: "x.mp3".into(), + status: DownloadStatusDto::Completed, + }) + .unwrap(), + ); + } + + #[test] + fn the_first_session_loss_is_the_one_reported() { + let mirror = Mirror::default(); + assert!(mirror.loss().is_none()); + mirror.apply(Event::SessionLoss, json!({ "loss": "displaced" })); + mirror.apply(Event::SessionLoss, json!({ "loss": "disconnected" })); + assert_eq!( + mirror.loss(), + Some(SessionLoss::Displaced), + "being displaced is the reason worth reporting; the dropped socket \ + is its consequence" + ); + } + + #[test] + fn an_unreadable_event_is_ignored_rather_than_fatal() { + // A newer daemon may send a shape this client does not know. Dropping + // one event beats dropping the connection. + let mirror = Mirror::default(); + mirror.apply(Event::Room, json!({ "event": "invented" })); + assert!(mirror.take_rooms().is_empty()); + } +} diff --git a/soulseek-rs/src/remote/mod.rs b/soulseek-rs/src/remote/mod.rs index 179adb7..ff2bec8 100644 --- a/soulseek-rs/src/remote/mod.rs +++ b/soulseek-rs/src/remote/mod.rs @@ -1 +1,887 @@ -//! placeholder +//! Talking to a daemon as if it were a local client. +//! +//! [`RemoteSession`] implements the same [`SessionApi`] the commands and the +//! TUI already use, so nothing above the seam knows whether the session is in +//! this process or another one. Request/response methods are RPCs; the +//! destructive `take_*` drains read from a mirror the daemon feeds by pushing. + +mod mirror; + +use crate::api::SessionApi; +use crate::daemon::proto::{ + Ack, AuthParams, AuthResult, CODE_APPLICATION, ChatMessageDto, + DaemonStatus, DirectoriesParams, DownloadDto, DownloadStartParams, + DownloadStarted, Downloads, Event, IntervalSeconds, Members, MessageParams, + Messages, Method, PROTOCOL_VERSION, QueryParams, Request, Response, + RoomRef, RpcError, SayParams, SearchResults, Seconds, SharesStatus, + SlotsParams, TransferRef, Uploads, UserRef, UserResult, +}; +use mirror::Mirror; +use serde::de::DeserializeOwned; +use soulseek_rs::types::{Download, DownloadMetadata}; +use soulseek_rs::{ + DownloadStatus, Result, RoomEvent, SearchResult, SessionLoss, + SharedDirectory, SoulseekRs, UploadInfo, UserInfo, UserMessage, +}; +use std::collections::HashMap; +use std::io::{BufRead, BufReader, Write}; +use std::net::TcpStream; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::mpsc::{Receiver, Sender, channel}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +/// How long to wait for the daemon to answer one request. Every method is +/// either a lock-and-read or a message onto an already-open socket, so a +/// daemon that has not answered by now is wedged, not busy. +const CALL_TIMEOUT: Duration = Duration::from_secs(30); + +/// Where a daemon is listening. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Endpoint { + /// A Unix socket path: local, and authenticated by its permissions. + Socket(PathBuf), + /// `host:port`: needs the token. + Tcp(String), +} + +impl Endpoint { + /// Read what the user wrote. The distinction is unambiguous because a + /// socket path always has a separator and `host:port` never does. + #[must_use] + pub fn parse(value: &str) -> Self { + if value.contains(std::path::MAIN_SEPARATOR) || value.contains('/') { + return Self::Socket(PathBuf::from(value)); + } + Self::Tcp(value.to_string()) + } + + /// Whether the transport itself proves who we are, in which case no token + /// has to be found. + const fn self_authenticating(&self) -> bool { + matches!(self, Self::Socket(_)) + } +} + +impl std::fmt::Display for Endpoint { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Socket(path) => write!(f, "{}", path.display()), + Self::Tcp(address) => f.write_str(address), + } + } +} + +/// Calls awaiting a reply, keyed by JSON-RPC id. +type Answer = std::result::Result; + +#[derive(Default)] +struct Pending(Mutex>>); + +impl Pending { + fn register(&self, id: u64) -> Receiver { + let (sender, receiver) = channel(); + if let Ok(mut pending) = self.0.lock() { + pending.insert(id, sender); + } + receiver + } + + fn settle(&self, id: u64, outcome: Answer) { + let sender = self.0.lock().ok().and_then(|mut p| p.remove(&id)); + if let Some(sender) = sender { + let _ = sender.send(outcome); + } + } + + fn abandon(&self, id: u64) { + if let Ok(mut pending) = self.0.lock() { + pending.remove(&id); + } + } + + /// Fail everything still waiting. Called when the reader stops, so callers + /// blocked on a dead connection are told rather than left to time out. + fn fail_all(&self) { + let Ok(mut pending) = self.0.lock() else { + return; + }; + for (_, sender) in pending.drain() { + let _ = sender.send(Err(RpcError::new( + CODE_APPLICATION, + "the daemon closed the connection", + ))); + } + } +} + +/// The half of a connection that can be closed from another thread. +/// +/// Dropping the write half is not enough to end the reader: it is blocked in +/// `read` on its own cloned handle, so the socket would stay open and the +/// daemon would keep a connection — and a hub subscription — for a client that +/// has gone. An explicit shutdown is what makes a dropped session actually +/// disconnect. +enum Closer { + Tcp(TcpStream), + #[cfg(unix)] + Unix(std::os::unix::net::UnixStream), +} + +impl Closer { + fn close(&self) { + let _ = match self { + Self::Tcp(stream) => stream.shutdown(std::net::Shutdown::Both), + #[cfg(unix)] + Self::Unix(stream) => stream.shutdown(std::net::Shutdown::Both), + }; + } +} + +/// One connection to a daemon, presented as a session. +pub struct RemoteSession { + outgoing: Mutex>, + closer: Closer, + pending: Arc, + mirror: Arc, + next_id: AtomicU64, + username: String, + /// Set when the reader thread stops, so a call fails immediately rather + /// than waiting out its timeout against a socket nobody is reading. + closed: Arc, +} + +impl RemoteSession { + /// Connect and authenticate. + /// + /// `token` is required on TCP and optional on a Unix socket, where the + /// socket's permissions have already answered the question. + pub fn connect(endpoint: &Endpoint, token: Option) -> Result { + let (read, write, closer) = open(endpoint)?; + + let pending = Arc::new(Pending::default()); + let mirror = Arc::new(Mirror::default()); + let closed = Arc::new(AtomicBool::new(false)); + { + let pending = Arc::clone(&pending); + let mirror = Arc::clone(&mirror); + let closed = Arc::clone(&closed); + std::thread::spawn(move || { + read_loop(read, &pending, &mirror); + closed.store(true, Ordering::Relaxed); + pending.fail_all(); + }); + } + + let mut session = Self { + outgoing: Mutex::new(write), + closer, + pending, + mirror, + next_id: AtomicU64::new(1), + username: String::new(), + closed, + }; + + let token = if endpoint.self_authenticating() { + token + } else { + Some(token.ok_or_else(|| { + network( + "a remote daemon needs a token; see `soulseek-rs daemon \ + token`", + ) + })?) + }; + let authenticated: AuthResult = session.request( + Method::Auth, + AuthParams { + token, + protocol: PROTOCOL_VERSION, + }, + )?; + if authenticated.protocol != PROTOCOL_VERSION { + return Err(network(format!( + "the daemon speaks protocol {} and this client speaks \ + {PROTOCOL_VERSION}; upgrade whichever is older", + authenticated.protocol + ))); + } + + session.username = authenticated.username; + Ok(session) + } + + /// Send a request and wait for its reply. + fn request( + &self, + method: Method, + params: P, + ) -> Result { + let value = self.call(method, params)?; + serde_json::from_value(value).map_err(|e| { + network(format!( + "the daemon's answer to {method} was unreadable: {e}" + )) + }) + } + + fn call( + &self, + method: Method, + params: P, + ) -> Result { + if self.closed.load(Ordering::Relaxed) { + return Err(SoulseekRs::ConnectionClosed); + } + let id = self.next_id.fetch_add(1, Ordering::Relaxed); + let request = Request { + jsonrpc: "2.0".to_string(), + id: Some(serde_json::json!(id)), + method: method.wire_name().to_string(), + params: serde_json::to_value(params) + .map_err(|e| network(format!("cannot encode {method}: {e}")))?, + }; + let line = serde_json::to_string(&request) + .map_err(|e| network(format!("cannot encode {method}: {e}")))?; + + let answer = self.pending.register(id); + { + let mut outgoing = + self.outgoing.lock().map_err(|_| SoulseekRs::LockPoisoned)?; + if writeln!(outgoing, "{line}").is_err() || outgoing.flush().is_err() + { + self.pending.abandon(id); + return Err(SoulseekRs::ConnectionClosed); + } + } + + match answer.recv_timeout(CALL_TIMEOUT) { + Ok(Ok(value)) => Ok(value), + Ok(Err(error)) => Err(network(error.message)), + Err(_) => { + self.pending.abandon(id); + Err(SoulseekRs::Timeout) + } + } + } + + /// A call whose reply carries nothing worth reading. + fn tell( + &self, + method: Method, + params: P, + ) -> Result<()> { + self.call(method, params).map(|_| ()) + } + + /// A call whose reply is a yes or a no. A daemon that cannot answer is a + /// "no": every caller of these already reads false as "nothing to do". + fn ask(&self, method: Method, params: P) -> bool { + self.request::(method, params) + .is_ok_and(|ack| ack.ok) + } + + fn status(&self) -> Option { + self.request(Method::DaemonStatus, ()).ok() + } + + fn shares(&self) -> SharesStatus { + self.request(Method::SharesStatusOf, ()) + .unwrap_or(SharesStatus { + folders: 0, + files: 0, + directories: Vec::new(), + }) + } + + fn transfer(username: &str, filename: &str) -> TransferRef { + TransferRef { + username: username.to_string(), + filename: filename.to_string(), + } + } + + /// Queue a transfer and hand back a channel that behaves like the library's. + /// + /// The daemon owns the real transfer; this mints the channel the caller + /// expects and registers it with the mirror, so pushed updates arrive on it + /// exactly as a local client's would. + fn start_download( + &self, + filename: String, + username: String, + size: u64, + metadata: DownloadMetadata, + ) -> Result<(Download, Receiver)> { + let (sender, receiver) = channel(); + // Register before asking: a transfer that fails instantly would + // otherwise report into a mirror that is not listening yet. + self.mirror + .expect_download(&username, &filename, sender.clone()); + + let started: DownloadStarted = self + .request( + Method::DownloadStart, + DownloadStartParams { + username: username.clone(), + filename: filename.clone(), + size, + // The daemon writes to its own filesystem, which a remote + // caller cannot see and so does not get to choose. + download_directory: None, + metadata: (&metadata).into(), + }, + ) + .inspect_err(|_| self.mirror.forget_download(&username, &filename))?; + + Ok((rebuild(started.download, sender, metadata), receiver)) + } + + /// One parameterless request, for the control commands that only need to + /// ask a daemon something once. + pub fn ask_for(&self, method: Method) -> Result { + self.request(method, ()) + } + + /// The daemon's chat history, which a remote client needs to rebuild a + /// conversation it was not attached for. Not part of [`SessionApi`] + /// because a local session keeps this in the TUI's own snapshot. + pub fn message_history(&self) -> Result> { + self.request::<_, Messages>(Method::MessageHistory, ()) + .map(|history| history.messages) + } +} + +/// Turn the daemon's view of a transfer back into the library's, with the +/// caller's own channel spliced in. +fn rebuild( + dto: DownloadDto, + sender: Sender, + metadata: DownloadMetadata, +) -> Download { + Download { + username: dto.username, + filename: dto.filename, + token: dto.token, + size: dto.size, + download_directory: dto.download_directory, + status: dto.status.into(), + sender, + queue_position: dto.queue_position, + metadata, + } +} + +fn network(message: impl Into) -> SoulseekRs { + SoulseekRs::InvalidMessage(message.into()) +} + +type Halves = (Box, Box, Closer); + +fn open(endpoint: &Endpoint) -> Result { + match endpoint { + Endpoint::Tcp(address) => { + let stream = TcpStream::connect(address).map_err(|e| { + network(format!("cannot reach the daemon at {address}: {e}")) + })?; + let read = stream.try_clone().map_err(SoulseekRs::NetworkError)?; + let closer = + Closer::Tcp(stream.try_clone().map_err(SoulseekRs::NetworkError)?); + Ok((Box::new(read), Box::new(stream), closer)) + } + Endpoint::Socket(path) => connect_socket(path), + } +} + +#[cfg(unix)] +fn connect_socket(path: &Path) -> Result { + let stream = std::os::unix::net::UnixStream::connect(path).map_err(|e| { + network(format!("cannot reach the daemon at {}: {e}", path.display())) + })?; + let read = stream.try_clone().map_err(SoulseekRs::NetworkError)?; + let closer = + Closer::Unix(stream.try_clone().map_err(SoulseekRs::NetworkError)?); + Ok((Box::new(read), Box::new(stream), closer)) +} + +#[cfg(not(unix))] +fn connect_socket(path: &Path) -> Result { + Err(network(format!( + "this platform has no Unix sockets, so {} cannot be a daemon address; \ + use host:port", + path.display() + ))) +} + +impl Drop for RemoteSession { + fn drop(&mut self) { + // Tell the daemon we are gone, so it can release the connection and + // stop fanning events at a client that is not there. + self.closer.close(); + } +} + +/// Route everything the daemon sends: replies to the call waiting for them, +/// events to the mirror. +fn read_loop( + read: Box, + pending: &Pending, + mirror: &Mirror, +) { + for line in BufReader::new(read).lines() { + let Ok(line) = line else { + return; + }; + if line.trim().is_empty() { + continue; + } + let Ok(response) = serde_json::from_str::(&line) else { + soulseek_rs::warn!("ignoring an unreadable line from the daemon"); + continue; + }; + + if let Some(method) = response.method.as_deref() { + if let Some(event) = Event::from_wire(method) { + mirror.apply(event, response.params.unwrap_or_default()); + } + continue; + } + + let Some(id) = response.id.as_ref().and_then(serde_json::Value::as_u64) + else { + continue; + }; + match (response.result, response.error) { + (Some(result), _) => pending.settle(id, Ok(result)), + (None, Some(error)) => pending.settle(id, Err(error)), + (None, None) => pending.settle(id, Ok(serde_json::Value::Null)), + } + } +} + +impl SessionApi for RemoteSession { + fn username(&self) -> String { + self.username.clone() + } + + fn listen_port(&self) -> Option { + self.status().and_then(|status| status.listen_port) + } + + fn session_loss(&self) -> Option { + // The pushed event is the prompt answer; the status call also covers a + // loss that happened before this client attached. + self.mirror.loss().or_else(|| { + self.status() + .and_then(|status| status.session_loss) + .map(Into::into) + }) + } + + fn search( + &self, + query: &str, + timeout: Duration, + ) -> Result> { + self.search_with_cancel(query, timeout, None) + } + + fn search_with_cancel( + &self, + query: &str, + timeout: Duration, + cancel: Option>, + ) -> Result> { + self.tell( + Method::SearchStart, + QueryParams { + query: query.to_string(), + }, + )?; + // The window is spent here rather than in the daemon: one client's + // search must not stop the daemon answering everybody else. + soulseek_rs::Client::collect_for(timeout, cancel); + Ok(self.get_search_results(query)) + } + + fn get_search_results(&self, key: &str) -> Vec { + self.request::<_, SearchResults>( + Method::SearchResults, + QueryParams { + query: key.to_string(), + }, + ) + .map(|found| found.results.into_iter().map(Into::into).collect()) + .unwrap_or_default() + } + + fn get_search_results_count(&self, key: &str) -> usize { + self.get_search_results(key).len() + } + + fn try_get_search_results(&self, key: &str) -> Option> { + Some(self.get_search_results(key)) + } + + fn start_wishlist_search(&self, query: &str) -> Result<()> { + self.tell( + Method::SearchWishlist, + QueryParams { + query: query.to_string(), + }, + ) + } + + fn wishlist_interval(&self) -> Duration { + self.request::<_, IntervalSeconds>(Method::SearchWishlistInterval, ()) + .map_or(Duration::from_mins(12), |interval| { + Duration::from_secs(interval.seconds) + }) + } + + fn download( + &self, + filename: String, + username: String, + size: u64, + _download_directory: String, + // the daemon's own download directory wins — a remote caller + // cannot see the filesystem the bytes land on. Plumb a per-transfer + // directory through if a client ever needs to sort into subfolders. + ) -> Result<(Download, Receiver)> { + self.start_download( + filename, + username, + size, + DownloadMetadata::default(), + ) + } + + fn download_with_metadata( + &self, + filename: String, + username: String, + size: u64, + _download_directory: String, + metadata: DownloadMetadata, + ) -> Result<(Download, Receiver)> { + self.start_download(filename, username, size, metadata) + } + + fn get_all_downloads(&self) -> Vec { + let Ok(listed) = self.request::<_, Downloads>(Method::DownloadList, ()) + else { + return Vec::new(); + }; + listed + .downloads + .into_iter() + .map(|dto| { + // Nothing reads this sender: a listing covers other clients' + // transfers too, and the ones this client started already have + // a live channel from `download`. + let (sender, _) = channel(); + let metadata = dto.metadata.clone().into(); + rebuild(dto, sender, metadata) + }) + .collect() + } + + fn pause_download(&self, username: &str, filename: &str) -> bool { + self.ask(Method::DownloadPause, Self::transfer(username, filename)) + } + + fn resume_download(&self, username: &str, filename: &str) -> bool { + self.ask(Method::DownloadResume, Self::transfer(username, filename)) + } + + fn remove_queued_download(&self, username: &str, filename: &str) -> bool { + self.ask( + Method::DownloadRemoveQueued, + Self::transfer(username, filename), + ) + } + + fn remove_download(&self, username: &str, filename: &str) -> bool { + self.mirror.forget_download(username, filename); + self.ask(Method::DownloadRemove, Self::transfer(username, filename)) + } + + fn uploads(&self) -> Vec { + self.request::<_, Uploads>(Method::UploadList, ()) + .map(|listed| listed.uploads.into_iter().map(Into::into).collect()) + .unwrap_or_default() + } + + fn take_upload_events(&self) -> Vec { + self.mirror.take_uploads() + } + + fn cancel_upload(&self, username: &str, filename: &str) -> bool { + self.ask(Method::UploadCancel, Self::transfer(username, filename)) + } + + fn set_upload_slots(&self, slots: usize) { + let _ = self.tell(Method::UploadSlots, SlotsParams { slots }); + } + + fn check_privileges(&self) -> Result<()> { + self.tell(Method::PrivilegesCheck, ()) + } + + fn own_privilege_seconds(&self) -> Option { + self.request::<_, Seconds>(Method::PrivilegesOwn, ()) + .ok()? + .seconds + } + + fn request_room_list(&self) -> Result<()> { + self.tell(Method::RoomListRequest, ()) + } + + fn join_room(&self, room: &str) -> Result<()> { + self.tell( + Method::RoomJoin, + RoomRef { + room: room.to_string(), + }, + ) + } + + fn leave_room(&self, room: &str) -> Result<()> { + self.tell( + Method::RoomLeave, + RoomRef { + room: room.to_string(), + }, + ) + } + + fn say_in_room(&self, room: &str, message: &str) -> Result<()> { + self.tell( + Method::RoomSay, + SayParams { + room: room.to_string(), + message: message.to_string(), + }, + ) + } + + fn room_members(&self, room: &str) -> Vec { + self.request::<_, Members>( + Method::RoomMembers, + RoomRef { + room: room.to_string(), + }, + ) + .map(|members| members.users) + .unwrap_or_default() + } + + fn take_room_events(&self) -> Vec { + self.mirror.take_rooms() + } + + fn send_private_message( + &self, + username: &str, + message: &str, + ) -> Result<()> { + self.tell( + Method::MessageSend, + MessageParams { + username: username.to_string(), + message: message.to_string(), + }, + ) + } + + fn take_private_messages(&self) -> Vec { + self.mirror.take_messages() + } + + fn browse_user(&self, username: &str) -> Result<()> { + self.tell( + Method::BrowseUser, + UserRef { + username: username.to_string(), + }, + ) + } + + fn take_browse_result( + &self, + username: &str, + ) -> Option> { + self.mirror.take_browse(username) + } + + fn request_user_info(&self, username: &str) -> Result<()> { + self.tell( + Method::UserRequest, + UserRef { + username: username.to_string(), + }, + ) + } + + fn user_info(&self, username: &str) -> Option { + self.request::<_, UserResult>( + Method::UserInfoOf, + UserRef { + username: username.to_string(), + }, + ) + .ok()? + .user + .map(Into::into) + } + + fn shared_counts(&self) -> (u32, u32) { + let shares = self.shares(); + (shares.folders, shares.files) + } + + fn shared_directories(&self) -> Vec { + self.shares().directories + } + + fn set_shared_directories(&self, directories: Vec) -> Result<()> { + self.tell(Method::SharesSet, DirectoriesParams { directories }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn an_endpoint_tells_a_socket_path_from_a_host_and_port() { + assert_eq!( + Endpoint::parse("/run/soulseek/daemon.sock"), + Endpoint::Socket(PathBuf::from("/run/soulseek/daemon.sock")) + ); + assert_eq!( + Endpoint::parse("127.0.0.1:5030"), + Endpoint::Tcp("127.0.0.1:5030".to_string()) + ); + assert_eq!( + Endpoint::parse("daemon.example:5030"), + Endpoint::Tcp("daemon.example:5030".to_string()) + ); + } + + #[test] + fn only_a_socket_authenticates_itself() { + assert!(Endpoint::parse("/tmp/daemon.sock").self_authenticating()); + assert!(!Endpoint::parse("host:5030").self_authenticating()); + } + + #[test] + fn a_reply_settles_the_call_that_is_waiting_for_it() { + let pending = Pending::default(); + let answer = pending.register(7); + pending.settle(7, Ok(serde_json::json!({ "ok": true }))); + assert_eq!( + answer.recv().unwrap().unwrap(), + serde_json::json!({ "ok": true }) + ); + } + + #[test] + fn a_reply_for_an_abandoned_call_is_dropped_quietly() { + let pending = Pending::default(); + let answer = pending.register(7); + pending.abandon(7); + pending.settle(7, Ok(serde_json::Value::Null)); + assert!(answer.recv().is_err()); + } + + #[test] + fn a_closed_connection_fails_every_waiting_call() { + // Otherwise each blocked caller would sit out the full timeout against + // a socket nobody is reading. + let pending = Pending::default(); + let first = pending.register(1); + let second = pending.register(2); + pending.fail_all(); + assert!(first.recv().unwrap().is_err()); + assert!(second.recv().unwrap().is_err()); + } + + #[test] + fn the_reader_routes_replies_to_calls_and_events_to_the_mirror() { + let pending = Pending::default(); + let mirror = Mirror::default(); + let answer = pending.register(1); + + let lines = concat!( + r#"{"jsonrpc":"2.0","method":"event.room","params":{"event":"left","room":"lobby"}}"#, + "\n", + r#"{"jsonrpc":"2.0","id":1,"result":{"ok":true}}"#, + "\n", + ); + read_loop(Box::new(lines.as_bytes()), &pending, &mirror); + + assert_eq!( + answer.recv().unwrap().unwrap(), + serde_json::json!({ "ok": true }) + ); + assert_eq!( + mirror.take_rooms(), + [RoomEvent::Left { + room: "lobby".into() + }] + ); + } + + #[test] + fn an_error_reply_reaches_the_caller_as_an_error() { + let pending = Pending::default(); + let mirror = Mirror::default(); + let answer = pending.register(1); + + read_loop( + Box::new( + concat!( + r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32000,"message":"nope","data":{"exit":4}}}"#, + "\n" + ) + .as_bytes(), + ), + &pending, + &mirror, + ); + + let error = answer.recv().unwrap().expect_err("an error reply"); + assert_eq!(error.message, "nope"); + assert_eq!(error.exit(), crate::output::Exit::NoResults); + } + + #[test] + fn a_junk_line_does_not_stop_the_reader() { + let pending = Pending::default(); + let mirror = Mirror::default(); + let answer = pending.register(1); + + read_loop( + Box::new( + concat!( + "not json at all\n", + r#"{"jsonrpc":"2.0","id":1,"result":{"ok":true}}"#, + "\n" + ) + .as_bytes(), + ), + &pending, + &mirror, + ); + + assert!( + answer.recv().unwrap().is_ok(), + "one bad line must not cost the reply behind it" + ); + } +}