ipodderx-rs/src/torrent.rs

//! Torrent enclosures via librqbit. Replaces iPXDownloader.getTorrent and the vendored
//! BitTorrent 4.2.1 + khashmir tree.

use anyhow::{Context, Result, bail};
use librqbit::{AddTorrent, AddTorrentOptions, AddTorrentResponse, Session, SessionOptions};
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, Instant};

use crate::config::Config;

pub struct Torrents {
    session: Arc<Session>,
}

/// How long a torrent may make no progress before we give up on it. The original called
/// this torrentMaxBeatTime and counted display ticks; a wall clock is easier to reason about.
fn stall_limit(cfg: &Config) -> Duration {
    Duration::from_secs(cfg.torrent.stall_mins.max(1) * 60)
}

impl Torrents {
    pub async fn new(cfg: &Config) -> Result<Self> {
        let (lo, _hi) = cfg.torrent.ports();
        let listen_addr: SocketAddr = format!("0.0.0.0:{lo}").parse()?;

        let opts = SessionOptions {
            listen: Some(librqbit::ListenerOptions {
                listen_addr,
                ..Default::default()
            }),
            client_name_and_version: Some(concat!("ipx ", env!("CARGO_PKG_VERSION")).into()),
            ..Default::default()
        };

        // The session root is only a fallback; every torrent names its own output folder.
        let session = Session::new_with_opts(cfg.general.download_dir.clone(), opts)
            .await
            .context("starting the torrent session")?;
        Ok(Self { session })
    }

    /// Downloads a torrent (a .torrent URL or a magnet link) straight into `dest_dir`,
    /// seeds it up to the configured limit, then stops.
    ///
    /// Downloading in place rather than into a staging dir is deliberate: seeding serves
    /// the files it downloaded, so moving them first would break it.
    pub async fn fetch(
        &self,
        cfg: &Config,
        url: &str,
        dest_dir: &Path,
        mut on_progress: impl FnMut(u64, u64),
    ) -> Result<(PathBuf, u64)> {
        tokio::fs::create_dir_all(dest_dir).await?;

        let limit = stall_limit(cfg);

        // Resolving a magnet's metadata happens inside add_torrent, and against a dead
        // swarm it never returns. The stall budget has to cover this phase too, or a
        // torrent nobody is seeding wedges the scan forever.
        let added = tokio::time::timeout(
            limit,
            self.session.add_torrent(
                AddTorrent::from_url(url),
                Some(AddTorrentOptions {
                    output_folder: Some(dest_dir.to_string_lossy().into_owned()),
                    overwrite: true,
                    ..Default::default()
                }),
            ),
        )
        .await
        .map_err(|_| {
            anyhow::anyhow!(
                "no metadata after {} minutes, gave up",
                limit.as_secs() / 60
            )
        })?
        .context("adding the torrent")?;

        let (id, handle) = match added {
            AddTorrentResponse::Added(id, h) | AddTorrentResponse::AlreadyManaged(id, h) => (id, h),
            AddTorrentResponse::ListOnly(_) => bail!("torrent added in list-only mode"),
        };

        let mut last_progress = 0u64;
        let mut last_move = Instant::now();

        loop {
            tokio::time::sleep(Duration::from_secs(1)).await;
            let stats = handle.stats();

            if let Some(err) = &stats.error {
                let _ = self.session.delete(id.into(), false).await;
                bail!("torrent failed: {err}");
            }
            on_progress(stats.progress_bytes, stats.total_bytes);

            if stats.finished {
                break;
            }
            if stats.progress_bytes > last_progress {
                last_progress = stats.progress_bytes;
                last_move = Instant::now();
            } else if last_move.elapsed() > limit {
                // Leave the partial files behind; the row records the failure.
                let _ = self.session.delete(id.into(), false).await;
                bail!("no progress for {} minutes, gave up", limit.as_secs() / 60);
            }
        }

        let name = handle.name().unwrap_or_else(|| "torrent".to_owned());
        let path = handle.output_folder().join(&name);
        let size = handle.stats().total_bytes;

        self.seed(cfg, &handle).await;
        self.session
            .delete(id.into(), false)
            .await
            .context("releasing the torrent")?;

        Ok((path, size))
    }

    /// Seeds until the ratio or the time limit is reached, whichever comes first.
    async fn seed(&self, cfg: &Config, handle: &Arc<librqbit::ManagedTorrent>) {
        if cfg.torrent.seed_ratio <= 0.0 || cfg.torrent.seed_time_mins == 0 {
            return;
        }
        let deadline = Instant::now() + Duration::from_secs(cfg.torrent.seed_time_mins * 60);
        loop {
            let stats = handle.stats();
            if ratio(stats.uploaded_bytes, stats.total_bytes) >= cfg.torrent.seed_ratio {
                tracing::info!(ratio = cfg.torrent.seed_ratio, "seed ratio reached");
                return;
            }
            if Instant::now() >= deadline {
                tracing::info!(mins = cfg.torrent.seed_time_mins, "seed time reached");
                return;
            }
            tokio::time::sleep(Duration::from_secs(5)).await;
        }
    }
}

/// Uploaded over total. A zero-byte torrent counts as fully seeded rather than dividing by zero.
pub fn ratio(uploaded: u64, total: u64) -> f64 {
    if total == 0 {
        return f64::INFINITY;
    }
    uploaded as f64 / total as f64
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn ratio_handles_the_empty_torrent() {
        assert_eq!(ratio(0, 100), 0.0);
        assert_eq!(ratio(50, 100), 0.5);
        assert_eq!(ratio(200, 100), 2.0);
        assert!(ratio(0, 0).is_infinite(), "must not divide by zero or seed forever");
    }

    #[test]
    fn stall_limit_is_never_zero() {
        let mut cfg = Config::default();
        cfg.torrent.stall_mins = 0;
        assert_eq!(stall_limit(&cfg), Duration::from_secs(60), "0 would abort instantly");
        cfg.torrent.stall_mins = 30;
        assert_eq!(stall_limit(&cfg), Duration::from_secs(1800));
    }
}