ipodderx-rs/src/main.rs

mod auth;
mod config;
mod db;
mod download;
mod feed;
mod ipc;
mod logbuf;
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 },
    /// Add, list or remove the accounts that can sign in to the web UI
    User {
        #[command(subcommand)]
        cmd: UserCmd,
    },
    /// 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>,
    },
}

#[derive(Subcommand)]
enum UserCmd {
    /// Create an account. The password is read from stdin: `echo -n hunter2 | ipx user add ray`
    Add {
        name: String,
        /// May manage other accounts, and is who the shared web token signs in as
        #[arg(long)]
        admin: bool,
        /// Sign-in comes from the proxy instead, so there is no password to set
        #[arg(long)]
        no_password: bool,
    },
    /// Show the accounts and how each one signs in
    List,
    /// Replace a password, read from stdin
    Passwd { name: String },
    /// Delete an account and everything it knows: its subscriptions and read state
    Rm { name: String },
}

/// What a brand new database starts with, so there is always a way in. Announced loudly
/// in the log, and the first thing the settings page nags about.
const DEFAULT_PASSWORD: &str = "ipodderx";

/// 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>,
    /// Caps how many torrents run at once when they are detached.
    pub torrent_slots: Arc<tokio::sync::Semaphore>,
    /// Needed because a subscribed OPML rewrites the feed list as it syncs.
    pub config_path: PathBuf,
    /// Run torrents off the command worker. A torrent takes minutes to fetch metadata and
    /// then seeds for up to an hour, and the worker is sequential -- inline, one torrent
    /// stops every feed scan, every HTTP download and every status command behind it.
    /// A one-shot CLI run keeps them inline, or the process would exit mid-download.
    pub detach_torrents: bool,
}

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();
    // Everything goes to stderr as before, and is mirrored into a ring the UI can read.
    {
        use tracing_subscriber::layer::SubscriberExt;
        use tracing_subscriber::util::SubscriberInitExt;
        use tracing_subscriber::Layer;
        // Two filters, deliberately different. stderr follows IPX_LOG; the in-app buffer
        // keeps debug as well, so the log view can show protocol traffic and routine
        // skips that would be noise on a terminal. IPX_UI_LOG overrides it.
        let stderr_filter = tracing_subscriber::EnvFilter::try_from_env("IPX_LOG")
            .unwrap_or_else(|_| "ipx=info".into());
        let ui_filter = tracing_subscriber::EnvFilter::try_from_env("IPX_UI_LOG")
            .unwrap_or_else(|_| "ipx=debug".into());
        tracing_subscriber::registry()
            .with(
                tracing_subscriber::fmt::layer()
                    .with_writer(std::io::stderr)
                    .with_filter(stderr_filter),
            )
            .with(logbuf::RingLayer.with_filter(ui_filter))
            .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::User { .. }
        | 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 is_daemon = matches!(cli.command, Command::Daemon { .. });
    let (events, _) = broadcast::channel(1024);
    let ctx = Arc::new(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: if is_daemon { Emitter::socket(events.clone(), false) } else { Emitter::terminal() },
        torrents: tokio::sync::OnceCell::new(),
        torrent_slots: Arc::new(tokio::sync::Semaphore::new(2)),
        config_path: config_path.clone(),
        detach_torrents: is_daemon,
    });

    match cli.command {
        Command::List => list(&ctx, &config_path),
        Command::Daemon { web } => daemon(ctx, config_path, web, events).await,
        Command::Add { url, folder, keywords } => {
            add(&ctx, &config_path, &url, folder, keywords).await
        }
        Command::Rm { feed } => rm(&ctx, &config_path, &feed),
        Command::User { cmd } => user_cmd(&ctx, cmd),
        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,
    }
}

