ipx's main.rs splits into leaf modules (#356)

main.rs held 3,200 lines: the command line, the shared context, the
daemon, scans, downloads, the catalogue and OPML, and their tests. The
style guide keeps a root module to declarations and a little glue, the
code in leaf modules named for what they hold:

  cli.rs        the commands, run and user_cmd, ago and duration
  ctx.rs        Ctx and CfgEdit, and Ctx::new, which main() built inline
  daemon.rs     the daemon's loop, web server, shutdown, status, reload
  scan.rs       due_at/due_after, scan_one, policy, downloads, torrents
  catalogue.rs  add/rm/leave/remove_user, subscriptions, retiring lists
  lists.rs      OPML import and syncing an OPML's or a creator's feeds
                (not opml.rs, which would shadow the opml crate)
  logging.rs    logging::init and the JSON lines with their trace
  test_util.rs  the context, feeds and feed server tests share

first_password moves to auth.rs and reap to retention.rs. Each test went
with what it tests; all 151 still run. main.rs is the module list and a
main() that hands the command on, 113 lines.

Moved verbatim, then made pub(crate) where a sibling calls it, with
calls across modules written module::function as the guide asks.
Everything the old main.rs defined is defined once in the new files.

With Ctx no longer a pub type at the crate root, Db's pub methods stop
counting as public API, and the compiler sees what nothing outside tests
calls: entries_in, count_in, unread_count and feed_description are now
#[cfg(test)], and media_feeds, called by nothing, is gone.

ipx depends on opentelemetry_sdk directly now (it was already in the
lock through ipx-directory), for logging::init's signature.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Cargo.lock

