ipx/src/main.rs

mod config;
mod db;
mod download;
mod feed;
mod ipc;
mod retention;
mod torrent;
mod web;

use anyhow::{Context, Result};
use clap::{Parser, Subcommand};
use ipc::{Command as Cmd, Emitter, Event};
use std::path::PathBuf;
use std::sync::Arc;
use std::future::Future;
use tokio::sync::{broadcast, mpsc, watch};

#[derive(Parser)]
#[command(name = "ipx", version, about = "A headless podcatcher")]
struct Cli {
    /// Config file (default: $XDG_CONFIG_HOME/ipx/config.toml)
    #[arg(long, global = true)]
    config: Option<PathBuf>,

    /// Do the work here even if a daemon is running
    #[arg(long, global = true)]
    local: bool,

    #[command(subcommand)]
    command: Command,
}

#[derive(Subcommand)]
enum Command {
    /// Show configured feeds and their state
    List,
    /// Scan feeds for new entries
    Fetch {
        /// Only this feed id
        feed: Option<String>,
        /// Poll even when the feed is not due yet
        #[arg(long)]
        force: bool,
    },
    /// Delete old or over-quota downloads
    Reap {
        /// Show what would go, delete nothing
        #[arg(long)]
        dry_run: bool,
    },
    /// 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<String>,
        /// Only take enclosures matching these keywords
        #[arg(long, value_delimiter = ',')]
        keywords: Vec<String>,
    },
    /// 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 {
        /// Serve the web UI on this address, overriding [web] in the config
        #[arg(long, value_name = "ADDR")]
        web: Option<String>,
    },
}

/// Everything a command needs. One per process.
pub struct Ctx {
    /// Swapped wholesale when the web UI rewrites config.toml, so a running daemon picks
    /// up feed changes without a restart. Callers take a snapshot; no guard is ever held
    /// across an await.
    pub cfg: std::sync::RwLock<std::sync::Arc<config::Config>>,
    pub db: db::Db,
    pub client: reqwest::Client,
    pub 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.
    pub torrents: tokio::sync::OnceCell<torrent::Torrents>,
}

impl Ctx {
    pub fn cfg(&self) -> std::sync::Arc<config::Config> {
        self.cfg.read().unwrap().clone()
    }

    /// Re-reads config.toml into the live snapshot.
    pub fn reload_cfg(&self, path: &std::path::Path) -> Result<()> {
        let fresh = config::Config::load(path)?;
        *self.cfg.write().unwrap() = std::sync::Arc::new(fresh);
        tracing::info!("config reloaded");
        Ok(())
    }

    async fn torrents(&self) -> Result<&torrent::Torrents> {
        let cfg = self.cfg();
        self.torrents
            .get_or_try_init(|| torrent::Torrents::new(&cfg))
            .await
    }
}

#[tokio::main]
async fn main() -> Result<()> {
    let cli = Cli::parse();
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_env("IPX_LOG")
                .unwrap_or_else(|_| "ipx=info".into()),
        )
        .with_writer(std::io::stderr)
        .init();

    let config_path = cli.config.clone().unwrap_or_else(config::config_path);
    let cfg = config::Config::load(&config_path)?;
    let db = db::Db::open(&config::data_dir().join("state.db"))?;

    // A daemon owns the state; don't have two processes downloading the same thing.
    let wire_cmd = match &cli.command {
        Command::Fetch { feed, force } => {
            Some(Cmd::Fetch { feed: feed.clone(), force: *force })
        }
        Command::Reap { dry_run } => Some(Cmd::Reap { dry_run: *dry_run }),
        Command::Status => Some(Cmd::Status),
        Command::List
        | Command::Daemon { .. }
        | Command::Add { .. }
        | Command::Rm { .. }
        | Command::Import { .. }
        | Command::Export { .. } => None,
    };
    if let Some(cmd) = &wire_cmd
        && !cli.local
        && ipc::daemon_is_live(&cfg.general.socket).await
    {
        return ipc::proxy(&cfg.general.socket, cmd).await;
    }

    let ctx = Ctx {
        cfg: std::sync::RwLock::new(std::sync::Arc::new(cfg)),
        db,
        client: reqwest::Client::builder()
            .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 { web } => daemon(ctx, config_path, web).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,
    }
}