/// Accounts. Passwords come in on stdin so they never reach a shell history or a `ps`
/// listing.
fn user_cmd(ctx: &Arc<Ctx>, cmd: UserCmd) -> Result<()> {
    let read_password = || -> Result<String> {
        use std::io::Read;
        let mut buf = String::new();
        std::io::stdin().read_to_string(&mut buf)?;
        let pw = buf.trim_end_matches(['\n', '\r']).to_string();
        if pw.is_empty() {
            anyhow::bail!("no password on stdin: try `echo -n secret | ipx user ...`");
        }
        Ok(pw)
    };

    match cmd {
        UserCmd::Add { name, admin, no_password } => {
            let name = name.trim().to_ascii_lowercase();
            if name.is_empty() {
                anyhow::bail!("a name is required");
            }
            if ctx.db.user_by_name(&name)?.is_some() {
                anyhow::bail!("{name} already exists");
            }
            let hash = if no_password {
                None
            } else {
                Some(crate::auth::hash_password(&read_password()?)?)
            };
            // The first account runs the place; there is nobody else to grant it.
            let first = ctx.db.users()?.is_empty();
            ctx.db.create_user(&name, hash.as_deref(), admin || first)?;
            println!(
                "added {name}{}{}",
                if admin || first { " (admin)" } else { "" },
                if no_password { ", signs in through the proxy" } else { "" }
            );
            Ok(())
        }
        UserCmd::List => {
            let users = ctx.db.users()?;
            if users.is_empty() {
                println!("no accounts yet: ipx user add <name>");
            }
            for u in users {
                println!(
                    "{:<20} {:<8} {}",
                    u.name,
                    if u.is_admin { "admin" } else { "" },
                    if u.pass_hash.is_some() { "password" } else { "proxy only" }
                );
            }
            Ok(())
        }
        UserCmd::Passwd { name } => {
            let name = name.trim().to_ascii_lowercase();
            let user = ctx
                .db
                .user_by_name(&name)?
                .ok_or_else(|| anyhow::anyhow!("no such account: {name}"))?;
            ctx.db.set_password(user.id, &crate::auth::hash_password(&read_password()?)?)?;
            println!("password changed for {name}");
            Ok(())
        }
        UserCmd::Rm { name } => {
            let name = name.trim().to_ascii_lowercase();
            let user = ctx
                .db
                .user_by_name(&name)?
                .ok_or_else(|| anyhow::anyhow!("no such account: {name}"))?;
            ctx.db.delete_user(user.id)?;
            println!("removed {name}");
            Ok(())
        }
    }
}

async fn run(ctx: &Arc<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()?;
            let feeds = subscriptions(ctx).map(|s| s.len()).unwrap_or(0);
            ctx.out.emit(Event::Status { feeds, pending, downloaded });
            Ok(())
        }
    }
}

async fn daemon(
    ctx: Arc<Ctx>,
    config_path: PathBuf,
    web_addr: Option<String>,
    events: broadcast::Sender<Event>,
) -> 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());
    }

    // A database with nobody in it cannot be signed into.
    if ctx.db.users()?.is_empty() {
        ctx.db.create_user("admin", Some(&crate::auth::hash_password(DEFAULT_PASSWORD)?), true)?;
        tracing::warn!(
            "no accounts yet: created 'admin' with the default password '{DEFAULT_PASSWORD}'. \
             Change it with `echo -n <password> | ipx user passwd admin`"
        );
    }

    if let Some(admin) = ctx.db.users()?.into_iter().find(|u| u.is_admin) {
        let catalogue: Vec<String> = ctx.cfg().feeds.keys().cloned().collect();
        match ctx.db.adopt_catalogue(admin.id, &catalogue) {
            Ok(0) => {}
            Ok(n) => tracing::info!(user = %admin.name, feeds = n, "subscribed the first admin to the catalogue"),
            Err(e) => tracing::error!(error = %e, "could not subscribe the first admin to the catalogue"),
        }
    }

    match ctx.db.requeue_interrupted() {
        Ok(n) if n > 0 => tracing::info!(count = n, "requeued downloads interrupted by a restart"),
        Ok(_) => {}
        Err(e) => tracing::warn!(error = ?e, "could not requeue interrupted downloads"),
    }

    let (tx_cmd, mut rx_cmd) = mpsc::channel::<Cmd>(64);

    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 = subscriptions(&ctx).map(|s| s.len()).unwrap_or(0),
        "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() => {
                // Both halves of the protocol are logged under one target so the UI can
                // show the conversation on its own: this is everything arriving, whatever
                // the source -- a socket client, the CLI proxying, or the web UI.
                tracing::info!(
                    target: "ipx::io",
                    "-> {}",
                    serde_json::to_string(&cmd).unwrap_or_else(|_| format!("{cmd:?}"))
                );
                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 = crate::auth::new_session_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();
    let url = &feed::expand_input(url);
    // Includes feeds derived from an OPML, or the same show could be added twice.
    if let Some(existing) = subscriptions(ctx)?.iter().find(|s| feed::same_feed(&s.cfg.url, url)) {
        anyhow::bail!("already subscribed as {:?}", existing.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(),
        group: None,
        media_types: None,
        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 {
        // An OPML subscription is named from its own <head><title>, not by trying to
        // parse it as a feed and falling back to the hostname.
        Ok(feed::Fetched::Body { bytes, .. }) if feed::is_opml(&bytes) => {
            feed::opml_title(&bytes).unwrap_or_else(|| url_stem(url))
        }
        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)
        }
    };

    // Slugs must be unique across derived feeds too, or a new feed can collide with one
    // an OPML already introduced.
    let mut taken: std::collections::BTreeMap<String, config::Feed> = subscriptions(ctx)?
        .into_iter()
        .map(|s| (s.id, s.cfg))
        .collect();
    // A removed feed keeps its rows, so its id is only free again for the same feed: re-adding
    // it gets its history back, and a different feed does not inherit someone else's.
    for (id, other) in ctx.db.feed_urls()? {
        if !feed::same_feed(&other, url) {
            taken.entry(id).or_insert_with(|| probe.clone());
        }
    }
    let id = config::unique_slug(&title, &taken);
    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() {
        // Derived from an OPML: drop it here, though the subscription will list it again
        // on the next read unless the OPML itself goes.
        ctx.db.drop_managed(feed)?;
        println!("removed {feed}; it came from an OPML subscription and may return on the next read");
        return Ok(());
    }
    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 text = std::fs::read_to_string(file)
        .with_context(|| format!("reading {}", file.display()))?;
    // The CLI speaks for the operator, as the shared web token does.
    let admin = ctx
        .db
        .users()?
        .into_iter()
        .find(|u| u.is_admin)
        .ok_or_else(|| anyhow::anyhow!("no admin account to subscribe: ipx user add <name> --admin"))?;
    let doc = opml::OPML::from_str(&text)
        .map_err(|e| anyhow::anyhow!("{} is not OPML: {e}", file.display()))?;
    let (added, had) = subscribe_opml(ctx, config_path, &doc, admin.id)?;
    println!("subscribed {} to {added} feed(s); {had} already there", admin.name);
    Ok(())
}