@@ -1875,6 +1875,7 @@ dependencies = [
  "jsonwebtoken",
  "librqbit",
  "opentelemetry",
+ "opentelemetry_sdk",
  "opml",
  "percent-encoding",
  "reqwest",

crates/ipx/Cargo.toml

@@ -21,6 +21,7 @@ futures-util.workspace = true
 jsonwebtoken.workspace = true
 librqbit.workspace = true
 opentelemetry.workspace = true
+opentelemetry_sdk.workspace = true
 opml.workspace = true
 percent-encoding.workspace = true
 reqwest.workspace = true

crates/ipx/src/auth.rs

@@ -2,11 +2,12 @@
 //! this -- Cloudflare Zero Trust on `ipodderx.sdf1.net`, which puts the authenticated
 //! address in `Cf-Access-Authenticated-User-Email`.
 
-use anyhow::{Result, bail};
+use anyhow::{Context, Result, bail};
 use argon2::Argon2;
 use argon2::password_hash::phc::PasswordHash;
 use argon2::password_hash::{PasswordHasher, PasswordVerifier};
-use std::fs::File;
+use std::fs::{File, OpenOptions};
+use std::path::Path;
 use std::sync::{Arc, LazyLock};
 use tokio::sync::Semaphore;
 
@@ -109,6 +110,36 @@ pub fn name_from_header(raw: &str) -> Option<String> {
     Some(name.to_ascii_lowercase())
 }
 
+/// The first admin's password: `IPX_ADMIN_PASSWORD` when set, or a random one written to
+/// `admin-password` beside config.toml, readable by the daemon's user alone. Every install began
+/// as admin / ipodderx, published in the README, and nothing made anyone change it (#321).
+/// Returns the password and where it can be found, for the log, which never shows it.
+pub(crate) fn first_password(config_dir: &Path, from_env: Option<String>) -> Result<(String, String)> {
+    if let Some(p) = from_env.filter(|p| !p.is_empty()) {
+        return Ok((p, "with the password in IPX_ADMIN_PASSWORD".into()));
+    }
+    let password = crate::auth::new_session_token()[..24].to_owned();
+    let file = config_dir.join("admin-password");
+    write_private(&file, &password)?;
+    Ok((password, format!("with the password in {}", file.display())))
+}
+
+/// Writes a secret only its owner may read: created 0600, never readable for a moment between.
+pub(crate) fn write_private(file: &Path, text: &str) -> Result<()> {
+    use std::io::Write;
+    use std::os::unix::fs::OpenOptionsExt;
+    std::fs::create_dir_all(file.parent().unwrap_or(Path::new(".")))?;
+    let _ = std::fs::remove_file(file);
+    let mut f = OpenOptions::new()
+        .write(true)
+        .create_new(true)
+        .mode(0o600)
+        .open(file)
+        .with_context(|| format!("writing {}", file.display()))?;
+    f.write_all(text.as_bytes())?;
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -164,4 +195,28 @@ mod tests {
         assert_eq!(name_from_header("ray\nadmin"), None);
         assert_eq!(name_from_header("ray;admin"), None);
     }
+
+    #[test]
+    fn the_first_admin_gets_a_password_nobody_else_knows() {
+        use std::os::unix::fs::PermissionsExt;
+        let dir = std::env::temp_dir().join(format!("ipx-first-password-{}", std::process::id()));
+        let _ = std::fs::remove_dir_all(&dir);
+
+        let (a, told) = first_password(&dir, None).unwrap();
+        let file = dir.join("admin-password");
+        assert_eq!(std::fs::read_to_string(&file).unwrap(), a);
+        assert_eq!(std::fs::metadata(&file).unwrap().permissions().mode() & 0o777, 0o600);
+        assert!(told.contains("admin-password") && !told.contains(&a), "the log says where, not what: {told}");
+        assert_eq!(a.len(), 24);
+        assert_ne!(a, "ipodderx");
+        assert_ne!(first_password(&dir, None).unwrap().0, a, "each new database gets its own");
+
+        std::fs::remove_dir_all(&dir).unwrap();
+        let (b, told) = first_password(&dir, Some("chosen".into())).unwrap();
+        assert_eq!(b, "chosen");
+        assert!(!told.contains("chosen"));
+        assert!(!file.exists(), "a password set in the environment is not written anywhere");
+        assert_eq!(first_password(&dir, Some(String::new())).unwrap().0.len(), 24, "empty means unset");
+        std::fs::remove_dir_all(&dir).unwrap();
+    }
 }

crates/ipx/src/catalogue.rs

@@ -0,0 +1,616 @@
+//! The feeds the server knows and who reads them: adding, removing, leaving, and what an
+//! OPML or Patreon list leaves behind when it goes.
+
+use crate::ctx::Ctx;
+use crate::ipc::Event;
+use crate::{cli, config, db, feed, lists};
+use anyhow::{Context, Result};
+use std::collections::BTreeMap;
+use std::path::Path;
+use url::Url;
+
+/// Subscribes to one feed, naming it from its own title.
+pub(crate) async fn add(
+    ctx: &Ctx,
+    url: &str,
+    folder: Option<String>,
+    keywords: Vec<String>,
+    category: Option<String>,
+    list: bool,
+) -> Result<()> {
+    let url = feed::find_feed(&ctx.client, &feed::expand_input(url)).await?;
+    let (id, existed) = add_to_catalogue(ctx, &url, folder, keywords).await?;
+    // Listing a feed the catalogue already has, or giving it a category, is the point of asking
+    // again: it changes those, and nothing else. A feed an OPML lists has no entry to change.
+    if list || category.is_some() {
+        let mut cfg = ctx.edit_cfg().await;
+        if let Some(f) = cfg.feeds.get_mut(&id) {
+            f.listed |= list;
+            if category.is_some() {
+                f.category = category;
+            }
+            ctx.store_cfg(cfg).await?;
+            if existed {
+                println!("{id} is already in the catalogue; updated its Directory listing");
+                return Ok(());
+            }
+        }
+    }
+    if existed {
+        anyhow::bail!("already subscribed as {id:?}");
+    }
+    println!("added {id}{}", if list { ", listed in the Directory" } else { "" });
+    Ok(())
+}
+
+/// Adds the feed at `url` to the catalogue unless it is there already, a feed an OPML lists
+/// included, so one show is never added twice: its id, and whether it was there. For `ipx add`
+/// and the page's Add feed alike, which were the same steps written twice (#161); each finds the
+/// feed first, to say its own way why an address is not one.
+pub async fn add_to_catalogue(
+    ctx: &Ctx,
+    url: &str,
+    folder: Option<String>,
+    keywords: Vec<String>,
+) -> Result<(String, bool)> {
+    // Taken before the duplicate check, so two people adding the same feed at once add it once.
+    let mut cfg = ctx.edit_cfg().await;
+    if let Some(existing) = subscriptions(ctx).await?.into_iter().find(|s| feed::same_feed(&s.cfg.url, url)) {
+        return Ok((existing.id, true));
+    }
+    let id = add_one(ctx, &mut cfg, url, folder, keywords).await?;
+    ctx.store_cfg(cfg).await?;
+    Ok((id, false))
+}
+
+/// Returns the new feed id. Adding checks the address is a feed first (`feed::find_feed`); this
+/// still names one it cannot read from its URL rather than failing.
+pub(crate) 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,
+        category: None,
+        listed: false,
+        private: false,
+    };
+
+    let title = match feed::fetch(&ctx.feed_client, &probe.url, None, None, None).await.map(|(got, _)| got) {
+        // 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: BTreeMap<String, config::Feed> =
+        subscriptions(ctx).await?.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().await? {
+        if !feed::same_feed(&other, url) {
+            taken.entry(id).or_insert_with(|| probe.clone());
+        }
+    }
+    let id = config::unique_slug(&title, |s| taken.contains_key(s));
+    cfg.feeds.insert(id.clone(), probe);
+    Ok(id)
+}
+
+/// Host plus last path segment, for naming a feed we could not read.
+pub(crate) fn url_stem(url: &str) -> String {
+    Url::parse(url).ok().and_then(|u| u.host_str().map(str::to_owned)).unwrap_or_else(|| url.to_owned())
+}
+
+pub(crate) async fn rm(ctx: &Ctx, feed: &str) -> Result<()> {
+    let mut cfg = ctx.edit_cfg().await;
+    let Some(parent) = cfg.feeds.remove(feed) else {
+        // 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).await?;
+        println!("removed {feed}; it came from an OPML subscription and may return on the next read");
+        return Ok(());
+    };
+    ctx.store_cfg(cfg).await?;
+    // State and files stay: re-adding the feed should not re-download its back catalogue.
+    println!("removed {feed}; downloads and history kept");
+    retire_group(ctx, feed, Some(&parent)).await?;
+    Ok(())
+}
+
+pub(crate) async fn import(ctx: &Ctx, file: &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()
+        .await?
+        .into_iter()
+        .find(|u| u.is_admin)
+        .ok_or_else(|| anyhow::anyhow!("no admin account to subscribe: ipx user add <name> --admin"))?;
+    let doc = feed::read_opml(&text).map_err(|e| anyhow::anyhow!("{} is not OPML: {e}", file.display()))?;
+    let (added, had) = lists::subscribe_opml(ctx, lists::opml_feeds(&doc)?, admin.id).await?;
+    println!("subscribed {} to {added} feed(s); {had} already there", admin.name);
+    Ok(())
+}
+
+/// The configuration ipx runs with: config.toml for where things are and who may sign in, the
+/// database for the feeds and the server settings (issue #18). The first time a database holds
+/// neither, it takes them from config.toml, which is then cut down to the rest, the original kept
+/// beside it as config.toml.pre-database.
+pub(crate) async fn assemble_config(db: &db::Db, mut cfg: config::Config, path: &Path) -> Result<config::Config> {
+    for _ in 0..2 {
+        if let Some((stored, feeds)) = db.stored_config().await? {
+            if config::Config::file_holds_stored(path) {
+                tracing::warn!(
+                    "config.toml still lists feeds or server settings; they are ignored, since the \
+                     database holds them now. Change them in the web UI, or with ipx add and rm."
+                );
+            }
+            stored.apply(&mut cfg);
+            cfg.feeds = feeds;
+            return Ok(cfg);
+        }
+        if db.import_config(&cfg).await? {
+            if path.exists() {
+                let original = path.with_extension("toml.pre-database");
+                if !original.exists() {
+                    std::fs::copy(path, &original)
+                        .with_context(|| format!("keeping the original as {}", original.display()))?;
+                }
+                cfg.save_bootstrap(path)?;
+            }
+            tracing::info!(
+                feeds = cfg.feeds.len(),
+                "moved the feeds and server settings from config.toml into the database"
+            );
+            return Ok(cfg);
+        }
+        // Another ipx imported between our look and our insert; the next pass reads its copy.
+    }
+    anyhow::bail!("the database says it holds the configuration and then that it does not")
+}
+
+pub(crate) async fn export(ctx: &Ctx, file: &std::path::Path) -> Result<()> {
+    let mut doc = opml::OPML {
+        head: Some(opml::Head { title: Some("ipx subscriptions".into()), ..Default::default() }),
+        ..Default::default()
+    };
+    for (id, feed) in &ctx.cfg().feeds {
+        let title = ctx.db.feed_summary(id).await.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(())
+}
+
+pub(crate) async fn list(ctx: &Ctx) -> Result<()> {
+    let cfg = ctx.cfg();
+    if cfg.feeds.is_empty() {
+        println!("No feeds configured. Add one with `ipx add <url>`, or in the web UI.");
+        return Ok(());
+    }
+    for (id, feed) in &cfg.feeds {
+        let s = ctx.db.feed_summary(id).await?;
+        // The id on a line of its own, labelled: first on the title's line, antirez.com's id
+        // "feed" read as a heading and 'ipx fetch antirez' was tried instead (issue #82).
+        println!("{}", s.title.as_deref().filter(|t| !t.is_empty()).unwrap_or(id));
+        println!("    id           {id}");
+        println!("    url          {}", feed.url);
+        println!("    last checked {}", cli::ago(s.last_checked));
+        println!("    entries      {} ({} downloaded)", s.entries, s.downloaded);
+        if let Some(err) = &s.last_error {
+            println!("    last error   {err}");
+        }
+    }
+    Ok(())
+}
+
+/// Removes feeds nobody subscribes to that are dead (failing for 30 days) or quiet (nothing new
+/// in a year), from the catalogue and the database, so the Directory lists feeds worth taking.
+/// A feed listed in it with no subscribers stays for as long as it works and publishes.
+pub(crate) async fn clean_directory(ctx: &Ctx) -> Result<()> {
+    const DEAD: i64 = 30 * 86_400;
+    const QUIET: i64 = 365 * 86_400;
+    let now = db::now();
+    // Once a day, for feeds dead a month or quiet a year: before every scan, its query took 1 to
+    // 12 s every few minutes, and the scans and downloads waited behind it (#193).
+    use std::sync::atomic::Ordering::Relaxed;
+    if now - ctx.directory_cleaned.load(Relaxed) < 86_400 {
+        return Ok(());
+    }
+    ctx.directory_cleaned.store(now, Relaxed);
+    let stale = ctx.db.stale_unsubscribed(now - DEAD, now - QUIET).await?;
+    if stale.is_empty() {
+        return Ok(());
+    }
+    let mut cfg = ctx.edit_cfg().await;
+    // Every one removed, so not `any`, which would stop at the first.
+    let mut removed = std::collections::HashMap::new();
+    for id in &stale {
+        if let Some(f) = cfg.feeds.remove(id) {
+            removed.insert(id.clone(), f);
+        }
+    }
+    if !removed.is_empty() {
+        ctx.store_cfg(cfg).await?;
+    }
+    for id in &stale {
+        retire_group(ctx, id, removed.get(id)).await?;
+        ctx.db.forget_feed(id).await?;
+        tracing::info!(
+            feed = id,
+            "removed a feed nobody subscribes to that is dead or has published nothing in a year"
+        );
+    }
+    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 async 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();
+
+    // A group's title, once a group: asked for each feed in it, three queries a feed, it was about
+    // 96 for The Glass Cannon Network's 32 shows on every feed list, live update and scan (#159).
+    let mut group_titles = std::collections::HashMap::<String, String>::new();
+    for m in ctx.db.managed_feeds().await? {
+        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);
+        if parent.is_none() {
+            // The OPML or Patreon feed this was derived from is no longer in config --
+            // removing it should have retired these rows too (see `retire_group`), but
+            // skip them here regardless so a row that slips through is never scanned.
+            continue;
+        }
+        let base = match parent.and_then(|p| p.folder.clone()) {
+            Some(folder) => folder,
+            None => {
+                if !group_titles.contains_key(&m.group_id) {
+                    let title = ctx.db.feed_summary(&m.group_id).await.ok().and_then(|s| s.title);
+                    group_titles.insert(m.group_id.clone(), title.unwrap_or_else(|| m.group_id.clone()));
+                }
+                group_titles[&m.group_id].clone()
+            }
+        };
+        let title = m.title.clone().unwrap_or_else(|| m.id.clone());
+        let (username, password, password_env) = parent.map(|p| p.credentials_for(&m.url)).unwrap_or_default();
+        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,
+                password,
+                password_env,
+                category: None,
+                listed: false,
+                private: false,
+            },
+            managed: true,
+        });
+    }
+    out.sort_by(|a, b| a.id.cmp(&b.id));
+    Ok(out)
+}
+
+/// Takes a feed off one person's list, and the feeds in it if it is a group. Everyone else's
+/// lists are left alone.
+pub async fn leave(ctx: &Ctx, user_id: i64, id: &str) -> Result<()> {
+    use tracing::Instrument;
+    ctx.db.unsubscribe(user_id, id).instrument(db::step("unsubscribe")).await?;
+    // What they kept stays on their list, a feed of its own once nobody reads the list (#280).
+    let kept = ctx.db.kept_feeds(user_id).instrument(db::step("kept")).await?;
+    let all = subscriptions(ctx).instrument(db::step("subscriptions")).await?;
+    let unkept = all.iter().filter(|s| s.cfg.group.as_deref() == Some(id) && !kept.contains(&s.id));
+    for child in unkept {
+        ctx.db.unsubscribe(user_id, &child.id).instrument(db::step("unsubscribe")).await?;
+    }
+    if ctx.db.subscriber_counts().instrument(db::step("subscriber_counts")).await?.contains_key(id) {
+        return Ok(());
+    }
+
+    // Nobody is left: the feed stops being scanned. Its files and history stay, so if
+    // someone subscribes again they do not pull the back catalogue a second time. A feed an
+    // admin listed stays in the Directory, for the next person.
+    let mut cfg = ctx.edit_cfg().await;
+    if cfg.feeds.get(id).is_some_and(|f| f.listed) {
+        return Ok(());
+    }
+    let Some(parent) = cfg.feeds.remove(id) else {
+        // A derived feed: forget it here, though the OPML will list it again on the next
+        // read unless you unsubscribe from the OPML itself.
+        return ctx.db.drop_managed(id).await;
+    };
+    ctx.store_cfg(cfg).await?;
+    retire_group(ctx, id, Some(&parent)).await
+}
+
+/// Removes an account, leaving each of its feeds first as unsubscribing does. The foreign keys
+/// took the subscriptions with the account, but not the feed: one only they read stayed in the
+/// catalogue with nobody subscribed, downloading every new episode for no one (#157).
+pub async fn remove_user(ctx: &Ctx, user_id: i64) -> Result<()> {
+    let catalogue = subscriptions(ctx).await?;
+    let mine: std::collections::HashSet<String> =
+        ctx.db.subscriptions_for(user_id).await?.into_iter().map(|s| s.feed_id).collect();
+    for id in &mine {
+        // A feed in a group of theirs goes with the group.
+        let in_theirs = catalogue.iter().any(|s| &s.id == id && s.cfg.group.as_ref().is_some_and(|g| mine.contains(g)));
+        if !in_theirs {
+            leave(ctx, user_id, id).await?;
+        }
+    }
+    ctx.db.delete_user(user_id).await
+}
+
+/// Retires every feed derived from `parent_id`, now that nothing subscribes to the OPML or
+/// Patreon feed that listed them: the same rule `sync_group` applies to one the list drops --
+/// removed if nothing was downloaded, orphaned and kept otherwise. Called once the parent
+/// itself is removed, since `subscriptions()` would otherwise keep scanning them under a
+/// fallback policy meant for a feed with no parent at all. A feed promoted to config is not
+/// derived any more, so it is only unmanaged. `parent` is the list's settings as they were before
+/// it was removed, None when that is no longer known.
+pub async fn retire_group(ctx: &Ctx, parent_id: &str, parent: Option<&config::Feed>) -> Result<()> {
+    let cfg = ctx.cfg();
+    let managed = ctx.db.managed_feeds().await?;
+    let derived = managed.into_iter().filter(|m| m.group_id == parent_id);
+    for m in derived {
+        if cfg.feeds.contains_key(&m.id) {
+            // Scanned from its config entry and still read. Dropped as derived, its stored
+            // entries would go with it: davewiner's 11 were promoted without being unmanaged.
+            ctx.db.unmanage(&m.id).await?;
+        } else if promote_kept(ctx, &m, parent).await? {
+            tracing::info!(feed = %m.id, "its list is gone but someone kept it; a feed of its own now");
+        } else if ctx.db.downloaded_count(&m.id).await.unwrap_or(1) > 0 {
+            ctx.db.set_orphaned(&m.id, true).await?;
+        } else {
+            ctx.db.drop_managed(&m.id).await?;
+        }
+    }
+    Ok(())
+}
+
+/// Makes a feed from an OPML or Patreon list a feed of its own, for those who kept it (#280): a
+/// catalogue entry in no group, which the list no longer adds or takes away. Whoever had it only
+/// through the list, without keeping it, loses it, as they would have when the list dropped it.
+/// False, changing nothing, when nobody kept it. `base` is the list's settings: most callers have
+/// removed the list from config by now, so they pass them.
+pub(crate) async fn promote_kept(ctx: &Ctx, m: &db::Managed, base: Option<&config::Feed>) -> Result<bool> {
+    let subs = ctx.db.subscribers_of(&m.id).await?;
+    if !subs.iter().any(|(_, kept)| *kept) {
+        return Ok(false);
+    }
+    let mut cfg = ctx.edit_cfg().await;
+    // The list's own settings carry over; its folder and group do not.
+    let (username, password, password_env) = base.map(|p| p.credentials_for(&m.url)).unwrap_or_default();
+    cfg.feeds.entry(m.id.clone()).or_insert_with(|| config::Feed {
+        url: m.url.clone(),
+        folder: None,
+        group: None,
+        media_types: base.and_then(|p| p.media_types.clone()),
+        schedule: base.and_then(|p| p.schedule.clone()),
+        keywords: base.map(|p| p.keywords.clone()).unwrap_or_default(),
+        allow_explicit: base.is_some_and(|p| p.allow_explicit),
+        auto_download: base.is_none_or(|p| p.auto_download),
+        max_new_per_check: base.and_then(|p| p.max_new_per_check),
+        username,
+        password,
+        password_env,
+        category: None,
+        listed: false,
+        // Looked up after the list was removed, private was always false, and a private OPML's
+        // kept show went to the directory (#332). A list no longer known may have been private.
+        private: base.is_none_or(|p| p.private),
+    });
+    ctx.store_cfg(cfg).await?;
+    ctx.db.unmanage(&m.id).await?;
+    ctx.db.set_orphaned(&m.id, false).await?;
+    for (user, kept) in subs {
+        if !kept {
+            ctx.db.unsubscribe(user, &m.id).await?;
+        }
+    }
+    Ok(true)
+}
+
+/// Retires every group whose parent is gone from config, and returns how many derived rows that
+/// dropped or unmanaged. An OPML removed before `retire_group` existed left its feeds behind:
+/// davewiner's 922 were skipped by every scan and never cleared, and their stale errors were
+/// most of the ones stored.
+pub(crate) async fn retire_stranded(ctx: &Ctx) -> Result<usize> {
+    let cfg = ctx.cfg();
+    let before = ctx.db.managed_feeds().await?;
+    let stranded: std::collections::BTreeSet<&str> =
+        before.iter().map(|m| m.group_id.as_str()).filter(|g| !cfg.feeds.contains_key(*g)).collect();
+    for group in stranded {
+        retire_group(ctx, group, None).await?;
+    }
+    Ok(before.len() - ctx.db.managed_feeds().await?.len())
+}
+
+/// A feed that answered from a new address after permanent redirects moves there in the
+/// catalogue, so it is read from there and no longer redirected every time. One an OPML lists
+/// is left alone, since the OPML would only put the old address back; so is a move onto an
+/// address another feed already has.
+pub(crate) async fn follow_move(ctx: &Ctx, id: &str, to: &str) -> Result<()> {
+    // Said in the event below, not logged here: the event is the log line.
+    if subscriptions(ctx).await?.iter().any(|s| s.id != id && feed::same_feed(&s.cfg.url, to)) {
+        tracing::info!(feed = id, to, "the feed moved to an address another feed already has; leaving it");
+        return Ok(());
+    }
+    let mut cfg = ctx.edit_cfg().await;
+    let Some(f) = cfg.feeds.get_mut(id) else { return Ok(()) };
+    let from = std::mem::replace(&mut f.url, to.to_owned());
+    ctx.store_cfg(cfg).await?;
+    ctx.out.emit(Event::FeedMoved { feed: id.to_owned(), from, to: to.to_owned() });
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::test_util::{feed, test_ctx};
+
+    #[tokio::test]
+    async fn the_first_start_moves_the_configuration_in_and_trims_the_file() {
+        let dir = std::env::temp_dir().join(format!("ipx-assemble-{}", std::process::id()));
+        std::fs::create_dir_all(&dir).unwrap();
+        let path = dir.join("config.toml");
+        std::fs::write(
+            &path,
+            "[general]\nschedule = \"every 2h\"\n[web]\ntoken = \"t\"\n[feeds.show]\nurl = \"http://x/show.xml\"\n",
+        )
+        .unwrap();
+        let db = db::Db::memory().await.unwrap();
+
+        let first = assemble_config(&db, config::Config::load(&path).unwrap(), &path).await.unwrap();
+        assert_eq!(first.feeds.keys().collect::<Vec<_>>(), ["show"]);
+        assert_eq!(first.general.schedule, "every 2h");
+        assert!(dir.join("config.toml.pre-database").exists(), "the original is kept");
+        assert!(!config::Config::file_holds_stored(&path), "and the file no longer lists them");
+
+        // The next start reads them from the database, the trimmed file notwithstanding.
+        let next = assemble_config(&db, config::Config::load(&path).unwrap(), &path).await.unwrap();
+        assert_eq!(next.feeds.keys().collect::<Vec<_>>(), ["show"]);
+        assert_eq!(next.general.schedule, "every 2h");
+        assert_eq!(next.web.token, "t");
+        std::fs::remove_dir_all(&dir).unwrap();
+    }
+
+    #[tokio::test]
+    async fn dead_feeds_leave_the_directory_once_a_day_not_before_every_scan() {
+        let ctx = test_ctx(config::Config::default()).await;
+        let dead = |id: &str| {
+            format!("INSERT INTO feeds (id, url, error_since, managed) VALUES ('{id}', 'u-{id}', 100, false)")
+        };
+        let listed = async || ctx.db.feed_urls().await.unwrap().into_iter().map(|(id, _)| id).collect::<Vec<_>>();
+        ctx.db.sql(&dead("a")).await.unwrap();
+        clean_directory(&ctx).await.unwrap();
+        assert!(listed().await.is_empty(), "the first pass since starting looks");
+        ctx.db.sql(&dead("b")).await.unwrap();
+        clean_directory(&ctx).await.unwrap();
+        assert_eq!(listed().await, ["b"], "not again the same day (#193)");
+        ctx.directory_cleaned.store(db::now() - 86_401, std::sync::atomic::Ordering::Relaxed);
+        clean_directory(&ctx).await.unwrap();
+        assert!(listed().await.is_empty(), "a day on, it looks again");
+    }
+
+    #[tokio::test]
+    async fn a_removed_account_takes_the_feeds_only_it_read_with_it() {
+        let ctx = test_ctx(config::Config::default()).await;
+        let mut cfg = ctx.edit_cfg().await;
+        for id in ["only-theirs", "shared", "listed"] {
+            cfg.feeds.insert(id.into(), config::Feed { listed: id == "listed", ..feed() });
+        }
+        ctx.store_cfg(cfg).await.unwrap();
+        let gone = ctx.db.create_user("gone", None, false).await.unwrap();
+        let stays = ctx.db.create_user("stays", None, true).await.unwrap();
+        for id in ["only-theirs", "shared", "listed"] {
+            ctx.db.subscribe(gone, id).await.unwrap();
+        }
+        ctx.db.subscribe(stays, "shared").await.unwrap();
+
+        remove_user(&ctx, gone).await.unwrap();
+        assert!(ctx.db.user_by_id(gone).await.unwrap().is_none());
+        // Nobody reads it now: out of the catalogue, not downloading for no one (#157).
+        assert_eq!(ctx.cfg().feeds.keys().collect::<Vec<_>>(), ["listed", "shared"]);
+        assert!(ctx.db.subscription(stays, "shared").await.unwrap().is_some());
+    }
+
+    #[tokio::test]
+    async fn a_derived_feed_is_not_scanned_once_its_opml_leaves_config() {
+        // davewiner: the OPML subscription left config.toml, but its 922 derived rows
+        // stayed in the database and kept being scanned under the no-parent fallback.
+        let ctx = test_ctx(config::Config::default()).await;
+        ctx.db.upsert_managed("child", "http://x/child.xml", "Child", "gone-opml").await.unwrap();
+        assert!(
+            subscriptions(&ctx).await.unwrap().iter().all(|s| s.id != "child"),
+            "a derived feed whose parent is gone from config must not be scanned"
+        );
+    }
+
+    #[tokio::test]
+    async fn retiring_a_group_drops_what_was_never_downloaded_and_orphans_the_rest() {
+        let ctx = test_ctx(config::Config::default()).await;
+        ctx.db.upsert_managed("empty", "http://x/empty.xml", "Empty", "parent").await.unwrap();
+        ctx.db.upsert_managed("has-file", "http://x/has-file.xml", "Has File", "parent").await.unwrap();
+        let enc = feed::Enclosure { url: "http://x/ep.mp3".into(), mime: None, length: None };
+        ctx.db.record_enclosure("has-file", "g1", &enc).await.unwrap();
+        ctx.db.mark_downloaded(&enc.url, std::path::Path::new("/downloads/ep.mp3"), 1).await.unwrap();
+
+        retire_group(&ctx, "parent", None).await.unwrap();
+
+        let managed = ctx.db.managed_feeds().await.unwrap();
+        assert!(!managed.iter().any(|m| m.id == "empty"), "nothing downloaded, so it is forgotten");
+        assert!(managed.iter().any(|m| m.id == "has-file"), "has a file on disk, so it is kept");
+        assert!(ctx.db.feed_summary("has-file").await.unwrap().orphaned, "and flagged as orphaned");
+    }
+
+    #[tokio::test]
+    async fn a_stranded_group_is_retired_but_a_promoted_feed_keeps_its_entries() {
+        // davewiner: the OPML left config before retire_group existed, and 11 of its feeds
+        // promoted to config since still said managed = 1.
+        let mut cfg = config::Config::default();
+        cfg.feeds.insert("promoted".into(), feed());
+        cfg.feeds.insert("live-opml".into(), feed());
+        let ctx = test_ctx(cfg).await;
+        ctx.db.upsert_managed("promoted", "http://x/p.xml", "Promoted", "gone-opml").await.unwrap();
+        ctx.db.upsert_managed("empty", "http://x/e.xml", "Empty", "gone-opml").await.unwrap();
+        ctx.db.upsert_managed("listed", "http://x/l.xml", "Listed", "live-opml").await.unwrap();
+        ctx.db.record_entry("promoted", &feed::Entry { guid: "g1".into(), ..Default::default() }).await.unwrap();
+
+        assert_eq!(retire_stranded(&ctx).await.unwrap(), 2, "empty dropped, promoted unmanaged");
+
+        let managed: Vec<String> = ctx.db.managed_feeds().await.unwrap().into_iter().map(|m| m.id).collect();
+        assert_eq!(managed, ["listed"], "a group still in config is left alone");
+        assert_eq!(ctx.db.feed_summary("promoted").await.unwrap().entries, 1, "its entries survive");
+    }
+}

