From b12e0c46ddf037a90aba5913057e69adeab90d42 Mon Sep 17 00:00:00 2001 From: rays Date: Wed, 9 Sep 2026 20:49:55 +0000 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_01RPyeapneuXrCdojsaiXGbe --- PROGRESS.md | 41 +++++- src/db.rs | 11 ++ src/ipc.rs | 277 +++++++++++++++++++++++++++++++++++++++ src/main.rs | 363 ++++++++++++++++++++++++++++++++++++---------------- 4 files changed, 579 insertions(+), 113 deletions(-) create mode 100644 src/ipc.rs diff --git a/PROGRESS.md b/PROGRESS.md index 07181d7..a9006e7 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -12,8 +12,7 @@ The full design and step list live in the plan file at - [x] **3. `feed.rs`** — conditional GET, RSS-then-Atom parse, persist entries. - [x] **4. `download.rs`** — downloads, filters, dedupe. - [x] **5. `retention.rs`** — oldest-first quota + age reaper. -- [ ] **6. `ipc.rs` + daemon** — broadcast event bus, UDS JSON-lines server, TTL scheduler, - CLI-proxies-to-daemon. *Done when:* smoke 3 passes. +- [x] **6. `ipc.rs` + daemon** — UDS JSON-lines server, TTL scheduler, CLI-proxies-to-daemon. - [ ] **7. `torrent.rs`** — librqbit, seed to ratio/time, stall abort. *Done when:* smoke 5 passes. - [ ] **8. OPML + polish** — import/export, add/rm/status, tracing setup, systemd unit, README. @@ -30,6 +29,44 @@ The full design and step list live in the plan file at --- +## 2026-09-09 — Step 6: ipc.rs + daemon + +`src/ipc.rs`: `Event` and `Command` as serde-tagged enums (`{"ev":...}` / `{"cmd":...}`), one JSON +object per line over a Unix socket. `Emitter` is the single output path — it broadcasts to socket +clients, prints the human rendering to a terminal, or both, so the scan code no longer knows how it +is being watched. That replaces `printMSG` and its `;;1;;1;;100.00;;42.31` sentinels. + +`main.rs` restructured around a `Ctx` (config, db, client, emitter). `ipx daemon` binds the socket, +serves clients, and runs a 60s ticker that defers to each feed's TTL. Commands from every client +funnel through one mpsc queue into a single worker, which is what stops two scans overlapping. +Any CLI subcommand with a wire form probes the socket first and proxies to a running daemon; +`--local` forces the work to happen in-process. SIGINT/SIGTERM remove the socket on the way out. + +**Bug the smoke test caught:** `fetch` runs a retention sweep first, and that sweep was emitting the +terminal `ReapDone`. A UI waiting for its fetch to finish would have stopped reading before the scan +started. Only a standalone `ipx reap` emits it now. + +Verified: `cargo test` 24/24. Smoke 3 in full — daemon starts and binds; a raw socket client sending +`{"cmd":"fetch","force":true}` gets `feed_start` → 14 throttled `progress` events → `download_done` +→ `feed_done` → `scan_done`; `{"cmd":"status"}` answers `{"ev":"status","feeds":1,...}`. With the +daemon up, `ipx fetch` from the CLI logged `command from a client cmd=Fetch { .. }` in the daemon +and rendered the streamed events, so it proxied rather than downloading in parallel. Socket removed +on SIGTERM; with no daemon the same command runs locally. + +Deferred: `{"cmd":"cancel","enclosure":N}` from the plan's protocol is **not implemented** — +downloads run sequentially in one worker, so there is nothing to cancel concurrently yet. It wants +a per-download cancellation token, which is worth doing when downloads go parallel. Say if you want +it sooner. + +Note: the event stream is a broadcast, so a CLI client seeing a busy daemon also sees that other +work. Fine for a terminal; a UI wanting strict request/response would want per-request ids. + +Also note: the daemon reads config once at startup — changing `config.toml` needs a restart. + +Next: step 7 — `torrent.rs`. + +--- + ## 2026-09-09 — Step 5: retention.rs `src/retention.rs`: reconcile pass (rows claiming a file that is gone become `reaped`, fixing the diff --git a/src/db.rs b/src/db.rs index 201f7ab..af09d0f 100644 --- a/src/db.rs +++ b/src/db.rs @@ -385,6 +385,17 @@ impl Db { } } +impl Db { + /// (pending, downloaded) across all feeds, for the status command. + pub fn counts(&self) -> Result<(i64, i64)> { + let conn = self.conn.lock().unwrap(); + Ok(( + conn.query_row("SELECT count(*) FROM enclosures WHERE state = 'pending'", [], |r| r.get(0))?, + conn.query_row("SELECT count(*) FROM enclosures WHERE path IS NOT NULL", [], |r| r.get(0))?, + )) + } +} + /// Unix seconds. Everything time-shaped in the DB is stored this way. pub fn now() -> i64 { std::time::SystemTime::now() diff --git a/src/ipc.rs b/src/ipc.rs new file mode 100644 index 0000000..b2c69c8 --- /dev/null +++ b/src/ipc.rs @@ -0,0 +1,277 @@ +//! Unix-socket control and event stream. Replaces printMSG's `;;1;;1;;100.00;;42.31`. + +use anyhow::{Context, Result}; +use serde::{Deserialize, Serialize}; +use std::path::{Path, PathBuf}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::{UnixListener, UnixStream}; +use tokio::sync::{broadcast, mpsc}; + +/// One JSON object per line, `ev` naming the variant. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "ev", rename_all = "snake_case")] +pub enum Event { + FeedStart { feed: String }, + FeedSkip { feed: String, reason: String }, + FeedDone { feed: String, new: usize, downloaded: usize, failed: usize, torrents: usize }, + FeedError { feed: String, msg: String }, + Progress { + feed: String, + url: String, + file: String, + done: u64, + #[serde(skip_serializing_if = "Option::is_none")] + total: Option, + }, + DownloadDone { feed: String, url: String, path: String, bytes: u64 }, + DownloadError { feed: String, url: String, msg: String }, + TorrentDeferred { feed: String, url: String }, + Reaped { path: String, bytes: u64 }, + /// Terminal: a client that asked for work stops reading here. + ScanDone { feeds: usize }, + ReapDone { files: usize, bytes: u64 }, + Status { feeds: usize, pending: i64, downloaded: i64 }, + Error { msg: String }, +} + +impl Event { + pub fn is_terminal(&self) -> bool { + matches!(self, Event::ScanDone { .. } | Event::ReapDone { .. } | Event::Status { .. }) + } + + /// The human rendering, for a terminal rather than a UI. + pub fn human(&self) -> Option { + Some(match self { + Event::FeedSkip { feed, reason } => format!("{feed}: {reason}"), + Event::FeedDone { feed, new, downloaded, failed, torrents } => format!( + "{feed}: {new} new entries, {downloaded} downloaded, {failed} failed, {torrents} torrents deferred" + ), + Event::FeedError { feed, msg } => format!("{feed}: error: {msg}"), + Event::Progress { file, done, total, .. } => match total { + Some(t) if *t > 0 => format!( + " {file}: {:.1}% ({:.1}/{:.1} MB)", + *done as f64 / *t as f64 * 100.0, + *done as f64 / 1_048_576.0, + *t as f64 / 1_048_576.0 + ), + _ => format!(" {file}: {:.1} MB", *done as f64 / 1_048_576.0), + }, + Event::DownloadDone { path, .. } => format!(" saved {path}"), + Event::DownloadError { url, msg, .. } => format!(" failed {url}: {msg}"), + Event::Reaped { path, bytes } => { + format!("reap {path} ({:.1} MB)", *bytes as f64 / 1_048_576.0) + } + Event::ReapDone { files, bytes } => format!( + "reaped {files} file(s), {:.1} MB", + *bytes as f64 / 1_048_576.0 + ), + Event::Status { feeds, pending, downloaded } => { + format!("{feeds} feeds, {pending} pending, {downloaded} downloaded") + } + Event::Error { msg } => format!("error: {msg}"), + // Noise in a terminal; a UI still gets them on the socket. + Event::FeedStart { .. } | Event::TorrentDeferred { .. } | Event::ScanDone { .. } => { + return None; + } + }) + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "cmd", rename_all = "snake_case")] +pub enum Command { + Fetch { + #[serde(default)] + feed: Option, + #[serde(default)] + force: bool, + }, + Reap { + #[serde(default)] + dry_run: bool, + }, + Status, +} + +/// Where events go: the socket, the terminal, or both. +#[derive(Clone)] +pub struct Emitter { + tx: Option>, + print: bool, +} + +impl Emitter { + pub fn terminal() -> Self { + Self { tx: None, print: true } + } + + pub fn socket(tx: broadcast::Sender, print: bool) -> Self { + Self { tx: Some(tx), print } + } + + pub fn emit(&self, e: Event) { + if let Some(tx) = &self.tx { + // An error here only means nobody is listening yet. + let _ = tx.send(e.clone()); + } + if self.print && let Some(line) = e.human() { + println!("{line}"); + } + } +} + +/// True when something is already listening -- i.e. a daemon owns this socket. +pub async fn daemon_is_live(path: &Path) -> bool { + UnixStream::connect(path).await.is_ok() +} + +/// Accepts connections, feeding commands to `cmds` and events from `events` back out. +pub async fn serve( + path: PathBuf, + events: broadcast::Sender, + cmds: mpsc::Sender, +) -> Result<()> { + // A socket file left by a crashed daemon would block the bind; a live one was already + // rejected by the caller's daemon_is_live() check. + if path.exists() { + std::fs::remove_file(&path) + .with_context(|| format!("removing stale socket {}", path.display()))?; + } + if let Some(dir) = path.parent() { + std::fs::create_dir_all(dir)?; + } + let listener = UnixListener::bind(&path) + .with_context(|| format!("binding {}", path.display()))?; + tracing::info!(socket = %path.display(), "listening"); + + loop { + let (stream, _) = listener.accept().await?; + let rx = events.subscribe(); + let cmds = cmds.clone(); + tokio::spawn(async move { + if let Err(e) = handle(stream, rx, cmds).await { + tracing::debug!(error = %e, "client gone"); + } + }); + } +} + +async fn handle( + stream: UnixStream, + mut rx: broadcast::Receiver, + cmds: mpsc::Sender, +) -> Result<()> { + let (read, mut write) = stream.into_split(); + + // Events out. + let writer = tokio::spawn(async move { + while let Ok(ev) = rx.recv().await { + let mut line = serde_json::to_string(&ev).unwrap_or_default(); + line.push('\n'); + if write.write_all(line.as_bytes()).await.is_err() { + break; + } + } + }); + + // Commands in. + let mut lines = BufReader::new(read).lines(); + while let Some(line) = lines.next_line().await? { + let line = line.trim(); + if line.is_empty() { + continue; + } + match serde_json::from_str::(line) { + Ok(cmd) => { + if cmds.send(cmd).await.is_err() { + break; // Worker is gone; so are we. + } + } + Err(e) => tracing::warn!(error = %e, line, "bad command"), + } + } + writer.abort(); + Ok(()) +} + +/// Sends one command to a running daemon and prints the events it produces. +pub async fn proxy(path: &Path, cmd: &Command) -> Result<()> { + let stream = UnixStream::connect(path).await?; + let (read, mut write) = stream.into_split(); + let mut line = serde_json::to_string(cmd)?; + line.push('\n'); + write.write_all(line.as_bytes()).await?; + + // ponytail: the event stream is a broadcast, so a busy daemon's other work shows up + // here too. Fine for a CLI; a UI that cares would want per-request ids. + let mut lines = BufReader::new(read).lines(); + while let Some(line) = lines.next_line().await? { + let Ok(ev) = serde_json::from_str::(&line) else { + continue; + }; + if let Some(text) = ev.human() { + println!("{text}"); + } + if ev.is_terminal() { + break; + } + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn commands_parse_from_the_wire_form() { + let got: Command = serde_json::from_str(r#"{"cmd":"fetch"}"#).unwrap(); + assert!(matches!(got, Command::Fetch { feed: None, force: false })); + + let got: Command = serde_json::from_str(r#"{"cmd":"fetch","feed":"atp","force":true}"#).unwrap(); + assert!(matches!(got, Command::Fetch { feed: Some(f), force: true } if f == "atp")); + + let got: Command = serde_json::from_str(r#"{"cmd":"reap","dry_run":true}"#).unwrap(); + assert!(matches!(got, Command::Reap { dry_run: true })); + + assert!(serde_json::from_str::(r#"{"cmd":"nope"}"#).is_err()); + } + + #[test] + fn events_serialise_to_the_documented_shape() { + let ev = Event::Progress { + feed: "atp".into(), + url: "https://x/ep.mp3".into(), + file: "ep.mp3".into(), + done: 10_485_760, + total: Some(52_428_800), + }; + let json = serde_json::to_string(&ev).unwrap(); + assert!(json.starts_with(r#"{"ev":"progress""#), "got {json}"); + assert!(json.contains(r#""done":10485760"#)); + + // total is omitted rather than null when the server sent no length. + let ev = Event::Progress { + feed: "a".into(), + url: "u".into(), + file: "f".into(), + done: 1, + total: None, + }; + assert!(!serde_json::to_string(&ev).unwrap().contains("total")); + } + + #[test] + fn only_completion_events_end_a_client_session() { + assert!(Event::ScanDone { feeds: 1 }.is_terminal()); + assert!(Event::ReapDone { files: 0, bytes: 0 }.is_terminal()); + assert!(!Event::FeedDone { + feed: "a".into(), + new: 0, + downloaded: 0, + failed: 0, + torrents: 0 + } + .is_terminal()); + } +} diff --git a/src/main.rs b/src/main.rs index fcf8e2a..a621136 100644 --- a/src/main.rs +++ b/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, + /// 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::(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 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 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> { 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 { +) -> 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) -> String {