/// Subscribes one person to every feed in an OPML document, for the CLI and the web alike.
/// A feed already in the catalogue costs nothing; an unknown one is added under the OPML's
/// title rather than refetching each. Returns (newly subscribed, already subscribed).
///
/// Before accounts, importing only added unknown URLs to config.toml. Once subscriptions
/// decided what each person sees, that imported nothing at all for a feed someone else
/// already had, and a new one had no subscriber, so it was never scanned.
///
/// The caller parses the document, so each refuses a file that is not OPML in its own terms,
/// before anything is touched: a 400 from the web, a message from the CLI.
pub fn subscribe_opml(
    ctx: &Ctx,
    config_path: &std::path::Path,
    doc: &opml::OPML,
    user_id: i64,
) -> Result<(usize, usize)> {
    let mut found = vec![];
    collect_outlines(&doc.body.outlines, &mut found);

    let known = subscriptions(ctx)?;
    let mut cfg = (*ctx.cfg()).clone();
    let mut ids = vec![];
    let mut grew = false;
    for (title, url) in found {
        let existing = known
            .iter()
            .find(|s| s.cfg.url == url)
            .map(|s| s.id.clone())
            // The same URL listed twice in one file.
            .or_else(|| cfg.feeds.iter().find(|(_, f)| f.url == url).map(|(id, _)| id.clone()));
        let id = match existing {
            Some(id) => id,
            None => {
                let id = config::unique_slug(&title, &cfg.feeds);
                cfg.feeds.insert(
                    id.clone(),
                    config::Feed {
                        url,
                        folder: None,
                        group: None,
                        media_types: None,
                        schedule: None,
                        keywords: vec![],
                        allow_explicit: false,
                        auto_download: true,
                        max_new_per_check: None,
                        username: None,
                        password: None,
                        password_env: None,
                    },
                );
                grew = true;
                id
            }
        };
        ids.push(id);
    }
    if grew {
        cfg.save(config_path)?;
        ctx.reload_cfg(config_path)?;
    }

    let (mut added, mut had) = (0, 0);
    for id in ids {
        if ctx.db.subscription(user_id, &id)?.is_some() {
            had += 1;
        } else {
            ctx.db.subscribe(user_id, &id)?;
            added += 1;
        }
    }
    Ok((added, had))
}

