soulseek-rs-modded/soulseek-rs/src/commands/peer.rs

//! Being a peer on the network: `serve`, `whoami`, and `user`.

use super::{Ctx, Session};
use crate::cli::{ServeArgs, UserArgs, WatchCommand};
use crate::output::{
    CliError, CliResult, UploadRecord, UserRecord, WatchedUserRecord,
    WhoamiRecord,
};
use soulseek_rs::types::{UploadInfo, UploadStatus, UserInfo};
use std::collections::HashMap;
use std::time::{Duration, Instant};

/// How often the serve loop samples the upload table.
const POLL: Duration = Duration::from_millis(500);

/// How long `whoami` waits for the server's privilege answer. It is one small
/// message on an already-open connection, so a server that has not replied by
/// now is not going to.
const PRIVILEGE_TIMEOUT: Duration = Duration::from_secs(5);

pub fn whoami(ctx: &Ctx) -> CliResult {
    let session = Session::open(ctx)?;
    let (folders, files) = session.client.shared_counts();
    // The bound port, not the configured one: a caller checking whether it is
    // reachable needs what is true, not what was asked for.
    let bound = session.client.listen_port();
    ctx.out.emit(&WhoamiRecord {
        // 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),
        shared_folders: folders,
        shared_files: files,
        download_dir: ctx.download_dir.clone(),
        privilege_seconds: own_privileges(session.client.as_ref()),
    });
    Ok(())
}

/// Ask the server how much privilege time this account has left.
///
/// A server that does not answer leaves the field null rather than reporting a
/// zero, because "no privileges" and "did not say" are different facts.
fn own_privileges(client: &dyn crate::api::SessionApi) -> Option<u32> {
    if client.check_privileges().is_err() {
        return None;
    }
    super::poll_until(PRIVILEGE_TIMEOUT, Duration::from_millis(50), || {
        client.own_privilege_seconds()
    })
}

