Step 6: control socket and daemon

Unix-socket JSON-lines protocol replacing the printMSG sentinels, with
an Emitter so scan code is agnostic about whether a terminal or a UI is
watching. Daemon serves clients and runs a TTL-aware scheduler; commands
funnel through one worker so scans cannot overlap, and the CLI proxies to
a running daemon rather than competing with it.

The pre-fetch retention sweep was emitting the terminal ReapDone event,
which would have ended a client's read before the scan began.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RPyeapneuXrCdojsaiXGbe

src/main.rs

@@ -2,11 +2,14 @@ mod config;
 mod db;
 mod download;
 mod feed;
+mod ipc;
 mod retention;
 
 use anyhow::Result;
 use clap::{Parser, Subcommand};
+use ipc::{Command as Cmd, Emitter, Event};
 use std::path::PathBuf;
+use tokio::sync::{broadcast, mpsc};
 
 #[derive(Parser)]
 #[command(name = "ipx", version, about = "A headless podcatcher")]
@@ -15,6 +18,10 @@ struct Cli {
     #[arg(long, global = true)]
     config: Option<PathBuf>,
 
+    /// Do the work here even if a daemon is running
+    #[arg(long, global = true)]
+    local: bool,
+
     #[command(subcommand)]
     command: Command,
 }