/// 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: &Arc<Ctx>, only: Option<&str>, force: bool) -> Result<()> {
    let cfg = ctx.cfg();
    let subs = subscriptions(ctx)?;
    if let Some(id) = only
        && !subs.iter().any(|s| s.id == id)
    {
        anyhow::bail!("no feed with id {id:?}");
    }

    let mut scanned = 0;
    let mut fresh: Vec<String> = vec![];
    for sub in subs.iter().filter(|s| only.is_none_or(|o| o == s.id)) {
        let (id, feed_cfg) = (&sub.id, &sub.cfg);
        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(Outcome::Feed(s)) => ctx.out.emit(Event::FeedDone {
                feed: id.clone(),
                new: s.new_entries,
                downloaded: s.downloaded,
                failed: s.failed,
                torrents: s.torrents,
            }),
            Ok(Outcome::NotModified) => ctx.out.emit(Event::FeedSkip {
                feed: id.clone(),
                reason: "not modified".into(),
            }),
            Ok(Outcome::Opml { added, removed, kept, total }) => {
                ctx.out.emit(Event::FeedSkip {
                    feed: id.clone(),
                    reason: format!(
                        "{total} feed(s) listed, {} added, {removed} unsubscribed, {kept} kept without a listing",
                        added.len()
                    ),
                });
                // Read them in the same pass, as the original did, rather than making
                // the user wait a whole interval for a newly listed show.
                fresh.extend(added);
            }
            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)?;
            }
        }
    }
    // Feeds a subscribed OPML just introduced: scan them now, in this run.
    if !fresh.is_empty() {
        let subs = subscriptions(ctx)?;
        for id in &fresh {
            let Some(feed_cfg) = subs.iter().find(|s| &s.id == id).map(|s| &s.cfg) else {
                continue;
            };
            scanned += 1;
            ctx.out.emit(Event::FeedStart { feed: id.clone() });
            let state = ctx.db.http_state(id)?;
            match scan_one(ctx, id, feed_cfg, &state).await {
                Ok(Outcome::Feed(s)) => ctx.out.emit(Event::FeedDone {
                    feed: id.clone(),
                    new: s.new_entries,
                    downloaded: s.downloaded,
                    failed: s.failed,
                    torrents: s.torrents,
                }),
                Ok(_) => {}
                Err(e) => {
                    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(())
}

/// One feed to scan: either an entry you wrote in config.toml, or one derived from an
/// OPML subscription and held only in the database.
pub struct Sub {
    pub id: String,
    pub cfg: config::Feed,
    /// True when it came from an OPML and has no config entry of its own.
    pub managed: bool,
}

/// Everything to scan: your config entries, plus whatever the OPML subscriptions listed.
///
/// A derived feed borrows its parent's settings wholesale. That is why it needs no config
/// entry -- there is nothing to store but its URL and where it came from.
pub fn subscriptions(ctx: &Ctx) -> Result<Vec<Sub>> {
    let cfg = ctx.cfg();
    let mut out: Vec<Sub> = cfg
        .feeds
        .iter()
        .map(|(id, f)| Sub { id: id.clone(), cfg: f.clone(), managed: false })
        .collect();

    for m in ctx.db.managed_feeds()? {
        if cfg.feeds.contains_key(&m.id) {
            continue; // promoted to config at some point; that entry wins
        }
        let parent = cfg.feeds.get(&m.group_id);
        let base = parent
            .and_then(|p| p.folder.clone())
            .or_else(|| ctx.db.feed_summary(&m.group_id).ok().and_then(|s| s.title))
            .unwrap_or_else(|| m.group_id.clone());
        let title = m.title.clone().unwrap_or_else(|| m.id.clone());
        out.push(Sub {
            id: m.id.clone(),
            cfg: config::Feed {
                url: m.url.clone(),
                folder: Some(format!("{base}/{title}")),
                group: Some(m.group_id.clone()),
                media_types: parent.and_then(|p| p.media_types.clone()),
                schedule: parent.and_then(|p| p.schedule.clone()),
                keywords: parent.map(|p| p.keywords.clone()).unwrap_or_default(),
                allow_explicit: parent.is_some_and(|p| p.allow_explicit),
                auto_download: parent.is_none_or(|p| p.auto_download),
                max_new_per_check: parent.and_then(|p| p.max_new_per_check),
                username: parent.and_then(|p| p.username.clone()),
                password: parent.and_then(|p| p.password.clone()),
                password_env: parent.and_then(|p| p.password_env.clone()),
            },
            managed: true,
        });
    }
    out.sort_by(|a, b| a.id.cmp(&b.id));
    Ok(out)
}

/// 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,
}

/// What a scan of one feed turned out to be.
enum Outcome {
    NotModified,
    Feed(Scan),
    /// The URL is a list of feeds rather than a feed: an OPML, or a Patreon creator's shows.
    Opml { added: Vec<String>, removed: usize, kept: usize, total: usize },
}

async fn scan_one(
    ctx: &Arc<Ctx>,
    id: &str,
    feed_cfg: &config::Feed,
    state: &db::HttpState,
) -> Result<Outcome> {
    // A Patreon creator with more than one show is a list of feeds, like an OPML.
    if feed::is_patreon_creator(&feed_cfg.url) {
        match feed::patreon_shows(&ctx.client, &feed_cfg.url).await {
            Ok((name, shows)) if shows.len() > 1 => {
                ctx.db.touch_feed(id, &feed_cfg.url)?;
                if let Some(name) = name {
                    ctx.db.set_title(id, &name)?;
                }
                // Read as one feed before it was split, it listed every show's items in one
                // heap. The items go; its files and read state move to each show as the show
                // lists them (`Db::adopt`), so no show comes up empty for want of a URL.
                ctx.db.clear_entries(id)?;
                return sync_group(ctx, id, feed_cfg, &shows).await;
            }
            Ok(_) => {} // One show: the creator's feed is that show.
            // Already split: keep the shows it has rather than read the creator as one heap.
            Err(e) if ctx.db.managed_feeds()?.iter().any(|m| m.group_id == id) => return Err(e),
            Err(e) => tracing::warn!(
                feed = id,
                error = %format!("{e:#}"),
                "could not list the Patreon shows; reading it as one feed"
            ),
        }
    }

    let mut fetched = feed::fetch(
        &ctx.client,
        feed_cfg,
        state.etag.as_deref(),
        state.last_modified.as_deref(),
    )
    .await?;

    // A 304 while nothing is stored means the validator has outlived the data -- a restore
    // from backup, a manual edit, a cleanup that removed entries. Believe the database over
    // the validator: drop it and ask again, or the feed stays empty until the publisher
    // happens to change something.
    if matches!(fetched, feed::Fetched::NotModified) && ctx.db.feed_summary(id)?.entries == 0 {
        tracing::info!(feed = id, "not modified, but nothing stored; refetching without the validator");
        ctx.db.clear_validators(id)?;
        fetched = feed::fetch(&ctx.client, feed_cfg, None, None).await?;
    }

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

    // A subscribed OPML is a list of feeds, not a feed. The original matched on a ".opml"
    // URL; sniffing the body also catches one served from a URL without that extension.
    if feed::is_opml(&bytes) {
        ctx.db.touch_feed(id, &feed_cfg.url)?;
        return sync_opml(ctx, id, feed_cfg, &bytes).await;
    }

    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 policy = policy_for(ctx, id, feed_cfg)?;
    if let Some(parent) = &feed_cfg.group {
        let listed: Vec<(&str, &str)> = parsed
            .entries
            .iter()
            .flat_map(|e| e.enclosures.iter().map(move |x| (e.guid.as_str(), x.url.as_str())))
            .collect();
        ctx.db.adopt(parent, id, &listed)?;
    }
    // Verdicts are recorded in `state`, so the download queue below is just "everything still
    // pending". A filter's verdict is looked at again on every scan, though: made once, at
    // discovery, it outlived the setting behind it, and allowing explicit items afterwards
    // changed nothing however often the feed was scanned.
    let skipped = ctx.db.skipped_by_filter(id)?;
    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 {
            let was = if ctx.db.record_enclosure(id, &entry.guid, enc)? {
                None
            } else if let Some(reason) = skipped.get(&enc.url) {
                Some(reason.as_str())
            } else {
                continue; // Settled: queued, downloaded, reaped, or another feed's file.
            };
            let now = reject(&ctx.cfg(), feed_cfg, &policy, entry, enc);
            if now != was {
                match now {
                    Some(reason) => ctx.db.mark_enclosure(&enc.url, "skipped", Some(reason))?,
                    None => ctx.db.mark_enclosure(&enc.url, "pending", None)?,
                }
            }
        }
    }

    let budget = policy.budget;
    if policy.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;
                }
                if ctx.detach_torrents {
                    // 'downloading' keeps the next scan from queueing it a second time.
                    ctx.db.mark_enclosure(&item.url, "downloading", None)?;
                    spawn_torrent(ctx, id.to_string(), item.id, item.url.clone(), dest_dir.clone());
                    scan.torrents += 1;
                    continue;
                }
                match torrent_one(ctx, id, item.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(),
                            enclosure: item.id,
                            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(),
                            enclosure: item.id,
                            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, item.id, feed_cfg, &item.url, &dest_dir).await {
                Ok((path, bytes)) => {
                    ctx.out.emit(Event::DownloadDone {
                        feed: id.to_string(),
                        enclosure: item.id,
                        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(),
                        enclosure: item.id,
                        url: item.url.clone(),
                        msg: msg.clone(),
                    });
                    ctx.db.mark_enclosure(&item.url, "error", Some(&msg))?;
                    scan.failed += 1;
                }
            }
        }
    }

    Ok(Outcome::Feed(scan))
}