async fn run(ctx: &Ctx, cmd: Cmd) -> Result<()> {
    match cmd {
        Cmd::Fetch { feed, force } => {
            // Make room before pulling more down, as the original did per download.
            reap(ctx, false, false)?;
            fetch(ctx, feed.as_deref(), force).await
        }
        Cmd::Reap { dry_run } => reap(ctx, dry_run, true),
        Cmd::Download { enclosure } => download_one(ctx, enclosure).await,
        Cmd::Status => {
            let (pending, downloaded) = ctx.db.counts()?;
            ctx.out.emit(Event::Status { feeds: ctx.cfg().feeds.len(), pending, downloaded });
            Ok(())
        }
    }
}

async fn daemon(ctx: Ctx, config_path: PathBuf, web_addr: Option<String>) -> Result<()> {
    let socket = ctx.cfg().general.socket.clone();
    if ipc::daemon_is_live(&socket).await {
        anyhow::bail!("a daemon is already listening on {}", socket.display());
    }

    let (events, _) = broadcast::channel(1024);
    let (tx_cmd, mut rx_cmd) = mpsc::channel::<Cmd>(64);
    let ctx = Arc::new(Ctx { out: Emitter::socket(events.clone(), false), ..ctx });

    let web = start_web(&ctx, &config_path, web_addr, &tx_cmd, &events).await?;
    let server = tokio::spawn(ipc::serve(socket.clone(), events.clone(), tx_cmd));

    // One command at a time: the queue is what keeps two scans from overlapping.
    let mut ticker = tokio::time::interval(std::time::Duration::from_secs(60));
    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    tracing::info!(feeds = ctx.cfg().feeds.len(), "daemon started");

    // A signal has to be able to interrupt work in progress, not just the wait between
    // jobs. Racing `shutdown()` in the outer select only cancels branch selection: once
    // inside a long download the daemon stopped listening and had to be SIGKILLed.
    let (tx_stop, rx_stop) = tokio::sync::watch::channel(false);
    tokio::spawn(async move {
        shutdown().await;
        let _ = tx_stop.send(true);
    });

    // Runs one job, abandoning it if a signal arrives. Returns false to end the loop.
    async fn until_stopped(
        ctx: &Ctx,
        rx: &watch::Receiver<bool>,
        job: impl Future<Output = Result<()>>,
    ) -> bool {
        let mut stop = rx.clone();
        tokio::select! {
            _ = stop.changed() => {
                tracing::info!("signal received; abandoning the job in progress");
                false
            }
            result = job => {
                if let Err(e) = result {
                    ctx.out.emit(Event::Error { msg: format!("{e:#}") });
                }
                true
            }
        }
    }

    let mut stop = rx_stop.clone();
    loop {
        if *stop.borrow() {
            break;
        }
        tokio::select! {
            biased;
            _ = stop.changed() => break,
            Some(cmd) = rx_cmd.recv() => {
                tracing::info!(?cmd, "command from a client");
                if !until_stopped(&ctx, &rx_stop, run(&ctx, cmd)).await {
                    break;
                }
            }
            _ = ticker.tick() => {
                // Per-feed schedule and TTL decide what actually gets polled.
                let job = run(&ctx, Cmd::Fetch { feed: None, force: false });
                if !until_stopped(&ctx, &rx_stop, job).await {
                    break;
                }
            }
        }
    }

    server.abort();
    if let Some(w) = web {
        w.abort();
    }
    let _ = std::fs::remove_file(&socket);
    tracing::info!("daemon stopped");
    Ok(())
}

/// Starts the web UI when it is switched on, minting and saving a token if there is none.
async fn start_web(
    ctx: &Arc<Ctx>,
    config_path: &std::path::Path,
    web_addr: Option<String>,
    cmds: &mpsc::Sender<Cmd>,
    events: &broadcast::Sender<Event>,
) -> Result<Option<tokio::task::JoinHandle<()>>> {
    let cfg = ctx.cfg();
    let enabled = cfg.web.enabled || web_addr.is_some();
    if !enabled {
        return Ok(None);
    }
    let bind = web_addr.unwrap_or_else(|| cfg.web.bind.clone());

    if cfg.web.token.is_empty() {
        let mut fresh = (*cfg).clone();
        fresh.web.enabled = true;
        fresh.web.bind = bind.clone();
        fresh.web.token = web::generate_token();
        fresh.save(config_path)?;
        ctx.reload_cfg(config_path)?;
        println!("web ui token generated. Open:\n  http://{bind}/?token={}", fresh.web.token);
    } else {
        println!(
            "web ui at http://{bind}/?token={}",
            ctx.cfg().web.token
        );
    }

    if ctx.cfg().web.binds_publicly() {
        tracing::warn!(bind, "web ui is reachable off this machine; the token is all that guards it");
    }

    let state = web::WebState {
        ctx: ctx.clone(),
        config_path: config_path.to_path_buf(),
        cmds: cmds.clone(),
        events: events.clone(),
    };
    Ok(Some(tokio::spawn(async move {
        if let Err(e) = web::serve(state, &bind).await {
            tracing::error!(error = ?e, "web ui stopped");
        }
    })))
}