crates/ipx/src/cli.rs

@@ -0,0 +1,234 @@
+//! The command line: its commands, and what each does when no daemon takes it.
+
+use crate::ctx::Ctx;
+use crate::ipc::Command as Cmd;
+use crate::{catalogue, daemon, db, metrics, retention, scan};
+use anyhow::Result;
+use chrono::DateTime;
+use clap::{Parser, Subcommand};
+use std::path::PathBuf;
+use std::sync::Arc;
+
+#[derive(Parser)]
+#[command(name = "ipx", version, about = "A headless podcatcher")]
+pub(crate) struct Cli {
+    /// Config file (default: $XDG_CONFIG_HOME/ipx/config.toml)
+    #[arg(long, global = true)]
+    pub(crate) config: Option<PathBuf>,
+
+    /// Do the work here even if a daemon is running
+    #[arg(long, global = true)]
+    pub(crate) local: bool,
+
+    #[command(subcommand)]
+    pub(crate) command: Command,
+}
+
+#[derive(Subcommand)]
+pub(crate) enum Command {
+    /// Show configured feeds and their state
+    List,
+    /// Scan feeds for new entries
+    Fetch {
+        /// Only this feed id
+        feed: Option<String>,
+        /// Read the feed in full now, even when it is not due and has not changed
+        #[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>,
+        /// The Directory's category, for a feed that names none of its own (News, Technology)
+        #[arg(long)]
+        category: Option<String>,
+        /// Put it in the Directory for anyone to subscribe to, and keep it there when nobody does
+        #[arg(long)]
+        list: bool,
+    },
+    /// 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)]
+pub(crate) 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 },
+    /// Rename an account, keeping its feeds, read state and admin rights. This is how an
+    /// account made before the proxy takes the name the proxy signs it in as
+    Rename { name: String, new_name: String },
+}
+
+/// Accounts. Passwords come in on stdin so they never reach a shell history or a `ps`
+/// listing.
+pub(crate) async 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).await?.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().await?.is_empty();
+            ctx.db.create_user(&name, hash.as_deref(), admin || first).await?;
+            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().await?;
+            if users.is_empty() {
+                println!("no accounts yet: ipx user add <name>");
+            }
+            for u in users {
+                let added = u
+                    .created
+                    .and_then(|t| DateTime::from_timestamp(t, 0))
+                    .map_or("?".into(), |d| d.format("%Y-%m-%d").to_string());
+                let seen = u.last_login.map_or("never signed in".into(), |t| format!("signed in {}", ago(Some(t))));
+                println!(
+                    "{:<20} {:<6} {:<11} added {added}  {seen}",
+                    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).await?.ok_or_else(|| anyhow::anyhow!("no such account: {name}"))?;
+            ctx.db.set_password(user.id, &crate::auth::hash_password(&read_password()?)?).await?;
+            println!("password changed for {name}, and every browser signed in as {name} signed out");
+            let tokens = ctx.db.api_tokens(user.id).await?.len();
+            if tokens > 0 {
+                println!(
+                    "{tokens} API token(s) of {name}'s still work: remove any that may be someone else's in Settings"
+                );
+            }
+            Ok(())
+        }
+        UserCmd::Rename { name, new_name } => {
+            let name = name.trim().to_ascii_lowercase();
+            // The same rules as a name the proxy vouches for, or the proxy would never find it.
+            let new_name = crate::auth::name_from_header(&new_name)
+                .ok_or_else(|| anyhow::anyhow!("not a usable name: no commas, semicolons or line breaks"))?;
+            let user = ctx.db.user_by_name(&name).await?.ok_or_else(|| anyhow::anyhow!("no such account: {name}"))?;
+            if ctx.db.user_by_name(&new_name).await?.is_some() {
+                anyhow::bail!("{new_name} already exists");
+            }
+            ctx.db.rename_user(user.id, &new_name).await?;
+            println!("renamed {name} to {new_name}");
+            Ok(())
+        }
+        UserCmd::Rm { name } => {
+            let name = name.trim().to_ascii_lowercase();
+            let user = ctx.db.user_by_name(&name).await?.ok_or_else(|| anyhow::anyhow!("no such account: {name}"))?;
+            catalogue::remove_user(ctx, user.id).await?;
+            println!("removed {name}");
+            Ok(())
+        }
+    }
+}
+
+pub(crate) async fn run(ctx: &Arc<Ctx>, cmd: Cmd) -> Result<()> {
+    // What the worker is on, for /metrics: the healthcheck answers while it is stuck (#195).
+    let _job = metrics::STATS.job(match &cmd {
+        Cmd::Fetch { .. } => "scan",
+        Cmd::Reap { .. } => "reap",
+        Cmd::Download { .. } => "download",
+        Cmd::Status => "status",
+        Cmd::Reload => "reload",
+    });
+    match cmd {
+        Cmd::Fetch { feed, force, feeds } => {
+            // Make room before pulling more down, as the original did per download.
+            retention::reap(ctx, /* dry_run */ false, /* standalone */ false).await?;
+            scan::fetch(ctx, feed.as_deref(), force, &feeds).await?;
+            metrics::STATS.last_scan_done.store(db::now(), std::sync::atomic::Ordering::Relaxed);
+            Ok(())
+        }
+        Cmd::Reap { dry_run } => retention::reap(ctx, dry_run, /* standalone */ true).await,
+        Cmd::Download { enclosure } => scan::download_one(ctx, enclosure).await,
+        Cmd::Status => {
+            ctx.out.emit(daemon::status(ctx).await);
+            Ok(())
+        }
+        Cmd::Reload => daemon::reload(ctx).await,
+    }
+}
+
+pub(crate) fn ago(t: Option<i64>) -> String {
+    let Some(t) = t else { return "never".into() };
+    format!("{} ago", duration((db::now() - t).max(0) as u64))
+}
+
+pub(crate) 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),
+    }
+}

crates/ipx/src/ctx.rs