async fn sync_opml(
    ctx: &Arc<Ctx>,
    parent_id: &str,
    parent: &config::Feed,
    bytes: &[u8],
) -> Result<Outcome> {
    let listed = feed::parse_opml(bytes)?;
    if let Some(title) = feed::opml_title(bytes) {
        ctx.db.set_title(parent_id, &title)?;
    }
    sync_group(ctx, parent_id, parent, &listed).await
}

/// Brings the feed list in step with a list of feeds: a subscribed OPML, or a Patreon
/// creator's shows.
///
/// New entries are added under the list's group and folder. An entry that has gone from
/// the list is unsubscribed *only if nothing was ever downloaded for it* -- otherwise it
/// is kept and flagged, because dropping it would orphan files on disk with nothing in
/// the UI to explain them.
async fn sync_group(
    ctx: &Arc<Ctx>,
    parent_id: &str,
    parent: &config::Feed,
    listed: &[(String, String)],
) -> Result<Outcome> {
    let cfg = ctx.cfg();
    let existing = ctx.db.managed_feeds()?;
    let mut added = vec![];

    for (title, url) in listed {
        // Already known, whether derived or promoted into the config.
        if let Some(m) = existing.iter().find(|m| &m.url == url) {
            ctx.db.upsert_managed(&m.id, url, title, parent_id)?;
            continue;
        }
        // A Patreon show you added by hand may be spelled differently from the one listed.
        if cfg.feeds.values().any(|f| feed::same_feed(&f.url, url)) {
            continue;
        }
        // A removed feed keeps its rows, so its id is only free again for the same feed.
        let known = ctx.db.feed_urls()?;
        let taken: std::collections::BTreeMap<String, config::Feed> = cfg
            .feeds
            .keys()
            .chain(existing.iter().map(|m| &m.id))
            .chain(added.iter())
            .chain(known.iter().filter(|(_, u)| !feed::same_feed(u, url)).map(|(id, _)| id))
            .map(|id| (id.clone(), parent.clone()))
            .collect();
        let id = config::unique_slug(title, &taken);
        ctx.db.upsert_managed(&id, url, title, parent_id)?;
        added.push(id);
    }

    // Whoever subscribes to the OPML subscribes to what it lists: that is what taking a
    // subscription means. Their own feeds are untouched.
    for id in ctx
        .db
        .managed_feeds()?
        .iter()
        .filter(|m| m.group_id == parent_id)
        .map(|m| m.id.clone())
        .chain(std::iter::once(parent_id.to_string()))
    {
        for user in ctx.db.users()? {
            if ctx.db.subscription(user.id, parent_id)?.is_some() {
                ctx.db.subscribe(user.id, &id)?;
            }
        }
    }

    // Anything in this group the OPML no longer lists.
    let mut removed = 0;
    let mut kept = 0;
    for m in existing.iter().filter(|m| m.group_id == parent_id) {
        if listed.iter().any(|(_, u)| u == &m.url) {
            continue;
        }
        if ctx.db.downloaded_count(&m.id).unwrap_or(1) > 0 {
            // Never orphan a downloaded file: keep the feed and say why in the UI.
            ctx.db.set_orphaned(&m.id, true)?;
            kept += 1;
            tracing::info!(feed = %m.id, "dropped from the OPML but has downloads; keeping it");
        } else {
            ctx.db.drop_managed(&m.id)?;
            removed += 1;
            tracing::info!(feed = %m.id, "dropped from the OPML with nothing downloaded; removed");
        }
    }

    Ok(Outcome::Opml { added, removed, kept, total: listed.len() })
}

