can't be bothered writing a commit message

breadarrd/src/sources/mod.rs

@@ -0,0 +1,83 @@
+pub mod rss;
+pub mod scrape;
+pub mod tpb;
+
+use anyhow::Result;
+use async_trait::async_trait;
+
+#[derive(Debug, Clone, PartialEq)]
+pub struct RawReleaseItem {
+    pub title: String,
+    /// Magnet URI or direct .torrent download URL — qBittorrent's add-by-URL
+    /// endpoint accepts either identically.
+    pub link: String,
+    pub guid: String,
+    pub size_bytes: Option<u64>,
+    pub seeders: Option<u32>,
+    pub leechers: Option<u32>,
+}
+
+#[async_trait]
+pub trait ReleaseSource {
+    /// Fetch current candidate releases. `query` is used by search-driven
+    /// sources (e.g. a scraped search page); feed-based sources ignore it
+    /// and return everything currently in the feed.
+    async fn fetch(&self, query: Option<&str>) -> Result<Vec<RawReleaseItem>>;
+}
+
+pub(crate) fn urlencode(s: &str) -> String {
+    s.chars()
+        .map(|c| {
+            if c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.' | '~') {
+                c.to_string()
+            } else if c == ' ' {
+                "%20".to_string()
+            } else {
+                let mut buf = [0u8; 4];
+                c.encode_utf8(&mut buf)
+                    .bytes()
+                    .map(|b| format!("%{b:02X}"))
+                    .collect()
+            }
+        })
+        .collect()
+}
+
+pub(crate) fn parse_human_size(s: &str) -> Option<u64> {
+    let s = s.trim();
+    let (num_part, unit) = s.split_once(' ')?;
+    let num: f64 = num_part.parse().ok()?;
+    let mult = match unit {
+        "B" => 1.0,
+        // 1337x labels these "KB/MB/GB" (decimal-looking) but the numbers
+        // are actually binary (1024-based), matching every other torrent
+        // site's convention — treat them the same as the KiB/MiB/GiB nyaa
+        // uses.
+        "KiB" | "KB" => 1024.0,
+        "MiB" | "MB" => 1024.0 * 1024.0,
+        "GiB" | "GB" => 1024.0 * 1024.0 * 1024.0,
+        "TiB" | "TB" => 1024.0_f64.powi(4),
+        _ => return None,
+    };
+    Some((num * mult) as u64)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parses_gib() {
+        assert_eq!(parse_human_size("38.1 GiB"), Some(40909563494));
+    }
+
+    #[test]
+    fn parses_mib() {
+        assert_eq!(parse_human_size("356.5 MiB"), Some(373817344));
+    }
+
+    #[test]
+    fn rejects_unknown_unit() {
+        assert_eq!(parse_human_size("5 XiB"), None);
+    }
+}

breadarrd/src/sources/rss.rs