@@ -0,0 +1,261 @@
+//! What every command shares: the configuration, the database, the clients, and the state the
+//! daemon's tasks hand each other.
+
+use crate::ipc::{Emitter, Event};
+use crate::{config, db, directory, feed, scan, torrent};
+use anyhow::Result;
+use std::collections::{HashMap, HashSet};
+use std::ops::{Deref, DerefMut};
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, AtomicI64};
+use std::time::Duration;
+use tokio::sync::broadcast;
+use tokio::sync::{MutexGuard, Notify, OnceCell, Semaphore};
+
+/// 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<Arc<config::Config>>,
+    pub db: db::Db,
+    /// For anything a feed or a person names: refuses internal addresses (#320).
+    pub client: reqwest::Client,
+    /// For the services this server is set up to call, which are on its own network by design:
+    /// the directory, as `http://ipx-directory:8100`. Never for an address a feed or a person names.
+    pub local_client: reqwest::Client,
+    /// For feeds only, refusing internal addresses as `client` does: follows no redirects, so `feed::fetch` sees each hop and can tell a feed
+    /// that moved for good from one sent elsewhere for now.
+    pub feed_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: OnceCell<torrent::Torrents>,
+    /// Caps how many torrents run at once when they are detached.
+    pub torrent_slots: Arc<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,
+    /// Feeds whose host could not be looked up for the moment, and when to try them again: a
+    /// failure that says nothing about the feed, kept off it and out of the database (#143).
+    /// Forgotten on a restart, which only makes the retry sooner.
+    pub retry: std::sync::Mutex<HashMap<String, i64>>,
+    /// Held from taking a copy of the config to storing it (`edit_cfg`). Each change stores the
+    /// whole catalogue from its copy, and two at once stored over each other: a feed added while
+    /// another add spent a minute fetching was gone once that one stored (#153).
+    // ponytail: this process only. `ipx add`, `rm` and `import` write from a process of their
+    // own, and a change in flight here can still store over theirs; take the copy from the
+    // database rather than memory if that ever happens.
+    pub cfg_write: tokio::sync::Mutex<()>,
+    /// Wakes DeadAir's maker when a show is turned on, so its copies start at once instead of
+    /// at the next pass (#171).
+    pub deadair: Notify,
+    /// Wakes the Signals judge when there are new items or a Signal is made or changed (#254).
+    pub signals: Notify,
+    /// People whose TypeSafe key was refused, passed over until they save another.
+    pub signals_refused: std::sync::Mutex<HashSet<i64>>,
+    /// When the Directory was last cleared of dead feeds (`clean_directory`), 0 for not since
+    /// starting.
+    pub directory_cleaned: AtomicI64,
+    /// Sends the snapshot to the directory now, as when its settings change (#205).
+    pub directory_wake: Notify,
+    /// When the snapshot last went and what came of it, for the admin page.
+    pub directory_shared: std::sync::Mutex<Option<directory::Shared>>,
+    /// Send the snapshot even if it has not changed: after an unblock or a dropped snapshot the
+    /// directory needs it again, though nothing here differs (#206).
+    pub directory_resend: AtomicBool,
+}
+
+/// A copy of the config to change, holding off every other change until `store_cfg` keeps it.
+pub struct CfgEdit<'a> {
+    pub(crate) cfg: config::Config,
+    pub(crate) _held: MutexGuard<'a, ()>,
+}
+
+impl Deref for CfgEdit<'_> {
+    type Target = config::Config;
+    fn deref(&self) -> &config::Config {
+        &self.cfg
+    }
+}
+
+impl DerefMut for CfgEdit<'_> {
+    fn deref_mut(&mut self) -> &mut config::Config {
+        &mut self.cfg
+    }
+}
+
+/// How soon a feed whose host could not be looked up is tried again (#143).
+pub(crate) const RETRY_LOOKUP_SECS: i64 = 600;
+
+impl Ctx {
+    /// Everything a command shares, built once at start: the clients refuse internal addresses
+    /// (#320) but `local_client`, and a daemon sends its events to the socket and the page.
+    pub(crate) fn new(
+        cfg: config::Config,
+        db: db::Db,
+        config_path: PathBuf,
+        is_daemon: bool,
+        events: &broadcast::Sender<Event>,
+    ) -> Result<Self> {
+        let guard = feed::guard::Guard::new(feed::guard::Allowed::parse(&cfg.general.fetch_internal)?);
+        Ok(Ctx {
+            cfg: std::sync::RwLock::new(Arc::new(cfg)),
+            db,
+            client: reqwest::Client::builder()
+                .user_agent(concat!("ipx/", env!("CARGO_PKG_VERSION")))
+                // Connecting only, so it bounds a download's start, not a long download (#108).
+                .connect_timeout(Duration::from_secs(10))
+                // Two minutes without a byte is a stalled download, however long it is: unbounded, one
+                // held the worker, and every scan behind it, for good, while the daemon read healthy (#152).
+                .read_timeout(Duration::from_secs(120))
+                .connector_layer(guard.clone())
+                .build()?,
+            local_client: reqwest::Client::builder()
+                .user_agent(concat!("ipx/", env!("CARGO_PKG_VERSION")))
+                .connect_timeout(Duration::from_secs(10))
+                .build()?,
+            feed_client: reqwest::Client::builder()
+                .user_agent(concat!("ipx/", env!("CARGO_PKG_VERSION")))
+                .connect_timeout(Duration::from_secs(10))
+                .read_timeout(feed::FEED_SILENCE)
+                .redirect(reqwest::redirect::Policy::none())
+                .connector_layer(guard)
+                .build()?,
+            out: if is_daemon {
+                Emitter::socket(events.clone(), /* print */ false)
+            } else {
+                Emitter::terminal()
+            },
+            torrents: OnceCell::new(),
+            torrent_slots: Arc::new(Semaphore::new(2)),
+            config_path,
+            detach_torrents: is_daemon,
+            retry: Default::default(),
+            cfg_write: Default::default(),
+            directory_cleaned: Default::default(),
+            deadair: Default::default(),
+            signals: Default::default(),
+            signals_refused: Default::default(),
+            directory_wake: Default::default(),
+            directory_shared: Default::default(),
+            directory_resend: Default::default(),
+        })
+    }
+
+    /// When a feed is next due, taking a lookup retry into account: not before it.
+    pub(crate) fn due(
+        &self,
+        cfg: &config::Config,
+        feed: &config::Feed,
+        id: &str,
+        state: &db::HttpState,
+        subscribed: bool,
+    ) -> Option<i64> {
+        let at = scan::due_at(cfg, feed, state, subscribed);
+        match self.retry.lock().expect("a panic elsewhere poisoned this lock").get(id) {
+            Some(&retry) => Some(at.unwrap_or(0).max(retry)),
+            None => at,
+        }
+    }
+
+    /// A scan's failure, recorded on the feed, unless it was only a moment's failure to look its
+    /// host up: on 2026-10-05 a DNS outage on Tower failed 204 feeds in a minute with
+    /// "Temporary failure in name resolution", and each, read once a day and backed off as a
+    /// failing feed, sat a day failing, unread and labelled gone (#143). Tried again in minutes
+    /// instead, with nothing on the feed to say it failed.
+    pub(crate) async fn feed_failed(&self, id: &str, url: &str, msg: &str) -> Result<()> {
+        if feed::is_temporary_lookup_failure(msg) {
+            self.retry
+                .lock()
+                .expect("a panic elsewhere poisoned this lock")
+                .insert(id.to_owned(), db::now() + RETRY_LOOKUP_SECS);
+            return Ok(());
+        }
+        self.db.set_feed_error(id, url, msg).await
+    }
+
+    pub fn cfg(&self) -> Arc<config::Config> {
+        self.cfg.read().expect("a panic elsewhere poisoned this lock").clone()
+    }
+
+    /// The config as it stands, to change and then keep with `store_cfg`; any other change waits
+    /// until then.
+    pub async fn edit_cfg(&self) -> CfgEdit<'_> {
+        // A step of its own: waiting here is waiting on another change to the catalogue.
+        let held = tracing::Instrument::instrument(self.cfg_write.lock(), db::step("cfg_lock")).await;
+        CfgEdit { cfg: (*self.cfg()).clone(), _held: held }
+    }
+
+    /// Keeps a changed catalogue or server settings: in the database, and for everything running
+    /// here from now on. What config.toml's save and reload did, before the database held them.
+    pub async fn store_cfg(&self, edit: CfgEdit<'_>) -> Result<()> {
+        tracing::Instrument::instrument(self.db.store_config(&edit.cfg), db::step("store_config")).await?;
+        self.set_cfg(edit.cfg);
+        Ok(())
+    }
+
+    pub(crate) fn set_cfg(&self, cfg: config::Config) {
+        *self.cfg.write().expect("a panic elsewhere poisoned this lock") = Arc::new(cfg);
+        tracing::info!("config reloaded");
+        // A feed added, removed, made private or given a category, or the directory's own
+        // settings changed: the directory hears of it now, not at the next half hour (#205).
+        self.directory_wake.notify_one();
+    }
+
+    pub(crate) async fn torrents(&self) -> Result<&torrent::Torrents> {
+        let cfg = self.cfg();
+        self.torrents.get_or_try_init(|| torrent::Torrents::new(&cfg)).await
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::test_util::{feed, test_ctx};
+
+    #[tokio::test]
+    async fn two_changes_to_the_catalogue_at_once_both_keep() {
+        let ctx = Arc::new(test_ctx(config::Config::default()).await);
+        // A slow change, as adding a feed is while it fetches, and a quick one made meanwhile.
+        let slow = {
+            let ctx = ctx.clone();
+            tokio::spawn(async move {
+                let mut cfg = ctx.edit_cfg().await;
+                tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+                cfg.feeds.insert("a".into(), feed());
+                ctx.store_cfg(cfg).await.unwrap();
+            })
+        };
+        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+        let mut cfg = ctx.edit_cfg().await;
+        cfg.feeds.insert("b".into(), feed());
+        ctx.store_cfg(cfg).await.unwrap();
+        slow.await.unwrap();
+        assert_eq!(ctx.cfg().feeds.keys().collect::<Vec<_>>(), ["a", "b"], "neither stored over the other (#153)");
+        let (_, stored) = ctx.db.stored_config().await.unwrap().unwrap();
+        assert_eq!(stored.keys().collect::<Vec<_>>(), ["a", "b"]);
+    }
+
+    #[tokio::test]
+    async fn a_moments_lookup_failure_is_tried_again_soon_and_not_held_against_the_feed() {
+        let ctx = test_ctx(config::Config::default()).await;
+        let blip = "connecting: dns error: failed to lookup address information: Temporary failure in name resolution";
+        ctx.feed_failed("npr", "https://feeds.npr.org/x", blip).await.unwrap();
+        // Nothing on the feed: no error, no run of failures to back off from or call it gone by.
+        let s = ctx.db.feed_summary("npr").await.unwrap();
+        assert!(s.last_error.is_none() && s.error_since.is_none(), "{:?}", s.last_error);
+        // Due again in minutes, though a feed nobody subscribes to is otherwise read once a day.
+        let state = db::HttpState { last_checked: Some(db::now() - 2 * 86_400), ..Default::default() };
+        let due = ctx.due(&config::Config::default(), &feed(), "npr", &state, false).unwrap();
+        assert!((due - db::now() - RETRY_LOOKUP_SECS).abs() <= 2, "{due}");
+        // A real failure is the feed's.
+        ctx.feed_failed("gone", "https://x/feed", "HTTP 404 Not Found").await.unwrap();
+        assert_eq!(ctx.db.feed_summary("gone").await.unwrap().last_error.as_deref(), Some("HTTP 404 Not Found"));
+    }
+}

crates/ipx/src/daemon.rs