@@ -23,12 +30,6 @@ struct Cli {
 enum Command {
     /// Show configured feeds and their state
     List,
-    /// Delete old or over-quota downloads
-    Reap {
-        /// Show what would go, delete nothing
-        #[arg(long)]
-        dry_run: bool,
-    },
     /// Scan feeds for new entries
     Fetch {
         /// Only this feed id
@@ -37,36 +38,149 @@ enum Command {
         #[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,
+    /// Run the scheduler and serve the control socket
+    Daemon,
+}
+
+/// Everything a command needs. One per process.
+struct Ctx {
+    cfg: config::Config,
+    db: db::Db,
+    client: reqwest::Client,
+    out: Emitter,
 }
 
 #[tokio::main]
 async fn main() -> Result<()> {
     let cli = Cli::parse();
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::try_from_env("IPX_LOG")
+                .unwrap_or_else(|_| "ipx=info".into()),
+        )
+        .with_writer(std::io::stderr)
+        .init();
+
     let config_path = cli.config.clone().unwrap_or_else(config::config_path);
     let cfg = config::Config::load(&config_path)?;
     let db = db::Db::open(&config::data_dir().join("state.db"))?;
 
-    match cli.command {
-        Command::List => list(&cfg, &db, &config_path),
+    // A daemon owns the state; don't have two processes downloading the same thing.
+    let wire_cmd = match &cli.command {
         Command::Fetch { feed, force } => {
-            // Make room before pulling more down, as the original did per download.
-            report_reap(&retention::run(&cfg, &db, false)?, false);
-            fetch(&cfg, &db, feed.as_deref(), force).await
+            Some(Cmd::Fetch { feed: feed.clone(), force: *force })
         }
-        Command::Reap { dry_run } => {
-            report_reap(&retention::run(&cfg, &db, dry_run)?, true);
+        Command::Reap { dry_run } => Some(Cmd::Reap { dry_run: *dry_run }),
+        Command::Status => Some(Cmd::Status),
+        Command::List | Command::Daemon => None,
+    };
+    if let Some(cmd) = &wire_cmd
+        && !cli.local
+        && ipc::daemon_is_live(&cfg.general.socket).await
+    {
+        return ipc::proxy(&cfg.general.socket, cmd).await;
+    }
+
+    let ctx = Ctx {
+        cfg,
+        db,
+        client: reqwest::Client::builder()
+            .user_agent(concat!("ipx/", env!("CARGO_PKG_VERSION")))
+            .build()?,
+        out: Emitter::terminal(),
+    };
+
+    match cli.command {
+        Command::List => list(&ctx, &config_path),
+        Command::Daemon => daemon(ctx).await,
+        _ => run(&ctx, wire_cmd.expect("only List and Daemon have no wire form")).await,
+    }
+}
+
+async fn run(ctx: &Ctx, cmd: Cmd) -> Result<()> {
+    match cmd {
+        Cmd::Fetch { feed, force } => {
+            // Make room before pulling more down, as the original did per download.
+            reap(ctx, false, false)?;
+            fetch(ctx, feed.as_deref(), force).await
+        }
+        Cmd::Reap { dry_run } => reap(ctx, dry_run, true),
+        Cmd::Status => {
+            let (pending, downloaded) = ctx.db.counts()?;
+            ctx.out.emit(Event::Status { feeds: ctx.cfg.feeds.len(), pending, downloaded });
             Ok(())
         }
     }
 }
 
-fn list(cfg: &config::Config, db: &db::Db, config_path: &std::path::Path) -> Result<()> {
-    if cfg.feeds.is_empty() {
+async fn daemon(ctx: Ctx) -> Result<()> {
+    let socket = ctx.cfg.general.socket.clone();
+    if ipc::daemon_is_live(&socket).await {
+        anyhow::bail!("a daemon is already listening on {}", socket.display());
+    }
+
+    let (events, _) = broadcast::channel(1024);
+    let (tx_cmd, mut rx_cmd) = mpsc::channel::<Cmd>(64);
+    let ctx = Ctx { out: Emitter::socket(events.clone(), false), ..ctx };
+
+    let server = tokio::spawn(ipc::serve(socket.clone(), events, tx_cmd));
+
+    // One command at a time: the queue is what keeps two scans from overlapping.
+    let mut ticker = tokio::time::interval(std::time::Duration::from_secs(60));
+    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    tracing::info!(feeds = ctx.cfg.feeds.len(), "daemon started");
+
+    loop {
+        tokio::select! {
+            Some(cmd) = rx_cmd.recv() => {
+                tracing::info!(?cmd, "command from a client");
+                if let Err(e) = run(&ctx, cmd).await {
+                    ctx.out.emit(Event::Error { msg: format!("{e:#}") });
+                }
+            }
+            _ = ticker.tick() => {
+                // Per-feed TTL decides what actually gets polled.
+                if let Err(e) = run(&ctx, Cmd::Fetch { feed: None, force: false }).await {
+                    ctx.out.emit(Event::Error { msg: format!("{e:#}") });
+                }
+            }
+            _ = shutdown() => break,
+        }
+    }
+
+    server.abort();
+    let _ = std::fs::remove_file(&socket);
+    tracing::info!("daemon stopped");
+    Ok(())
+}
+
+async fn shutdown() {
+    use tokio::signal::unix::{SignalKind, signal};
+    let mut term = match signal(SignalKind::terminate()) {
+        Ok(s) => s,
+        Err(_) => return std::future::pending().await,
+    };
+    tokio::select! {
+        _ = tokio::signal::ctrl_c() => {}
+        _ = term.recv() => {}
+    }
+}
+
+fn list(ctx: &Ctx, config_path: &std::path::Path) -> Result<()> {
+    if ctx.cfg.feeds.is_empty() {
         println!("No feeds configured in {}", config_path.display());
         return Ok(());
     }
-    for (id, feed) in &cfg.feeds {
-        let s = db.feed_summary(id)?;
+    for (id, feed) in &ctx.cfg.feeds {
+        let s = ctx.db.feed_summary(id)?;
         println!("{id}  {}", s.title.as_deref().unwrap_or("-"));
         println!("    url          {}", feed.url);
         println!("    last checked {}", ago(s.last_checked));
@@ -78,63 +192,98 @@ fn list(cfg: &config::Config, db: &db::Db, config_path: &std::path::Path) -> Res
     Ok(())
 }
 
-async fn fetch(
-    cfg: &config::Config,
-    db: &db::Db,
-    only: Option<&str>,
-    force: bool,
-) -> Result<()> {
-    let client = reqwest::Client::builder()
-        .user_agent(concat!("ipx/", env!("CARGO_PKG_VERSION")))
-        .build()?;
-
-    if let Some(id) = only
-        && !cfg.feeds.contains_key(id)
-    {
-        anyhow::bail!("no feed with id {id:?}");
+/// `standalone` false means this is the sweep that runs before a scan: it reports what it
+/// deleted, but must not emit the terminal ReapDone, or a client waiting on its `fetch`
+/// would stop reading before the scan had even started.
+fn reap(ctx: &Ctx, dry_run: bool, standalone: bool) -> Result<()> {
+    let r = retention::run(&ctx.cfg, &ctx.db, dry_run)?;
+    for c in r.aged_out.iter().chain(r.over_quota.iter()) {
+        ctx.out.emit(Event::Reaped {
+            path: c.path.clone(),
+            bytes: c.bytes.max(0) as u64,
+        });
     }
-
-    for (id, feed_cfg) in cfg.feeds.iter().filter(|(id, _)| only.is_none_or(|o| o == *id)) {
-        let state = db.http_state(id)?;
-
-        // TTL: the feed's own <ttl> wins when it is longer than our poll interval.
-        if !force && let Some(last) = state.last_checked {
-            let wait = state.ttl_mins.unwrap_or(0).max(cfg.general.interval_mins) * 60;
-            let due = last + wait as i64;
-            if due > db::now() {
-                println!("{id}: not due for {}", duration((due - db::now()) as u64));
-                continue;
-            }
-        }
-
-        match scan_one(&client, cfg, db, id, feed_cfg, &state).await {
-            Ok(Some(s)) => println!(
-                "{id}: {} new entries, {} downloaded, {} failed, {} torrents deferred",
-                s.new_entries, s.downloaded, s.failed, s.torrents
-            ),
-            Ok(None) => println!("{id}: not modified"),
-            Err(e) => {
-                // One bad feed must not end the scan.
-                let msg = format!("{e:#}");
-                println!("{id}: error: {msg}");
-                db.set_feed_error(id, &feed_cfg.url, &msg)?;
-            }
-        }
+    if standalone {
+        ctx.out.emit(Event::ReapDone {
+            files: r.aged_out.len() + r.over_quota.len(),
+            bytes: r.bytes_freed,
+        });
     }
     Ok(())
 }
 
+async fn fetch(ctx: &Ctx, only: Option<&str>, force: bool) -> Result<()> {
+    if let Some(id) = only
+        && !ctx.cfg.feeds.contains_key(id)
+    {
+        anyhow::bail!("no feed with id {id:?}");
+    }
+
+    let mut scanned = 0;
+    for (id, feed_cfg) in ctx
+        .cfg
+        .feeds
+        .iter()
+        .filter(|(id, _)| only.is_none_or(|o| o == *id))
+    {
+        let state = ctx.db.http_state(id)?;
+
+        // TTL: the feed's own <ttl> wins when it is longer than our poll interval.
+        if !force && let Some(last) = state.last_checked {
+            let wait = state.ttl_mins.unwrap_or(0).max(ctx.cfg.general.interval_mins) * 60;
+            let due = last + wait as i64;
+            if due > db::now() {
+                ctx.out.emit(Event::FeedSkip {
+                    feed: id.clone(),
+                    reason: format!("not due for {}", duration((due - db::now()) as u64)),
+                });
+                continue;
+            }
+        }
+
+        scanned += 1;
+        ctx.out.emit(Event::FeedStart { feed: id.clone() });
+        match scan_one(ctx, id, feed_cfg, &state).await {
+            Ok(Some(s)) => ctx.out.emit(Event::FeedDone {
+                feed: id.clone(),
+                new: s.new_entries,
+                downloaded: s.downloaded,
+                failed: s.failed,
+                torrents: s.torrents,
+            }),
+            Ok(None) => ctx.out.emit(Event::FeedSkip {
+                feed: id.clone(),
+                reason: "not modified".into(),
+            }),
+            Err(e) => {
+                // One bad feed must not end the scan.
+                let msg = format!("{e:#}");
+                ctx.out.emit(Event::FeedError { feed: id.clone(), msg: msg.clone() });
+                ctx.db.set_feed_error(id, &feed_cfg.url, &msg)?;
+            }
+        }
+    }
+    ctx.out.emit(Event::ScanDone { feeds: scanned });
+    Ok(())
+}
+
+#[derive(Default)]
+struct Scan {
+    new_entries: usize,
+    downloaded: usize,
+    failed: usize,
+    torrents: usize,
+}
+
 /// Ok(None) means 304.
 async fn scan_one(
-    client: &reqwest::Client,
-    cfg: &config::Config,
-    db: &db::Db,
+    ctx: &Ctx,
     id: &str,
     feed_cfg: &config::Feed,
     state: &db::HttpState,
 ) -> Result<Option<Scan>> {
     let fetched = feed::fetch(
-        client,
+        &ctx.client,
         feed_cfg,
         state.etag.as_deref(),
         state.last_modified.as_deref(),
@@ -143,14 +292,14 @@ async fn scan_one(
 
     let (bytes, etag, last_modified) = match fetched {
         feed::Fetched::NotModified => {
-            db.touch_feed(id, &feed_cfg.url)?;
+            ctx.db.touch_feed(id, &feed_cfg.url)?;
             return Ok(None);
         }
         feed::Fetched::Body { bytes, etag, last_modified } => (bytes, etag, last_modified),
     };
 
     let parsed = feed::parse(&bytes)?;
-    db.record_feed(
+    ctx.db.record_feed(
         id,
         &feed_cfg.url,
         parsed.title.as_deref(),
@@ -161,42 +310,55 @@ async fn scan_one(
 
     let mut scan = Scan::default();
     for entry in &parsed.entries {
-        if db.record_entry(id, entry)? {
+        if ctx.db.record_entry(id, entry)? {
             scan.new_entries += 1;
         }
         for enc in &entry.enclosures {
-            if !db.record_enclosure(id, &entry.guid, enc)? {
+            if !ctx.db.record_enclosure(id, &entry.guid, enc)? {
                 continue; // Seen before: downloaded, skipped or deliberately reaped.
             }
             // Filters run once, at discovery, and are recorded in `state`. The download
             // queue below is then just "everything still pending".
             if let Some(reason) = reject(feed_cfg, entry, &enc.url) {
-                db.mark_enclosure(&enc.url, "skipped", Some(reason))?;
+                ctx.db.mark_enclosure(&enc.url, "skipped", Some(reason))?;
             }
         }
     }
 
     let budget = feed_cfg.max_new_per_check.unwrap_or(usize::MAX);
     if feed_cfg.auto_download && budget > 0 {
-        let folder = download::folder_for(cfg, id, feed_cfg, parsed.title.as_deref());
-        let dest_dir = cfg.general.download_dir.join(&folder);
+        let folder = download::folder_for(&ctx.cfg, id, feed_cfg, parsed.title.as_deref());
+        let dest_dir = ctx.cfg.general.download_dir.join(&folder);
 
-        for item in db.pending(id, budget)? {
+        for item in ctx.db.pending(id, budget)? {
             if download::looks_like_torrent(&item.url, item.mime.as_deref()) {
                 // Step 7 owns these; recorded so nothing re-queues them meanwhile.
-                db.mark_enclosure(&item.url, "torrent", None)?;
+                ctx.db.mark_enclosure(&item.url, "torrent", None)?;
+                ctx.out.emit(Event::TorrentDeferred {
+                    feed: id.to_string(),
+                    url: item.url.clone(),
+                });
                 scan.torrents += 1;
                 continue;
             }
-            match fetch_one(client, cfg, db, feed_cfg, &item.url, &dest_dir).await {
-                Ok(path) => {
-                    println!("    saved {}", path.display());
+            match fetch_one(ctx, id, feed_cfg, &item.url, &dest_dir).await {
+                Ok((path, bytes)) => {
+                    ctx.out.emit(Event::DownloadDone {
+                        feed: id.to_string(),
+                        url: item.url.clone(),
+                        path: path.display().to_string(),
+                        bytes,
+                    });
                     scan.downloaded += 1;
                 }
                 Err(e) => {
                     let msg = format!("{e:#}");
-                    println!("    failed {}: {msg}", item.url);
-                    db.mark_enclosure(&item.url, "error", Some(&msg))?;
+                    ctx.out.emit(Event::DownloadError {
+                        feed: id.to_string(),
+                        url: item.url.clone(),
+                        msg: msg.clone(),
+                    });
+                    ctx.db.mark_enclosure(&item.url, "error", Some(&msg))?;
                     scan.failed += 1;
                 }
             }
@@ -206,14 +368,6 @@ async fn scan_one(
     Ok(Some(scan))
 }
 
-#[derive(Default)]
-struct Scan {
-    new_entries: usize,
-    downloaded: usize,
-    failed: usize,
-    torrents: usize,
-}
-
 /// Why this enclosure should not be downloaded, if it should not be.
 fn reject(feed_cfg: &config::Feed, entry: &feed::Entry, url: &str) -> Option<&'static str> {
     if !feed_cfg.auto_download {
@@ -236,22 +390,27 @@ fn reject(feed_cfg: &config::Feed, entry: &feed::Entry, url: &str) -> Option<&'s
 }
 
 async fn fetch_one(
-    client: &reqwest::Client,
-    cfg: &config::Config,
-    db: &db::Db,
+    ctx: &Ctx,
+    feed_id: &str,
     feed_cfg: &config::Feed,
     url: &str,
     dest_dir: &std::path::Path,
-) -> Result<std::path::PathBuf> {
+) -> Result<(PathBuf, u64)> {
     // Throttled to whole percents, as the original's lastDLStepSize guard did.
     let mut last_pct = -1i64;
     let name = download::filename_for(url, None);
-    let got = download::download(client, cfg, feed_cfg, url, |done, total| {
+    let got = download::download(&ctx.client, &ctx.cfg, feed_cfg, url, |done, total| {
         if let Some(t) = total.filter(|t| *t > 0) {
             let pct = (done * 100 / t) as i64;
             if pct > last_pct {
                 last_pct = pct;
-                println!("{}", download::progress_line(&name, done, total));
+                ctx.out.emit(Event::Progress {
+                    feed: feed_id.to_string(),
+                    url: url.to_string(),
+                    file: name.clone(),
+                    done,
+                    total,
+                });
             }
         }
     })
@@ -260,31 +419,13 @@ async fn fetch_one(
     if matches!(got.kind, download::Kind::Torrent) {
         // The MIME lied. Don't file a .torrent as an episode.
         let _ = tokio::fs::remove_file(&got.tmp).await;
-        db.mark_enclosure(url, "torrent", None)?;
+        ctx.db.mark_enclosure(url, "torrent", None)?;
         anyhow::bail!("body is a torrent, deferred to the torrent downloader");
     }
 
     let path = download::place(&got, dest_dir).await?;
-    db.mark_downloaded(url, &path, got.bytes)?;
-    Ok(path)
-}
-
-fn report_reap(r: &retention::Report, verbose: bool) {
-    for c in r.aged_out.iter().chain(r.over_quota.iter()) {
-        println!("reap {} ({:.1} MB)", c.path, c.bytes.max(0) as f64 / 1_048_576.0);
-    }
-    if r.reconciled > 0 {
-        println!("{} row(s) pointed at files that were already gone", r.reconciled);
-    }
-    let total = r.aged_out.len() + r.over_quota.len();
-    if total > 0 || verbose {
-        println!(
-            "reaped {total} file(s), {:.1} MB, {} stale entr{} pruned",
-            r.bytes_freed as f64 / 1_048_576.0,
-            r.entries_pruned,
-            if r.entries_pruned == 1 { "y" } else { "ies" }
-        );
-    }
+    ctx.db.mark_downloaded(url, &path, got.bytes)?;
+    Ok((path, got.bytes))
 }
 
 fn ago(t: Option<i64>) -> String {