@@ -0,0 +1,196 @@
+use anyhow::Result;
+use async_trait::async_trait;
+use quick_xml::escape::unescape;
+use quick_xml::events::Event;
+use quick_xml::reader::Reader;
+
+use super::{parse_human_size, urlencode, RawReleaseItem, ReleaseSource};
+
+/// RSS source for nyaa.si-style feeds, whose `nyaa:seeders`/`nyaa:leechers`/
+/// `nyaa:size` custom-namespaced fields carry the seeder data this project
+/// needs — generic feed libraries (e.g. feed-rs) don't surface arbitrary
+/// vendor namespaces, so this parses the raw XML directly.
+pub struct RssSource {
+    feed_url: String,
+    client: reqwest::Client,
+}
+
+impl RssSource {
+    pub fn new(feed_url: impl Into<String>) -> Self {
+        Self {
+            feed_url: feed_url.into(),
+            // The background loop holds the DB mutex across this fetch — no
+            // total timeout (reqwest's default) means a stalled connection
+            // hangs the whole daemon, not just this one request.
+            client: reqwest::Client::builder()
+                .timeout(std::time::Duration::from_secs(30))
+                .build()
+                .expect("reqwest client build"),
+        }
+    }
+}
+
+#[async_trait]
+impl ReleaseSource for RssSource {
+    async fn fetch(&self, query: Option<&str>) -> Result<Vec<RawReleaseItem>> {
+        let url = build_search_url(&self.feed_url, query);
+        let bytes = self.client.get(&url).send().await?.bytes().await?;
+        parse_nyaa_rss(&bytes)
+    }
+}
+
+/// nyaa's search *is* its RSS feed with a `q` param appended — same
+/// endpoint, same custom-namespace fields, so the parser needs no changes
+/// at all for search-driven use.
+fn build_search_url(feed_url: &str, query: Option<&str>) -> String {
+    match query {
+        Some(q) => {
+            let sep = if feed_url.contains('?') { '&' } else { '?' };
+            format!("{feed_url}{sep}q={}", urlencode(q))
+        }
+        None => feed_url.to_string(),
+    }
+}
+
+fn parse_nyaa_rss(bytes: &[u8]) -> Result<Vec<RawReleaseItem>> {
+    let mut reader = Reader::from_reader(bytes);
+    reader.config_mut().trim_text(true);
+
+    let mut items = Vec::new();
+    let mut buf = Vec::new();
+
+    let mut in_item = false;
+    let mut cur_tag = String::new();
+    let mut title = None;
+    let mut link = None;
+    let mut guid = None;
+    let mut seeders = None;
+    let mut leechers = None;
+    let mut size_bytes = None;
+
+    loop {
+        match reader.read_event_into(&mut buf)? {
+            Event::Eof => break,
+            Event::Start(e) => {
+                let name = String::from_utf8_lossy(e.name().as_ref()).into_owned();
+                if name == "item" {
+                    in_item = true;
+                    title = None;
+                    link = None;
+                    guid = None;
+                    seeders = None;
+                    leechers = None;
+                    size_bytes = None;
+                }
+                cur_tag = name;
+            }
+            Event::Text(t) if in_item => {
+                let raw = t.decode()?;
+                let text = unescape(&raw)?.into_owned();
+                match cur_tag.as_str() {
+                    "title" => title = Some(text),
+                    "link" => link = Some(text),
+                    "guid" => guid = Some(text),
+                    "nyaa:seeders" => seeders = text.parse().ok(),
+                    "nyaa:leechers" => leechers = text.parse().ok(),
+                    "nyaa:size" => size_bytes = parse_human_size(&text),
+                    _ => {}
+                }
+            }
+            Event::End(e) => {
+                let name = String::from_utf8_lossy(e.name().as_ref()).into_owned();
+                if name == "item" {
+                    if let (Some(title), Some(link), Some(guid)) =
+                        (title.take(), link.take(), guid.take())
+                    {
+                        items.push(RawReleaseItem {
+                            title,
+                            link,
+                            guid,
+                            size_bytes,
+                            seeders,
+                            leechers,
+                        });
+                    }
+                    in_item = false;
+                }
+            }
+            _ => {}
+        }
+        buf.clear();
+    }
+
+    Ok(items)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const SAMPLE: &str = r#"<?xml version="1.0" encoding="UTF-8"?>
+<rss xmlns:atom="http://www.w3.org/2005/Atom" xmlns:nyaa="https://nyaa.si/xmlns/nyaa" version="2.0">
+<channel>
+<title>Nyaa - Home - Torrent File RSS</title>
+<item>
+    <title>[Group] Some Show - 05 [1080p]</title>
+    <link>https://nyaa.si/download/2130903.torrent</link>
+    <guid isPermaLink="true">https://nyaa.si/view/2130903</guid>
+    <pubDate>Sat, 11 Jul 2026 08:51:04 -0000</pubDate>
+    <nyaa:seeders>12</nyaa:seeders>
+    <nyaa:leechers>3</nyaa:leechers>
+    <nyaa:size>356.5 MiB</nyaa:size>
+</item>
+</channel>
+</rss>"#;
+
+    #[test]
+    fn parses_nyaa_item_with_custom_fields() {
+        let items = parse_nyaa_rss(SAMPLE.as_bytes()).unwrap();
+        assert_eq!(items.len(), 1);
+        let item = &items[0];
+        assert_eq!(item.title, "[Group] Some Show - 05 [1080p]");
+        assert_eq!(item.link, "https://nyaa.si/download/2130903.torrent");
+        assert_eq!(item.guid, "https://nyaa.si/view/2130903");
+        assert_eq!(item.seeders, Some(12));
+        assert_eq!(item.leechers, Some(3));
+        assert_eq!(item.size_bytes, Some(373817344));
+    }
+
+    #[test]
+    fn build_search_url_appends_query_param() {
+        assert_eq!(
+            build_search_url("https://nyaa.si/?page=rss&c=1_2", Some("Some Movie 2016")),
+            "https://nyaa.si/?page=rss&c=1_2&q=Some%20Movie%202016"
+        );
+    }
+
+    #[test]
+    fn build_search_url_handles_a_feed_url_with_no_existing_query_string() {
+        assert_eq!(
+            build_search_url("https://nyaa.si/rss", Some("Some Movie")),
+            "https://nyaa.si/rss?q=Some%20Movie"
+        );
+    }
+
+    #[test]
+    fn build_search_url_is_unchanged_without_a_query() {
+        assert_eq!(
+            build_search_url("https://nyaa.si/?page=rss", None),
+            "https://nyaa.si/?page=rss"
+        );
+    }
+
+    /// Hits the real nyaa.si feed — not run by default. `cargo test -- --ignored`
+    /// to sanity-check the parser against live markup if nyaa changes their format.
+    #[tokio::test]
+    #[ignore]
+    async fn parses_live_nyaa_feed() {
+        let source = RssSource::new("https://nyaa.si/?page=rss");
+        let items = source.fetch(None).await.unwrap();
+        assert!(
+            !items.is_empty(),
+            "expected at least one item from live feed"
+        );
+        assert!(items.iter().any(|i| i.seeders.is_some()));
+    }
+}