@@ -0,0 +1,356 @@
+//! The daemon: its worker, its web server, and how it stops.
+
+use crate::ctx::Ctx;
+use crate::ipc::{Command as Cmd, Event};
+use crate::{access, auth, catalogue, cli, config, db, deadair, directory, ipc, metrics, signals, web};
+use anyhow::Result;
+use std::future::Future;
+use std::path::Path;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+use tokio::runtime::Handle;
+use tokio::sync::{broadcast, mpsc, watch};
+use tokio::task::JoinHandle;
+
+/// The catalogue and server settings as the database now has them, in place of the copy in
+/// memory, for a daemon told another ipx changed them (#120); config.toml, read again, for where
+/// things are and who gets in.
+pub(crate) async fn reload(ctx: &Ctx) -> Result<()> {
+    let _held = ctx.cfg_write.lock().await;
+    let cfg = catalogue::assemble_config(&ctx.db, config::Config::load(&ctx.config_path)?, &ctx.config_path).await?;
+    ctx.set_cfg(cfg);
+    Ok(())
+}
+
+/// The counts `ipx status` prints, and /api/status serves. A running daemon's socket answers with
+/// this directly rather than through the job queue.
+pub(crate) async fn status(ctx: &Ctx) -> Event {
+    match tracing::Instrument::instrument(ctx.db.counts(), db::step("counts")).await {
+        Ok((pending, downloaded)) => {
+            let feeds = tracing::Instrument::instrument(catalogue::subscriptions(ctx), db::step("subscriptions"))
+                .await
+                .map(|s| s.len())
+                .unwrap_or(0);
+            // Kept for /metrics, which asks the database nothing (#195); the daemon counts every 30 s.
+            use std::sync::atomic::Ordering::Relaxed;
+            let st = &metrics::STATS;
+            st.feeds.store(feeds as i64, Relaxed);
+            st.pending.store(pending, Relaxed);
+            st.downloaded.store(downloaded, Relaxed);
+            Event::Status { feeds, pending, downloaded }
+        }
+        Err(e) => Event::Error { msg: format!("{e:#}") },
+    }
+}
+
+pub(crate) 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().await?.is_empty() {
+        let dir = ctx.config_path.parent().unwrap_or(Path::new("."));
+        let (password, told) = auth::first_password(dir, std::env::var("IPX_ADMIN_PASSWORD").ok())?;
+        ctx.db.create_user("admin", Some(&crate::auth::hash_password(&password)?), /* admin */ true).await?;
+        tracing::warn!(
+            "no accounts yet: created 'admin' {told}. Change it with `echo -n <password> | ipx user passwd admin`"
+        );
+    }
+
+    // Unread lists start from here, and a reap may be a day off (#347). Slower without, never wrong.
+    if let Err(e) = ctx.db.advance_unread_from().await {
+        tracing::warn!(error = %format!("{e:#}"), "could not work out where unread items start");
+    }
+
+    if let Some(admin) = ctx.db.users().await?.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).await {
+            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().await {
+        Ok(n) if n > 0 => tracing::info!(count = n, "requeued downloads interrupted by a restart"),
+        Ok(_) => {}
+        Err(e) => tracing::warn!(error = %format!("{e:#}"), "could not requeue interrupted downloads"),
+    }
+
+    match catalogue::retire_stranded(&ctx).await {
+        Ok(0) => {}
+        Ok(n) => tracing::info!(feeds = n, "retired feeds whose OPML is no longer in config"),
+        Err(e) => tracing::warn!(error = %format!("{e:#}"), "could not retire feeds whose OPML is no longer in config"),
+    }
+
+    // Room for any one request's commands: a web request waits inside itself for a free place,
+    // and Download latest sends up to 100. With 64 the rest waited on downloads an hour long, and
+    // every other request that queues work waited behind them (#154).
+    let (tx_cmd, mut rx_cmd) = mpsc::channel::<Cmd>(1024);
+
+    let web = start_web(&ctx, &config_path, web_addr, &tx_cmd, &events).await?;
+    // status and reload are answered by the socket itself; everything else waits its turn in the
+    // queue.
+    let answer: ipc::StatusFn = {
+        let ctx = ctx.clone();
+        Arc::new(move |cmd| {
+            let ctx = ctx.clone();
+            Box::pin(async move {
+                if matches!(cmd, Cmd::Reload)
+                    && let Err(e) = reload(&ctx).await
+                {
+                    return Event::Error { msg: format!("reading the catalogue again: {e:#}") };
+                }
+                status(&ctx).await
+            })
+        })
+    };
+    let server = tokio::spawn(ipc::serve(socket.clone(), events.clone(), tx_cmd, answer));
+
+    // One command at a time: the queue is what keeps two scans from overlapping.
+    tracing::info!(feeds = catalogue::subscriptions(&ctx).await.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);
+    });
+    // Off the worker: the first pass over a library is hours of ffmpeg, and scans and downloads
+    // must not wait behind it.
+    tokio::spawn(deadair::run(ctx.clone(), rx_stop.clone()));
+    tokio::spawn(signals::run(ctx.clone(), rx_stop.clone()));
+    tokio::spawn(directory::run(ctx.clone(), rx_stop.clone()));
+    // The library's counts for /metrics. The Docker healthcheck asked for them every 30 s until
+    // it was removed: Docker runs it with exec, which failed on Tower whenever root was logged
+    // out (#218).
+    {
+        let (ctx, mut stop) = (ctx.clone(), rx_stop.clone());
+        tokio::spawn(async move {
+            let mut every = tokio::time::interval(Duration::from_secs(30));
+            loop {
+                tokio::select! {
+                    _ = every.tick() => { status(&ctx).await; }
+                    _ = stop.changed() => break,
+                }
+            }
+        });
+    }
+
+    // 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();
+    // The first pass at once, as the minute's tick did: what came due while it was down.
+    let mut first = true;
+    loop {
+        if *stop.borrow() {
+            break;
+        }
+        let wait = if std::mem::take(&mut first) { Duration::ZERO } else { until_next_scan(&ctx).await };
+        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, cli::run(&ctx, cmd)).await {
+                    break;
+                }
+            }
+            _ = tokio::time::sleep(wait) => {
+                // Per-feed schedule and TTL decide what actually gets polled.
+                let job = cli::run(&ctx, Cmd::Fetch { feed: None, force: false, feeds: vec![] });
+                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.
+pub(crate) async fn start_web(
+    ctx: &Arc<Ctx>,
+    config_path: &Path,
+    web_addr: Option<String>,
+    cmds: &mpsc::Sender<Cmd>,
+    events: &broadcast::Sender<Event>,
+) -> Result<Option<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();
+        // The token is config.toml's, not the database's: it decides who gets in.
+        fresh.save_bootstrap(config_path)?;
+        ctx.set_cfg(fresh.clone());
+        // The token signs in as the admin, and whatever reads this process's output (docker logs,
+        // for one) is wider than who reads config.toml. So say where it is, never what it is.
+        // Logged, not printed, so a JSON log stays one object a line (#91).
+        tracing::info!(
+            "web ui token generated and saved to {} as [web] token. Open http://{bind}/?token=<that token>",
+            config_path.display()
+        );
+    } else {
+        tracing::info!("web ui at http://{bind}/ (the sign-in token is [web] token in {})", config_path.display());
+    }
+
+    // Bound beyond localhost, as a container has to be for its port to be published. Worth a
+    // warning only when nothing but a password stands in front of it: it said "the token is all
+    // that guards it" on every start of production, behind Cloudflare Access, and was the only
+    // warning in a healthy log (#106).
+    if ctx.cfg().web.binds_publicly() {
+        if ctx.cfg().web.access().is_some() {
+            tracing::info!(
+                bind,
+                "web ui is reachable off this machine; signing in takes an account's password or the admin token, or Cloudflare Access through a trusted proxy"
+            );
+        } else {
+            tracing::warn!(
+                bind,
+                "web ui is reachable off this machine; an account's password or the admin token is all that guards it"
+            );
+        }
+    }
+
+    let access = Arc::new(access::Keys::default());
+    if let Some((team, _)) = ctx.cfg().web.access() {
+        let (access, ctx, team) = (access.clone(), ctx.clone(), team.to_owned());
+        tokio::spawn(async move { access.prefetch(&ctx.client, &team).await });
+    }
+    let playing = deadair::Playing::default();
+    if !cfg.web.metrics_bind.is_empty() {
+        let sources = metrics::Sources {
+            ctx: ctx.clone(),
+            cmds: cmds.clone(),
+            events: events.clone(),
+            playing: playing.clone(),
+            runtime: Handle::current(),
+            started: Instant::now(),
+        };
+        let at = metrics::spawn(&cfg.web.metrics_bind, sources)?;
+        tracing::info!(%at, "serving metrics for Prometheus at /metrics");
+    }
+    let state = web::WebState { ctx: ctx.clone(), cmds: cmds.clone(), events: events.clone(), access, playing };
+    Ok(Some(tokio::spawn(async move {
+        if let Err(e) = web::serve(state, &bind).await {
+            tracing::error!(error = %format!("{e:#}"), "web ui stopped");
+        }
+    })))
+}
+
+pub(crate) async fn shutdown() {
+    use tokio::signal::unix;
+    use tokio::signal::unix::SignalKind;
+    let mut term = match unix::signal(SignalKind::terminate()) {
+        Ok(s) => s,
+        Err(_) => return std::future::pending().await,
+    };
+    tokio::select! {
+        _ = tokio::signal::ctrl_c() => {}
+        _ = term.recv() => {}
+    }
+}
+
+/// How long the daemon may sleep before a feed is due: until the earliest one, at least 30 s and
+/// at most 10 minutes. It ticked every minute and ran a scan pass each time, 80% of them finding
+/// nothing due (#114). The floor keeps a feed that never gets a check time from spinning it; the
+/// ceiling picks up within ten minutes what no command announces, such as `ipx add` or a
+/// shorter schedule. A command, a refresh or a feed added on the page, wakes it at once anyway.
+pub(crate) async fn until_next_scan(ctx: &Ctx) -> std::time::Duration {
+    let next = async {
+        let cfg = ctx.cfg();
+        let subscribed = ctx.db.subscriber_counts().await?;
+        let states = ctx.db.http_states().await?;
+        anyhow::Ok(
+            catalogue::subscriptions(ctx)
+                .await?
+                .iter()
+                // A feed with no row yet has never been checked: due now.
+                .map(|s| {
+                    let never = db::HttpState::default();
+                    let state = states.get(&s.id).unwrap_or(&never);
+                    ctx.due(&cfg, &s.cfg, &s.id, state, subscribed.contains_key(&s.id)).unwrap_or(0)
+                })
+                .min(),
+        )
+    };
+    let wait = match next.await {
+        Ok(Some(at)) => (at - db::now()).max(0) as u64,
+        Ok(None) => u64::MAX,
+        Err(e) => {
+            tracing::warn!(error = %format!("{e:#}"), "could not work out when the next feed is due");
+            0
+        }
+    };
+    std::time::Duration::from_secs(wait.clamp(30, 600))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::test_util::{feed, test_ctx};
+
+    #[tokio::test]
+    async fn a_reload_takes_the_catalogue_another_ipx_wrote() {
+        let with = |id: &str| {
+            let mut cfg = config::Config::default();
+            cfg.feeds.insert(id.into(), config::Feed { url: format!("http://x/{id}.xml"), ..feed() });
+            cfg
+        };
+        let held = with("kept-in-memory");
+        let ctx = test_ctx(held.clone()).await;
+        ctx.db.store_config(&held).await.unwrap();
+        // `ipx add` and `ipx rm`, from a process of their own (#120).
+        ctx.db.store_config(&with("added-by-the-cli")).await.unwrap();
+        reload(&ctx).await.unwrap();
+        let urls: Vec<String> = ctx.cfg().feeds.values().map(|f| f.url.clone()).collect();
+        assert_eq!(urls, vec!["http://x/added-by-the-cli.xml".to_string()]);
+    }
+}

crates/ipx/src/db.rs

@@ -497,6 +497,7 @@ impl Db {
     }
 
     /// What a feed says it is, for its page in the Directory.
+    #[cfg(test)]
     pub async fn feed_description(&self, feed_id: &str) -> Result<Option<String>> {
         Ok(feeds::Entity::find_by_id(feed_id.to_owned()).one(&self.orm).await?.and_then(|f| f.description))
     }