async fn shutdown() {
    use tokio::signal::unix::{SignalKind, signal};
    let mut term = match signal(SignalKind::terminate()) {
        Ok(s) => s,
        Err(_) => return std::future::pending().await,
    };
    tokio::select! {
        _ = tokio::signal::ctrl_c() => {}
        _ = term.recv() => {}
    }
}

/// Subscribes to one feed, naming it from its own title.
async fn add(
    ctx: Ctx,
    config_path: &std::path::Path,
    url: &str,
    folder: Option<String>,
    keywords: Vec<String>,
) -> Result<()> {
    let mut cfg = (*ctx.cfg()).clone();
    if let Some((id, _)) = cfg.feeds.iter().find(|(_, f)| f.url == url) {
        anyhow::bail!("already subscribed as {id:?}");
    }
    let id = add_one(&ctx, &mut cfg, url, folder, keywords).await?;
    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.
pub async fn add_one(
    ctx: &Ctx,
    cfg: &mut config::Config,
    url: &str,
    folder: Option<String>,
    keywords: Vec<String>,
) -> Result<String> {
    let probe = config::Feed {
        url: url.to_owned(),
        folder: folder.clone(),
        schedule: None,
        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, &cfg.feeds);
    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(ctx: Ctx, config_path: &std::path::Path, feed: &str) -> Result<()> {
    let mut cfg = (*ctx.cfg()).clone();
    if cfg.feeds.remove(feed).is_none() {
        anyhow::bail!("no feed with id {feed:?}");
    }
    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(ctx: Ctx, config_path: &std::path::Path, file: &std::path::Path) -> Result<()> {
    let mut cfg = (*ctx.cfg()).clone();
    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 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, &cfg.feeds);
        cfg.feeds.insert(
            id.clone(),
            config::Feed {
                url,
                folder: None,
                schedule: 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;
    }
    cfg.save(config_path)?;
    println!("{added} feed(s) imported");
    Ok(())
}

/// OPML nests feeds inside folder outlines, so this walks the whole tree.
pub 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<()> {
    let cfg = ctx.cfg();
    if cfg.feeds.is_empty() {
        println!("No feeds configured in {}", config_path.display());
        return Ok(());
    }
    for (id, feed) in &cfg.feeds {
        let s = ctx.db.feed_summary(id)?;
        println!("{id}  {}", s.title.as_deref().unwrap_or("-"));
        println!("    url          {}", feed.url);
        println!("    last checked {}", ago(s.last_checked));
        println!("    entries      {} ({} downloaded)", s.entries, s.downloaded);
        if let Some(err) = &s.last_error {
            println!("    last error   {err}");
        }
    }
    Ok(())
}

/// `standalone` false means this is the sweep that runs before a scan: it reports what it
/// deleted, but must not emit the terminal ReapDone, or a client waiting on its `fetch`
/// would stop reading before the scan had even started.
fn reap(ctx: &Ctx, dry_run: bool, standalone: bool) -> Result<()> {
    let r = retention::run(&ctx.cfg(), &ctx.db, dry_run)?;
    for c in r.aged_out.iter().chain(r.over_quota.iter()) {
        ctx.out.emit(Event::Reaped {
            path: c.path.clone(),
            bytes: c.bytes.max(0) as u64,
        });
    }
    if standalone {
        ctx.out.emit(Event::ReapDone {
            files: r.aged_out.len() + r.over_quota.len(),
            bytes: r.bytes_freed,
        });
    }
    Ok(())
}

async fn fetch(ctx: &Ctx, only: Option<&str>, force: bool) -> Result<()> {
    let cfg = ctx.cfg();
    if let Some(id) = only
        && !cfg.feeds.contains_key(id)
    {
        anyhow::bail!("no feed with id {id:?}");
    }

    let mut scanned = 0;
    for (id, feed_cfg) in cfg
        .feeds
        .iter()
        .filter(|(id, _)| only.is_none_or(|o| o == *id))
    {
        let state = ctx.db.http_state(id)?;

        if !force && let Some(last) = state.last_checked {
            let due = last + due_after(&cfg, feed_cfg, state.ttl_mins) as i64;
            if due > db::now() {
                ctx.out.emit(Event::FeedSkip {
                    feed: id.clone(),
                    reason: format!("not due for {}", duration((due - db::now()) as u64)),
                });
                continue;
            }
        }

        scanned += 1;
        ctx.out.emit(Event::FeedStart { feed: id.clone() });
        match scan_one(ctx, id, feed_cfg, &state).await {
            Ok(Some(s)) => ctx.out.emit(Event::FeedDone {
                feed: id.clone(),
                new: s.new_entries,
                downloaded: s.downloaded,
                failed: s.failed,
                torrents: s.torrents,
            }),
            Ok(None) => ctx.out.emit(Event::FeedSkip {
                feed: id.clone(),
                reason: "not modified".into(),
            }),
            Err(e) => {
                // One bad feed must not end the scan.
                let msg = format!("{e:#}");
                ctx.out.emit(Event::FeedError { feed: id.clone(), msg: msg.clone() });
                ctx.db.set_feed_error(id, &feed_cfg.url, &msg)?;
            }
        }
    }
    ctx.out.emit(Event::ScanDone { feeds: scanned });
    Ok(())
}

/// Seconds to wait before re-checking a feed.
///
/// A per-feed schedule is an explicit instruction and wins outright. Without one, the
/// global schedule applies, but the feed's own <ttl> raises it when the publisher asks to
/// be polled less often.
pub fn due_after(cfg: &config::Config, feed: &config::Feed, ttl_mins: Option<u64>) -> u64 {
    let mins = match feed.schedule.as_deref().and_then(config::parse_interval) {
        Some(explicit) => explicit,
        None => ttl_mins.unwrap_or(0).max(cfg.general.interval()),
    };
    mins * 60
}

#[derive(Default)]
struct Scan {
    new_entries: usize,
    downloaded: usize,
    failed: usize,
    torrents: usize,
}

/// Ok(None) means 304.
async fn scan_one(
    ctx: &Ctx,
    id: &str,
    feed_cfg: &config::Feed,
    state: &db::HttpState,
) -> Result<Option<Scan>> {
    let fetched = feed::fetch(
        &ctx.client,
        feed_cfg,
        state.etag.as_deref(),
        state.last_modified.as_deref(),
    )
    .await?;

    let (bytes, etag, last_modified) = match fetched {
        feed::Fetched::NotModified => {
            ctx.db.touch_feed(id, &feed_cfg.url)?;
            return Ok(None);
        }
        feed::Fetched::Body { bytes, etag, last_modified } => (bytes, etag, last_modified),
    };

    let parsed = feed::parse(&bytes)?;
    ctx.db.record_feed(
        id,
        &feed_cfg.url,
        parsed.title.as_deref(),
        etag.as_deref(),
        last_modified.as_deref(),
        parsed.ttl_mins,
        parsed.image.as_deref(),
    )?;

    let mut scan = Scan::default();
    for entry in &parsed.entries {
        if ctx.db.record_entry(id, entry)? {
            scan.new_entries += 1;
        }
        for enc in &entry.enclosures {
            if !ctx.db.record_enclosure(id, &entry.guid, enc)? {
                continue; // Seen before: downloaded, skipped or deliberately reaped.
            }
            // Filters run once, at discovery, and are recorded in `state`. The download
            // queue below is then just "everything still pending".
            if let Some(reason) = reject(feed_cfg, entry, &enc.url) {
                ctx.db.mark_enclosure(&enc.url, "skipped", Some(reason))?;
            }
        }
    }

    let budget = feed_cfg.max_new_per_check.unwrap_or(usize::MAX);
    if feed_cfg.auto_download && budget > 0 {
        let cfg = ctx.cfg();
        let folder = download::folder_for(&cfg, id, feed_cfg, parsed.title.as_deref());
        let dest_dir = cfg.general.download_dir.join(&folder);

        for item in ctx.db.pending(id, budget)? {
            if download::looks_like_torrent(&item.url, item.mime.as_deref()) {
                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 {
                Ok((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;
                }
            }
        }
    }

    Ok(Some(scan))
}

/// Why this enclosure should not be downloaded, if it should not be.
fn reject(feed_cfg: &config::Feed, entry: &feed::Entry, url: &str) -> Option<&'static str> {
    if !feed_cfg.auto_download {
        return Some("auto_download is off");
    }
    if entry.explicit && !feed_cfg.allow_explicit {
        return Some("explicit");
    }
    let categories = entry.categories.join(" ");
    let haystacks = [
        url,
        entry.title.as_deref().unwrap_or(""),
        entry.description.as_deref().unwrap_or(""),
        categories.as_str(),
    ];
    if !download::matches_keywords(&feed_cfg.keywords, &haystacks) {
        return Some("no keyword match");
    }
    None
}

async fn fetch_one(
    ctx: &Ctx,
    feed_id: &str,
    feed_cfg: &config::Feed,
    url: &str,
    dest_dir: &std::path::Path,
) -> Result<(PathBuf, u64)> {
    // Throttled to whole percents, as the original's lastDLStepSize guard did.
    let mut last_pct = -1i64;
    let name = download::filename_for(url, None);
    let cfg = ctx.cfg();
    let got = download::download(&ctx.client, &cfg, feed_cfg, url, |done, total| {
        if let Some(t) = total.filter(|t| *t > 0) {
            let pct = (done * 100 / t) 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,
                });
            }
        }
    })
    .await?;

    if matches!(got.kind, download::Kind::Torrent) {
        // 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;
        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?;
    ctx.db.mark_downloaded(url, &path, got.bytes)?;
    Ok((path, got.bytes))
}

/// Downloads one specific enclosure immediately, whatever the per-scan cap says and
/// wherever it sits in the queue.
async fn download_one(ctx: &Ctx, id: i64) -> Result<()> {
    let cfg = ctx.cfg();
    let enc = ctx
        .db
        .enclosure(id)?
        .ok_or_else(|| anyhow::anyhow!("no enclosure {id}"))?;
    if enc.path.is_some() {
        return Ok(()); // Already here.
    }
    let feed_cfg = cfg
        .feeds
        .get(&enc.feed_id)
        .ok_or_else(|| anyhow::anyhow!("enclosure {id} belongs to unsubscribed feed {:?}", enc.feed_id))?;

    let title = ctx.db.feed_summary(&enc.feed_id)?.title;
    let folder = download::folder_for(&cfg, &enc.feed_id, feed_cfg, title.as_deref());
    let dest_dir = cfg.general.download_dir.join(&folder);

    ctx.out.emit(Event::FeedStart { feed: enc.feed_id.clone() });
    let is_torrent = download::looks_like_torrent(&enc.url, enc.mime.as_deref());
    let result = if is_torrent {
        if !cfg.torrent.enabled {
            Err(anyhow::anyhow!("torrents are disabled"))
        } else {
            torrent_one(ctx, &enc.feed_id, &enc.url, &dest_dir).await
        }
    } else {
        fetch_one(ctx, &enc.feed_id, feed_cfg, &enc.url, &dest_dir).await
    };

    match result {
        Ok((path, bytes)) => {
            ctx.db.mark_downloaded(&enc.url, &path, bytes)?;
            ctx.out.emit(Event::DownloadDone {
                feed: enc.feed_id.clone(),
                url: enc.url.clone(),
                path: path.display().to_string(),
                bytes,
            });
        }
        Err(e) => {
            let msg = format!("{e:#}");
            ctx.db.mark_enclosure(&enc.url, "error", Some(&msg))?;
            ctx.out.emit(Event::DownloadError {
                feed: enc.feed_id.clone(),
                url: enc.url.clone(),
                msg,
            });
        }
    }
    // Terminal, so a UI waiting on this request stops here.
    ctx.out.emit(Event::ScanDone { feeds: 1 });
    Ok(())
}

/// 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;
    let cfg = ctx.cfg();
    ctx.torrents()
        .await?
        .fetch(&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<i64>) -> String {
    let Some(t) = t else { return "never".into() };
    format!("{} ago", duration((db::now() - t).max(0) as u64))
}

fn duration(secs: u64) -> String {
    match secs {
        s if s < 90 => format!("{s}s"),
        s if s < 5400 => format!("{}m", s / 60),
        s if s < 172_800 => format!("{}h", s / 3600),
        s => format!("{}d", s / 86_400),
    }
}