breadarrd/src/sources/scrape.rs

@@ -0,0 +1,549 @@
+use std::sync::Mutex;
+use std::time::{Duration, Instant};
+
+use anyhow::{bail, Context, Result};
+use async_trait::async_trait;
+use scraper::{Html, Selector};
+
+use super::{parse_human_size, urlencode, RawReleaseItem, ReleaseSource};
+
+const BASE_COOLDOWN_SECS: u64 = 5 * 60;
+const MAX_COOLDOWN: Duration = Duration::from_secs(6 * 60 * 60);
+const RATE_LIMIT_MIN_COOLDOWN: Duration = Duration::from_secs(30 * 60);
+const RATE_LIMIT_MAX_RETRY_AFTER: Duration = Duration::from_secs(60 * 60);
+/// Hard cap on the tracked failure streak — well past the point where
+/// `BASE_COOLDOWN_SECS * 4^(streak-1)` already exceeds `MAX_COOLDOWN`, it
+/// exists purely so the exponent can never grow large enough to overflow.
+const MAX_FAILURE_STREAK: u32 = 10;
+
+/// 1337x's main domain (1337x.to) bans IPs at the Cloudflare WAF level
+/// after bursts of automated traffic — a network-level block that no
+/// amount of browser-fingerprint evasion gets around. Its community
+/// mirrors run on separate domains/Cloudflare zones, so a ban on one
+/// doesn't carry over. Requests are round-robined across mirrors (not just
+/// tried in a fixed fallback order) so no single domain absorbs the bulk of
+/// traffic, and a mirror that fails is demoted into a cooldown rather than
+/// re-probed on the very next search.
+pub struct ScrapeSource {
+    mirrors: Vec<String>,
+    client: reqwest::Client,
+    ring: Mutex<MirrorRing>,
+}
+
+struct MirrorRing {
+    next: usize,
+    cooldown_until: Vec<Option<Instant>>,
+    failure_streak: Vec<u32>,
+}
+
+impl MirrorRing {
+    fn new(count: usize) -> Self {
+        Self {
+            next: 0,
+            cooldown_until: vec![None; count],
+            failure_streak: vec![0; count],
+        }
+    }
+}
+
+/// Distinguishes *why* a mirror attempt failed, since the right cooldown
+/// differs: a WAF-issued rate-limit/block response means "this domain is
+/// now watching you" and gets a firm minimum cooldown regardless of streak,
+/// while a timeout or a challenge page gets the milder exponential ladder.
+#[derive(Debug)]
+enum MirrorError {
+    /// No results table found at all — most likely a Cloudflare challenge
+    /// page served with an HTTP 200 (so `error_for_status` wouldn't catch
+    /// it), or the mirror's HTML layout has drifted.
+    Challenge,
+    /// 403/429/503 — an explicit throttle/block signal from the WAF, not a
+    /// generic network failure.
+    RateLimited {
+        retry_after_secs: Option<u64>,
+    },
+    Other(anyhow::Error),
+}
+
+impl std::fmt::Display for MirrorError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            MirrorError::Challenge => write!(f, "no results table found (challenge page?)"),
+            MirrorError::RateLimited { retry_after_secs } => {
+                write!(f, "rate limited (retry_after={retry_after_secs:?})")
+            }
+            MirrorError::Other(e) => write!(f, "{e}"),
+        }
+    }
+}
+
+/// Returned by [`ScrapeSource::fetch`] when every mirror is either on
+/// cooldown or failed this attempt — a distinct type (rather than a plain
+/// string error) so callers can `downcast_ref` to trigger cycle-level
+/// backoff without string-matching an error message.
+#[derive(Debug)]
+pub struct AllMirrorsFailed;
+
+impl std::fmt::Display for AllMirrorsFailed {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "all 1337x mirrors failed or are in cooldown")
+    }
+}
+
+impl std::error::Error for AllMirrorsFailed {}
+
+impl ScrapeSource {
+    pub fn new(mirrors: Vec<String>) -> Self {
+        let count = mirrors.len();
+        Self {
+            mirrors,
+            client: reqwest::Client::builder()
+                .user_agent("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36")
+                .timeout(Duration::from_secs(30))
+                // Cloudflare's cf_clearance cookie is per-zone (per mirror
+                // domain) — holding it means a mirror that challenged once
+                // can pass on subsequent requests instead of re-challenging
+                // every single time.
+                .cookie_store(true)
+                .build()
+                .expect("reqwest client build"),
+            ring: Mutex::new(MirrorRing::new(count)),
+        }
+    }
+
+    /// Mirror indices to try, starting from the round-robin cursor and
+    /// wrapping, skipping any still in cooldown. Advances the cursor
+    /// unconditionally (not just on success) so consecutive searches keep
+    /// moving through the ring instead of retrying the same start point.
+    fn candidate_order(&self) -> Vec<usize> {
+        let mut ring = self.ring.lock().expect("mirror ring poisoned");
+        let start = ring.next;
+        if !self.mirrors.is_empty() {
+            ring.next = (ring.next + 1) % self.mirrors.len();
+        }
+        compute_candidate_order(
+            self.mirrors.len(),
+            start,
+            &ring.cooldown_until,
+            Instant::now(),
+        )
+    }
+
+    fn record_success(&self, idx: usize) {
+        let mut ring = self.ring.lock().expect("mirror ring poisoned");
+        ring.failure_streak[idx] = 0;
+    }
+
+    fn demote(&self, idx: usize, err: &MirrorError) {
+        let mut ring = self.ring.lock().expect("mirror ring poisoned");
+        let cooldown = match err {
+            MirrorError::RateLimited { retry_after_secs } => {
+                // A rate-limit response isn't generic flakiness — don't let
+                // it ratchet the exponential streak, just apply a firm
+                // minimum (or the server's own Retry-After, capped).
+                let retry_after = retry_after_secs
+                    .map(|s| Duration::from_secs(s).min(RATE_LIMIT_MAX_RETRY_AFTER));
+                retry_after
+                    .unwrap_or(RATE_LIMIT_MIN_COOLDOWN)
+                    .max(RATE_LIMIT_MIN_COOLDOWN)
+            }
+            MirrorError::Challenge | MirrorError::Other(_) => {
+                let streak = (ring.failure_streak[idx] + 1).min(MAX_FAILURE_STREAK);
+                ring.failure_streak[idx] = streak;
+                let secs = BASE_COOLDOWN_SECS.saturating_mul(4u64.saturating_pow(streak - 1));
+                Duration::from_secs(secs).min(MAX_COOLDOWN)
+            }
+        };
+        ring.cooldown_until[idx] = Some(Instant::now() + cooldown);
+    }
+
+    async fn search_mirror(
+        &self,
+        mirror: &str,
+        query: &str,
+    ) -> std::result::Result<Vec<RawReleaseItem>, MirrorError> {
+        let url = format!(
+            "{}/search/{}/1/",
+            mirror.trim_end_matches('/'),
+            urlencode(query)
+        );
+        let resp = self.client.get(&url).send().await.map_err(|e| {
+            MirrorError::Other(anyhow::Error::new(e).context(format!("request to {url} failed")))
+        })?;
+
+        let status = resp.status();
+        if matches!(status.as_u16(), 403 | 429 | 503) {
+            let retry_after_secs = resp
+                .headers()
+                .get(reqwest::header::RETRY_AFTER)
+                .and_then(|v| v.to_str().ok())
+                .and_then(|v| v.parse::<u64>().ok());
+            return Err(MirrorError::RateLimited { retry_after_secs });
+        }
+        let resp = resp.error_for_status().map_err(|e| {
+            MirrorError::Other(
+                anyhow::Error::new(e).context(format!("{url} returned an error status")),
+            )
+        })?;
+        let body = resp.text().await.map_err(|e| {
+            MirrorError::Other(anyhow::Error::new(e).context("failed to read response body"))
+        })?;
+        parse_search_results(&body, mirror).ok_or(MirrorError::Challenge)
+    }
+}
+
+/// Mirror indices to try, in round-robin order starting at `start`,
+/// excluding any whose cooldown hasn't elapsed yet. Pure and separately
+/// testable from the ring's locking/mutation.
+fn compute_candidate_order(
+    len: usize,
+    start: usize,
+    cooldown_until: &[Option<Instant>],
+    now: Instant,
+) -> Vec<usize> {
+    if len == 0 {
+        return Vec::new();
+    }
+    (0..len)
+        .map(|offset| (start + offset) % len)
+        .filter(|&i| cooldown_until[i].is_none_or(|until| now >= until))
+        .collect()
+}
+
+/// True if `link` is already directly usable by qBittorrent's add-by-URL
+/// endpoint (a magnet URI or a direct .torrent download link) — false for
+/// a detail-page URL that needs resolving first (what search results give
+/// us, since the results table doesn't carry the magnet directly).
+pub fn needs_resolution(link: &str) -> bool {
+    !(link.starts_with("magnet:") || link.ends_with(".torrent"))
+}
+
+/// The search results table only links to a torrent's detail page, not its
+/// magnet URI directly — fetches that page and extracts the magnet link.
+/// A free function (not tied to a `ScrapeSource`/mirror list) because the
+/// detail URL captured at search time already has the right mirror domain
+/// baked in, and this needs to be callable from the grab path generically
+/// (including review-queue approval, which only has a stored link, not a
+/// `ScrapeSource` instance) — called only for the one candidate actually
+/// being grabbed, not for every search result, to keep request volume low.
+pub async fn resolve_magnet(client: &reqwest::Client, detail_url: &str) -> Result<String> {
+    let resp = client
+        .get(detail_url)
+        .send()
+        .await
+        .with_context(|| format!("request to {detail_url} failed"))?
+        .error_for_status()
+        .with_context(|| format!("{detail_url} returned an error status"))?;
+    let body = resp.text().await.context("failed to read response body")?;
+    extract_magnet(&body).with_context(|| format!("no magnet link found on {detail_url}"))
+}
+
+#[async_trait]
+impl ReleaseSource for ScrapeSource {
+    async fn fetch(&self, query: Option<&str>) -> Result<Vec<RawReleaseItem>> {
+        let Some(query) = query else {
+            bail!(
+                "ScrapeSource requires a search query (this is a search-driven source, not a feed)"
+            );
+        };
+        if self.mirrors.is_empty() {
+            bail!("no 1337x mirrors configured");
+        }
+
+        for idx in self.candidate_order() {
+            let mirror = &self.mirrors[idx];
+            match self.search_mirror(mirror, query).await {
+                Ok(items) => {
+                    self.record_success(idx);
+                    return Ok(items);
+                }
+                Err(e) => {
+                    tracing::warn!(mirror, error = %e, "1337x mirror failed, trying next");
+                    self.demote(idx, &e);
+                }
+            }
+        }
+        Err(anyhow::Error::new(AllMirrorsFailed))
+    }
+}
+
+/// Pulls the numeric torrent ID out of a 1337x listing/detail href of the
+/// shape `/torrent/<id>/<slug>/` (relative) or `https://<mirror>/torrent/
+/// <id>/<slug>/` (absolute) — the one part of the URL that's identical
+/// across every mirror, unlike the domain. `None` if the href doesn't
+/// contain `/torrent/` at all (an unexpected HTML shape), in which case
+/// the caller falls back to the full URL rather than losing the item.
+fn extract_torrent_id(href: &str) -> Option<&str> {
+    href.split("/torrent/")
+        .nth(1)?
+        .split('/')
+        .next()
+        .filter(|s| !s.is_empty())
+}
+
+/// `None` when the results table itself is missing from the document —
+/// most likely a Cloudflare challenge page served with a 200 status (so it
+/// never trips `error_for_status`), or the mirror's HTML layout drifted.
+/// `Some(vec![])` is a genuine, trustworthy "no results" — those two cases
+/// must not be conflated, since the former should demote the mirror and
+/// the latter should not.
+fn parse_search_results(html: &str, mirror: &str) -> Option<Vec<RawReleaseItem>> {
+    let doc = Html::parse_document(html);
+    let table_sel = Selector::parse("table.table-list").unwrap();
+    doc.select(&table_sel).next()?;
+
+    let row_sel = Selector::parse("table.table-list tbody tr").unwrap();
+    let name_link_sel = Selector::parse("td.coll-1.name a:not(.icon)").unwrap();
+    let seeds_sel = Selector::parse("td.coll-2").unwrap();
+    let leeches_sel = Selector::parse("td.coll-3").unwrap();
+    let size_sel = Selector::parse("td.coll-4").unwrap();
+
+    let mirror = mirror.trim_end_matches('/');
+    let mut items = Vec::new();
+
+    for row in doc.select(&row_sel) {
+        let Some(link_el) = row.select(&name_link_sel).next() else {
+            continue;
+        };
+        let Some(href) = link_el.value().attr("href") else {
+            continue;
+        };
+        let title: String = link_el.text().collect();
+        let title = title.trim().to_string();
+        if title.is_empty() {
+            continue;
+        }
+        let detail_url = if href.starts_with("http") {
+            href.to_string()
+        } else {
+            format!("{mirror}{href}")
+        };
+        // The dedup key must not depend on which mirror answered: the same
+        // physical torrent's `href` path (`/torrent/<id>/<slug>/`) is
+        // identical across every mirror, but `detail_url` bakes in
+        // whichever mirror happened to serve this particular search — with
+        // ~10 mirrors round-robined for ban resilience, using the full URL
+        // as `guid` meant the same release looked "new" up to 10 times
+        // over, defeating `is_seen`/`mark_seen` dedup entirely and flooding
+        // the review queue with duplicates of the same low-confidence
+        // match (verified live: one title queued 13 times). The numeric
+        // torrent ID is the stable, mirror-invariant identity instead.
+        let guid = extract_torrent_id(href)
+            .map(|id| format!("1337x:{id}"))
+            .unwrap_or_else(|| detail_url.clone());
+
+        let seeders = row
+            .select(&seeds_sel)
+            .next()
+            .and_then(|e| e.text().collect::<String>().trim().parse().ok());
+        let leechers = row
+            .select(&leeches_sel)
+            .next()
+            .and_then(|e| e.text().collect::<String>().trim().parse().ok());
+        let size_bytes = row
+            .select(&size_sel)
+            .next()
+            .and_then(|e| parse_human_size(e.text().collect::<String>().trim()));
+
+        items.push(RawReleaseItem {
+            title,
+            // Not directly grabbable yet — a detail-page URL, resolved to
+            // a real magnet link via `resolve_magnet` right before the
+            // winning candidate is actually sent to qBittorrent.
+            link: detail_url,
+            guid,
+            size_bytes,
+            seeders,
+            leechers,
+        });
+    }
+
+    Some(items)
+}
+
+fn extract_magnet(html: &str) -> Option<String> {
+    let doc = Html::parse_document(html);
+    let sel = Selector::parse(r#"a[href^="magnet:"]"#).unwrap();
+    doc.select(&sel)
+        .next()
+        .and_then(|e| e.value().attr("href"))
+        .map(str::to_string)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const SAMPLE_ROW: &str = r#"
+<table class="table-list table table-responsive table-striped">
+<thead><tr><th class="coll-1 name">name</th><th class="coll-2">se</th><th class="coll-3">le</th><th class="coll-date">time</th><th class="coll-4"><span class="size">size</span></th><th class="coll-5">uploader</th></tr></thead>
+<tbody>
+<tr>
+<td class="coll-1 name"><a href="/sub/tv/HD/1/" class="icon"><i class="flaticon-hd"></i></a><a href="/torrent/3250239/The-Big-Bang-Theory-S12E01-720p-HDTV-x264-KILLERS-eztv/">The Big Bang Theory S12E01 720p HDTV x264-KILLERS [eztv]</a></td>
+<td class="coll-2 seeds">3978</td>
+<td class="coll-3 leeches">1443</td>
+<td class="coll-date">Sep. 25th  '18</td>
+<td class="coll-4 size mob-uploader">559 MB</td>
+<td class="coll-5 uploader"><a href="/user/EZTVag/">EZTVag</a></td>
+</tr>
+</tbody>
+</table>"#;
+
+    #[test]
+    fn parses_a_real_captured_result_row() {
+        let items = parse_search_results(SAMPLE_ROW, "https://13377x.info").unwrap();
+        assert_eq!(items.len(), 1);
+        let item = &items[0];
+        assert_eq!(
+            item.title,
+            "The Big Bang Theory S12E01 720p HDTV x264-KILLERS [eztv]"
+        );
+        assert_eq!(
+            item.link,
+            "https://13377x.info/torrent/3250239/The-Big-Bang-Theory-S12E01-720p-HDTV-x264-KILLERS-eztv/"
+        );
+        assert_eq!(item.guid, "1337x:3250239");
+        assert_eq!(item.seeders, Some(3978));
+        assert_eq!(item.leechers, Some(1443));
+        assert_eq!(item.size_bytes, Some(586153984));
+    }
+
+    #[test]
+    fn guid_is_identical_across_different_mirrors_for_the_same_torrent() {
+        let a = parse_search_results(SAMPLE_ROW, "https://13377x.info").unwrap();
+        let b = parse_search_results(SAMPLE_ROW, "https://1337x.maskbay.info").unwrap();
+        // The `link` legitimately differs (it's used to actually fetch from
+        // whichever mirror answered this search) but the dedup `guid` must
+        // not, or mirror rotation defeats `is_seen`/`mark_seen` entirely —
+        // this was a real bug that flooded the review queue with up to 13
+        // duplicate entries for the same release.
+        assert_ne!(a[0].link, b[0].link);
+        assert_eq!(a[0].guid, b[0].guid);
+    }
+
+    #[test]
+    fn extract_torrent_id_handles_relative_and_absolute_hrefs() {
+        assert_eq!(
+            extract_torrent_id("/torrent/3602010/Instant-Family-2018/"),
+            Some("3602010")
+        );
+        assert_eq!(
+            extract_torrent_id("https://13377x.info/torrent/3602010/Instant-Family-2018/"),
+            Some("3602010")
+        );
+    }
+
+    #[test]
+    fn extract_torrent_id_is_none_for_an_unexpected_shape() {
+        assert_eq!(extract_torrent_id("/sub/tv/HD/1/"), None);
+    }
+
+    #[test]
+    fn returns_none_when_results_table_is_absent() {
+        let challenge_page = "<html><body><h1>Just a moment...</h1></body></html>";
+        assert_eq!(
+            parse_search_results(challenge_page, "https://13377x.info"),
+            None
+        );
+    }
+
+    #[test]
+    fn returns_empty_vec_for_a_genuine_no_results_page() {
+        let empty_results = r#"<table class="table-list table table-responsive table-striped">
+<thead><tr><th class="coll-1 name">name</th></tr></thead>
+<tbody></tbody>
+</table>"#;
+        assert_eq!(
+            parse_search_results(empty_results, "https://13377x.info"),
+            Some(vec![])
+        );
+    }
+
+    #[test]
+    fn extracts_magnet_from_detail_page() {
+        let html = r#"<a href="magnet:?xt=urn:btih:F30F455DC4C64A38E18C79C853B2A80B417C2343&dn=test">Magnet Download</a>"#;
+        assert_eq!(
+            extract_magnet(html).as_deref(),
+            Some("magnet:?xt=urn:btih:F30F455DC4C64A38E18C79C853B2A80B417C2343&dn=test")
+        );
+    }
+
+    #[test]
+    fn urlencode_handles_spaces_and_unicode() {
+        assert_eq!(urlencode("Big Buck Bunny"), "Big%20Buck%20Bunny");
+    }
+
+    #[test]
+    fn candidate_order_round_robins_from_start() {
+        let cooldowns = vec![None, None, None, None];
+        let order = compute_candidate_order(4, 2, &cooldowns, Instant::now());
+        assert_eq!(order, vec![2, 3, 0, 1]);
+    }
+
+    #[test]
+    fn candidate_order_skips_mirrors_still_in_cooldown() {
+        let now = Instant::now();
+        let cooldowns = vec![
+            None,
+            Some(now + Duration::from_secs(600)), // still cooling down
+            None,
+            Some(now - Duration::from_secs(1)), // cooldown already elapsed
+        ];
+        let order = compute_candidate_order(4, 0, &cooldowns, now);
+        assert_eq!(order, vec![0, 2, 3]);
+    }
+
+    #[test]
+    fn candidate_order_empty_when_every_mirror_cooling_down() {
+        let now = Instant::now();
+        let cooldowns = vec![Some(now + Duration::from_secs(60)); 3];
+        let order = compute_candidate_order(3, 0, &cooldowns, now);
+        assert!(order.is_empty());
+    }
+
+    #[test]
+    fn demote_applies_a_firm_minimum_cooldown_for_rate_limiting() {
+        let source = ScrapeSource::new(vec!["https://a".into(), "https://b".into()]);
+        source.demote(
+            0,
+            &MirrorError::RateLimited {
+                retry_after_secs: None,
+            },
+        );
+        let ring = source.ring.lock().unwrap();
+        let until = ring.cooldown_until[0].expect("should be cooling down");
+        assert!(until >= Instant::now() + Duration::from_secs(29 * 60));
+    }
+
+    #[test]
+    fn demote_caps_rate_limit_retry_after_at_one_hour() {
+        let source = ScrapeSource::new(vec!["https://a".into()]);
+        source.demote(
+            0,
+            &MirrorError::RateLimited {
+                retry_after_secs: Some(999_999),
+            },
+        );
+        let ring = source.ring.lock().unwrap();
+        let until = ring.cooldown_until[0].expect("should be cooling down");
+        assert!(until <= Instant::now() + RATE_LIMIT_MAX_RETRY_AFTER + Duration::from_secs(5));
+    }
+
+    #[test]
+    fn demote_escalates_generic_failures_exponentially() {
+        let source = ScrapeSource::new(vec!["https://a".into()]);
+        source.demote(0, &MirrorError::Challenge);
+        let first = source.ring.lock().unwrap().cooldown_until[0].unwrap();
+        source.demote(0, &MirrorError::Challenge);
+        let second = source.ring.lock().unwrap().cooldown_until[0].unwrap();
+        assert!(second > first);
+    }
+
+    #[test]
+    fn record_success_resets_failure_streak() {
+        let source = ScrapeSource::new(vec!["https://a".into()]);
+        source.demote(0, &MirrorError::Challenge);
+        assert_eq!(source.ring.lock().unwrap().failure_streak[0], 1);
+        source.record_success(0);
+        assert_eq!(source.ring.lock().unwrap().failure_streak[0], 0);
+    }
+}

breadarrd/src/sources/tpb.rs

@@ -0,0 +1,132 @@
+use anyhow::{Context, Result};
+use async_trait::async_trait;
+use serde::Deserialize;
+
+use super::{urlencode, RawReleaseItem, ReleaseSource};
+
+/// A community-run JSON API mirror of The Pirate Bay's search — unlike
+/// 1337x, this is a genuine machine-readable API (not HTML scraping), and
+/// unlike 1337x's HTML results table, `info_hash` is enough to build a
+/// magnet directly: no second detail-page fetch needed to resolve a link
+/// before grabbing. Also proved dramatically more precise in practice —
+/// its search actually ranks by relevance, where 1337x's is a pure
+/// seeder-count sort that buries genuine matches for common-word titles
+/// under unrelated, much-more-seeded content.
+pub struct TpbSource {
+    api_url: String,
+    client: reqwest::Client,
+}
+
+#[derive(Deserialize)]
+struct TpbResult {
+    id: String,
+    name: String,
+    info_hash: String,
+    seeders: String,
+    leechers: String,
+    size: String,
+}
+
+const TRACKERS: &[&str] = &[
+    "udp://tracker.opentrackr.org:1337/announce",
+    "udp://open.stealth.si:80/announce",
+    "udp://tracker.torrent.eu.org:451/announce",
+    "udp://tracker.openbittorrent.com:6969/announce",
+    "udp://exodus.desync.com:6969/announce",
+];
+
+fn build_magnet(info_hash: &str, name: &str) -> String {
+    let mut magnet = format!("magnet:?xt=urn:btih:{info_hash}&dn={}", urlencode(name));
+    for t in TRACKERS {
+        magnet.push_str("&tr=");
+        magnet.push_str(&urlencode(t));
+    }
+    magnet
+}
+
+impl TpbSource {
+    pub fn new(api_url: impl Into<String>) -> Self {
+        Self {
+            api_url: api_url.into(),
+            // No total timeout is reqwest's default — the background loop
+            // holds the DB mutex across this fetch, so a stalled connection
+            // would hang the whole daemon.
+            client: reqwest::Client::builder()
+                .timeout(std::time::Duration::from_secs(30))
+                .build()
+                .expect("reqwest client build"),
+        }
+    }
+}
+
+#[async_trait]
+impl ReleaseSource for TpbSource {
+    async fn fetch(&self, query: Option<&str>) -> Result<Vec<RawReleaseItem>> {
+        let Some(query) = query else {
+            anyhow::bail!(
+                "TpbSource requires a search query (this is a search-driven source, not a feed)"
+            );
+        };
+        let url = format!("{}?q={}", self.api_url, urlencode(query));
+        let results: Vec<TpbResult> = self
+            .client
+            .get(&url)
+            .send()
+            .await
+            .with_context(|| format!("request to {url} failed"))?
+            .error_for_status()
+            .with_context(|| format!("{url} returned an error status"))?
+            .json()
+            .await
+            .context("failed to parse apibay response as JSON")?;
+
+        Ok(results
+            .into_iter()
+            // A query with no matches returns a single sentinel row
+            // (id="0", an all-zero info_hash) rather than an empty array —
+            // has to be filtered out explicitly or it'd be treated as one
+            // real (and completely bogus) result.
+            .filter(|r| r.id != "0" && !r.info_hash.chars().all(|c| c == '0'))
+            .map(|r| RawReleaseItem {
+                title: r.name.clone(),
+                link: build_magnet(&r.info_hash, &r.name),
+                guid: r.info_hash,
+                size_bytes: r.size.parse().ok(),
+                seeders: r.seeders.parse().ok(),
+                leechers: r.leechers.parse().ok(),
+            })
+            .collect())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn build_magnet_includes_hash_name_and_trackers() {
+        let magnet = build_magnet("ABC123", "Some.Show.S01E01");
+        assert!(magnet.starts_with("magnet:?xt=urn:btih:ABC123&dn=Some.Show.S01E01"));
+        assert!(magnet.contains("tracker.opentrackr.org"));
+    }
+
+    #[test]
+    fn parses_a_real_captured_response() {
+        let body = r#"[{"id":"51630137","name":"Modern Family S01 (1080p BluRay)","info_hash":"8F87C7C186172F17E35F4512BB1A3E93B614ADED","leechers":"309","seeders":"379","size":"5395577424","num_files":"28","username":"rjaa","added":"1629816694","status":"vip","category":"208","imdb":""}]"#;
+        let results: Vec<TpbResult> = serde_json::from_str(body).unwrap();
+        assert_eq!(results.len(), 1);
+        assert_eq!(results[0].name, "Modern Family S01 (1080p BluRay)");
+        assert_eq!(results[0].seeders, "379");
+    }
+
+    #[test]
+    fn filters_out_the_no_results_sentinel() {
+        let body = r#"[{"id":"0","name":"No results returned","info_hash":"0000000000000000000000000000000000000000","leechers":"0","seeders":"0","size":"0","num_files":"0","username":"","added":"0","status":"","category":"0","imdb":""}]"#;
+        let results: Vec<TpbResult> = serde_json::from_str(body).unwrap();
+        let filtered: Vec<_> = results
+            .into_iter()
+            .filter(|r| r.id != "0" && !r.info_hash.chars().all(|c| c == '0'))
+            .collect();
+        assert!(filtered.is_empty());
+    }
+}