/// Why this enclosure should not be downloaded, if it should not be.
fn reject(
    cfg: &config::Config,
    feed_cfg: &config::Feed,
    policy: &Policy,
    entry: &feed::Entry,
    enc: &feed::Enclosure,
) -> Option<&'static str> {
    let url = enc.url.as_str();
    if !policy.auto_download {
        return Some("auto_download is off");
    }
    // Blog feeds put the article's header image in an <enclosure>; without this a text
    // feed reads as a podcast full of episodes and fills the disk with artwork.
    let wanted = feed_cfg
        .media_types
        .as_deref()
        .unwrap_or(&cfg.general.media_types);
    if !config::wanted_media(enc.mime.as_deref(), wanted) {
        return Some("not audio or video");
    }
    if entry.explicit && !policy.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(),
    ];
    // One file serves everyone subscribed, so an item is wanted if it is wanted by
    // anyone: any one person's keyword set matching is enough.
    let wanted_by_someone = policy.keyword_sets.is_empty()
        || policy
            .keyword_sets
            .iter()
            .any(|set| download::matches_keywords(set, &haystacks));
    if !wanted_by_someone {
        return Some("no keyword match");
    }
    None
}

/// What the scanner should do for a feed, merged across everyone subscribed to it. The
/// feed is fetched once and its files are downloaded once, so the merge is a union: if
/// one person wants a thing, it is fetched, and everyone else simply sees it listed.
///
/// With no subscribers at all -- a hand-written config entry nobody has claimed yet --
/// the feed's own settings stand, which is how a single-user install behaves.
pub struct Policy {
    pub auto_download: bool,
    pub allow_explicit: bool,
    /// Empty means take everything. Otherwise one set per subscriber who filters.
    pub keyword_sets: Vec<Vec<String>>,
    pub budget: usize,
}