@@ -1140,6 +1141,7 @@ impl Db {
 }
 
 impl Db {
+    #[cfg(test)]
     pub async fn unread_count(&self, user_id: i64, feed_id: &str) -> Result<i64> {
         let r = self
             .rows(
@@ -1400,6 +1402,7 @@ impl Db {
     /// person subscribes to when `feed_id` is None (All Subscriptions). `search` matches title
     /// and description, case-insensitively.
     #[allow(clippy::too_many_arguments)]
+    #[cfg(test)]
     pub async fn entries_in(
         &self,
         user_id: i64,
@@ -1557,6 +1560,7 @@ impl Db {
     }
 
     /// How many entries `entries_in` would page through, so the UI knows whether there is more.
+    #[cfg(test)]
     pub async fn count_in(
         &self,
         user_id: i64,
@@ -1795,16 +1799,6 @@ impl Db {
             .collect()
     }
 
-    /// Feeds with any audio or video enclosure: the Directory's Podcasts, with the rest Blogs.
-    /// Reaped files keep their rows, so a show whose files have all been purged still counts.
-    pub async fn media_feeds(&self) -> Result<std::collections::HashSet<String>> {
-        self.rows("SELECT DISTINCT feed_id FROM enclosures WHERE mime LIKE 'audio/%' OR mime LIKE 'video/%'", vec![])
-            .await?
-            .iter()
-            .map(|r| Ok(r.try_get("", "feed_id")?))
-            .collect()
-    }
-
     /// Who else would miss this file: subscribers other than `user_id` who have starred
     /// the item or have not read it yet. Deleting is deleting their copy too.
     pub async fn others_wanting(&self, enclosure_id: i64, user_id: i64) -> Result<(i64, i64)> {

crates/ipx/src/deadair.rs

@@ -25,7 +25,7 @@ use crate::db;
 use anyhow::{Context, Result, bail};
 use serde::{Deserialize, Serialize};
 
-use crate::Ctx;
+use crate::ctx::Ctx;
 use axum::body::Bytes;
 use std::collections::{HashSet, VecDeque};
 use std::io::SeekFrom;
@@ -805,6 +805,133 @@ mod tests {
         }
         std::fs::remove_dir_all(&dir).unwrap();
     }
+
+    use crate::config;
+    use crate::test_util::test_ctx;
+
+    #[tokio::test]
+    async fn deadair_hands_out_each_persons_clock_and_saves_in_the_originals() {
+        let ctx = test_ctx(config::Config::default()).await;
+        let db = &ctx.db;
+        let u = db.create_user("dee", None, false).await.unwrap();
+        db.subscribe(u, "f").await.unwrap();
+        for sql in [
+            "INSERT INTO entries (feed_id, guid, first_seen, duration) VALUES ('f', 'g', 1, 600)",
+            "INSERT INTO enclosures (id, feed_id, guid, url, path, state, downloaded_at)
+             VALUES (7, 'f', 'g', 'http://x/a.mp3', '/x/a.mp3', 'done', 100)",
+        ] {
+            db.sql(sql).await.unwrap();
+        }
+        // Two minutes cut at the usual pause, both before the place saved; 40 s at the longest.
+        let plans = r#"{"300": {"cuts": [[60000,120000],[200000,260000]], "pieces": []},
+                        "500": {"cuts": [[60000,100000]], "pieces": []}}"#;
+        db.store_plans(7, plans).await.unwrap();
+        db.set_position(u, "f", "g", 300, None).await.unwrap();
+        let order = db::order_sql("published", "desc");
+        let page = || {
+            let (order, ctx) = (order.clone(), &ctx);
+            async move {
+                let mut rows = ctx.db.entries_in(u, Some("f"), db::Filter::All, None, 0, 10, &order).await.unwrap();
+                for_reader(ctx, u, &mut rows).await.unwrap();
+                rows.remove(0)
+            }
+        };
+        let off = page().await;
+        assert_eq!((off.position, off.enclosures[0].deadair), (300, None), "off until turned on for the feed");
+        assert!(!off.enclosures[0].deadair_pending);
+
+        db.set_deadair(u, "f", Some(true)).await.unwrap();
+        db.sql("DELETE FROM deadair_plans").await.unwrap();
+        let waiting = page().await;
+        assert!(waiting.enclosures[0].deadair_pending, "on, and its silences not found yet");
+        db.store_plans(7, plans).await.unwrap();
+        let on = page().await;
+        assert!(!on.enclosures[0].deadair_pending);
+        assert_eq!((on.position, on.duration, on.enclosures[0].deadair), (180, Some(480), Some(120)));
+        // A longer pause of their own: their clock, and the playlist, are that plan's (#182).
+        db.set_deadair_pause(u, "f", 500).await.unwrap();
+        let long = page().await;
+        assert_eq!((long.position, long.duration, long.enclosures[0].deadair), (260, Some(560), Some(40)));
+        assert_eq!(plan_for(&ctx, u, "f", 7).await.unwrap().unwrap().saved_ms(), 40_000);
+        db.set_deadair_pause(u, "f", 300).await.unwrap();
+
+        // Saved from the playlist at 3:20 of its 8:00: 5:20 of the original's 10:00, short of 75%.
+        let playing = Playing::default();
+        let plan = plan_for(&ctx, u, "f", 7).await.unwrap().unwrap();
+        playing.lock().unwrap().insert((u, "f".into(), "g".into()), Arc::new(plan));
+        let saved = on_save(&ctx, &playing, u, "f", "g", 200, Some(480)).await.unwrap();
+        assert_eq!(saved, (320, Some(600)));
+        assert_eq!(db.deadair_total(u).await.unwrap(), 0);
+        // Past 75% (6:00 of the playlist is 8:00 of the original), counted once.
+        for _ in 0..2 {
+            on_save(&ctx, &playing, u, "f", "g", 360, None).await.unwrap();
+        }
+        assert_eq!(db.deadair_total(u).await.unwrap(), 120_000);
+        // Playing the original again: saved as it is.
+        playing.lock().unwrap().clear();
+        assert_eq!(on_save(&ctx, &playing, u, "f", "g", 200, Some(600)).await.unwrap(), (200, Some(600)));
+        // Turned off again: the original's clock, and no playlist.
+        db.set_deadair(u, "f", Some(false)).await.unwrap();
+        assert_eq!(page().await.enclosures[0].deadair, None);
+        assert!(plan_for(&ctx, u, "f", 7).await.unwrap().is_none());
+    }
+
+    #[tokio::test]
+    async fn two_people_play_one_file_each_in_their_own_version_and_place() {
+        let ctx = test_ctx(config::Config::default()).await;
+        let db = &ctx.db;
+        for sql in [
+            "INSERT INTO entries (feed_id, guid, first_seen, duration) VALUES ('f', 'g', 1, 600)",
+            "INSERT INTO enclosures (id, feed_id, guid, url, path, state, downloaded_at)
+             VALUES (7, 'f', 'g', 'http://x/a.mp3', '/x/a.mp3', 'done', 100)",
+        ] {
+            db.sql(sql).await.unwrap();
+        }
+        db.store_plans(
+            7,
+            r#"{"300": {"cuts": [[60000,120000],[200000,260000]], "pieces": []},
+                              "500": {"cuts": [[60000,100000]], "pieces": []}}"#,
+        )
+        .await
+        .unwrap();
+        // One file on disk, three listeners at once: DeadAir at the usual pause, at a longer
+        // one, and the original.
+        let (usual, long, plain) = (
+            db.create_user("usual", None, false).await.unwrap(),
+            db.create_user("long", None, false).await.unwrap(),
+            db.create_user("plain", None, false).await.unwrap(),
+        );
+        for u in [usual, long, plain] {
+            db.subscribe(u, "f").await.unwrap();
+        }
+        for u in [usual, long] {
+            db.set_deadair(u, "f", Some(true)).await.unwrap();
+        }
+        db.set_deadair_pause(long, "f", 500).await.unwrap();
+        // Each is handed their own playlist, as /media/7/deadair.m3u8 does.
+        let playing = Playing::default();
+        for u in [usual, long] {
+            let plan = plan_for(&ctx, u, "f", 7).await.unwrap().unwrap();
+            playing.lock().unwrap().insert((u, "f".into(), "g".into()), Arc::new(plan));
+        }
+        assert!(plan_for(&ctx, plain, "f", 7).await.unwrap().is_none());
+        // All three save 3:20 of what they hear, as the position route does: each is read in
+        // that person's own clock and kept as a place in the original, theirs alone.
+        let mut kept = vec![];
+        for u in [usual, long, plain] {
+            let (secs, len) = on_save(&ctx, &playing, u, "f", "g", 200, None).await.unwrap();
+            db.set_position(u, "f", "g", secs, len).await.unwrap();
+            kept.push(secs);
+        }
+        assert_eq!(kept, [320, 240, 200], "two minutes cut before 3:20, forty seconds, none");
+        // Each comes back to 3:20 of their own version, whatever the others saved.
+        let order = db::order_sql("published", "desc");
+        for (u, len) in [(usual, 480), (long, 560), (plain, 600)] {
+            let mut rows = db.entries_in(u, Some("f"), db::Filter::All, None, 0, 10, &order).await.unwrap();
+            for_reader(&ctx, u, &mut rows).await.unwrap();
+            assert_eq!((rows[0].position, rows[0].duration), (200, Some(len)), "user {u}");
+        }
+    }
 }
 
 /// A real episode, for trying the cut on something other than a test tone:

crates/ipx/src/directory.rs

@@ -7,7 +7,9 @@ use std::collections::{HashMap, HashSet};
 use ipx_directory::{FeedPage, Listing, Me, SnapshotEntry, SnapshotResult};
 use serde::de::DeserializeOwned;
 
-use crate::{Ctx, Sub, config, feed};
+use crate::catalogue::Sub;
+use crate::ctx::Ctx;
+use crate::{config, feed};
 use std::fmt::{Display, Formatter};
 use std::sync::{Arc, LazyLock};
 use std::time::{Duration, Instant};
@@ -226,7 +228,7 @@ pub fn snapshot_entries(catalogue: &[Sub], counts: &HashMap<String, i64>) -> Vec
 
 /// Sends the snapshot now.
 async fn snapshot(ctx: &Ctx) -> Result<Vec<SnapshotEntry>, DirError> {
-    let catalogue = crate::subscriptions(ctx).await.map_err(|e| DirError::Unreachable(format!("{e:#}")))?;
+    let catalogue = crate::catalogue::subscriptions(ctx).await.map_err(|e| DirError::Unreachable(format!("{e:#}")))?;
     let counts = ctx.db.subscriber_counts().await.map_err(|e| DirError::Unreachable(format!("{e:#}")))?;
     Ok(snapshot_entries(&catalogue, &counts))
 }

crates/ipx/src/lists.rs

@@ -0,0 +1,270 @@
+//! OPML: importing a file's feeds, and keeping a subscribed list's feeds in step with it.
+
+use crate::ctx::Ctx;
+use crate::scan::Outcome;
+use crate::{catalogue, config, feed};
+use anyhow::Result;
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+
+/// The feeds an OPML document lists, up to as many as a subscribed list may bring in: each becomes
+/// a feed in the catalogue, scanned for everyone, and one upload of tens of thousands held the
+/// catalogue for minutes and stayed in it (#338).
+pub fn opml_feeds(doc: &opml::OPML) -> Result<Vec<(String, String)>> {
+    let mut found = vec![];
+    feed::collect_outlines(&doc.body.outlines, &mut found);
+    anyhow::ensure!(
+        found.len() <= MAX_LISTED,
+        "it lists {} feeds, more than the {MAX_LISTED} one may have",
+        found.len()
+    );
+    Ok(found)
+}
+
+/// 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 and lists its feeds (`opml_feeds`), so each refuses a file in
+/// its own terms, before anything is touched: a 400 from the web, a message from the CLI.
+pub async fn subscribe_opml(ctx: &Ctx, found: Vec<(String, String)>, user_id: i64) -> Result<(usize, usize)> {
+    let mut cfg = ctx.edit_cfg().await;
+    // By address, once: each feed looked through every subscription and the whole catalogue, and
+    // the catalogue's lock is held throughout (#338).
+    let mut known = HashMap::new();
+    for s in catalogue::subscriptions(ctx).await? {
+        // The first with an address wins, as a search in order found it.
+        known.entry(s.cfg.url).or_insert(s.id);
+    }
+    let mut ids = vec![];
+    let mut grew = false;
+    for (title, url) in found {
+        // Known already, or the same URL listed twice in one file.
+        let existing = known.get(&url).cloned();
+        let id = match existing {
+            Some(id) => id,
+            None => {
+                let id = config::unique_slug(&title, |s| cfg.feeds.contains_key(s));
+                known.insert(url.clone(), id.clone());
+                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,
+                        category: None,
+                        listed: false,
+                        private: false,
+                    },
+                );
+                grew = true;
+                id
+            }
+        };
+        ids.push(id);
+    }
+    if grew {
+        ctx.store_cfg(cfg).await?;
+    }
+
+    let (mut added, mut had) = (0, 0);
+    for id in ids {
+        if ctx.db.subscription(user_id, &id).await?.is_some() {
+            had += 1;
+        } else {
+            ctx.db.subscribe(user_id, &id).await?;
+            added += 1;
+        }
+    }
+    Ok((added, had))
+}
+
+pub(crate) async fn sync_opml(ctx: &Arc<Ctx>, parent_id: &str, 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).await?;
+    }
+    sync_group(ctx, parent_id, &listed).await
+}
+
+/// How many feeds an OPML or Patreon list may bring in.
+pub(crate) const MAX_LISTED: usize = 2000;
+
+/// 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.
+pub(crate) async fn sync_group(ctx: &Arc<Ctx>, parent_id: &str, listed: &[(String, String)]) -> Result<Outcome> {
+    // Each is a feed scanned for everyone subscribed to the list, and the list is its author's to
+    // write: one of tens of thousands held the scan for hours (#337). davewiner's had 922.
+    anyhow::ensure!(
+        listed.len() <= MAX_LISTED,
+        "it lists {} feeds, more than the {MAX_LISTED} a list may have",
+        listed.len()
+    );
+    let cfg = ctx.cfg();
+    let existing = ctx.db.managed_feeds().await?;
+    let mut added = vec![];
+    // Read once, not once a new feed: each read is the whole feeds table.
+    let known = ctx.db.feed_urls().await?;
+    let fixed: HashSet<&str> = cfg.feeds.keys().chain(existing.iter().map(|m| &m.id)).map(String::as_str).collect();
+    let known_ids: HashSet<&str> = known.iter().map(|(id, _)| id.as_str()).collect();
+
+    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).await?;
+            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 its_own: Vec<&str> =
+            known.iter().filter(|(_, u)| feed::same_feed(u, url)).map(|(id, _)| id.as_str()).collect();
+        let id = config::unique_slug(title, |s| {
+            fixed.contains(s) || added.iter().any(|a| a == s) || (known_ids.contains(s) && !its_own.contains(&s))
+        });
+        ctx.db.upsert_managed(&id, url, title, parent_id).await?;
+        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.
+    let managed = ctx.db.managed_feeds().await?;
+    let list_and_feeds = managed
+        .iter()
+        .filter(|m| m.group_id == parent_id)
+        .map(|m| m.id.clone())
+        .chain(std::iter::once(parent_id.to_string()));
+    for id in list_and_feeds {
+        for user in ctx.db.users().await? {
+            if ctx.db.subscription(user.id, parent_id).await?.is_some() {
+                ctx.db.subscribe(user.id, &id).await?;
+            }
+        }
+    }
+
+    // Anything in this group the OPML no longer lists.
+    let mut removed = 0;
+    let mut kept = 0;
+    let parent = ctx.cfg().feeds.get(parent_id).cloned();
+    let in_group = existing.iter().filter(|m| m.group_id == parent_id);
+    for m in in_group {
+        if listed.iter().any(|(_, u)| u == &m.url) {
+            continue;
+        }
+        if catalogue::promote_kept(ctx, m, parent.as_ref()).await? {
+            kept += 1;
+            tracing::info!(feed = %m.id, "dropped from the OPML but someone kept it; a feed of its own now");
+            continue;
+        }
+        if ctx.db.downloaded_count(&m.id).await.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).await?;
+            kept += 1;
+            tracing::info!(feed = %m.id, "dropped from the OPML but has downloads; keeping it");
+        } else {
+            ctx.db.drop_managed(&m.id).await?;
+            removed += 1;
+            tracing::info!(feed = %m.id, "dropped from the OPML with nothing downloaded; removed");
+        }
+    }
+
+    Ok(Outcome::Opml { added, removed, kept, total: listed.len() })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::test_util::{feed, test_ctx};
+
+    #[tokio::test]
+    async fn an_opml_upload_is_capped_and_takes_a_repeated_address_once() {
+        let opml = |n: usize| {
+            let outlines: String =
+                (0..n).map(|i| format!(r#"<outline text="Same" xmlUrl="http://x/{}.xml"/>"#, i % 3)).collect();
+            feed::read_opml(&format!(r#"<opml version="2.0"><head/><body>{outlines}</body></opml>"#)).unwrap()
+        };
+        assert!(opml_feeds(&opml(MAX_LISTED + 1)).unwrap_err().to_string().contains("more than the 2000"));
+        let ctx = test_ctx(config::Config::default()).await;
+        let me = ctx.db.create_user("me", None, false).await.unwrap();
+        // Six outlines, three addresses, one title: three feeds, named apart.
+        let (added, had) = subscribe_opml(&ctx, opml_feeds(&opml(6)).unwrap(), me).await.unwrap();
+        let ids: std::collections::BTreeSet<String> = ctx.cfg().feeds.keys().cloned().collect();
+        assert_eq!(ids, ["same", "same-2", "same-3"].map(String::from).into());
+        assert_eq!((added, had), (3, 3));
+    }
+
+    #[tokio::test]
+    async fn a_list_of_more_feeds_than_the_limit_is_refused() {
+        let ctx = Arc::new(test_ctx(config::Config::default()).await);
+        let listed: Vec<(String, String)> =
+            (0..=MAX_LISTED).map(|n| (format!("Show {n}"), format!("http://x/{n}.xml"))).collect();
+        let err = sync_group(&ctx, "opml", &listed).await.err().expect("refused");
+        assert!(err.to_string().contains("more than the 2000"), "{err:#}");
+        assert!(ctx.db.managed_feeds().await.unwrap().is_empty(), "nothing taken from it");
+        // At the limit, all of them, each its own id.
+        sync_group(&ctx, "opml", &listed[..MAX_LISTED]).await.unwrap();
+        let ids: std::collections::HashSet<String> =
+            ctx.db.managed_feeds().await.unwrap().into_iter().map(|m| m.id).collect();
+        assert_eq!(ids.len(), MAX_LISTED);
+    }
+
+    #[tokio::test]
+    async fn a_kept_feed_outlives_its_opml_for_whoever_kept_it() {
+        let mut cfg = config::Config::default();
+        cfg.feeds.insert("opml".into(), config::Feed { private: true, ..feed() });
+        let ctx = Arc::new(test_ctx(cfg).await);
+        let (me, you) = (
+            ctx.db.create_user("me", None, false).await.unwrap(),
+            ctx.db.create_user("you", None, false).await.unwrap(),
+        );
+        for id in ["opml", "a", "b"] {
+            if id != "opml" {
+                ctx.db.upsert_managed(id, &format!("http://x/{id}.xml"), id, "opml").await.unwrap();
+            }
+            for u in [me, you] {
+                ctx.db.subscribe(u, id).await.unwrap();
+            }
+        }
+        assert!(ctx.db.set_kept(me, "a", true).await.unwrap());
+        assert!(ctx.db.set_kept(me, "b", true).await.unwrap());
+
+        // The OPML drops a: mine to keep, a feed of its own; yours, unkept, goes.
+        sync_group(&ctx, "opml", &[("b".into(), "http://x/b.xml".into())]).await.unwrap();
+        let a = ctx.cfg().feeds.get("a").cloned().expect("a is in the catalogue");
+        assert_eq!((a.url.as_str(), a.group, a.private), ("http://x/a.xml", None, true));
+        assert!(!ctx.db.managed_feeds().await.unwrap().iter().any(|m| m.id == "a"));
+        assert!(ctx.db.subscription(me, "a").await.unwrap().is_some());
+        assert!(ctx.db.subscription(you, "a").await.unwrap().is_none());
+
+        // Leaving the OPML takes what I did not keep and leaves b, still the OPML's while you read it.
+        catalogue::leave(&ctx, me, "opml").await.unwrap();
+        assert!(ctx.db.subscription(me, "b").await.unwrap().is_some());
+        assert!(ctx.db.managed_feeds().await.unwrap().iter().any(|m| m.id == "b"));
+        // Once nobody reads it, b is mine alone, and yours goes with the OPML.
+        catalogue::leave(&ctx, you, "opml").await.unwrap();
+        // Still private: the OPML is gone from config by the time b is promoted (#332).
+        assert!(ctx.cfg().feeds.get("b").is_some_and(|f| f.group.is_none() && f.private));
+        assert!(ctx.db.subscription(me, "b").await.unwrap().is_some());
+        assert!(ctx.db.subscription(you, "b").await.unwrap().is_none());
+    }
+}

crates/ipx/src/logging.rs

@@ -0,0 +1,89 @@
+//! Log lines as JSON for Loki, carrying the trace and span they belong to.
+
+use crate::logbuf;
+use std::io::IsTerminal;
+use tracing::Subscriber;
+use tracing_subscriber::EnvFilter;
+use tracing_subscriber::fmt::format::JsonFields;
+use tracing_subscriber::fmt::{FmtContext, FormatEvent, FormatFields};
+use tracing_subscriber::registry::LookupSpan;
+
+/// The JSON log line with the trace and span it belongs to (#91), so a line in Loki leads to its
+/// trace in Tempo. The JSON formatter cannot take a field of its own, so the ids go on the end of
+/// the object it writes. A line outside any traced span, or with no trace exporter, is unchanged.
+pub(crate) struct WithTrace<F>(F);
+
+impl<S, N, F> FormatEvent<S, N> for WithTrace<F>
+where
+    F: FormatEvent<S, N>,
+    S: Subscriber + for<'a> LookupSpan<'a>,
+    N: for<'w> FormatFields<'w> + 'static,
+{
+    fn format_event(
+        &self,
+        ctx: &FmtContext<'_, S, N>,
+        mut w: tracing_subscriber::fmt::format::Writer<'_>,
+        ev: &tracing::Event<'_>,
+    ) -> std::fmt::Result {
+        use opentelemetry::trace::TraceContextExt;
+        use tracing_opentelemetry::OpenTelemetrySpanExt;
+        let current = tracing::Span::current();
+        let sc = if current.is_none() { None } else { Some(current.context().span().span_context().clone()) };
+        let Some(sc) = sc.filter(|c| c.is_valid()) else {
+            return self.0.format_event(ctx, w, ev);
+        };
+        let mut line = String::new();
+        self.0.format_event(ctx, tracing_subscriber::fmt::format::Writer::new(&mut line), ev)?;
+        match line.trim_end().strip_suffix('}') {
+            Some(body) => writeln!(w, r#"{body},"trace_id":"{}","span_id":"{}"}}"#, sc.trace_id(), sc.span_id()),
+            None => w.write_str(&line),
+        }
+    }
+}
+
+/// Sets up logging: to stderr, as text or as JSON (`IPX_LOG_FORMAT`), into the ring the page's log
+/// reads, and to the trace exporter when there is one.
+pub(crate) fn init(otel: Option<&opentelemetry_sdk::trace::SdkTracerProvider>) {
+    // Everything goes to stderr as before, and is mirrored into a ring the UI can read.
+    use tracing_subscriber::Layer;
+    use tracing_subscriber::layer::SubscriberExt;
+    use tracing_subscriber::util::SubscriberInitExt;
+    // 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 = || EnvFilter::try_from_env("IPX_LOG").unwrap_or_else(|_| "ipx=info".into());
+    // One JSON object a line for Loki (#91), with each event's fields as its own; text
+    // otherwise, for someone reading a terminal.
+    let json = std::env::var("IPX_LOG_FORMAT").is_ok_and(|f| f.eq_ignore_ascii_case("json"));
+    let ui_filter = EnvFilter::try_from_env("IPX_UI_LOG").unwrap_or_else(|_| "ipx=debug".into());
+    tracing_subscriber::registry()
+        .with((!json).then(|| {
+            tracing_subscriber::fmt::layer()
+                .with_writer(std::io::stderr)
+                // Colour for a terminal only: in docker logs and Loki the escapes are noise
+                // every query has to strip (#88).
+                .with_ansi(IsTerminal::is_terminal(&std::io::stderr()))
+                .with_filter(stderr_filter())
+        }))
+        .with(json.then(|| {
+            tracing_subscriber::fmt::layer()
+                .fmt_fields(JsonFields::new())
+                .event_format(WithTrace(
+                    tracing_subscriber::fmt::format()
+                        .json()
+                        .flatten_event(true)
+                        .with_current_span(true)
+                        .with_span_list(false),
+                ))
+                .with_writer(std::io::stderr)
+                .with_filter(stderr_filter())
+        }))
+        .with(logbuf::RingLayer.with_filter(ui_filter))
+        .with(otel.as_ref().map(|p| {
+            use opentelemetry::trace::TracerProvider;
+            tracing_opentelemetry::layer()
+                .with_tracer(p.tracer("ipx"))
+                .with_filter(EnvFilter::new("ipx=info,ipx_feed=info"))
+        }))
+        .init();
+}

crates/ipx/src/metrics.rs

@@ -144,7 +144,7 @@ impl Drop for Busy {
 
 /// What the metrics are read from.
 pub struct Sources {
-    pub ctx: Arc<crate::Ctx>,
+    pub ctx: Arc<crate::ctx::Ctx>,
     pub cmds: tokio::sync::mpsc::Sender<crate::ipc::Command>,
     pub events: tokio::sync::broadcast::Sender<crate::ipc::Event>,
     pub playing: crate::deadair::Playing,
@@ -519,4 +519,51 @@ mod tests {
         let (rss, fds) = ipx_directory::metrics::process();
         assert!(rss.unwrap() > 0.0 && fds.unwrap() > 0.0);
     }
+
+    use crate::config;
+    use crate::test_util::test_ctx;
+    use tokio::sync::{broadcast, mpsc};
+
+    #[tokio::test]
+    async fn metrics_answer_from_a_thread_of_their_own() {
+        let ctx = Arc::new(test_ctx(config::Config::default()).await);
+        let (cmds, _rx) = mpsc::channel(8);
+        let (events, _) = broadcast::channel(8);
+        let sources = Sources {
+            ctx: ctx.clone(),
+            cmds,
+            events,
+            playing: Default::default(),
+            runtime: tokio::runtime::Handle::current(),
+            started: std::time::Instant::now(),
+        };
+        let at = spawn("127.0.0.1:0", sources).unwrap();
+        let text = ctx.client.get(format!("http://{at}/metrics")).send().await.unwrap().text().await.unwrap();
+        // Every sample a name, optional labels and a number, as Prometheus parses them (#194).
+        let samples = text.lines().filter(|l| !l.starts_with('#'));
+        for line in samples {
+            let (name, value) = line.rsplit_once(' ').unwrap();
+            assert!(name.starts_with("ipx_") && value.parse::<f64>().is_ok(), "{line}");
+        }
+        for name in
+            ["ipx_db_connections{state=\"in_use\"}", "ipx_db_connections_max", "ipx_worker_queue", "ipx_runtime_tasks"]
+        {
+            assert!(text.contains(name), "{name}");
+        }
+        assert!(
+            spawn(
+                &at.to_string(),
+                Sources {
+                    ctx: ctx.clone(),
+                    cmds: mpsc::channel(1).0,
+                    events: broadcast::channel(1).0,
+                    playing: Default::default(),
+                    runtime: tokio::runtime::Handle::current(),
+                    started: std::time::Instant::now(),
+                }
+            )
+            .is_err(),
+            "a port already taken is an error at start"
+        );
+    }
 }

crates/ipx/src/retention.rs

@@ -4,8 +4,11 @@ use anyhow::Result;
 use tracing::Instrument;
 
 use crate::config::Config;
+use crate::ctx::Ctx;
 use crate::db;
 use crate::db::{Candidate, Db};
+use crate::ipc::Event;
+use crate::{art, catalogue};
 use std::io::ErrorKind;
 use std::path::Path;
 
@@ -115,6 +118,38 @@ async fn remove(db: &Db, c: &Candidate, dry_run: bool) -> Result<u64> {
     Ok(size)
 }
 
+/// `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.
+#[tracing::instrument(name = "reap", skip_all, fields(dry_run = dry_run, files = tracing::field::Empty, bytes = tracing::field::Empty, entries = tracing::field::Empty))]
+pub(crate) async fn reap(ctx: &Ctx, dry_run: bool, standalone: bool) -> Result<()> {
+    let r = run(&ctx.cfg(), &ctx.db, dry_run).await?;
+    let span = tracing::Span::current();
+    span.record("files", r.aged_out.len() + r.over_quota.len());
+    span.record("bytes", r.bytes_freed);
+    span.record("entries", r.entries_pruned);
+    if !dry_run {
+        tracing::Instrument::instrument(catalogue::clean_directory(ctx), db::step("clean_directory")).await?;
+        let moved =
+            tracing::Instrument::instrument(ctx.db.advance_unread_from(), db::step("advance_unread_from")).await?;
+        tracing::debug!(moved, "unread lists start from each person's oldest unread item");
+        // Off the async threads: it walks every file kept, on FUSE (#345).
+        let (span, limit) = (db::step("trim_artwork"), ctx.cfg().general.art_cache_mb * 1_048_576);
+        let gone = tokio::task::spawn_blocking(move || span.in_scope(|| art::trim(&art::dir(), limit))).await?;
+        if gone > 0 {
+            tracing::debug!(gone, "trimmed the artwork kept on disk");
+        }
+    }
+    let removed = r.aged_out.iter().chain(r.over_quota.iter());
+    for c in removed {
+        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(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/ipx/src/signals.rs

@@ -14,7 +14,7 @@ use tracing::Instrument;
 
 use std::sync::atomic::Ordering::Relaxed;
 
-use crate::Ctx;
+use crate::ctx::Ctx;
 use crate::db;
 use crate::db::{Ask, Unjudged};
 use crate::feed;

crates/ipx/src/test_util.rs

@@ -0,0 +1,104 @@
+//! What several modules' tests share: a context on a scratch database, feeds to put in it, and a
+//! server answering with feeds.
+
+use crate::ctx::Ctx;
+use crate::ipc::Emitter;
+use crate::{config, db};
+use std::path::PathBuf;
+use std::sync::Arc;
+
+pub(crate) fn feed() -> config::Feed {
+    // Whatever `ipx add` would write, which is the shape every code path sees.
+    let mut cfg = config::Config::default();
+
+    add_one_cfg(&mut cfg, "http://x/f.xml", None, vec![])
+}
+
+/// The feed entry `add` builds, without the network round trip it does for a title.
+pub(crate) 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,
+        category: None,
+        listed: false,
+        private: false,
+    }
+}
+
+pub(crate) 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,
+        blocked: vec![],
+    }
+}
+
+pub(crate) async fn test_ctx(cfg: config::Config) -> Ctx {
+    Ctx {
+        cfg: std::sync::RwLock::new(Arc::new(cfg)),
+        db: db::Db::memory().await.unwrap(),
+        client: reqwest::Client::new(),
+        local_client: reqwest::Client::new(),
+        feed_client: reqwest::Client::builder().redirect(reqwest::redirect::Policy::none()).build().unwrap(),
+        out: Emitter::terminal(),
+        torrents: tokio::sync::OnceCell::new(),
+        torrent_slots: Arc::new(tokio::sync::Semaphore::new(2)),
+        config_path: PathBuf::new(),
+        detach_torrents: false,
+        retry: Default::default(),
+        cfg_write: Default::default(),
+        deadair: Default::default(),
+        signals: Default::default(),
+        signals_refused: Default::default(),
+        directory_wake: Default::default(),
+        directory_shared: Default::default(),
+        directory_resend: Default::default(),
+        directory_cleaned: Default::default(),
+    }
+}
+
+/// Answers /empty with a 200 and nothing, /fixed with a feed of one item as it should read,
+/// anything else with the same item's text broken.
+pub(crate) async fn empty_or_feed_server() -> String {
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    tokio::spawn(async move {
+        loop {
+            let Ok((mut sock, _)) = listener.accept().await else { return };
+            tokio::spawn(async move {
+                let mut buf = [0u8; 2048];
+                let n = sock.read(&mut buf).await.unwrap_or(0);
+                let req = String::from_utf8_lossy(&buf[..n]).into_owned();
+                let body = if req.starts_with("GET /empty") {
+                    "\n"
+                } else if req.starts_with("GET /fixed") {
+                    "<?xml version=\"1.0\"?><rss version=\"2.0\"><channel><title>T</title><item><title>One</title><guid>g1</guid><description>As it should read.</description></item></channel></rss>"
+                } else {
+                    "<?xml version=\"1.0\"?><rss version=\"2.0\"><channel><title>T</title><item><title>One</title><guid>g1</guid><description>Somebody else's text.</description></item></channel></rss>"
+                };
+                let resp = format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n{body}", body.len());
+                let _ = sock.write_all(resp.as_bytes()).await;
+            });
+        }
+    });
+    format!("http://{addr}")
+}

crates/ipx/src/web.rs

@@ -20,7 +20,7 @@ use tokio::sync::{broadcast, mpsc};
 use tower::ServiceExt;
 use tower_http::services::ServeFile;
 
-use crate::Ctx;
+use crate::ctx::Ctx;
 use crate::db;
 use crate::ipc::{Command, Event};
 use axum::Extension;
@@ -768,7 +768,7 @@ async fn remove_user(
             target.name
         )));
     }
-    crate::remove_user(&state.ctx, id).await?;
+    crate::catalogue::remove_user(&state.ctx, id).await?;
     tracing::info!(by = %user.name, user = %target.name, "account removed");
     Ok(StatusCode::NO_CONTENT)
 }
