diff --git a/PROGRESS.md b/PROGRESS.md index a9006e7..f6ea08c 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -13,8 +13,9 @@ The full design and step list live in the plan file at - [x] **4. `download.rs`** — downloads, filters, dedupe. - [x] **5. `retention.rs`** — oldest-first quota + age reaper. - [x] **6. `ipc.rs` + daemon** — UDS JSON-lines server, TTL scheduler, CLI-proxies-to-daemon. -- [ ] **7. `torrent.rs`** — librqbit, seed to ratio/time, stall abort. *Done when:* smoke 5 passes. -- [ ] **8. OPML + polish** — import/export, add/rm/status, tracing setup, systemd unit, README. +- [x] **7. `torrent.rs`** — librqbit, seed to ratio/time, stall abort. (swarm download unverified — + see the step 7 entry) +- [x] **8. OPML + polish** — import/export, add/rm/status, tracing setup, systemd units, README. ## Smoke tests @@ -29,6 +30,54 @@ The full design and step list live in the plan file at --- +## 2026-09-09 — Step 8: OPML and polish + +`ipx add ` fetches the feed to name it from its own title (`Accidental Tech Podcast` -> +`accidental-tech-podcast`); a feed that cannot be reached is still added, named from its URL, rather +than refused. `ipx rm` leaves downloads and history alone, so re-adding a feed does not re-pull its +back catalogue. `ipx import`/`export` walk nested OPML folder outlines and skip URLs already +subscribed. `tracing` logs to stderr, `IPX_LOG` sets the filter. `contrib/` has a systemd user unit +for the daemon, plus a timer and one-shot service for the no-daemon style (with the caveat that +without a daemon there is no socket for a UI). + +Verified: `cargo test` 27/27. Round-trip — exported 3 feeds with real titles, removed one, +re-imported: exactly the missing one came back, no duplicates, config still mode 0600. Quickstart +from a genuinely empty home with no env overrides: `list` on a missing config, `add`, `fetch` +(1 downloaded, 1 explicit skipped, 1 torrent 404), `list`; every path resolved under `$HOME`. + +--- + +## 2026-09-09 — Step 7: torrent.rs + +`src/torrent.rs`: a librqbit `Session` started lazily on first torrent (binding ports and starting a +DHT for a config that has never seen a torrent would be rude). `.torrent` URLs and magnets both go +through `AddTorrent::from_url`. Progress is polled once a second and reported through the same +`Progress` events HTTP downloads use. + +Downloads go **straight into the feed folder** rather than staging and moving. The plan said move +then seed, which cannot work — seeding serves the files it downloaded, so moving them first breaks +it. In-place also removes a copy the original had to do. + +Seeding stops at `seed_ratio` or `seed_time_mins`, whichever comes first, then the torrent is +released from the session (files kept). + +**Bug the smoke test caught:** the stall budget only covered the download loop, but resolving a +magnet's metadata happens inside `add_torrent`, which against a dead swarm never returns — a +torrent nobody seeds wedged the scan indefinitely. `add_torrent` is now wrapped in the same budget. + +Verified: `cargo test` 27/27 (ratio incl. the divide-by-zero case, stall_mins = 0 not meaning +"abort instantly"). Routing: with `enabled = false` a torrent enclosure is marked +`skipped/torrents disabled` and never attempted. Session startup works here. Stall abort measured +end to end: with `stall_mins = 1`, a dead magnet failed at 20:58:23 -> 20:59:24, exactly 61s, and +the row recorded `error / no metadata after 1 minutes, gave up`. + +**Not verified: an actual successful swarm download.** This sandbox has no reachable peers, so +smoke 5's happy path — payload lands, seeding stops at the ratio — has not been run. The code paths +either side of it are tested; the swarm itself needs a real network. Worth running once against a +live torrent feed before trusting it. + +--- + ## 2026-09-09 — Step 6: ipc.rs + daemon `src/ipc.rs`: `Event` and `Command` as serde-tagged enums (`{"ev":...}` / `{"cmd":...}`), one JSON diff --git a/README.md b/README.md index 60f6b17..5f8815b 100644 --- a/README.md +++ b/README.md @@ -11,21 +11,100 @@ Python 2 engine behind **iPodderX** (2004-2008, Ray Slakinski & August Trometer) open-sourced under the MIT License in 2010. What carries over: the feed scan and TTL handling, GUID/URL dedupe, per-feed and per-date download -folders, keyword filters, the explicit-content filter, torrent enclosures, and "SmartSpace" — the +folders, keyword filters, the explicit-content filter, torrent enclosures, and "SmartSpace" -- the oldest-first disk quota reaper. What does not: iTunes and iPhoto export via AppleScript, text-to-speech enclosures, the Windows WMP/COM paths, XML plists and Python pickles for state, the `directory.iPodderX.com` survey ping, 3DES-encrypted preferences, and the `printMSG` stdout protocol (replaced by a JSON-lines socket). -## Status +## Quickstart -Under construction. See [PROGRESS.md](PROGRESS.md) for what works today. +```sh +cargo install --path . + +ipx add https://atp.fm/rss # names the feed from its own title +ipx list +ipx fetch # scan now +ipx daemon # or run continuously, honouring each feed's +``` + +Config lives at `~/.config/ipx/config.toml` (mode 0600, since it may hold feed passwords); +state at `~/.local/share/ipx/state.db`. Override with `IPX_CONFIG` and `IPX_DATA_DIR`. +Set `IPX_LOG=ipx=debug` for verbose logging on stderr. + +## Commands + +| command | what it does | +|---|---| +| `ipx add [--folder X] [--keywords a,b]` | subscribe; the id comes from the feed title | +| `ipx rm ` | unsubscribe; downloads and history are kept | +| `ipx list` / `ipx status` | subscriptions and their state | +| `ipx fetch [FEED] [--force]` | scan; `--force` ignores the TTL | +| `ipx reap [--dry-run]` | run retention now | +| `ipx import/export ` | move subscriptions in or out | +| `ipx daemon` | scheduler plus the control socket | + +Any command with a wire form probes the socket first: if a daemon is running it does the work, +and the CLI just renders the events it streams back. `--local` forces in-process execution. ## Configuration -`~/.config/ipx/config.toml`, state in `~/.local/share/ipx/state.db`. See PROGRESS.md until the -usage section lands. +```toml +[general] +download_dir = "~/Podcasts" +socket = "/run/user/1000/ipx.sock" # default: $XDG_RUNTIME_DIR/ipx.sock +interval_mins = 60 # default poll; a feed's own wins when longer +organize = "feed" # "feed" | "date" +max_total_gb = 50 # 0 = unlimited +max_age_days = 30 # 0 = keep forever + +[torrent] +enabled = true +seed_ratio = 1.0 # stop seeding at this ratio ... +seed_time_mins = 60 # ... or after this long, whichever comes first +port_range = "6881-6889" +stall_mins = 30 # give up on a torrent making no progress + +[feeds.atp] +url = "https://atp.fm/rss" +folder = "Accidental Tech Podcast" # default: the feed title +keywords = ["deep dive"] # OR across keywords, AND within one +allow_explicit = false +auto_download = true +max_new_per_check = 3 # the rest wait for the next scan +username = "ray" # optional HTTP basic auth +password_env = "IPX_ATP_PASS" # or a literal `password` +``` + +Retention keeps files that are `flagged` in the database, and deletes read episodes before unread +ones, oldest first. + +## Socket protocol + +Newline-delimited JSON over a Unix socket, both directions. + +```sh +$ printf '{"cmd":"fetch","force":true}\n' | socat - UNIX-CONNECT:$XDG_RUNTIME_DIR/ipx.sock +{"ev":"feed_start","feed":"atp"} +{"ev":"progress","feed":"atp","url":"...","file":"ep1.mp3","done":8192,"total":3000000} +{"ev":"download_done","feed":"atp","url":"...","path":"...","bytes":3000000} +{"ev":"feed_done","feed":"atp","new":1,"downloaded":1,"failed":0,"torrents":0} +{"ev":"scan_done","feeds":1} +``` + +Commands: `fetch` (optional `feed`, `force`), `reap` (optional `dry_run`), `status`. +Events: `feed_start`, `feed_skip`, `feed_done`, `feed_error`, `progress`, `download_done`, +`download_error`, `torrent_deferred`, `reaped`, `reap_done`, `scan_done`, `status`, `error`. +`scan_done`, `reap_done` and `status` are terminal -- a client that asked for work stops there. + +Progress is throttled to whole percents. The stream is a broadcast, so a client attached to a busy +daemon also sees that daemon's other work. + +## Running it as a service + +`contrib/` has a systemd user unit for the daemon, and a timer plus one-shot service if you would +rather run periodic scans with no daemon (in which case there is no socket for a UI to attach to). ## License diff --git a/contrib/ipx-scan.service b/contrib/ipx-scan.service new file mode 100644 index 0000000..eafb932 --- /dev/null +++ b/contrib/ipx-scan.service @@ -0,0 +1,7 @@ +[Unit] +Description=ipx feed scan (one shot) + +[Service] +Type=oneshot +ExecStart=%h/.cargo/bin/ipx fetch +Environment=IPX_LOG=ipx=info diff --git a/contrib/ipx.service b/contrib/ipx.service new file mode 100644 index 0000000..a166ac4 --- /dev/null +++ b/contrib/ipx.service @@ -0,0 +1,18 @@ +# User unit: install to ~/.config/systemd/user/ipx.service, then +# systemctl --user enable --now ipx +# The socket lands in $XDG_RUNTIME_DIR/ipx.sock by default, so a UI running as the +# same user can attach without extra configuration. +[Unit] +Description=ipx podcatcher +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +ExecStart=%h/.cargo/bin/ipx daemon +Restart=on-failure +RestartSec=30 +Environment=IPX_LOG=ipx=info + +[Install] +WantedBy=default.target diff --git a/contrib/ipx.timer b/contrib/ipx.timer new file mode 100644 index 0000000..89fe5b8 --- /dev/null +++ b/contrib/ipx.timer @@ -0,0 +1,16 @@ +# Alternative to the daemon: a periodic one-shot scan, closer to how the original +# iPodderX agent was driven. Use this OR ipx.service, not both -- with no daemon +# running there is no socket, so a UI cannot attach. +# +# Install ipx-scan.service and ipx.timer to ~/.config/systemd/user/, then +# systemctl --user enable --now ipx.timer +[Unit] +Description=Periodic ipx feed scan + +[Timer] +OnBootSec=5min +OnUnitActiveSec=1h +Persistent=true + +[Install] +WantedBy=timers.target diff --git a/src/config.rs b/src/config.rs index b5ac5d0..e42265f 100644 --- a/src/config.rs +++ b/src/config.rs @@ -188,6 +188,33 @@ fn default_socket() -> PathBuf { } } +/// Feed ids are the TOML table key, so they must be readable and punctuation-free. +pub fn slug(text: &str) -> String { + let mut out = String::new(); + for c in text.chars() { + if c.is_ascii_alphanumeric() { + out.push(c.to_ascii_lowercase()); + } else if c.is_alphanumeric() { + out.push(c); // Keep non-ASCII letters; TOML bare keys are stricter, but we quote. + } else if !out.ends_with('-') { + out.push('-'); + } + } + let out = out.trim_matches('-').to_owned(); + let out: String = out.chars().take(40).collect(); + let out = out.trim_matches('-').to_owned(); + if out.is_empty() { "feed".into() } else { out } +} + +/// `slug`, with a numeric suffix if that id is taken. +pub fn unique_slug(text: &str, taken: &BTreeMap) -> String { + let base = slug(text); + if !taken.contains_key(&base) { + return base; + } + (2..).map(|n| format!("{base}-{n}")).find(|s| !taken.contains_key(s)).unwrap() +} + fn home() -> PathBuf { dirs::home_dir().unwrap_or_else(|| PathBuf::from(".")) } @@ -241,6 +268,24 @@ mod tests { assert_eq!(t.ports(), (6881, 6889), "reversed range is not a range"); } + #[test] + fn slugs_are_readable_and_unique() { + assert_eq!(slug("Accidental Tech Podcast"), "accidental-tech-podcast"); + assert_eq!(slug(" The Daily!! "), "the-daily"); + assert_eq!(slug("99% Invisible"), "99-invisible"); + assert_eq!(slug("///"), "feed"); + assert_eq!(slug("").len(), 4); + assert!(slug(&"x".repeat(100)).len() <= 40); + + let mut taken = BTreeMap::new(); + taken.insert("the-daily".to_string(), Feed { + url: "u".into(), folder: None, keywords: vec![], allow_explicit: false, + auto_download: true, max_new_per_check: None, username: None, + password: None, password_env: None, + }); + assert_eq!(unique_slug("The Daily", &taken), "the-daily-2"); + } + #[test] fn password_env_wins_over_literal() { let mut f = Feed { diff --git a/src/download.rs b/src/download.rs index 10f4f51..7615d4c 100644 --- a/src/download.rs +++ b/src/download.rs @@ -258,20 +258,6 @@ pub fn matches_keywords(keywords: &[String], haystacks: &[&str]) -> bool { }) } -/// Formats a progress line, throttled by the caller to ~1% steps as the original's -/// lastDLStepSize guard did. -pub fn progress_line(name: &str, done: u64, total: Option) -> String { - match total { - Some(t) if t > 0 => format!( - " {name}: {:.1}% ({:.1}/{:.1} MB)", - done as f64 / t as f64 * 100.0, - done as f64 / 1_048_576.0, - t as f64 / 1_048_576.0 - ), - _ => format!(" {name}: {:.1} MB", done as f64 / 1_048_576.0), - } -} - #[cfg(test)] mod tests { use super::*; diff --git a/src/feed.rs b/src/feed.rs index d64115a..d7e68cb 100644 --- a/src/feed.rs +++ b/src/feed.rs @@ -9,10 +9,7 @@ use crate::config::Feed as FeedCfg; #[derive(Debug, Default)] pub struct ParsedFeed { pub title: Option, - pub link: Option, pub ttl_mins: Option, - /// Feed-level explicit flag; per the original, it overrides the entry level. - pub explicit: bool, pub entries: Vec, } @@ -148,9 +145,7 @@ fn from_rss(ch: rss::Channel) -> ParsedFeed { ParsedFeed { title: non_empty(Some(ch.title())), - link: non_empty(Some(ch.link())), ttl_mins: ch.ttl().and_then(|t| t.trim().parse().ok()), - explicit, entries, } } @@ -205,13 +200,7 @@ fn from_atom(feed: atom_syndication::Feed) -> ParsedFeed { ParsedFeed { title: non_empty(Some(feed.title().as_str())), - link: feed - .links() - .iter() - .find(|l| l.rel() == "alternate") - .map(|l| l.href().to_owned()), ttl_mins: None, - explicit: false, entries, } } @@ -261,7 +250,6 @@ mod tests { assert_eq!(feed.title.as_deref(), Some("Test Cast")); assert_eq!(feed.ttl_mins, Some(45)); - assert!(!feed.explicit, "feed-level explicit is 'no'"); assert_eq!(feed.entries.len(), 3); let ep = &feed.entries[0]; @@ -296,8 +284,10 @@ mod tests { "#; let feed = parse(xml).unwrap(); - assert!(feed.explicit); - assert!(feed.entries[0].explicit, "feed level must win"); + assert!( + feed.entries[0].explicit, + "the entry says nothing; the feed-level flag must still mark it explicit" + ); } #[test] diff --git a/src/main.rs b/src/main.rs index a621136..3bf6bef 100644 --- a/src/main.rs +++ b/src/main.rs @@ -4,8 +4,9 @@ mod download; mod feed; mod ipc; mod retention; +mod torrent; -use anyhow::Result; +use anyhow::{Context, Result}; use clap::{Parser, Subcommand}; use ipc::{Command as Cmd, Emitter, Event}; use std::path::PathBuf; @@ -46,6 +47,22 @@ enum Command { }, /// Counts of feeds, pending and downloaded enclosures Status, + /// Subscribe to a feed + Add { + url: String, + /// Download folder name (default: the feed title) + #[arg(long)] + folder: Option, + /// Only take enclosures matching these keywords + #[arg(long, value_delimiter = ',')] + keywords: Vec, + }, + /// Unsubscribe. Downloads and history are left alone. + Rm { feed: String }, + /// Add every feed in an OPML file + Import { file: PathBuf }, + /// Write subscriptions out as OPML + Export { file: PathBuf }, /// Run the scheduler and serve the control socket Daemon, } @@ -56,6 +73,17 @@ struct Ctx { db: db::Db, client: reqwest::Client, out: Emitter, + /// Started on first use: a BitTorrent session binds ports and starts a DHT, which is + /// rude to do for a config that has never seen a torrent. + torrents: tokio::sync::OnceCell, +} + +impl Ctx { + async fn torrents(&self) -> Result<&torrent::Torrents> { + self.torrents + .get_or_try_init(|| torrent::Torrents::new(&self.cfg)) + .await + } } #[tokio::main] @@ -80,7 +108,12 @@ async fn main() -> Result<()> { } Command::Reap { dry_run } => Some(Cmd::Reap { dry_run: *dry_run }), Command::Status => Some(Cmd::Status), - Command::List | Command::Daemon => None, + Command::List + | Command::Daemon + | Command::Add { .. } + | Command::Rm { .. } + | Command::Import { .. } + | Command::Export { .. } => None, }; if let Some(cmd) = &wire_cmd && !cli.local @@ -96,11 +129,18 @@ async fn main() -> Result<()> { .user_agent(concat!("ipx/", env!("CARGO_PKG_VERSION"))) .build()?, out: Emitter::terminal(), + torrents: tokio::sync::OnceCell::new(), }; match cli.command { Command::List => list(&ctx, &config_path), Command::Daemon => daemon(ctx).await, + Command::Add { url, folder, keywords } => { + add(ctx, &config_path, &url, folder, keywords).await + } + Command::Rm { feed } => rm(ctx, &config_path, &feed), + Command::Import { file } => import(ctx, &config_path, &file).await, + Command::Export { file } => export(&ctx, &file), _ => run(&ctx, wire_cmd.expect("only List and Daemon have no wire form")).await, } } @@ -174,6 +214,146 @@ async fn shutdown() { } } +/// Subscribes to one feed, naming it from its own title. +async fn add( + mut ctx: Ctx, + config_path: &std::path::Path, + url: &str, + folder: Option, + keywords: Vec, +) -> Result<()> { + if let Some((id, _)) = ctx.cfg.feeds.iter().find(|(_, f)| f.url == url) { + anyhow::bail!("already subscribed as {id:?}"); + } + let id = add_one(&mut ctx, url, folder, keywords).await?; + ctx.cfg.save(config_path)?; + println!("added {id}"); + Ok(()) +} + +/// Returns the new feed id. The title needs a fetch, so a feed that cannot be reached is +/// still added -- under a slug derived from its URL -- rather than refused. +async fn add_one( + ctx: &mut Ctx, + url: &str, + folder: Option, + keywords: Vec, +) -> Result { + let probe = config::Feed { + url: url.to_owned(), + folder: folder.clone(), + keywords: keywords.clone(), + allow_explicit: false, + auto_download: true, + max_new_per_check: None, + username: None, + password: None, + password_env: None, + }; + + let title = match feed::fetch(&ctx.client, &probe, None, None).await { + Ok(feed::Fetched::Body { bytes, .. }) => feed::parse(&bytes) + .ok() + .and_then(|f| f.title) + .unwrap_or_else(|| url_stem(url)), + _ => { + tracing::warn!(url, "could not read the feed; naming it from its URL"); + url_stem(url) + } + }; + + let id = config::unique_slug(&title, &ctx.cfg.feeds); + ctx.cfg.feeds.insert(id.clone(), probe); + Ok(id) +} + +/// Host plus last path segment, for naming a feed we could not read. +fn url_stem(url: &str) -> String { + url::Url::parse(url) + .ok() + .and_then(|u| u.host_str().map(str::to_owned)) + .unwrap_or_else(|| url.to_owned()) +} + +fn rm(mut ctx: Ctx, config_path: &std::path::Path, feed: &str) -> Result<()> { + if ctx.cfg.feeds.remove(feed).is_none() { + anyhow::bail!("no feed with id {feed:?}"); + } + ctx.cfg.save(config_path)?; + // State and files stay: re-adding the feed should not re-download its back catalogue. + println!("removed {feed}; downloads and history kept"); + Ok(()) +} + +async fn import(mut ctx: Ctx, config_path: &std::path::Path, file: &std::path::Path) -> Result<()> { + let text = std::fs::read_to_string(file) + .with_context(|| format!("reading {}", file.display()))?; + let doc = opml::OPML::from_str(&text).map_err(|e| anyhow::anyhow!("parsing OPML: {e}"))?; + + let mut found = vec![]; + collect_outlines(&doc.body.outlines, &mut found); + + let mut added = 0; + for (title, url) in found { + if ctx.cfg.feeds.values().any(|f| f.url == url) { + continue; + } + // Name it from the OPML title rather than refetching every feed. + let id = config::unique_slug(&title, &ctx.cfg.feeds); + ctx.cfg.feeds.insert( + id.clone(), + config::Feed { + url, + folder: None, + keywords: vec![], + allow_explicit: false, + auto_download: true, + max_new_per_check: None, + username: None, + password: None, + password_env: None, + }, + ); + println!("added {id}"); + added += 1; + } + ctx.cfg.save(config_path)?; + println!("{added} feed(s) imported"); + Ok(()) +} + +/// OPML nests feeds inside folder outlines, so this walks the whole tree. +fn collect_outlines(outlines: &[opml::Outline], out: &mut Vec<(String, String)>) { + for o in outlines { + if let Some(url) = &o.xml_url { + let title = o.title.clone().unwrap_or_else(|| o.text.clone()); + out.push((title, url.clone())); + } + collect_outlines(&o.outlines, out); + } +} + +fn export(ctx: &Ctx, file: &std::path::Path) -> Result<()> { + let mut doc = opml::OPML::default(); + doc.head = Some(opml::Head { + title: Some("ipx subscriptions".into()), + ..Default::default() + }); + for (id, feed) in &ctx.cfg.feeds { + let title = ctx + .db + .feed_summary(id) + .ok() + .and_then(|s| s.title) + .unwrap_or_else(|| id.clone()); + doc.add_feed(&title, &feed.url); + } + let xml = doc.to_string().map_err(|e| anyhow::anyhow!("writing OPML: {e}"))?; + std::fs::write(file, xml).with_context(|| format!("writing {}", file.display()))?; + println!("exported {} feed(s) to {}", ctx.cfg.feeds.len(), file.display()); + Ok(()) +} + fn list(ctx: &Ctx, config_path: &std::path::Path) -> Result<()> { if ctx.cfg.feeds.is_empty() { println!("No feeds configured in {}", config_path.display()); @@ -332,13 +512,37 @@ async fn scan_one( for item in ctx.db.pending(id, budget)? { if download::looks_like_torrent(&item.url, item.mime.as_deref()) { - // Step 7 owns these; recorded so nothing re-queues them meanwhile. - ctx.db.mark_enclosure(&item.url, "torrent", None)?; - ctx.out.emit(Event::TorrentDeferred { - feed: id.to_string(), - url: item.url.clone(), - }); - scan.torrents += 1; + if !ctx.cfg.torrent.enabled { + ctx.db.mark_enclosure(&item.url, "skipped", Some("torrents disabled"))?; + ctx.out.emit(Event::TorrentDeferred { + feed: id.to_string(), + url: item.url.clone(), + }); + scan.torrents += 1; + continue; + } + match torrent_one(ctx, id, &item.url, &dest_dir).await { + Ok((path, bytes)) => { + ctx.db.mark_downloaded(&item.url, &path, bytes)?; + ctx.out.emit(Event::DownloadDone { + feed: id.to_string(), + url: item.url.clone(), + path: path.display().to_string(), + bytes, + }); + scan.downloaded += 1; + } + Err(e) => { + let msg = format!("{e:#}"); + ctx.out.emit(Event::DownloadError { + feed: id.to_string(), + url: item.url.clone(), + msg: msg.clone(), + }); + ctx.db.mark_enclosure(&item.url, "error", Some(&msg))?; + scan.failed += 1; + } + } continue; } match fetch_one(ctx, id, feed_cfg, &item.url, &dest_dir).await { @@ -417,10 +621,14 @@ async fn fetch_one( .await?; if matches!(got.kind, download::Kind::Torrent) { - // The MIME lied. Don't file a .torrent as an episode. + // The MIME lied. Hand the URL to the torrent session instead of filing a .torrent + // as if it were an episode. let _ = tokio::fs::remove_file(&got.tmp).await; - ctx.db.mark_enclosure(url, "torrent", None)?; - anyhow::bail!("body is a torrent, deferred to the torrent downloader"); + if !ctx.cfg.torrent.enabled { + ctx.db.mark_enclosure(url, "skipped", Some("torrents disabled"))?; + anyhow::bail!("body is a torrent and torrents are disabled"); + } + return torrent_one(ctx, feed_id, url, dest_dir).await; } let path = download::place(&got, dest_dir).await?; @@ -428,6 +636,36 @@ async fn fetch_one( Ok((path, got.bytes)) } +/// Torrent progress is reported the same way an HTTP download's is, throttled to whole +/// percents so a UI is not flooded. +async fn torrent_one( + ctx: &Ctx, + feed_id: &str, + url: &str, + dest_dir: &std::path::Path, +) -> Result<(PathBuf, u64)> { + let name = download::filename_for(url, None); + let mut last_pct = -1i64; + ctx.torrents() + .await? + .fetch(&ctx.cfg, url, dest_dir, |done, total| { + if total > 0 { + let pct = (done * 100 / total) as i64; + if pct > last_pct { + last_pct = pct; + ctx.out.emit(Event::Progress { + feed: feed_id.to_string(), + url: url.to_string(), + file: name.clone(), + done, + total: Some(total), + }); + } + } + }) + .await +} + fn ago(t: Option) -> String { let Some(t) = t else { return "never".into() }; format!("{} ago", duration((db::now() - t).max(0) as u64)) diff --git a/src/torrent.rs b/src/torrent.rs new file mode 100644 index 0000000..a989966 --- /dev/null +++ b/src/torrent.rs @@ -0,0 +1,176 @@ +//! 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, +} + +/// 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 { + 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) { + 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)); + } +}