pub fn user(ctx: &Ctx, args: &UserArgs) -> CliResult {
    let session = Session::open(ctx)?;
    session.client.request_user_info(&args.user).map_err(|e| {
        CliError::connection(format!("cannot ask about {}: {e}", args.user))
    })?;
    ctx.out
        .status(&format!("asking the server about {}…", args.user));

    // Presence and statistics are two replies; wait for both, but report
    // whichever arrived rather than defaulting the other to something that
    // reads like an answer.
    let deadline = Instant::now() + Duration::from_secs(args.timeout);
    let mut latest = None;
    while Instant::now() < deadline {
        latest = session.client.user_info(&args.user);
        if latest.as_ref().is_some_and(UserInfo::is_complete) {
            break;
        }
        std::thread::sleep(Duration::from_millis(100));
    }

    let Some(info) = latest else {
        return Err(CliError::timeout(format!(
            "the server said nothing about {} within {}s",
            args.user, args.timeout
        )));
    };
    if !info.is_complete() {
        ctx.out
            .warn("the server answered only partly; unknown fields are null");
    }
    ctx.out.emit(&UserRecord {
        user: info.username,
        status: info.presence.map(|p| p.status.to_string()),
        privileged: info.presence.map(|p| p.privileged),
        average_speed: info.stats.map(|s| s.average_speed),
        shared_files: info.stats.map(|s| s.shared_files),
        shared_folders: info.stats.map(|s| s.shared_folders),
    });
    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(
    ctx: &Ctx,
    args: &ServeArgs,
    mut sweeper: Option<crate::commands::wish::Sweeper>,
) -> CliResult {
    // Cheap check first, so a misconfigured local run is rejected before it
    // bothers the server with a login.
    if ctx.daemon.is_none() {
        if ctx.settings.shared_directories.is_empty() {
            return Err(nothing_to_share());
        }
        if !ctx.settings.enable_listen {
            return Err(needs_a_listener());
        }
    }

    let session = Session::open(ctx)?;
    // What the session actually offers is what decides. Routed to a daemon
    // that is what the daemon shares, which is the only thing peers can see —
    // this run's own configuration has no bearing on it.
    if session.client.shared_directories().is_empty() {
        return Err(nothing_to_share());
    }
    if session.client.listen_port().is_none() {
        return Err(needs_a_listener());
    }
    if let Some(slots) = args.upload_slots {
        session.client.set_upload_slots(usize::from(slots));
        ctx.out.status(&format!("{slots} upload slots"));
    }
    let (folders, files) = session.client.shared_counts();
    ctx.out.status(&format!(
        "serving {files} files in {folders} folders on port {}",
        session.client.listen_port().unwrap_or_default()
    ));
    if files == 0 {
        ctx.out
            .warn("the share index is empty — peers will find nothing");
    }

    let span = super::listen_span(args.duration, args.follow);
    ctx.out.status(&match span {
        Some(span) => format!("online for {}s…", span.as_secs()),
        None => "online until interrupted…".to_string(),
    });

    let deadline = span.map(|span| Instant::now() + span);
    let mut reported: HashMap<(String, String), UploadRecord> = HashMap::new();
    while deadline.is_none_or(|deadline| Instant::now() < deadline) {
        // Queued states first, and from the client's own record rather than
        // this snapshot: with few slots and small files a peer can queue and be
        // served well inside one poll, and "it had to wait" is precisely what a
        // transfer log exists to report.
        for queued in session.client.take_upload_events() {
            emit_changed(&mut reported, upload_record(&queued), &ctx.out);
        }
        for upload in session.client.uploads() {
            emit_changed(&mut reported, upload_record(&upload), &ctx.out);
        }
        if let Some(sweeper) = sweeper.as_mut() {
            sweeper.tick(ctx, session.client.as_ref());
        }
        std::thread::sleep(POLL);
    }
    Ok(())
}

fn nothing_to_share() -> CliError {
    CliError::usage(
        "nothing to share: pass --shared-dir, or add one with `shares add`",
    )
}

fn needs_a_listener() -> CliError {
    CliError::usage("serving needs the listener: drop --no-listener")
}

/// Flatten a live upload into the record `serve` reports.
fn upload_record(upload: &UploadInfo) -> UploadRecord {
    let (status, reason) = match &upload.status {
        UploadStatus::Queued(place) => {
            ("queued", Some(format!("place {place}")))
        }
        UploadStatus::InProgress => ("uploading", None),
        UploadStatus::Completed => ("completed", None),
        UploadStatus::Cancelled => ("cancelled", None),
        UploadStatus::Failed(why) => ("failed", Some(why.clone())),
    };
    UploadRecord {
        user: upload.username.clone(),
        path: upload.filename.clone(),
        size: upload.size,
        bytes_sent: upload.bytes_sent,
        speed: upload.speed_bytes_per_sec,
        status: status.to_string(),
        reason,
    }
}

/// Emit `record` only when it differs from the last one reported for that
/// (user, file).
fn emit_changed(
    reported: &mut HashMap<(String, String), UploadRecord>,
    record: UploadRecord,
    out: &crate::output::Out,
) -> bool {
    let key = (record.user.clone(), record.path.clone());
    if reported.get(&key) == Some(&record) {
        return false;
    }
    out.emit(&record);
    reported.insert(key, record);
    true
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::output::{Out, Record};

    fn upload(status: UploadStatus, bytes_sent: u64) -> UploadInfo {
        UploadInfo {
            username: "bob".to_string(),
            filename: "@@x\\track.mp3".to_string(),
            size: 100,
            bytes_sent,
            status,
            speed_bytes_per_sec: 1.0,
        }
    }

    #[test]
    fn an_upload_maps_to_its_wire_status() {
        assert_eq!(
            upload_record(&upload(UploadStatus::InProgress, 5)).status,
            "uploading"
        );
        assert_eq!(
            upload_record(&upload(UploadStatus::Completed, 100)).status,
            "completed"
        );
        assert_eq!(
            upload_record(&upload(UploadStatus::Cancelled, 5)).status,
            "cancelled"
        );

        let failed =
            upload_record(&upload(UploadStatus::Failed("gone".into()), 5));
        assert_eq!(failed.status, "failed");
        assert_eq!(failed.reason.as_deref(), Some("gone"));
    }

    #[test]
    fn the_upload_line_leads_with_status_and_progress() {
        let record = upload_record(&upload(UploadStatus::InProgress, 42));
        assert_eq!(record.text(), "uploading\tbob\t42/100\t@@x\\track.mp3");
    }

    #[test]
    fn an_unchanged_upload_is_reported_once() {
        let out = Out::new(false, true);
        let mut reported = HashMap::new();
        let record = upload_record(&upload(UploadStatus::InProgress, 42));

        assert!(emit_changed(&mut reported, record.clone(), &out));
        assert!(
            !emit_changed(&mut reported, record, &out),
            "polling must not re-report an upload that has not moved"
        );
    }

    #[test]
    fn progress_and_status_changes_are_both_reported() {
        let out = Out::new(false, true);
        let mut reported = HashMap::new();
        emit_changed(
            &mut reported,
            upload_record(&upload(UploadStatus::InProgress, 10)),
            &out,
        );
        assert!(
            emit_changed(
                &mut reported,
                upload_record(&upload(UploadStatus::InProgress, 60)),
                &out
            ),
            "more bytes sent is news"
        );
        assert!(
            emit_changed(
                &mut reported,
                upload_record(&upload(UploadStatus::Completed, 100)),
                &out
            ),
            "finishing is news"
        );
    }
}