rays
5190a29cdb
Deleting entries during a cleanup left each feed's ETag in place, so the rescan got 304s, skipped parsing, and 57 feeds stayed empty until a publisher happened to change something. A 304 while the feed holds zero entries means the validator has outlived the data, so the daemon drops it and asks again. "not a wanted media type" was jargon, and storing it in last_error painted an ordinary filter decision red. It reads "not audio or video" now, and a reason is only shown as an error when the state actually is one. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01AdXho5tTkjFLeUXKbEjKBh
1237 lines
44 KiB
Rust
1237 lines
44 KiB
Rust
mod config;
|
|
mod db;
|
|
mod download;
|
|
mod feed;
|
|
mod ipc;
|
|
mod logbuf;
|
|
mod retention;
|
|
mod torrent;
|
|
mod web;
|
|
|
|
use anyhow::{Context, Result};
|
|
use clap::{Parser, Subcommand};
|
|
use ipc::{Command as Cmd, Emitter, Event};
|
|
use std::path::PathBuf;
|
|
use std::sync::Arc;
|
|
use std::future::Future;
|
|
use tokio::sync::{broadcast, mpsc, watch};
|
|
|
|
#[derive(Parser)]
|
|
#[command(name = "ipx", version, about = "A headless podcatcher")]
|
|
struct Cli {
|
|
/// Config file (default: $XDG_CONFIG_HOME/ipx/config.toml)
|
|
#[arg(long, global = true)]
|
|
config: Option<PathBuf>,
|
|
|
|
/// Do the work here even if a daemon is running
|
|
#[arg(long, global = true)]
|
|
local: bool,
|
|
|
|
#[command(subcommand)]
|
|
command: Command,
|
|
}
|
|
|
|
#[derive(Subcommand)]
|
|
enum Command {
|
|
/// Show configured feeds and their state
|
|
List,
|
|
/// Scan feeds for new entries
|
|
Fetch {
|
|
/// Only this feed id
|
|
feed: Option<String>,
|
|
/// Poll even when the feed is not due yet
|
|
#[arg(long)]
|
|
force: bool,
|
|
},
|
|
/// Delete old or over-quota downloads
|
|
Reap {
|
|
/// Show what would go, delete nothing
|
|
#[arg(long)]
|
|
dry_run: bool,
|
|
},
|
|
/// Counts of feeds, pending and downloaded enclosures
|
|
Status,
|
|
/// Subscribe to a feed
|
|
Add {
|
|
url: String,
|
|
/// Download folder name (default: the feed title)
|
|
#[arg(long)]
|
|
folder: Option<String>,
|
|
/// Only take enclosures matching these keywords
|
|
#[arg(long, value_delimiter = ',')]
|
|
keywords: Vec<String>,
|
|
},
|
|
/// Unsubscribe. Downloads and history are left alone.
|
|
Rm { feed: String },
|
|
/// Add every feed in an OPML file
|
|
Import { file: PathBuf },
|
|
/// Write subscriptions out as OPML
|
|
Export { file: PathBuf },
|
|
/// Run the scheduler and serve the control socket
|
|
Daemon {
|
|
/// Serve the web UI on this address, overriding [web] in the config
|
|
#[arg(long, value_name = "ADDR")]
|
|
web: Option<String>,
|
|
},
|
|
}
|
|
|
|
/// Everything a command needs. One per process.
|
|
pub struct Ctx {
|
|
/// Swapped wholesale when the web UI rewrites config.toml, so a running daemon picks
|
|
/// up feed changes without a restart. Callers take a snapshot; no guard is ever held
|
|
/// across an await.
|
|
pub cfg: std::sync::RwLock<std::sync::Arc<config::Config>>,
|
|
pub db: db::Db,
|
|
pub client: reqwest::Client,
|
|
pub out: Emitter,
|
|
/// Started on first use: a BitTorrent session binds ports and starts a DHT, which is
|
|
/// rude to do for a config that has never seen a torrent.
|
|
pub torrents: tokio::sync::OnceCell<torrent::Torrents>,
|
|
/// Caps how many torrents run at once when they are detached.
|
|
pub torrent_slots: Arc<tokio::sync::Semaphore>,
|
|
/// Needed because a subscribed OPML rewrites the feed list as it syncs.
|
|
pub config_path: PathBuf,
|
|
/// Run torrents off the command worker. A torrent takes minutes to fetch metadata and
|
|
/// then seeds for up to an hour, and the worker is sequential -- inline, one torrent
|
|
/// stops every feed scan, every HTTP download and every status command behind it.
|
|
/// A one-shot CLI run keeps them inline, or the process would exit mid-download.
|
|
pub detach_torrents: bool,
|
|
}
|
|
|
|
impl Ctx {
|
|
pub fn cfg(&self) -> std::sync::Arc<config::Config> {
|
|
self.cfg.read().unwrap().clone()
|
|
}
|
|
|
|
/// Re-reads config.toml into the live snapshot.
|
|
pub fn reload_cfg(&self, path: &std::path::Path) -> Result<()> {
|
|
let fresh = config::Config::load(path)?;
|
|
*self.cfg.write().unwrap() = std::sync::Arc::new(fresh);
|
|
tracing::info!("config reloaded");
|
|
Ok(())
|
|
}
|
|
|
|
async fn torrents(&self) -> Result<&torrent::Torrents> {
|
|
let cfg = self.cfg();
|
|
self.torrents
|
|
.get_or_try_init(|| torrent::Torrents::new(&cfg))
|
|
.await
|
|
}
|
|
}
|
|
|
|
#[tokio::main]
|
|
async fn main() -> Result<()> {
|
|
let cli = Cli::parse();
|
|
// Everything goes to stderr as before, and is mirrored into a ring the UI can read.
|
|
{
|
|
use tracing_subscriber::layer::SubscriberExt;
|
|
use tracing_subscriber::util::SubscriberInitExt;
|
|
use tracing_subscriber::Layer;
|
|
// Two filters, deliberately different. stderr follows IPX_LOG; the in-app buffer
|
|
// keeps debug as well, so the log view can show protocol traffic and routine
|
|
// skips that would be noise on a terminal. IPX_UI_LOG overrides it.
|
|
let stderr_filter = tracing_subscriber::EnvFilter::try_from_env("IPX_LOG")
|
|
.unwrap_or_else(|_| "ipx=info".into());
|
|
let ui_filter = tracing_subscriber::EnvFilter::try_from_env("IPX_UI_LOG")
|
|
.unwrap_or_else(|_| "ipx=debug".into());
|
|
tracing_subscriber::registry()
|
|
.with(
|
|
tracing_subscriber::fmt::layer()
|
|
.with_writer(std::io::stderr)
|
|
.with_filter(stderr_filter),
|
|
)
|
|
.with(logbuf::RingLayer.with_filter(ui_filter))
|
|
.init();
|
|
}
|
|
|
|
let config_path = cli.config.clone().unwrap_or_else(config::config_path);
|
|
let cfg = config::Config::load(&config_path)?;
|
|
let db = db::Db::open(&config::data_dir().join("state.db"))?;
|
|
|
|
// A daemon owns the state; don't have two processes downloading the same thing.
|
|
let wire_cmd = match &cli.command {
|
|
Command::Fetch { feed, force } => {
|
|
Some(Cmd::Fetch { feed: feed.clone(), force: *force })
|
|
}
|
|
Command::Reap { dry_run } => Some(Cmd::Reap { dry_run: *dry_run }),
|
|
Command::Status => Some(Cmd::Status),
|
|
Command::List
|
|
| Command::Daemon { .. }
|
|
| Command::Add { .. }
|
|
| Command::Rm { .. }
|
|
| Command::Import { .. }
|
|
| Command::Export { .. } => None,
|
|
};
|
|
if let Some(cmd) = &wire_cmd
|
|
&& !cli.local
|
|
&& ipc::daemon_is_live(&cfg.general.socket).await
|
|
{
|
|
return ipc::proxy(&cfg.general.socket, cmd).await;
|
|
}
|
|
|
|
let is_daemon = matches!(cli.command, Command::Daemon { .. });
|
|
let (events, _) = broadcast::channel(1024);
|
|
let ctx = Arc::new(Ctx {
|
|
cfg: std::sync::RwLock::new(std::sync::Arc::new(cfg)),
|
|
db,
|
|
client: reqwest::Client::builder()
|
|
.user_agent(concat!("ipx/", env!("CARGO_PKG_VERSION")))
|
|
.build()?,
|
|
out: if is_daemon { Emitter::socket(events.clone(), false) } else { Emitter::terminal() },
|
|
torrents: tokio::sync::OnceCell::new(),
|
|
torrent_slots: Arc::new(tokio::sync::Semaphore::new(2)),
|
|
config_path: config_path.clone(),
|
|
detach_torrents: is_daemon,
|
|
});
|
|
|
|
match cli.command {
|
|
Command::List => list(&ctx, &config_path),
|
|
Command::Daemon { web } => daemon(ctx, config_path, web, events).await,
|
|
Command::Add { url, folder, keywords } => {
|
|
add(&ctx, &config_path, &url, folder, keywords).await
|
|
}
|
|
Command::Rm { feed } => rm(&ctx, &config_path, &feed),
|
|
Command::Import { file } => import(&ctx, &config_path, &file).await,
|
|
Command::Export { file } => export(&ctx, &file),
|
|
_ => run(&ctx, wire_cmd.expect("only List and Daemon have no wire form")).await,
|
|
}
|
|
}
|
|
|
|
async fn run(ctx: &Arc<Ctx>, cmd: Cmd) -> Result<()> {
|
|
match cmd {
|
|
Cmd::Fetch { feed, force } => {
|
|
// Make room before pulling more down, as the original did per download.
|
|
reap(ctx, false, false)?;
|
|
fetch(ctx, feed.as_deref(), force).await
|
|
}
|
|
Cmd::Reap { dry_run } => reap(ctx, dry_run, true),
|
|
Cmd::Download { enclosure } => download_one(ctx, enclosure).await,
|
|
Cmd::Status => {
|
|
let (pending, downloaded) = ctx.db.counts()?;
|
|
let feeds = subscriptions(ctx).map(|s| s.len()).unwrap_or(0);
|
|
ctx.out.emit(Event::Status { feeds, pending, downloaded });
|
|
Ok(())
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn daemon(
|
|
ctx: Arc<Ctx>,
|
|
config_path: PathBuf,
|
|
web_addr: Option<String>,
|
|
events: broadcast::Sender<Event>,
|
|
) -> Result<()> {
|
|
let socket = ctx.cfg().general.socket.clone();
|
|
if ipc::daemon_is_live(&socket).await {
|
|
anyhow::bail!("a daemon is already listening on {}", socket.display());
|
|
}
|
|
|
|
match migrate_opml_children(&ctx) {
|
|
Ok(n) if n > 0 => tracing::info!(count = n, "moved OPML feeds out of config.toml into the database"),
|
|
Ok(_) => {}
|
|
Err(e) => tracing::warn!(error = ?e, "could not tidy OPML feeds out of the config"),
|
|
}
|
|
|
|
match ctx.db.requeue_interrupted() {
|
|
Ok(n) if n > 0 => tracing::info!(count = n, "requeued downloads interrupted by a restart"),
|
|
Ok(_) => {}
|
|
Err(e) => tracing::warn!(error = ?e, "could not requeue interrupted downloads"),
|
|
}
|
|
|
|
let (tx_cmd, mut rx_cmd) = mpsc::channel::<Cmd>(64);
|
|
|
|
let web = start_web(&ctx, &config_path, web_addr, &tx_cmd, &events).await?;
|
|
let server = tokio::spawn(ipc::serve(socket.clone(), events.clone(), tx_cmd));
|
|
|
|
// One command at a time: the queue is what keeps two scans from overlapping.
|
|
let mut ticker = tokio::time::interval(std::time::Duration::from_secs(60));
|
|
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
|
|
tracing::info!(
|
|
feeds = subscriptions(&ctx).map(|s| s.len()).unwrap_or(0),
|
|
"daemon started"
|
|
);
|
|
|
|
// A signal has to be able to interrupt work in progress, not just the wait between
|
|
// jobs. Racing `shutdown()` in the outer select only cancels branch selection: once
|
|
// inside a long download the daemon stopped listening and had to be SIGKILLed.
|
|
let (tx_stop, rx_stop) = tokio::sync::watch::channel(false);
|
|
tokio::spawn(async move {
|
|
shutdown().await;
|
|
let _ = tx_stop.send(true);
|
|
});
|
|
|
|
// Runs one job, abandoning it if a signal arrives. Returns false to end the loop.
|
|
async fn until_stopped(
|
|
ctx: &Ctx,
|
|
rx: &watch::Receiver<bool>,
|
|
job: impl Future<Output = Result<()>>,
|
|
) -> bool {
|
|
let mut stop = rx.clone();
|
|
tokio::select! {
|
|
_ = stop.changed() => {
|
|
tracing::info!("signal received; abandoning the job in progress");
|
|
false
|
|
}
|
|
result = job => {
|
|
if let Err(e) = result {
|
|
ctx.out.emit(Event::Error { msg: format!("{e:#}") });
|
|
}
|
|
true
|
|
}
|
|
}
|
|
}
|
|
|
|
let mut stop = rx_stop.clone();
|
|
loop {
|
|
if *stop.borrow() {
|
|
break;
|
|
}
|
|
tokio::select! {
|
|
biased;
|
|
_ = stop.changed() => break,
|
|
Some(cmd) = rx_cmd.recv() => {
|
|
// Both halves of the protocol are logged under one target so the UI can
|
|
// show the conversation on its own: this is everything arriving, whatever
|
|
// the source -- a socket client, the CLI proxying, or the web UI.
|
|
tracing::info!(
|
|
target: "ipx::io",
|
|
"-> {}",
|
|
serde_json::to_string(&cmd).unwrap_or_else(|_| format!("{cmd:?}"))
|
|
);
|
|
if !until_stopped(&ctx, &rx_stop, run(&ctx, cmd)).await {
|
|
break;
|
|
}
|
|
}
|
|
_ = ticker.tick() => {
|
|
// Per-feed schedule and TTL decide what actually gets polled.
|
|
let job = run(&ctx, Cmd::Fetch { feed: None, force: false });
|
|
if !until_stopped(&ctx, &rx_stop, job).await {
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
server.abort();
|
|
if let Some(w) = web {
|
|
w.abort();
|
|
}
|
|
let _ = std::fs::remove_file(&socket);
|
|
tracing::info!("daemon stopped");
|
|
Ok(())
|
|
}
|
|
|
|
/// Starts the web UI when it is switched on, minting and saving a token if there is none.
|
|
async fn start_web(
|
|
ctx: &Arc<Ctx>,
|
|
config_path: &std::path::Path,
|
|
web_addr: Option<String>,
|
|
cmds: &mpsc::Sender<Cmd>,
|
|
events: &broadcast::Sender<Event>,
|
|
) -> Result<Option<tokio::task::JoinHandle<()>>> {
|
|
let cfg = ctx.cfg();
|
|
let enabled = cfg.web.enabled || web_addr.is_some();
|
|
if !enabled {
|
|
return Ok(None);
|
|
}
|
|
let bind = web_addr.unwrap_or_else(|| cfg.web.bind.clone());
|
|
|
|
if cfg.web.token.is_empty() {
|
|
let mut fresh = (*cfg).clone();
|
|
fresh.web.enabled = true;
|
|
fresh.web.bind = bind.clone();
|
|
fresh.web.token = web::generate_token();
|
|
fresh.save(config_path)?;
|
|
ctx.reload_cfg(config_path)?;
|
|
println!("web ui token generated. Open:\n http://{bind}/?token={}", fresh.web.token);
|
|
} else {
|
|
println!(
|
|
"web ui at http://{bind}/?token={}",
|
|
ctx.cfg().web.token
|
|
);
|
|
}
|
|
|
|
if ctx.cfg().web.binds_publicly() {
|
|
tracing::warn!(bind, "web ui is reachable off this machine; the token is all that guards it");
|
|
}
|
|
|
|
let state = web::WebState {
|
|
ctx: ctx.clone(),
|
|
config_path: config_path.to_path_buf(),
|
|
cmds: cmds.clone(),
|
|
events: events.clone(),
|
|
};
|
|
Ok(Some(tokio::spawn(async move {
|
|
if let Err(e) = web::serve(state, &bind).await {
|
|
tracing::error!(error = ?e, "web ui stopped");
|
|
}
|
|
})))
|
|
}
|
|
|
|
async fn shutdown() {
|
|
use tokio::signal::unix::{SignalKind, signal};
|
|
let mut term = match signal(SignalKind::terminate()) {
|
|
Ok(s) => s,
|
|
Err(_) => return std::future::pending().await,
|
|
};
|
|
tokio::select! {
|
|
_ = tokio::signal::ctrl_c() => {}
|
|
_ = term.recv() => {}
|
|
}
|
|
}
|
|
|
|
/// Subscribes to one feed, naming it from its own title.
|
|
async fn add(
|
|
ctx: &Ctx,
|
|
config_path: &std::path::Path,
|
|
url: &str,
|
|
folder: Option<String>,
|
|
keywords: Vec<String>,
|
|
) -> Result<()> {
|
|
let mut cfg = (*ctx.cfg()).clone();
|
|
// Includes feeds derived from an OPML, or the same show could be added twice.
|
|
if let Some(existing) = subscriptions(ctx)?.iter().find(|s| s.cfg.url == url) {
|
|
anyhow::bail!("already subscribed as {:?}", existing.id);
|
|
}
|
|
let id = add_one(ctx, &mut cfg, url, folder, keywords).await?;
|
|
cfg.save(config_path)?;
|
|
println!("added {id}");
|
|
Ok(())
|
|
}
|
|
|
|
/// Returns the new feed id. The title needs a fetch, so a feed that cannot be reached is
|
|
/// still added -- under a slug derived from its URL -- rather than refused.
|
|
pub async fn add_one(
|
|
ctx: &Ctx,
|
|
cfg: &mut config::Config,
|
|
url: &str,
|
|
folder: Option<String>,
|
|
keywords: Vec<String>,
|
|
) -> Result<String> {
|
|
let probe = config::Feed {
|
|
url: url.to_owned(),
|
|
folder: folder.clone(),
|
|
group: None,
|
|
media_types: None,
|
|
schedule: None,
|
|
keywords: keywords.clone(),
|
|
allow_explicit: false,
|
|
auto_download: true,
|
|
max_new_per_check: None,
|
|
username: None,
|
|
password: None,
|
|
password_env: None,
|
|
};
|
|
|
|
let title = match feed::fetch(&ctx.client, &probe, None, None).await {
|
|
// An OPML subscription is named from its own <head><title>, not by trying to
|
|
// parse it as a feed and falling back to the hostname.
|
|
Ok(feed::Fetched::Body { bytes, .. }) if feed::is_opml(&bytes) => {
|
|
feed::opml_title(&bytes).unwrap_or_else(|| url_stem(url))
|
|
}
|
|
Ok(feed::Fetched::Body { bytes, .. }) => feed::parse(&bytes)
|
|
.ok()
|
|
.and_then(|f| f.title)
|
|
.unwrap_or_else(|| url_stem(url)),
|
|
_ => {
|
|
tracing::warn!(url, "could not read the feed; naming it from its URL");
|
|
url_stem(url)
|
|
}
|
|
};
|
|
|
|
// Slugs must be unique across derived feeds too, or a new feed can collide with one
|
|
// an OPML already introduced.
|
|
let taken: std::collections::BTreeMap<String, config::Feed> = subscriptions(ctx)?
|
|
.into_iter()
|
|
.map(|s| (s.id, s.cfg))
|
|
.collect();
|
|
let id = config::unique_slug(&title, &taken);
|
|
cfg.feeds.insert(id.clone(), probe);
|
|
Ok(id)
|
|
}
|
|
|
|
/// Host plus last path segment, for naming a feed we could not read.
|
|
fn url_stem(url: &str) -> String {
|
|
url::Url::parse(url)
|
|
.ok()
|
|
.and_then(|u| u.host_str().map(str::to_owned))
|
|
.unwrap_or_else(|| url.to_owned())
|
|
}
|
|
|
|
fn rm(ctx: &Ctx, config_path: &std::path::Path, feed: &str) -> Result<()> {
|
|
let mut cfg = (*ctx.cfg()).clone();
|
|
if cfg.feeds.remove(feed).is_none() {
|
|
// Derived from an OPML: drop it here, though the subscription will list it again
|
|
// on the next read unless the OPML itself goes.
|
|
ctx.db.drop_managed(feed)?;
|
|
println!("removed {feed}; it came from an OPML subscription and may return on the next read");
|
|
return Ok(());
|
|
}
|
|
cfg.save(config_path)?;
|
|
// State and files stay: re-adding the feed should not re-download its back catalogue.
|
|
println!("removed {feed}; downloads and history kept");
|
|
Ok(())
|
|
}
|
|
|
|
async fn import(ctx: &Ctx, config_path: &std::path::Path, file: &std::path::Path) -> Result<()> {
|
|
let mut cfg = (*ctx.cfg()).clone();
|
|
let text = std::fs::read_to_string(file)
|
|
.with_context(|| format!("reading {}", file.display()))?;
|
|
let doc = opml::OPML::from_str(&text).map_err(|e| anyhow::anyhow!("parsing OPML: {e}"))?;
|
|
|
|
let mut found = vec![];
|
|
collect_outlines(&doc.body.outlines, &mut found);
|
|
|
|
let mut added = 0;
|
|
for (title, url) in found {
|
|
if cfg.feeds.values().any(|f| f.url == url) {
|
|
continue;
|
|
}
|
|
// Name it from the OPML title rather than refetching every feed.
|
|
let id = config::unique_slug(&title, &cfg.feeds);
|
|
cfg.feeds.insert(
|
|
id.clone(),
|
|
config::Feed {
|
|
url,
|
|
folder: None,
|
|
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,
|
|
},
|
|
);
|
|
println!("added {id}");
|
|
added += 1;
|
|
}
|
|
cfg.save(config_path)?;
|
|
println!("{added} feed(s) imported");
|
|
Ok(())
|
|
}
|
|
|
|
/// OPML nests feeds inside folder outlines, so this walks the whole tree.
|
|
pub fn collect_outlines(outlines: &[opml::Outline], out: &mut Vec<(String, String)>) {
|
|
for o in outlines {
|
|
if let Some(url) = &o.xml_url {
|
|
let title = o.title.clone().unwrap_or_else(|| o.text.clone());
|
|
out.push((title, url.clone()));
|
|
}
|
|
collect_outlines(&o.outlines, out);
|
|
}
|
|
}
|
|
|
|
fn export(ctx: &Ctx, file: &std::path::Path) -> Result<()> {
|
|
let mut doc = opml::OPML::default();
|
|
doc.head = Some(opml::Head {
|
|
title: Some("ipx subscriptions".into()),
|
|
..Default::default()
|
|
});
|
|
for (id, feed) in &ctx.cfg().feeds {
|
|
let title = ctx
|
|
.db
|
|
.feed_summary(id)
|
|
.ok()
|
|
.and_then(|s| s.title)
|
|
.unwrap_or_else(|| id.clone());
|
|
doc.add_feed(&title, &feed.url);
|
|
}
|
|
let xml = doc.to_string().map_err(|e| anyhow::anyhow!("writing OPML: {e}"))?;
|
|
std::fs::write(file, xml).with_context(|| format!("writing {}", file.display()))?;
|
|
println!("exported {} feed(s) to {}", ctx.cfg().feeds.len(), file.display());
|
|
Ok(())
|
|
}
|
|
|
|
fn list(ctx: &Ctx, config_path: &std::path::Path) -> Result<()> {
|
|
let cfg = ctx.cfg();
|
|
if cfg.feeds.is_empty() {
|
|
println!("No feeds configured in {}", config_path.display());
|
|
return Ok(());
|
|
}
|
|
for (id, feed) in &cfg.feeds {
|
|
let s = ctx.db.feed_summary(id)?;
|
|
println!("{id} {}", s.title.as_deref().unwrap_or("-"));
|
|
println!(" url {}", feed.url);
|
|
println!(" last checked {}", ago(s.last_checked));
|
|
println!(" entries {} ({} downloaded)", s.entries, s.downloaded);
|
|
if let Some(err) = &s.last_error {
|
|
println!(" last error {err}");
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// `standalone` false means this is the sweep that runs before a scan: it reports what it
|
|
/// deleted, but must not emit the terminal ReapDone, or a client waiting on its `fetch`
|
|
/// would stop reading before the scan had even started.
|
|
fn reap(ctx: &Ctx, dry_run: bool, standalone: bool) -> Result<()> {
|
|
let r = retention::run(&ctx.cfg(), &ctx.db, dry_run)?;
|
|
for c in r.aged_out.iter().chain(r.over_quota.iter()) {
|
|
ctx.out.emit(Event::Reaped {
|
|
path: c.path.clone(),
|
|
bytes: c.bytes.max(0) as u64,
|
|
});
|
|
}
|
|
if standalone {
|
|
ctx.out.emit(Event::ReapDone {
|
|
files: r.aged_out.len() + r.over_quota.len(),
|
|
bytes: r.bytes_freed,
|
|
});
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
async fn fetch(ctx: &Arc<Ctx>, only: Option<&str>, force: bool) -> Result<()> {
|
|
let cfg = ctx.cfg();
|
|
let subs = subscriptions(ctx)?;
|
|
if let Some(id) = only
|
|
&& !subs.iter().any(|s| s.id == id)
|
|
{
|
|
anyhow::bail!("no feed with id {id:?}");
|
|
}
|
|
|
|
let mut scanned = 0;
|
|
let mut fresh: Vec<String> = vec![];
|
|
for sub in subs.iter().filter(|s| only.is_none_or(|o| o == s.id)) {
|
|
let (id, feed_cfg) = (&sub.id, &sub.cfg);
|
|
let state = ctx.db.http_state(id)?;
|
|
|
|
if !force && let Some(last) = state.last_checked {
|
|
let due = last + due_after(&cfg, feed_cfg, state.ttl_mins) as i64;
|
|
if due > db::now() {
|
|
ctx.out.emit(Event::FeedSkip {
|
|
feed: id.clone(),
|
|
reason: format!("not due for {}", duration((due - db::now()) as u64)),
|
|
});
|
|
continue;
|
|
}
|
|
}
|
|
|
|
scanned += 1;
|
|
ctx.out.emit(Event::FeedStart { feed: id.clone() });
|
|
match scan_one(ctx, id, feed_cfg, &state).await {
|
|
Ok(Outcome::Feed(s)) => ctx.out.emit(Event::FeedDone {
|
|
feed: id.clone(),
|
|
new: s.new_entries,
|
|
downloaded: s.downloaded,
|
|
failed: s.failed,
|
|
torrents: s.torrents,
|
|
}),
|
|
Ok(Outcome::NotModified) => ctx.out.emit(Event::FeedSkip {
|
|
feed: id.clone(),
|
|
reason: "not modified".into(),
|
|
}),
|
|
Ok(Outcome::Opml { added, removed, kept, total }) => {
|
|
ctx.out.emit(Event::FeedSkip {
|
|
feed: id.clone(),
|
|
reason: format!(
|
|
"OPML: {total} feed(s) listed, {} added, {removed} unsubscribed, {kept} kept without a listing",
|
|
added.len()
|
|
),
|
|
});
|
|
// Read them in the same pass, as the original did, rather than making
|
|
// the user wait a whole interval for a newly listed show.
|
|
fresh.extend(added);
|
|
}
|
|
Err(e) => {
|
|
// One bad feed must not end the scan.
|
|
let msg = format!("{e:#}");
|
|
ctx.out.emit(Event::FeedError { feed: id.clone(), msg: msg.clone() });
|
|
ctx.db.set_feed_error(id, &feed_cfg.url, &msg)?;
|
|
}
|
|
}
|
|
}
|
|
// Feeds a subscribed OPML just introduced: scan them now, in this run.
|
|
if !fresh.is_empty() {
|
|
let subs = subscriptions(ctx)?;
|
|
for id in &fresh {
|
|
let Some(feed_cfg) = subs.iter().find(|s| &s.id == id).map(|s| &s.cfg) else {
|
|
continue;
|
|
};
|
|
scanned += 1;
|
|
ctx.out.emit(Event::FeedStart { feed: id.clone() });
|
|
let state = ctx.db.http_state(id)?;
|
|
match scan_one(ctx, id, feed_cfg, &state).await {
|
|
Ok(Outcome::Feed(s)) => ctx.out.emit(Event::FeedDone {
|
|
feed: id.clone(),
|
|
new: s.new_entries,
|
|
downloaded: s.downloaded,
|
|
failed: s.failed,
|
|
torrents: s.torrents,
|
|
}),
|
|
Ok(_) => {}
|
|
Err(e) => {
|
|
let msg = format!("{e:#}");
|
|
ctx.out.emit(Event::FeedError { feed: id.clone(), msg: msg.clone() });
|
|
ctx.db.set_feed_error(id, &feed_cfg.url, &msg)?;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
ctx.out.emit(Event::ScanDone { feeds: scanned });
|
|
Ok(())
|
|
}
|
|
|
|
/// One feed to scan: either an entry you wrote in config.toml, or one derived from an
|
|
/// OPML subscription and held only in the database.
|
|
pub struct Sub {
|
|
pub id: String,
|
|
pub cfg: config::Feed,
|
|
/// True when it came from an OPML and has no config entry of its own.
|
|
pub managed: bool,
|
|
}
|
|
|
|
/// Everything to scan: your config entries, plus whatever the OPML subscriptions listed.
|
|
///
|
|
/// A derived feed borrows its parent's settings wholesale. That is why it needs no config
|
|
/// entry -- there is nothing to store but its URL and where it came from.
|
|
pub fn subscriptions(ctx: &Ctx) -> Result<Vec<Sub>> {
|
|
let cfg = ctx.cfg();
|
|
let mut out: Vec<Sub> = cfg
|
|
.feeds
|
|
.iter()
|
|
.map(|(id, f)| Sub { id: id.clone(), cfg: f.clone(), managed: false })
|
|
.collect();
|
|
|
|
for m in ctx.db.managed_feeds()? {
|
|
if cfg.feeds.contains_key(&m.id) {
|
|
continue; // promoted to config at some point; that entry wins
|
|
}
|
|
let parent = cfg.feeds.get(&m.group_id);
|
|
let base = parent
|
|
.and_then(|p| p.folder.clone())
|
|
.or_else(|| ctx.db.feed_summary(&m.group_id).ok().and_then(|s| s.title))
|
|
.unwrap_or_else(|| m.group_id.clone());
|
|
let title = m.title.clone().unwrap_or_else(|| m.id.clone());
|
|
out.push(Sub {
|
|
id: m.id.clone(),
|
|
cfg: config::Feed {
|
|
url: m.url.clone(),
|
|
folder: Some(format!("{base}/{title}")),
|
|
group: Some(m.group_id.clone()),
|
|
media_types: parent.and_then(|p| p.media_types.clone()),
|
|
schedule: parent.and_then(|p| p.schedule.clone()),
|
|
keywords: parent.map(|p| p.keywords.clone()).unwrap_or_default(),
|
|
allow_explicit: parent.is_some_and(|p| p.allow_explicit),
|
|
auto_download: parent.is_none_or(|p| p.auto_download),
|
|
max_new_per_check: parent.and_then(|p| p.max_new_per_check),
|
|
username: parent.and_then(|p| p.username.clone()),
|
|
password: parent.and_then(|p| p.password.clone()),
|
|
password_env: parent.and_then(|p| p.password_env.clone()),
|
|
},
|
|
managed: true,
|
|
});
|
|
}
|
|
out.sort_by(|a, b| a.id.cmp(&b.id));
|
|
Ok(out)
|
|
}
|
|
|
|
/// Moves OPML children that older versions wrote into config.toml over to the database.
|
|
/// They were never yours to edit, and 80-odd of them made the file unreadable.
|
|
fn migrate_opml_children(ctx: &Ctx) -> Result<usize> {
|
|
let cfg = (*ctx.cfg()).clone();
|
|
let children: Vec<(String, config::Feed)> = cfg
|
|
.feeds
|
|
.iter()
|
|
.filter(|(_, f)| f.group.is_some())
|
|
.map(|(id, f)| (id.clone(), f.clone()))
|
|
.collect();
|
|
if children.is_empty() {
|
|
return Ok(0);
|
|
}
|
|
let mut fresh = cfg.clone();
|
|
for (id, f) in &children {
|
|
let group = f.group.clone().unwrap_or_default();
|
|
let title = ctx
|
|
.db
|
|
.feed_summary(id)
|
|
.ok()
|
|
.and_then(|s| s.title)
|
|
.unwrap_or_else(|| id.clone());
|
|
ctx.db.upsert_managed(id, &f.url, &title, &group)?;
|
|
fresh.feeds.remove(id);
|
|
}
|
|
fresh.save(&ctx.config_path)?;
|
|
ctx.reload_cfg(&ctx.config_path)?;
|
|
Ok(children.len())
|
|
}
|
|
|
|
/// Seconds to wait before re-checking a feed.
|
|
///
|
|
/// A per-feed schedule is an explicit instruction and wins outright. Without one, the
|
|
/// global schedule applies, but the feed's own <ttl> raises it when the publisher asks to
|
|
/// be polled less often.
|
|
pub fn due_after(cfg: &config::Config, feed: &config::Feed, ttl_mins: Option<u64>) -> u64 {
|
|
let mins = match feed.schedule.as_deref().and_then(config::parse_interval) {
|
|
Some(explicit) => explicit,
|
|
None => ttl_mins.unwrap_or(0).max(cfg.general.interval()),
|
|
};
|
|
mins * 60
|
|
}
|
|
|
|
#[derive(Default)]
|
|
struct Scan {
|
|
new_entries: usize,
|
|
downloaded: usize,
|
|
failed: usize,
|
|
torrents: usize,
|
|
}
|
|
|
|
/// What a scan of one feed turned out to be.
|
|
enum Outcome {
|
|
NotModified,
|
|
Feed(Scan),
|
|
/// The URL served an OPML document, so it is a subscription list rather than a feed.
|
|
Opml { added: Vec<String>, removed: usize, kept: usize, total: usize },
|
|
}
|
|
|
|
async fn scan_one(
|
|
ctx: &Arc<Ctx>,
|
|
id: &str,
|
|
feed_cfg: &config::Feed,
|
|
state: &db::HttpState,
|
|
) -> Result<Outcome> {
|
|
let mut fetched = feed::fetch(
|
|
&ctx.client,
|
|
feed_cfg,
|
|
state.etag.as_deref(),
|
|
state.last_modified.as_deref(),
|
|
)
|
|
.await?;
|
|
|
|
// A 304 while nothing is stored means the validator has outlived the data -- a restore
|
|
// from backup, a manual edit, a cleanup that removed entries. Believe the database over
|
|
// the validator: drop it and ask again, or the feed stays empty until the publisher
|
|
// happens to change something.
|
|
if matches!(fetched, feed::Fetched::NotModified) && ctx.db.feed_summary(id)?.entries == 0 {
|
|
tracing::info!(feed = id, "not modified, but nothing stored; refetching without the validator");
|
|
ctx.db.clear_validators(id)?;
|
|
fetched = feed::fetch(&ctx.client, feed_cfg, None, None).await?;
|
|
}
|
|
|
|
let (bytes, etag, last_modified) = match fetched {
|
|
feed::Fetched::NotModified => {
|
|
ctx.db.touch_feed(id, &feed_cfg.url)?;
|
|
return Ok(Outcome::NotModified);
|
|
}
|
|
feed::Fetched::Body { bytes, etag, last_modified } => (bytes, etag, last_modified),
|
|
};
|
|
|
|
// A subscribed OPML is a list of feeds, not a feed. The original matched on a ".opml"
|
|
// URL; sniffing the body also catches one served from a URL without that extension.
|
|
if feed::is_opml(&bytes) {
|
|
ctx.db.touch_feed(id, &feed_cfg.url)?;
|
|
return sync_opml(ctx, id, feed_cfg, &bytes).await;
|
|
}
|
|
|
|
let parsed = feed::parse(&bytes)?;
|
|
ctx.db.record_feed(
|
|
id,
|
|
&feed_cfg.url,
|
|
parsed.title.as_deref(),
|
|
etag.as_deref(),
|
|
last_modified.as_deref(),
|
|
parsed.ttl_mins,
|
|
parsed.image.as_deref(),
|
|
)?;
|
|
|
|
let mut scan = Scan::default();
|
|
for entry in &parsed.entries {
|
|
if ctx.db.record_entry(id, entry)? {
|
|
scan.new_entries += 1;
|
|
}
|
|
for enc in &entry.enclosures {
|
|
if !ctx.db.record_enclosure(id, &entry.guid, enc)? {
|
|
continue; // Seen before: downloaded, skipped or deliberately reaped.
|
|
}
|
|
// Filters run once, at discovery, and are recorded in `state`. The download
|
|
// queue below is then just "everything still pending".
|
|
if let Some(reason) = reject(&ctx.cfg(), feed_cfg, entry, enc) {
|
|
ctx.db.mark_enclosure(&enc.url, "skipped", Some(reason))?;
|
|
}
|
|
}
|
|
}
|
|
|
|
// An unset per-feed cap follows the global one; 0 there means unlimited.
|
|
let budget = feed_cfg.max_new_per_check.unwrap_or_else(|| {
|
|
let g = ctx.cfg().general.max_new_per_check;
|
|
if g == 0 { usize::MAX } else { g }
|
|
});
|
|
if feed_cfg.auto_download && budget > 0 {
|
|
let cfg = ctx.cfg();
|
|
let folder = download::folder_for(&cfg, id, feed_cfg, parsed.title.as_deref());
|
|
let dest_dir = cfg.general.download_dir.join(&folder);
|
|
|
|
for item in ctx.db.pending(id, budget)? {
|
|
if download::looks_like_torrent(&item.url, item.mime.as_deref()) {
|
|
if !ctx.cfg().torrent.enabled {
|
|
ctx.db.mark_enclosure(&item.url, "skipped", Some("torrents disabled"))?;
|
|
ctx.out.emit(Event::TorrentDeferred {
|
|
feed: id.to_string(),
|
|
url: item.url.clone(),
|
|
});
|
|
scan.torrents += 1;
|
|
continue;
|
|
}
|
|
if ctx.detach_torrents {
|
|
// 'downloading' keeps the next scan from queueing it a second time.
|
|
ctx.db.mark_enclosure(&item.url, "downloading", None)?;
|
|
spawn_torrent(ctx, id.to_string(), item.id, item.url.clone(), dest_dir.clone());
|
|
scan.torrents += 1;
|
|
continue;
|
|
}
|
|
match torrent_one(ctx, id, item.id, &item.url, &dest_dir).await {
|
|
Ok((path, bytes)) => {
|
|
ctx.db.mark_downloaded(&item.url, &path, bytes)?;
|
|
ctx.out.emit(Event::DownloadDone {
|
|
feed: id.to_string(),
|
|
enclosure: item.id,
|
|
url: item.url.clone(),
|
|
path: path.display().to_string(),
|
|
bytes,
|
|
});
|
|
scan.downloaded += 1;
|
|
}
|
|
Err(e) => {
|
|
let msg = format!("{e:#}");
|
|
ctx.out.emit(Event::DownloadError {
|
|
feed: id.to_string(),
|
|
enclosure: item.id,
|
|
url: item.url.clone(),
|
|
msg: msg.clone(),
|
|
});
|
|
ctx.db.mark_enclosure(&item.url, "error", Some(&msg))?;
|
|
scan.failed += 1;
|
|
}
|
|
}
|
|
continue;
|
|
}
|
|
match fetch_one(ctx, id, item.id, feed_cfg, &item.url, &dest_dir).await {
|
|
Ok((path, bytes)) => {
|
|
ctx.out.emit(Event::DownloadDone {
|
|
feed: id.to_string(),
|
|
enclosure: item.id,
|
|
url: item.url.clone(),
|
|
path: path.display().to_string(),
|
|
bytes,
|
|
});
|
|
scan.downloaded += 1;
|
|
}
|
|
Err(e) => {
|
|
let msg = format!("{e:#}");
|
|
ctx.out.emit(Event::DownloadError {
|
|
feed: id.to_string(),
|
|
enclosure: item.id,
|
|
url: item.url.clone(),
|
|
msg: msg.clone(),
|
|
});
|
|
ctx.db.mark_enclosure(&item.url, "error", Some(&msg))?;
|
|
scan.failed += 1;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
Ok(Outcome::Feed(scan))
|
|
}
|
|
|
|
/// Brings the feed list in step with a subscribed OPML.
|
|
///
|
|
/// New entries are added under the OPML's group and folder. An entry that has gone from
|
|
/// the OPML is unsubscribed *only if nothing was ever downloaded for it* -- otherwise it
|
|
/// is kept and flagged, because dropping it would orphan files on disk with nothing in
|
|
/// the UI to explain them.
|
|
async fn sync_opml(
|
|
ctx: &Arc<Ctx>,
|
|
parent_id: &str,
|
|
parent: &config::Feed,
|
|
bytes: &[u8],
|
|
) -> Result<Outcome> {
|
|
let listed = feed::parse_opml(bytes)?;
|
|
if let Some(title) = feed::opml_title(bytes) {
|
|
ctx.db.set_title(parent_id, &title)?;
|
|
}
|
|
|
|
let cfg = ctx.cfg();
|
|
let existing = ctx.db.managed_feeds()?;
|
|
let mut added = vec![];
|
|
|
|
for (title, url) in &listed {
|
|
// Already known, whether derived or promoted into the config.
|
|
if let Some(m) = existing.iter().find(|m| &m.url == url) {
|
|
ctx.db.upsert_managed(&m.id, url, title, parent_id)?;
|
|
continue;
|
|
}
|
|
if cfg.feeds.values().any(|f| &f.url == url) {
|
|
continue;
|
|
}
|
|
let taken: std::collections::BTreeMap<String, config::Feed> = cfg
|
|
.feeds
|
|
.keys()
|
|
.chain(existing.iter().map(|m| &m.id))
|
|
.chain(added.iter())
|
|
.map(|id| (id.clone(), parent.clone()))
|
|
.collect();
|
|
let id = config::unique_slug(title, &taken);
|
|
ctx.db.upsert_managed(&id, url, title, parent_id)?;
|
|
added.push(id);
|
|
}
|
|
|
|
// Anything in this group the OPML no longer lists.
|
|
let mut removed = 0;
|
|
let mut kept = 0;
|
|
for m in existing.iter().filter(|m| m.group_id == parent_id) {
|
|
if listed.iter().any(|(_, u)| u == &m.url) {
|
|
continue;
|
|
}
|
|
if ctx.db.downloaded_count(&m.id).unwrap_or(1) > 0 {
|
|
// Never orphan a downloaded file: keep the feed and say why in the UI.
|
|
ctx.db.set_orphaned(&m.id, true)?;
|
|
kept += 1;
|
|
tracing::info!(feed = %m.id, "dropped from the OPML but has downloads; keeping it");
|
|
} else {
|
|
ctx.db.drop_managed(&m.id)?;
|
|
removed += 1;
|
|
tracing::info!(feed = %m.id, "dropped from the OPML with nothing downloaded; removed");
|
|
}
|
|
}
|
|
|
|
Ok(Outcome::Opml { added, removed, kept, total: listed.len() })
|
|
}
|
|
|
|
/// Why this enclosure should not be downloaded, if it should not be.
|
|
fn reject(
|
|
cfg: &config::Config,
|
|
feed_cfg: &config::Feed,
|
|
entry: &feed::Entry,
|
|
enc: &feed::Enclosure,
|
|
) -> Option<&'static str> {
|
|
let url = enc.url.as_str();
|
|
if !feed_cfg.auto_download {
|
|
return Some("auto_download is off");
|
|
}
|
|
// Blog feeds put the article's header image in an <enclosure>; without this a text
|
|
// feed reads as a podcast full of episodes and fills the disk with artwork.
|
|
let wanted = feed_cfg
|
|
.media_types
|
|
.as_deref()
|
|
.unwrap_or(&cfg.general.media_types);
|
|
if !config::wanted_media(enc.mime.as_deref(), wanted) {
|
|
return Some("not audio or video");
|
|
}
|
|
if entry.explicit && !feed_cfg.allow_explicit {
|
|
return Some("explicit");
|
|
}
|
|
let categories = entry.categories.join(" ");
|
|
let haystacks = [
|
|
url,
|
|
entry.title.as_deref().unwrap_or(""),
|
|
entry.description.as_deref().unwrap_or(""),
|
|
categories.as_str(),
|
|
];
|
|
if !download::matches_keywords(&feed_cfg.keywords, &haystacks) {
|
|
return Some("no keyword match");
|
|
}
|
|
None
|
|
}
|
|
|
|
async fn fetch_one(
|
|
ctx: &Arc<Ctx>,
|
|
feed_id: &str,
|
|
enclosure: i64,
|
|
feed_cfg: &config::Feed,
|
|
url: &str,
|
|
dest_dir: &std::path::Path,
|
|
) -> Result<(PathBuf, u64)> {
|
|
// Throttled to whole percents, as the original's lastDLStepSize guard did.
|
|
let mut last_pct = -1i64;
|
|
let name = download::filename_for(url, None);
|
|
let cfg = ctx.cfg();
|
|
let got = download::download(&ctx.client, &cfg, feed_cfg, url, |done, total| {
|
|
if let Some(t) = total.filter(|t| *t > 0) {
|
|
let pct = (done * 100 / t) as i64;
|
|
if pct > last_pct {
|
|
last_pct = pct;
|
|
ctx.out.emit(Event::Progress {
|
|
feed: feed_id.to_string(),
|
|
enclosure,
|
|
url: url.to_string(),
|
|
file: name.clone(),
|
|
done,
|
|
total,
|
|
});
|
|
}
|
|
}
|
|
})
|
|
.await?;
|
|
|
|
if matches!(got.kind, download::Kind::Torrent) {
|
|
// The MIME lied. Hand the URL to the torrent session instead of filing a .torrent
|
|
// as if it were an episode.
|
|
let _ = tokio::fs::remove_file(&got.tmp).await;
|
|
if !ctx.cfg().torrent.enabled {
|
|
ctx.db.mark_enclosure(url, "skipped", Some("torrents disabled"))?;
|
|
anyhow::bail!("body is a torrent and torrents are disabled");
|
|
}
|
|
return torrent_one(ctx, feed_id, enclosure, url, dest_dir).await;
|
|
}
|
|
|
|
let path = download::place(&got, dest_dir).await?;
|
|
ctx.db.mark_downloaded(url, &path, got.bytes)?;
|
|
Ok((path, got.bytes))
|
|
}
|
|
|
|
/// Runs a torrent off the command worker, so feed scans and HTTP downloads keep moving
|
|
/// while it fetches metadata, transfers and then seeds.
|
|
fn spawn_torrent(ctx: &Arc<Ctx>, feed_id: String, enclosure: i64, url: String, dest_dir: PathBuf) {
|
|
let ctx = ctx.clone();
|
|
tokio::spawn(async move {
|
|
// Held for the whole job, so a feed full of torrents cannot open hundreds at once.
|
|
let _permit = match ctx.torrent_slots.clone().acquire_owned().await {
|
|
Ok(p) => p,
|
|
Err(_) => return,
|
|
};
|
|
let outcome = torrent_one(&ctx, &feed_id, enclosure, &url, &dest_dir).await;
|
|
let db = &ctx.db;
|
|
match outcome {
|
|
Ok((path, bytes)) => {
|
|
if let Err(e) = db.mark_downloaded(&url, &path, bytes) {
|
|
tracing::warn!(error = ?e, "could not record the finished torrent");
|
|
}
|
|
ctx.out.emit(Event::DownloadDone {
|
|
feed: feed_id,
|
|
enclosure,
|
|
url,
|
|
path: path.display().to_string(),
|
|
bytes,
|
|
});
|
|
}
|
|
Err(e) => {
|
|
let msg = format!("{e:#}");
|
|
let _ = db.mark_enclosure(&url, "error", Some(&msg));
|
|
ctx.out.emit(Event::DownloadError { feed: feed_id, enclosure, url, msg });
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
/// Downloads one specific enclosure immediately, whatever the per-scan cap says and
|
|
/// wherever it sits in the queue.
|
|
async fn download_one(ctx: &Arc<Ctx>, id: i64) -> Result<()> {
|
|
let cfg = ctx.cfg();
|
|
let enc = ctx
|
|
.db
|
|
.enclosure(id)?
|
|
.ok_or_else(|| anyhow::anyhow!("no enclosure {id}"))?;
|
|
if enc.path.is_some() {
|
|
return Ok(()); // Already here.
|
|
}
|
|
// Must look through the derived feeds too: anything inside an OPML subscription has
|
|
// no config entry, so a config-only lookup called every one of them "unsubscribed".
|
|
let subs = subscriptions(ctx)?;
|
|
let feed_cfg = subs
|
|
.iter()
|
|
.find(|s| s.id == enc.feed_id)
|
|
.map(|s| s.cfg.clone())
|
|
.ok_or_else(|| anyhow::anyhow!("enclosure {id} belongs to unsubscribed feed {:?}", enc.feed_id))?;
|
|
let feed_cfg = &feed_cfg;
|
|
|
|
let title = ctx.db.feed_summary(&enc.feed_id)?.title;
|
|
let folder = download::folder_for(&cfg, &enc.feed_id, feed_cfg, title.as_deref());
|
|
let dest_dir = cfg.general.download_dir.join(&folder);
|
|
|
|
ctx.out.emit(Event::FeedStart { feed: enc.feed_id.clone() });
|
|
let is_torrent = download::looks_like_torrent(&enc.url, enc.mime.as_deref());
|
|
if is_torrent && cfg.torrent.enabled && ctx.detach_torrents {
|
|
ctx.db.mark_enclosure(&enc.url, "downloading", None)?;
|
|
spawn_torrent(ctx, enc.feed_id.clone(), enc.id, enc.url.clone(), dest_dir);
|
|
return Ok(());
|
|
}
|
|
let result = if is_torrent {
|
|
if !cfg.torrent.enabled {
|
|
Err(anyhow::anyhow!("torrents are disabled"))
|
|
} else {
|
|
torrent_one(ctx, &enc.feed_id, enc.id, &enc.url, &dest_dir).await
|
|
}
|
|
} else {
|
|
fetch_one(ctx, &enc.feed_id, enc.id, feed_cfg, &enc.url, &dest_dir).await
|
|
};
|
|
|
|
match result {
|
|
Ok((path, bytes)) => {
|
|
ctx.db.mark_downloaded(&enc.url, &path, bytes)?;
|
|
ctx.out.emit(Event::DownloadDone {
|
|
feed: enc.feed_id.clone(),
|
|
enclosure: enc.id,
|
|
url: enc.url.clone(),
|
|
path: path.display().to_string(),
|
|
bytes,
|
|
});
|
|
}
|
|
Err(e) => {
|
|
let msg = format!("{e:#}");
|
|
ctx.db.mark_enclosure(&enc.url, "error", Some(&msg))?;
|
|
ctx.out.emit(Event::DownloadError {
|
|
feed: enc.feed_id.clone(),
|
|
enclosure: enc.id,
|
|
url: enc.url.clone(),
|
|
msg,
|
|
});
|
|
}
|
|
}
|
|
// Terminal, so a UI waiting on this request stops here.
|
|
ctx.out.emit(Event::ScanDone { feeds: 1 });
|
|
Ok(())
|
|
}
|
|
|
|
/// Torrent progress is reported the same way an HTTP download's is, throttled to whole
|
|
/// percents so a UI is not flooded.
|
|
async fn torrent_one(
|
|
ctx: &Arc<Ctx>,
|
|
feed_id: &str,
|
|
enclosure: i64,
|
|
url: &str,
|
|
dest_dir: &std::path::Path,
|
|
) -> Result<(PathBuf, u64)> {
|
|
let name = download::filename_for(url, None);
|
|
let mut last_pct = -1i64;
|
|
let cfg = ctx.cfg();
|
|
ctx.torrents()
|
|
.await?
|
|
.fetch(&cfg, url, dest_dir, |done, total| {
|
|
if total > 0 {
|
|
let pct = (done * 100 / total) as i64;
|
|
if pct > last_pct {
|
|
last_pct = pct;
|
|
ctx.out.emit(Event::Progress {
|
|
feed: feed_id.to_string(),
|
|
enclosure,
|
|
url: url.to_string(),
|
|
file: name.clone(),
|
|
done,
|
|
total: Some(total),
|
|
});
|
|
}
|
|
}
|
|
})
|
|
.await
|
|
}
|
|
|
|
fn ago(t: Option<i64>) -> String {
|
|
let Some(t) = t else { return "never".into() };
|
|
format!("{} ago", duration((db::now() - t).max(0) as u64))
|
|
}
|
|
|
|
fn duration(secs: u64) -> String {
|
|
match secs {
|
|
s if s < 90 => format!("{s}s"),
|
|
s if s < 5400 => format!("{}m", s / 60),
|
|
s if s < 172_800 => format!("{}h", s / 3600),
|
|
s => format!("{}d", s / 86_400),
|
|
}
|
|
}
|