fn policy_for(ctx: &Ctx, id: &str, feed_cfg: &config::Feed) -> Result<Policy> {
    let global = ctx.cfg().general.max_new_per_check;
    Ok(merge_policy(&ctx.db.subscribers(id, feed_cfg.group.as_deref())?, feed_cfg, global))
}

fn merge_policy(subs: &[db::Sub], feed_cfg: &config::Feed, global: usize) -> Policy {
    let cap = |n: Option<usize>| n.unwrap_or(if global == 0 { usize::MAX } else { global });

    if subs.is_empty() {
        return Policy {
            auto_download: feed_cfg.auto_download,
            allow_explicit: feed_cfg.allow_explicit,
            keyword_sets: if feed_cfg.keywords.is_empty() {
                vec![]
            } else {
                vec![feed_cfg.keywords.clone()]
            },
            budget: cap(feed_cfg.max_new_per_check),
        };
    }

    let mut policy = Policy {
        auto_download: false,
        allow_explicit: false,
        keyword_sets: vec![],
        budget: 0,
    };
    for sub in subs {
        if !sub.auto_download.unwrap_or(feed_cfg.auto_download) {
            continue; // Not fetching for this person, so their wants add nothing.
        }
        policy.auto_download = true;
        policy.allow_explicit |= sub.allow_explicit.unwrap_or(feed_cfg.allow_explicit);
        policy.budget = policy
            .budget
            .max(cap(sub.max_new_per_check.map(|n| n as usize).or(feed_cfg.max_new_per_check)));
        let kw = sub.keywords.clone().unwrap_or_else(|| feed_cfg.keywords.clone());
        if kw.is_empty() {
            // Somebody takes everything, so no filter can apply to the shared copy.
            return Policy { keyword_sets: vec![], ..policy };
        }
        policy.keyword_sets.push(kw);
    }
    policy
}

async fn fetch_one(
    ctx: &Arc<Ctx>,
    feed_id: &str,
    enclosure: i64,
    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(),
                    enclosure,
                    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, enclosure, url, dest_dir).await;
    }

    let path = download::place(&got, dest_dir).await?;
    ctx.db.mark_downloaded(url, &path, got.bytes)?;
    Ok((path, got.bytes))
}

/// Runs a torrent off the command worker, so feed scans and HTTP downloads keep moving
/// while it fetches metadata, transfers and then seeds.
fn spawn_torrent(ctx: &Arc<Ctx>, feed_id: String, enclosure: i64, url: String, dest_dir: PathBuf) {
    let ctx = ctx.clone();
    tokio::spawn(async move {
        // Held for the whole job, so a feed full of torrents cannot open hundreds at once.
        let _permit = match ctx.torrent_slots.clone().acquire_owned().await {
            Ok(p) => p,
            Err(_) => return,
        };
        let outcome = torrent_one(&ctx, &feed_id, enclosure, &url, &dest_dir).await;
        let db = &ctx.db;
        match outcome {
            Ok((path, bytes)) => {
                if let Err(e) = db.mark_downloaded(&url, &path, bytes) {
                    tracing::warn!(error = ?e, "could not record the finished torrent");
                }
                ctx.out.emit(Event::DownloadDone {
                    feed: feed_id,
                    enclosure,
                    url,
                    path: path.display().to_string(),
                    bytes,
                });
            }
            Err(e) => {
                let msg = format!("{e:#}");
                let _ = db.mark_enclosure(&url, "error", Some(&msg));
                ctx.out.emit(Event::DownloadError { feed: feed_id, enclosure, url, msg });
            }
        }
    });
}