@@ -895,7 +895,7 @@ async fn feed_rows(state: &WebState, user: &crate::db::User, only: Option<&str>)
     let cfg = state.ctx.cfg();
     // Config entries plus the feeds derived from OPML subscriptions -- the catalogue.
     // What comes back is only the part of it this person subscribes to.
-    let subs = crate::subscriptions(&state.ctx).instrument(db::step("catalogue")).await?;
+    let subs = crate::catalogue::subscriptions(&state.ctx).instrument(db::step("catalogue")).await?;
     let mine: HashMap<String, crate::db::Sub> = state
         .ctx
         .db
@@ -965,11 +965,11 @@ async fn feed_rows(state: &WebState, user: &crate::db::User, only: Option<&str>)
             managed: sub.managed,
             schedule: feed.schedule.clone(),
             schedule_mins: feed.schedule.as_deref().and_then(crate::config::parse_interval),
-            every_mins: crate::due_after(&cfg, feed, ttl_mins, None, None, 0) / 60,
+            every_mins: crate::scan::due_after(&cfg, feed, ttl_mins, None, None, 0) / 60,
             last_checked: s.last_checked,
             next_check: s
                 .last_checked
-                .map(|t| t + crate::due_after(&cfg, feed, ttl_mins, s.error_since, s.same_since, t) as i64),
+                .map(|t| t + crate::scan::due_after(&cfg, feed, ttl_mins, s.error_since, s.same_since, t) as i64),
             failing: s
                 .error_since
                 .filter(|since| crate::db::now() - since >= FLAG_AFTER_SECS)