/// Downloads one specific enclosure immediately, whatever the per-scan cap says and
/// wherever it sits in the queue.
async fn download_one(ctx: &Arc<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.
    }
    // Must look through the derived feeds too: anything inside an OPML subscription has
    // no config entry, so a config-only lookup called every one of them "unsubscribed".
    let subs = subscriptions(ctx)?;
    let feed_cfg = subs
        .iter()
        .find(|s| s.id == enc.feed_id)
        .map(|s| s.cfg.clone())
        .ok_or_else(|| anyhow::anyhow!("enclosure {id} belongs to unsubscribed feed {:?}", enc.feed_id))?;
    let feed_cfg = &feed_cfg;

    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());
    if is_torrent && cfg.torrent.enabled && ctx.detach_torrents {
        ctx.db.mark_enclosure(&enc.url, "downloading", None)?;
        spawn_torrent(ctx, enc.feed_id.clone(), enc.id, enc.url.clone(), dest_dir);
        return Ok(());
    }
    let result = if is_torrent {
        if !cfg.torrent.enabled {
            Err(anyhow::anyhow!("torrents are disabled"))
        } else {
            torrent_one(ctx, &enc.feed_id, enc.id, &enc.url, &dest_dir).await
        }
    } else {
        fetch_one(ctx, &enc.feed_id, enc.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(),
                enclosure: enc.id,
                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(),
                enclosure: enc.id,
                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: &Arc<Ctx>,
    feed_id: &str,
    enclosure: i64,
    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(),
                        enclosure,
                        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),
    }
}

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

    fn feed() -> config::Feed {
        // Whatever `ipx add` would write, which is the shape every code path sees.
        let mut cfg = config::Config::default();
        let f = add_one_cfg(&mut cfg, "http://x/f.xml", None, vec![]);
        f
    }

    /// The feed entry `add` builds, without the network round trip it does for a title.
    fn add_one_cfg(
        _cfg: &mut config::Config,
        url: &str,
        folder: Option<String>,
        keywords: Vec<String>,
    ) -> config::Feed {
        config::Feed {
            url: url.into(),
            folder,
            keywords,
            allow_explicit: false,
            auto_download: true,
            group: None,
            media_types: None,
            schedule: None,
            max_new_per_check: None,
            username: None,
            password: None,
            password_env: None,
        }
    }

    fn sub(kw: Option<&[&str]>, auto: Option<bool>, max: Option<i64>) -> db::Sub {
        db::Sub {
            feed_id: "f".into(),
            keywords: kw.map(|k| k.iter().map(|s| s.to_string()).collect()),
            auto_download: auto,
            allow_explicit: None,
            max_new_per_check: max,
        }
    }

    #[test]
    fn a_shared_feed_is_fetched_for_whoever_wants_the_most() {
        // Nobody subscribed: the feed's own settings stand, as in a single-user install.
        let p = merge_policy(&[], &feed(), 3);
        assert!(p.auto_download);
        assert_eq!(p.budget, 3);
        assert!(p.keyword_sets.is_empty());

        // Two filters: an item wanted by either of them is fetched, since one file serves
        // both. The larger per-scan cap wins for the same reason.
        let p = merge_policy(
            &[sub(Some(&["rust"]), None, Some(2)), sub(Some(&["sqlite"]), None, Some(9))],
            &feed(),
            3,
        );
        assert_eq!(p.keyword_sets.len(), 2);
        assert_eq!(p.budget, 9);

        // One person taking everything removes the filter for the shared copy.
        let p = merge_policy(&[sub(Some(&["rust"]), None, None), sub(Some(&[]), None, None)], &feed(), 3);
        assert!(p.keyword_sets.is_empty());

        // Everyone has auto-download off: nothing is fetched automatically.
        let p = merge_policy(&[sub(None, Some(false), None), sub(None, Some(false), None)], &feed(), 3);
        assert!(!p.auto_download);

        // One of them wants it, so it is fetched.
        let p = merge_policy(&[sub(None, Some(false), None), sub(None, Some(true), None)], &feed(), 3);
        assert!(p.auto_download);
    }
}