@@ -1039,7 +1039,7 @@ async fn rows_for(state: &WebState, user_id: i64, listed: Vec<ipx_directory::Lis
         .into_iter()
         .map(|s| s.feed_id)
         .collect();
-    let here: HashMap<String, String> = crate::subscriptions(&state.ctx)
+    let here: HashMap<String, String> = crate::catalogue::subscriptions(&state.ctx)
         .instrument(db::step("catalogue"))
         .await?
         .into_iter()
@@ -1216,7 +1216,7 @@ async fn subscribe_popular(
         crate::directory::DirError::Refused(404, _) => ApiError::bad_request(format!("{id:?} is not in the directory")),
         e => dir_error(e),
     })?;
-    let (feed_id, existed) = crate::add_to_catalogue(&state.ctx, &page.listing.url, None, vec![]).await?;
+    let (feed_id, existed) = crate::catalogue::add_to_catalogue(&state.ctx, &page.listing.url, None, vec![]).await?;
     let already = existed && state.ctx.db.subscription(user.id, &feed_id).await?.is_some();
     state.ctx.db.subscribe(user.id, &feed_id).await?;
     if !already {
@@ -1706,7 +1706,7 @@ async fn add_feed(
         .map_err(|e| ApiError::bad_request(format!("{e:#}")))?;
     // Someone else may already have it. Then adding costs nothing: no second fetch, no
     // second copy on disk, just another name against the same feed.
-    let (id, existed) = crate::add_to_catalogue(&state.ctx, &url, body.folder, body.keywords).await?;
+    let (id, existed) = crate::catalogue::add_to_catalogue(&state.ctx, &url, body.folder, body.keywords).await?;
     let already = existed && state.ctx.db.subscription(user.id, &id).await?.is_some();
     state.ctx.db.subscribe(user.id, &id).await?;
     if !already {
@@ -1880,7 +1880,7 @@ async fn patch_feed(
     // Derived feeds have no config entry. Editing one is the moment it earns a real
     // entry: promote it, so the config holds your decisions and nothing else.
     if !cfg.feeds.contains_key(&id) {
-        let subs = crate::subscriptions(&state.ctx).instrument(db::step("subscriptions")).await?;
+        let subs = crate::catalogue::subscriptions(&state.ctx).instrument(db::step("subscriptions")).await?;
         let found =
             subs.iter().find(|s| s.id == id).ok_or_else(|| ApiError::bad_request(format!("no feed with id {id:?}")))?;
         cfg.feeds.insert(id.clone(), found.cfg.clone());
@@ -1934,7 +1934,7 @@ async fn remove_feed(
     Path(id): Path<String>,
     user: crate::db::User,
 ) -> Result<StatusCode, ApiError> {
-    crate::leave(&state.ctx, user.id, &id).await?;
+    crate::catalogue::leave(&state.ctx, user.id, &id).await?;
     Ok(StatusCode::NO_CONTENT)
 }
 
@@ -2442,7 +2442,7 @@ async fn read_all(
     // A subscription's own row has no entries, so marking it read means everything under it.
     let mut ids = vec![id.clone()];
     ids.extend(
-        crate::subscriptions(&state.ctx)
+        crate::catalogue::subscriptions(&state.ctx)
             .await?
             .into_iter()
             .filter(|s| s.cfg.group.as_deref() == Some(id.as_str()))
@@ -2504,7 +2504,7 @@ async fn export_opml(State(state): State<WebState>, user: crate::db::User) -> Re
         head: Some(opml::Head { title: Some("ipx subscriptions".into()), ..Default::default() }),
         ..Default::default()
     };
-    for s in crate::subscriptions(&state.ctx).await? {
+    for s in crate::catalogue::subscriptions(&state.ctx).await? {
         // A feed from an OPML subscription comes back with the OPML itself.
         if s.managed || !mine.contains(&s.id) {
             continue;
@@ -2538,8 +2538,8 @@ async fn import_opml(
     // file arrives as text, is read here, and is gone when the request ends.
     let doc = crate::feed::read_opml(&body.xml)
         .map_err(|e| ApiError::bad_request(format!("that is not an OPML file: {e}")))?;
-    let found = crate::opml_feeds(&doc).map_err(|e| ApiError::bad_request(format!("{e:#}")))?;
+    let found = crate::lists::opml_feeds(&doc).map_err(|e| ApiError::bad_request(format!("{e:#}")))?;
-    let (added, already) = crate::subscribe_opml(&state.ctx, found, user.id).await?;
+    let (added, already) = crate::lists::subscribe_opml(&state.ctx, found, user.id).await?;
     if added > 0 {
         scan_soon(&state, None).await;
     }
@@ -2549,7 +2549,7 @@ async fn import_opml(
 /// The numbers `ipx status` prints, and the version, for a dashboard such as Homepage's
 /// customapi widget. Behind sign-in like the rest of /api; Homepage sends the shared token.
 async fn status(State(state): State<WebState>) -> Response {
-    match crate::status(&state.ctx).await {
+    match crate::daemon::status(&state.ctx).await {
         Event::Status { feeds, pending, downloaded } => Json(serde_json::json!({
             "feeds": feeds,
             "pending": pending,

docs/architecture.md

@@ -27,7 +27,14 @@ id, and Caddy's request lines theirs. ipx also serves Prometheus metrics (`[web]
 
 | File | Responsibility | What it replaced in the Python |
 |---|---|---|
-| `crates/ipx/src/main.rs` | CLI, dispatch, scan loop, download policy | `iPXAgent.py` |
+| `crates/ipx/src/main.rs` | The module list and `main`: reads the command line and hands it on | `iPXAgent.py` |
+| `crates/ipx/src/cli.rs` | The commands, and what each does when no daemon takes it | `iPXAgent.py` |
+| `crates/ipx/src/ctx.rs` | `Ctx`, what every command shares: configuration, database, clients | — |
+| `crates/ipx/src/daemon.rs` | The daemon: worker loop, web server, shutdown, `status` | `iPXAgent.py` |
+| `crates/ipx/src/scan.rs` | When a feed is due, reading it, download policy, downloads and torrents | `iPXAgent.py` |
+| `crates/ipx/src/catalogue.rs` | Adding, removing and leaving feeds; subscriptions; retiring a list's feeds | — |
+| `crates/ipx/src/lists.rs` | OPML import, and keeping an OPML's or a Patreon creator's feeds in step | — |
+| `crates/ipx/src/logging.rs` | Logging setup, and JSON lines carrying their trace | — |
 | `crates/ipx/src/config.rs` | TOML load/save, `General`/`Feed`/`Web`, intervals, slugs | `iPXSettings.py`, `feeds.plist` |
 | `crates/ipx/src/db.rs` | Every query, through SeaORM; creates missing tables | `.ipxd` plists, `history.dat`, `qmcache.dat` |
 | `crates/ipx/src/entity.rs` | The tables, as SeaORM entities: the schema | — |