rays
4f8b3d6a1d
The log was text, so the Grafana dashboard picked lines apart with regular expressions, and a change of wording would have blanked its panels. With IPX_LOG_FORMAT=json each line is one JSON object: the access log carries method, path, route, status and ms as fields (the route passed from the routing layer in the response's extensions), and each wire event its ev, feed, new, downloaded, failed, bytes, msg and the rest (log_wire), beside the old message. The two startup lines that were println! are logged, so no line breaks the JSON. Text stays the default, for a terminal. The dashboard reads the fields with Loki's json parser, and groups requests by route rather than path. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1975 lines
77 KiB
Rust
1975 lines
77 KiB
Rust
mod access;
|
|
mod auth;
|
|
mod config;
|
|
mod db;
|
|
mod entity;
|
|
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,
|
|
/// Copy everything from a SQLite state.db into the database IPX_DATABASE_URL names, which
|
|
/// must be empty: the one-off move to Postgres
|
|
CopyDb {
|
|
/// The SQLite file to copy from
|
|
from: PathBuf,
|
|
},
|
|
/// 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>,
|
|
},
|
|
/// Unsubscribe. Downloads and history are left alone.
|
|
Rm { feed: String },
|
|
/// Add every feed in an OPML file
|
|
Import { file: PathBuf },
|
|
/// Write subscriptions out as OPML
|
|
Export { file: PathBuf },
|
|
/// Add, list or remove the accounts that can sign in to the web UI
|
|
User {
|
|
#[command(subcommand)]
|
|
cmd: UserCmd,
|
|
},
|
|
/// Run the scheduler and serve the control socket
|
|
Daemon {
|
|
/// Serve the web UI on this address, overriding [web] in the config
|
|
#[arg(long, value_name = "ADDR")]
|
|
web: Option<String>,
|
|
},
|
|
}
|
|
|
|
#[derive(Subcommand)]
|
|
enum UserCmd {
|
|
/// Create an account. The password is read from stdin: `echo -n hunter2 | ipx user add ray`
|
|
Add {
|
|
name: String,
|
|
/// May manage other accounts, and is who the shared web token signs in as
|
|
#[arg(long)]
|
|
admin: bool,
|
|
/// Sign-in comes from the proxy instead, so there is no password to set
|
|
#[arg(long)]
|
|
no_password: bool,
|
|
},
|
|
/// Show the accounts and how each one signs in
|
|
List,
|
|
/// Replace a password, read from stdin
|
|
Passwd { name: String },
|
|
/// Delete an account and everything it knows: its subscriptions and read state
|
|
Rm { name: String },
|
|
/// 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 },
|
|
}
|
|
|
|
/// What a brand new database starts with, so there is always a way in. Announced loudly
|
|
/// in the log, and the first thing the settings page nags about.
|
|
const DEFAULT_PASSWORD: &str = "ipodderx";
|
|
|
|
/// Everything a command needs. One per process.
|
|
pub struct Ctx {
|
|
/// Swapped wholesale when the web UI rewrites config.toml, so a running daemon picks
|
|
/// up feed changes without a restart. Callers take a snapshot; no guard is ever held
|
|
/// across an await.
|
|
pub cfg: std::sync::RwLock<std::sync::Arc<config::Config>>,
|
|
pub db: db::Db,
|
|
pub client: reqwest::Client,
|
|
pub out: Emitter,
|
|
/// Started on first use: a BitTorrent session binds ports and starts a DHT, which is
|
|
/// rude to do for a config that has never seen a torrent.
|
|
pub torrents: tokio::sync::OnceCell<torrent::Torrents>,
|
|
/// Caps how many torrents run at once when they are detached.
|
|
pub torrent_slots: Arc<tokio::sync::Semaphore>,
|
|
/// Needed because a subscribed OPML rewrites the feed list as it syncs.
|
|
pub config_path: PathBuf,
|
|
/// Run torrents off the command worker. A torrent takes minutes to fetch metadata and
|
|
/// then seeds for up to an hour, and the worker is sequential -- inline, one torrent
|
|
/// stops every feed scan, every HTTP download and every status command behind it.
|
|
/// A one-shot CLI run keeps them inline, or the process would exit mid-download.
|
|
pub detach_torrents: bool,
|
|
}
|
|
|
|
impl Ctx {
|
|
pub fn cfg(&self) -> std::sync::Arc<config::Config> {
|
|
self.cfg.read().unwrap().clone()
|
|
}
|
|
|
|
/// Re-reads config.toml into the live snapshot.
|
|
/// 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, cfg: config::Config) -> Result<()> {
|
|
self.db.store_config(&cfg).await?;
|
|
self.set_cfg(cfg);
|
|
Ok(())
|
|
}
|
|
|
|
fn set_cfg(&self, cfg: config::Config) {
|
|
*self.cfg.write().unwrap() = std::sync::Arc::new(cfg);
|
|
tracing::info!("config reloaded");
|
|
}
|
|
|
|
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();
|
|
// The daemon only: `ipx status` runs every half minute as the healthcheck, and a trace
|
|
// for each would bury the ones worth looking at.
|
|
let otel = match cli.command {
|
|
Command::Daemon { .. } => otel_provider()?,
|
|
_ => None,
|
|
};
|
|
// 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())
|
|
};
|
|
// 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 = tracing_subscriber::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(std::io::IsTerminal::is_terminal(&std::io::stderr()))
|
|
.with_filter(stderr_filter())
|
|
}))
|
|
.with(json.then(|| {
|
|
tracing_subscriber::fmt::layer()
|
|
.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(tracing_subscriber::EnvFilter::new("ipx=info"))
|
|
}))
|
|
.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(&db::location()).await?;
|
|
|
|
// 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, feeds: vec![] })
|
|
}
|
|
Command::Reap { dry_run } => Some(Cmd::Reap { dry_run: *dry_run }),
|
|
Command::Status => Some(Cmd::Status),
|
|
Command::List
|
|
| Command::Daemon { .. }
|
|
| Command::User { .. }
|
|
| Command::Add { .. }
|
|
| Command::Rm { .. }
|
|
| Command::Import { .. }
|
|
| Command::Export { .. }
|
|
| Command::CopyDb { .. } => 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;
|
|
}
|
|
|
|
// copy-db fills an empty database from another, configuration included; taking config.toml
|
|
// into it first would have the copy collide with it.
|
|
let cfg = if matches!(cli.command, Command::CopyDb { .. }) {
|
|
cfg
|
|
} else {
|
|
assemble_config(&db, cfg, &config_path).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,
|
|
});
|
|
|
|
let result = match cli.command {
|
|
Command::List => list(&ctx).await,
|
|
Command::Daemon { web } => daemon(ctx, config_path, web, events).await,
|
|
Command::Add { url, folder, keywords } => {
|
|
add(&ctx, &url, folder, keywords).await
|
|
}
|
|
Command::Rm { feed } => rm(&ctx, &feed).await,
|
|
Command::User { cmd } => user_cmd(&ctx, cmd).await,
|
|
Command::Import { file } => import(&ctx, &file).await,
|
|
Command::Export { file } => export(&ctx, &file).await,
|
|
Command::CopyDb { from } => copy_db(&ctx, &from).await,
|
|
_ => run(&ctx, wire_cmd.expect("only List and Daemon have no wire form")).await,
|
|
};
|
|
// The batch exporter holds the last few seconds of spans; without this they are lost.
|
|
if let Some(p) = otel {
|
|
let _ = p.shutdown();
|
|
}
|
|
result
|
|
}
|
|
|
|
/// Traces over OTLP, to Tempo for one, when OTEL_EXPORTER_OTLP_ENDPOINT names a collector
|
|
/// (`http://host:4318`: the exporter speaks OTLP over HTTP and adds `/v1/traces`). The exporter
|
|
/// reads that and the other `OTEL_` variables itself.
|
|
fn otel_provider() -> Result<Option<opentelemetry_sdk::trace::SdkTracerProvider>> {
|
|
if std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").unwrap_or_default().is_empty() {
|
|
return Ok(None);
|
|
}
|
|
let exporter = opentelemetry_otlp::SpanExporter::builder().with_http().build()?;
|
|
let resource = opentelemetry_sdk::Resource::builder().with_service_name("ipx").build();
|
|
Ok(Some(
|
|
opentelemetry_sdk::trace::SdkTracerProvider::builder()
|
|
.with_batch_exporter(exporter)
|
|
.with_resource(resource)
|
|
.build(),
|
|
))
|
|
}
|
|
|
|
/// Accounts. Passwords come in on stdin so they never reach a shell history or a `ps`
|
|
/// listing.
|
|
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| chrono::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}");
|
|
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}"))?;
|
|
ctx.db.delete_user(user.id).await?;
|
|
println!("removed {name}");
|
|
Ok(())
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn run(ctx: &Arc<Ctx>, cmd: Cmd) -> Result<()> {
|
|
match cmd {
|
|
Cmd::Fetch { feed, force, feeds } => {
|
|
// Make room before pulling more down, as the original did per download.
|
|
reap(ctx, false, false).await?;
|
|
fetch(ctx, feed.as_deref(), force, &feeds).await
|
|
}
|
|
Cmd::Reap { dry_run } => reap(ctx, dry_run, true).await,
|
|
Cmd::Download { enclosure } => download_one(ctx, enclosure).await,
|
|
Cmd::Status => {
|
|
ctx.out.emit(status(ctx).await);
|
|
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 ctx.db.counts().await {
|
|
Ok((pending, downloaded)) => {
|
|
let feeds = subscriptions(ctx).await.map(|s| s.len()).unwrap_or(0);
|
|
Event::Status { feeds, pending, downloaded }
|
|
}
|
|
Err(e) => Event::Error { msg: format!("{e:#}") },
|
|
}
|
|
}
|
|
|
|
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() {
|
|
ctx.db.create_user("admin", Some(&crate::auth::hash_password(DEFAULT_PASSWORD)?), true).await?;
|
|
tracing::warn!(
|
|
"no accounts yet: created 'admin' with the default password '{DEFAULT_PASSWORD}'. \
|
|
Change it with `echo -n <password> | ipx user passwd admin`"
|
|
);
|
|
}
|
|
|
|
if let Some(admin) = ctx.db.users().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 = ?e, "could not requeue interrupted downloads"),
|
|
}
|
|
|
|
match 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 = ?e, "could not retire feeds whose OPML is no longer in config"),
|
|
}
|
|
|
|
let (tx_cmd, mut rx_cmd) = mpsc::channel::<Cmd>(64);
|
|
|
|
let web = start_web(&ctx, &config_path, web_addr, &tx_cmd, &events).await?;
|
|
// status is answered by the socket itself; everything else waits its turn in the queue.
|
|
let answer: ipc::StatusFn = {
|
|
let ctx = ctx.clone();
|
|
Arc::new(move || {
|
|
let ctx = ctx.clone();
|
|
Box::pin(async move { 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.
|
|
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).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);
|
|
});
|
|
|
|
// 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, 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.
|
|
async fn start_web(
|
|
ctx: &Arc<Ctx>,
|
|
config_path: &std::path::Path,
|
|
web_addr: Option<String>,
|
|
cmds: &mpsc::Sender<Cmd>,
|
|
events: &broadcast::Sender<Event>,
|
|
) -> Result<Option<tokio::task::JoinHandle<()>>> {
|
|
let cfg = ctx.cfg();
|
|
let enabled = cfg.web.enabled || web_addr.is_some();
|
|
if !enabled {
|
|
return Ok(None);
|
|
}
|
|
let bind = web_addr.unwrap_or_else(|| cfg.web.bind.clone());
|
|
|
|
if cfg.web.token.is_empty() {
|
|
let mut fresh = (*cfg).clone();
|
|
fresh.web.enabled = true;
|
|
fresh.web.bind = bind.clone();
|
|
fresh.web.token = crate::auth::new_session_token();
|
|
// 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()
|
|
);
|
|
}
|
|
|
|
if ctx.cfg().web.binds_publicly() {
|
|
tracing::warn!(bind, "web ui is reachable off this machine; the 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 state = web::WebState {
|
|
ctx: ctx.clone(),
|
|
cmds: cmds.clone(),
|
|
events: events.clone(),
|
|
access,
|
|
};
|
|
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,
|
|
url: &str,
|
|
folder: Option<String>,
|
|
keywords: Vec<String>,
|
|
) -> Result<()> {
|
|
let mut cfg = (*ctx.cfg()).clone();
|
|
let url = &feed::feed_behind_page(&ctx.client, &feed::expand_input(url)).await;
|
|
// Includes feeds derived from an OPML, or the same show could be added twice.
|
|
if let Some(existing) = subscriptions(ctx).await?.iter().find(|s| feed::same_feed(&s.cfg.url, url)) {
|
|
anyhow::bail!("already subscribed as {:?}", existing.id);
|
|
}
|
|
let id = add_one(ctx, &mut cfg, url, folder, keywords).await?;
|
|
ctx.store_cfg(cfg).await?;
|
|
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,
|
|
category: None,
|
|
};
|
|
|
|
let title = match feed::fetch(&ctx.client, &probe, None, None).await {
|
|
// An OPML subscription is named from its own <head><title>, not by trying to
|
|
// parse it as a feed and falling back to the hostname.
|
|
Ok(feed::Fetched::Body { bytes, .. }) if feed::is_opml(&bytes) => {
|
|
feed::opml_title(&bytes).unwrap_or_else(|| url_stem(url))
|
|
}
|
|
Ok(feed::Fetched::Body { bytes, .. }) => feed::parse(&bytes)
|
|
.ok()
|
|
.and_then(|f| f.title)
|
|
.unwrap_or_else(|| url_stem(url)),
|
|
_ => {
|
|
tracing::warn!(url, "could not read the feed; naming it from its URL");
|
|
url_stem(url)
|
|
}
|
|
};
|
|
|
|
// Slugs must be unique across derived feeds too, or a new feed can collide with one
|
|
// an OPML already introduced.
|
|
let mut taken: std::collections::BTreeMap<String, config::Feed> = subscriptions(ctx).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, &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())
|
|
}
|
|
|
|
async fn rm(ctx: &Ctx, 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).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).await?;
|
|
Ok(())
|
|
}
|
|
|
|
async fn import(ctx: &Ctx, file: &std::path::Path) -> Result<()> {
|
|
let text = std::fs::read_to_string(file)
|
|
.with_context(|| format!("reading {}", file.display()))?;
|
|
// The CLI speaks for the operator, as the shared web token does.
|
|
let admin = ctx
|
|
.db
|
|
.users().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 = opml::OPML::from_str(&text)
|
|
.map_err(|e| anyhow::anyhow!("{} is not OPML: {e}", file.display()))?;
|
|
let (added, had) = subscribe_opml(ctx, &doc, admin.id).await?;
|
|
println!("subscribed {} to {added} feed(s); {had} already there", admin.name);
|
|
Ok(())
|
|
}
|
|
|
|
/// Subscribes one person to every feed in an OPML document, for the CLI and the web alike.
|
|
/// A feed already in the catalogue costs nothing; an unknown one is added under the OPML's
|
|
/// title rather than refetching each. Returns (newly subscribed, already subscribed).
|
|
///
|
|
/// Before accounts, importing only added unknown URLs to config.toml. Once subscriptions
|
|
/// decided what each person sees, that imported nothing at all for a feed someone else
|
|
/// already had, and a new one had no subscriber, so it was never scanned.
|
|
///
|
|
/// The caller parses the document, so each refuses a file that is not OPML in its own terms,
|
|
/// before anything is touched: a 400 from the web, a message from the CLI.
|
|
pub async fn subscribe_opml(
|
|
ctx: &Ctx,
|
|
doc: &opml::OPML,
|
|
user_id: i64,
|
|
) -> Result<(usize, usize)> {
|
|
let mut found = vec![];
|
|
collect_outlines(&doc.body.outlines, &mut found);
|
|
|
|
let known = subscriptions(ctx).await?;
|
|
let mut cfg = (*ctx.cfg()).clone();
|
|
let mut ids = vec![];
|
|
let mut grew = false;
|
|
for (title, url) in found {
|
|
let existing = known
|
|
.iter()
|
|
.find(|s| s.cfg.url == url)
|
|
.map(|s| s.id.clone())
|
|
// The same URL listed twice in one file.
|
|
.or_else(|| cfg.feeds.iter().find(|(_, f)| f.url == url).map(|(id, _)| id.clone()));
|
|
let id = match existing {
|
|
Some(id) => id,
|
|
None => {
|
|
let id = config::unique_slug(&title, &cfg.feeds);
|
|
cfg.feeds.insert(
|
|
id.clone(),
|
|
config::Feed {
|
|
url,
|
|
folder: None,
|
|
group: None,
|
|
media_types: None,
|
|
schedule: None,
|
|
keywords: vec![],
|
|
allow_explicit: false,
|
|
auto_download: true,
|
|
max_new_per_check: None,
|
|
username: None,
|
|
password: None,
|
|
password_env: None,
|
|
category: None,
|
|
},
|
|
);
|
|
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))
|
|
}
|
|
|
|
/// 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);
|
|
}
|
|
}
|
|
|
|
/// 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.
|
|
async fn assemble_config(db: &db::Db, mut cfg: config::Config, path: &std::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")
|
|
}
|
|
|
|
async fn copy_db(ctx: &Ctx, from: &std::path::Path) -> Result<()> {
|
|
anyhow::ensure!(from.exists(), "{} does not exist", from.display());
|
|
let source = db::Db::open(&from.display().to_string()).await?;
|
|
for (table, n) in ctx.db.copy_from(&source).await? {
|
|
println!("{table:14} {n}");
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
async 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).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(())
|
|
}
|
|
|
|
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?;
|
|
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.
|
|
#[tracing::instrument(name = "reap", skip_all, fields(dry_run = dry_run))]
|
|
async fn reap(ctx: &Ctx, dry_run: bool, standalone: bool) -> Result<()> {
|
|
let r = retention::run(&ctx.cfg(), &ctx.db, dry_run).await?;
|
|
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(())
|
|
}
|
|
|
|
/// `scope`, when not empty, narrows the scan to those feeds and the feeds inside any of them.
|
|
#[tracing::instrument(name = "scan", skip_all, fields(only = ?only, force = force))]
|
|
async fn fetch(ctx: &Arc<Ctx>, only: Option<&str>, force: bool, scope: &[String]) -> Result<()> {
|
|
let cfg = ctx.cfg();
|
|
let subs = subscriptions(ctx).await?;
|
|
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![];
|
|
let in_scope = |s: &Sub| {
|
|
scope.is_empty() || scope.contains(&s.id) || s.cfg.group.as_ref().is_some_and(|g| scope.contains(g))
|
|
};
|
|
for sub in subs.iter().filter(|s| only.is_none_or(|o| o == s.id) && in_scope(s)) {
|
|
let (id, feed_cfg) = (&sub.id, &sub.cfg);
|
|
let mut state = ctx.db.http_state(id).await?;
|
|
// A scan someone asked for reads the feed in full. With the validators it only skipped the
|
|
// wait: a feed that had not changed answered 304 and nothing was read (#77).
|
|
if force {
|
|
state.etag = None;
|
|
state.last_modified = None;
|
|
}
|
|
|
|
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::Empty) => ctx.out.emit(Event::FeedSkip {
|
|
feed: id.clone(),
|
|
reason: "nothing yet".into(),
|
|
}),
|
|
Ok(Outcome::Opml { added, removed, kept, total }) => {
|
|
ctx.out.emit(Event::FeedSkip {
|
|
feed: id.clone(),
|
|
reason: format!(
|
|
"{total} feed(s) listed, {} added, {removed} unsubscribed, {kept} kept without a listing",
|
|
added.len()
|
|
),
|
|
});
|
|
// Read them in the same pass, as the original did, rather than making
|
|
// the user wait a whole interval for a newly listed show.
|
|
fresh.extend(added);
|
|
}
|
|
Err(e) => {
|
|
// One bad feed must not end the scan.
|
|
let msg = format!("{e:#}");
|
|
ctx.out.emit(Event::FeedError { feed: id.clone(), msg: msg.clone() });
|
|
ctx.db.set_feed_error(id, &feed_cfg.url, &msg).await?;
|
|
}
|
|
}
|
|
}
|
|
// Feeds a subscribed OPML just introduced: scan them now, in this run.
|
|
if !fresh.is_empty() {
|
|
let subs = subscriptions(ctx).await?;
|
|
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).await?;
|
|
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).await?;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
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 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();
|
|
|
|
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 => ctx.db.feed_summary(&m.group_id).await.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()),
|
|
category: None,
|
|
},
|
|
managed: true,
|
|
});
|
|
}
|
|
out.sort_by(|a, b| a.id.cmp(&b.id));
|
|
Ok(out)
|
|
}
|
|
|
|
/// 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.
|
|
pub async fn retire_group(ctx: &Ctx, parent_id: &str) -> Result<()> {
|
|
let cfg = ctx.cfg();
|
|
for m in ctx.db.managed_feeds().await?.into_iter().filter(|m| m.group_id == parent_id) {
|
|
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 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(())
|
|
}
|
|
|
|
/// 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.
|
|
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).await?;
|
|
}
|
|
Ok(before.len() - ctx.db.managed_feeds().await?.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,
|
|
/// A response with nothing in it -- the British Antarctic Survey answers a 202 with an
|
|
/// empty body when it has nothing new to publish. Not a parse failure; try again later.
|
|
Empty,
|
|
Feed(Scan),
|
|
/// The URL is a list of feeds rather than a feed: an OPML, or a Patreon creator's shows.
|
|
Opml { added: Vec<String>, removed: usize, kept: usize, total: usize },
|
|
}
|
|
|
|
#[tracing::instrument(name = "feed", skip_all, fields(feed = id))]
|
|
async fn scan_one(
|
|
ctx: &Arc<Ctx>,
|
|
id: &str,
|
|
feed_cfg: &config::Feed,
|
|
state: &db::HttpState,
|
|
) -> Result<Outcome> {
|
|
// A Patreon creator with more than one show is a list of feeds, like an OPML.
|
|
if feed::is_patreon_creator(&feed_cfg.url) {
|
|
match feed::patreon_shows(&ctx.client, &feed_cfg.url).await {
|
|
Ok((name, shows)) if shows.len() > 1 => {
|
|
ctx.db.touch_feed(id, &feed_cfg.url).await?;
|
|
if let Some(name) = name {
|
|
ctx.db.set_title(id, &name).await?;
|
|
}
|
|
// Read as one feed before it was split, it listed every show's items in one
|
|
// heap. The items go; its files and read state move to each show as the show
|
|
// lists them (`Db::adopt`), so no show comes up empty for want of a URL.
|
|
ctx.db.clear_entries(id).await?;
|
|
return sync_group(ctx, id, feed_cfg, &shows).await;
|
|
}
|
|
Ok(_) => {} // One show: the creator's feed is that show.
|
|
// Already split: keep the shows it has rather than read the creator as one heap.
|
|
Err(e) if ctx.db.managed_feeds().await?.iter().any(|m| m.group_id == id) => return Err(e),
|
|
Err(e) => tracing::warn!(
|
|
feed = id,
|
|
error = %format!("{e:#}"),
|
|
"could not list the Patreon shows; reading it as one feed"
|
|
),
|
|
}
|
|
}
|
|
|
|
let mut fetched = feed::fetch(
|
|
&ctx.client,
|
|
feed_cfg,
|
|
state.etag.as_deref(),
|
|
state.last_modified.as_deref(),
|
|
)
|
|
.await?;
|
|
|
|
// A 304 while nothing is stored means the validator has outlived the data -- a restore
|
|
// from backup, a manual edit, a cleanup that removed entries. Believe the database over
|
|
// the validator: drop it and ask again, or the feed stays empty until the publisher
|
|
// happens to change something. The same for artwork never looked for: a feed from before
|
|
// site icons (#73) would otherwise wait for its next post to get one.
|
|
let stored = ctx.db.feed_summary(id).await?;
|
|
if matches!(fetched, feed::Fetched::NotModified) && (stored.entries == 0 || stored.image.is_none()) {
|
|
tracing::info!(feed = id, "not modified, but nothing stored; refetching without the validator");
|
|
ctx.db.clear_validators(id).await?;
|
|
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).await?;
|
|
return Ok(Outcome::NotModified);
|
|
}
|
|
feed::Fetched::Body { bytes, etag, last_modified } => (bytes, etag, last_modified),
|
|
};
|
|
|
|
if bytes.iter().all(u8::is_ascii_whitespace) {
|
|
ctx.db.touch_feed(id, &feed_cfg.url).await?;
|
|
return Ok(Outcome::Empty);
|
|
}
|
|
|
|
// 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).await?;
|
|
return sync_opml(ctx, id, feed_cfg, &bytes).await;
|
|
}
|
|
|
|
let mut parsed = feed::parse(&bytes)?;
|
|
// Looked for again whenever the feed is read in full, which is when it has changed or
|
|
// someone asked for a refresh, so an icon the site changes or fixes follows it (#80). A miss
|
|
// is stored as "", which the page draws as no art, and which stops the refetch above.
|
|
// ponytail: a feed without validators is read in full every scan and asks its site each
|
|
// time too; keep a checked-at time per feed if that shows up in anyone's logs.
|
|
if parsed.image.is_none() {
|
|
parsed.image = Some(match &parsed.site {
|
|
Some(site) => feed::site_icon(&ctx.client, site).await.unwrap_or_default(),
|
|
None => String::new(),
|
|
});
|
|
}
|
|
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(),
|
|
parsed.category.as_deref(),
|
|
).await?;
|
|
|
|
let policy = policy_for(ctx, id, feed_cfg).await?;
|
|
if let Some(parent) = &feed_cfg.group {
|
|
let listed: Vec<(&str, &str)> = parsed
|
|
.entries
|
|
.iter()
|
|
.flat_map(|e| e.enclosures.iter().map(move |x| (e.guid.as_str(), x.url.as_str())))
|
|
.collect();
|
|
ctx.db.adopt(parent, id, &listed).await?;
|
|
}
|
|
// Verdicts are recorded in `state`, so the download queue below is just "everything still
|
|
// pending". A filter's verdict is looked at again on every scan, though: made once, at
|
|
// discovery, it outlived the setting behind it, and allowing explicit items afterwards
|
|
// changed nothing however often the feed was scanned.
|
|
let skipped = ctx.db.skipped_by_filter(id).await?;
|
|
let mut scan = Scan::default();
|
|
for entry in &parsed.entries {
|
|
if ctx.db.record_entry(id, entry).await? {
|
|
scan.new_entries += 1;
|
|
}
|
|
for enc in &entry.enclosures {
|
|
let was = if ctx.db.record_enclosure(id, &entry.guid, enc).await? {
|
|
None
|
|
} else if let Some(reason) = skipped.get(&enc.url) {
|
|
Some(reason.as_str())
|
|
} else {
|
|
continue; // Settled: queued, downloaded, reaped, or another feed's file.
|
|
};
|
|
let now = reject(&ctx.cfg(), feed_cfg, &policy, entry, enc);
|
|
if now != was {
|
|
match now {
|
|
Some(reason) => ctx.db.mark_enclosure(&enc.url, "skipped", Some(reason)).await?,
|
|
None => ctx.db.mark_enclosure(&enc.url, "pending", None).await?,
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
if scan.new_entries > 0 {
|
|
ctx.db.rehide(id).await?; // what is new may hold someone's blocked words
|
|
}
|
|
|
|
let budget = policy.budget;
|
|
if policy.auto_download && budget > 0 {
|
|
let cfg = ctx.cfg();
|
|
let folder = download::folder_for(&cfg, id, feed_cfg, parsed.title.as_deref());
|
|
let dest_dir = cfg.general.download_dir.join(&folder);
|
|
|
|
for item in ctx.db.pending(id, budget).await? {
|
|
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")).await?;
|
|
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).await?;
|
|
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).await?;
|
|
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)).await?;
|
|
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)).await?;
|
|
scan.failed += 1;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
Ok(Outcome::Feed(scan))
|
|
}
|
|
|
|
async fn sync_opml(
|
|
ctx: &Arc<Ctx>,
|
|
parent_id: &str,
|
|
parent: &config::Feed,
|
|
bytes: &[u8],
|
|
) -> Result<Outcome> {
|
|
let listed = feed::parse_opml(bytes)?;
|
|
if let Some(title) = feed::opml_title(bytes) {
|
|
ctx.db.set_title(parent_id, &title).await?;
|
|
}
|
|
sync_group(ctx, parent_id, parent, &listed).await
|
|
}
|
|
|
|
/// Brings the feed list in step with a list of feeds: a subscribed OPML, or a Patreon
|
|
/// creator's shows.
|
|
///
|
|
/// New entries are added under the list's group and folder. An entry that has gone from
|
|
/// the list is unsubscribed *only if nothing was ever downloaded for it* -- otherwise it
|
|
/// is kept and flagged, because dropping it would orphan files on disk with nothing in
|
|
/// the UI to explain them.
|
|
async fn sync_group(
|
|
ctx: &Arc<Ctx>,
|
|
parent_id: &str,
|
|
parent: &config::Feed,
|
|
listed: &[(String, String)],
|
|
) -> Result<Outcome> {
|
|
let cfg = ctx.cfg();
|
|
let existing = ctx.db.managed_feeds().await?;
|
|
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).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 known = ctx.db.feed_urls().await?;
|
|
let taken: std::collections::BTreeMap<String, config::Feed> = cfg
|
|
.feeds
|
|
.keys()
|
|
.chain(existing.iter().map(|m| &m.id))
|
|
.chain(added.iter())
|
|
.chain(known.iter().filter(|(_, u)| !feed::same_feed(u, url)).map(|(id, _)| id))
|
|
.map(|id| (id.clone(), parent.clone()))
|
|
.collect();
|
|
let id = config::unique_slug(title, &taken);
|
|
ctx.db.upsert_managed(&id, url, title, parent_id).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.
|
|
for id in ctx
|
|
.db
|
|
.managed_feeds().await?
|
|
.iter()
|
|
.filter(|m| m.group_id == parent_id)
|
|
.map(|m| m.id.clone())
|
|
.chain(std::iter::once(parent_id.to_string()))
|
|
{
|
|
for user in ctx.db.users().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;
|
|
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).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() })
|
|
}
|
|
|
|
/// Why this enclosure should not be downloaded, if it should not be.
|
|
fn reject(
|
|
cfg: &config::Config,
|
|
feed_cfg: &config::Feed,
|
|
policy: &Policy,
|
|
entry: &feed::Entry,
|
|
enc: &feed::Enclosure,
|
|
) -> Option<&'static str> {
|
|
let url = enc.url.as_str();
|
|
if !policy.auto_download {
|
|
return Some("auto_download is off");
|
|
}
|
|
// Blog feeds put the article's header image in an <enclosure>; without this a text
|
|
// feed reads as a podcast full of episodes and fills the disk with artwork.
|
|
let wanted = feed_cfg
|
|
.media_types
|
|
.as_deref()
|
|
.unwrap_or(&cfg.general.media_types);
|
|
if !config::wanted_media(enc.mime.as_deref(), wanted) {
|
|
return Some("not audio or video");
|
|
}
|
|
if entry.explicit && !policy.allow_explicit {
|
|
return Some("explicit");
|
|
}
|
|
let categories = entry.categories.join(" ");
|
|
let haystacks = [
|
|
url,
|
|
entry.title.as_deref().unwrap_or(""),
|
|
entry.description.as_deref().unwrap_or(""),
|
|
categories.as_str(),
|
|
];
|
|
let text = [entry.title.as_deref().unwrap_or(""), entry.description.as_deref().unwrap_or("")];
|
|
if !policy.wanted(&haystacks, &text) {
|
|
// Worded for whichever filter could have let it through, the keywords if none blocked it.
|
|
return Some(if policy.wanted(&haystacks, &[]) { "blocked word" } else { "no keyword match" });
|
|
}
|
|
None
|
|
}
|
|
|
|
/// What the scanner should do for a feed, merged across everyone subscribed to it. The
|
|
/// feed is fetched once and its files are downloaded once, so the merge is a union: if
|
|
/// one person wants a thing, it is fetched, and everyone else simply sees it listed.
|
|
///
|
|
/// With no subscribers at all -- a hand-written config entry nobody has claimed yet --
|
|
/// the feed's own settings stand, which is how a single-user install behaves.
|
|
pub struct Policy {
|
|
pub auto_download: bool,
|
|
pub allow_explicit: bool,
|
|
/// What each subscriber fetching the feed wants.
|
|
pub wants: Vec<Want>,
|
|
pub budget: usize,
|
|
}
|
|
|
|
/// One person's filters: an item is theirs if it matches their keywords (all of it, with none)
|
|
/// and none of their blocked words.
|
|
pub struct Want {
|
|
pub keywords: Vec<String>,
|
|
pub blocked: Vec<String>,
|
|
}
|
|
|
|
impl Policy {
|
|
/// One file serves everyone subscribed, so an item is wanted if anyone wants it. Keywords
|
|
/// look at `haystacks`, blocked words at `text`, the item's title and body only.
|
|
fn wanted(&self, haystacks: &[&str], text: &[&str]) -> bool {
|
|
self.wants.iter().any(|w| {
|
|
download::matches_keywords(&w.keywords, haystacks) && !download::blocked(&w.blocked, text)
|
|
})
|
|
}
|
|
}
|
|
|
|
async fn policy_for(ctx: &Ctx, id: &str, feed_cfg: &config::Feed) -> Result<Policy> {
|
|
let global = ctx.cfg().general.max_new_per_check;
|
|
Ok(merge_policy(&ctx.db.subscribers(id, feed_cfg.group.as_deref()).await?, feed_cfg, global))
|
|
}
|
|
|
|
fn merge_policy(subs: &[db::Sub], feed_cfg: &config::Feed, global: usize) -> Policy {
|
|
let cap = |n: Option<usize>| n.unwrap_or(if global == 0 { usize::MAX } else { global });
|
|
|
|
if subs.is_empty() {
|
|
return Policy {
|
|
auto_download: feed_cfg.auto_download,
|
|
allow_explicit: feed_cfg.allow_explicit,
|
|
wants: vec![Want { keywords: feed_cfg.keywords.clone(), blocked: vec![] }],
|
|
budget: cap(feed_cfg.max_new_per_check),
|
|
};
|
|
}
|
|
|
|
let mut policy = Policy {
|
|
auto_download: false,
|
|
allow_explicit: false,
|
|
wants: vec![],
|
|
budget: 0,
|
|
};
|
|
for sub in subs {
|
|
if !sub.auto_download.unwrap_or(feed_cfg.auto_download) {
|
|
continue; // Not fetching for this person, so their wants add nothing.
|
|
}
|
|
policy.auto_download = true;
|
|
policy.allow_explicit |= sub.allow_explicit.unwrap_or(feed_cfg.allow_explicit);
|
|
policy.budget = policy
|
|
.budget
|
|
.max(cap(sub.max_new_per_check.map(|n| n as usize).or(feed_cfg.max_new_per_check)));
|
|
policy.wants.push(Want {
|
|
keywords: sub.keywords.clone().unwrap_or_else(|| feed_cfg.keywords.clone()),
|
|
blocked: sub.blocked.clone(),
|
|
});
|
|
}
|
|
policy
|
|
}
|
|
|
|
#[tracing::instrument(name = "download", skip_all, fields(feed = feed_id, enclosure = enclosure, url = url))]
|
|
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")).await?;
|
|
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).await?;
|
|
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).await {
|
|
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)).await;
|
|
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.
|
|
#[tracing::instrument(skip(ctx))]
|
|
async fn download_one(ctx: &Arc<Ctx>, id: i64) -> Result<()> {
|
|
let cfg = ctx.cfg();
|
|
let enc = ctx
|
|
.db
|
|
.enclosure(id).await?
|
|
.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).await?;
|
|
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).await?.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).await?;
|
|
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).await?;
|
|
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)).await?;
|
|
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.
|
|
#[tracing::instrument(name = "torrent", skip_all, fields(feed = feed_id, enclosure = enclosure))]
|
|
async fn torrent_one(
|
|
ctx: &Arc<Ctx>,
|
|
feed_id: &str,
|
|
enclosure: i64,
|
|
url: &str,
|
|
dest_dir: &std::path::Path,
|
|
) -> Result<(PathBuf, u64)> {
|
|
let name = download::filename_for(url, None);
|
|
let mut last_pct = -1i64;
|
|
let cfg = ctx.cfg();
|
|
ctx.torrents()
|
|
.await?
|
|
.fetch(&cfg, url, dest_dir, |done, total| {
|
|
if total > 0 {
|
|
let pct = (done * 100 / total) as i64;
|
|
if pct > last_pct {
|
|
last_pct = pct;
|
|
ctx.out.emit(Event::Progress {
|
|
feed: feed_id.to_string(),
|
|
enclosure,
|
|
url: url.to_string(),
|
|
file: name.clone(),
|
|
done,
|
|
total: Some(total),
|
|
});
|
|
}
|
|
}
|
|
})
|
|
.await
|
|
}
|
|
|
|
fn ago(t: Option<i64>) -> String {
|
|
let Some(t) = t else { return "never".into() };
|
|
format!("{} ago", duration((db::now() - t).max(0) as u64))
|
|
}
|
|
|
|
fn duration(secs: u64) -> String {
|
|
match secs {
|
|
s if s < 90 => format!("{s}s"),
|
|
s if s < 5400 => format!("{}m", s / 60),
|
|
s if s < 172_800 => format!("{}h", s / 3600),
|
|
s => format!("{}d", s / 86_400),
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[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();
|
|
}
|
|
|
|
fn feed() -> config::Feed {
|
|
// Whatever `ipx add` would write, which is the shape every code path sees.
|
|
let mut cfg = config::Config::default();
|
|
let f = add_one_cfg(&mut cfg, "http://x/f.xml", None, vec![]);
|
|
f
|
|
}
|
|
|
|
/// The feed entry `add` builds, without the network round trip it does for a title.
|
|
fn add_one_cfg(
|
|
_cfg: &mut config::Config,
|
|
url: &str,
|
|
folder: Option<String>,
|
|
keywords: Vec<String>,
|
|
) -> config::Feed {
|
|
config::Feed {
|
|
url: url.into(),
|
|
folder,
|
|
keywords,
|
|
allow_explicit: false,
|
|
auto_download: true,
|
|
group: None,
|
|
media_types: None,
|
|
schedule: None,
|
|
max_new_per_check: None,
|
|
username: None,
|
|
password: None,
|
|
password_env: None,
|
|
category: None,
|
|
}
|
|
}
|
|
|
|
fn sub(kw: Option<&[&str]>, auto: Option<bool>, max: Option<i64>) -> db::Sub {
|
|
db::Sub {
|
|
feed_id: "f".into(),
|
|
keywords: kw.map(|k| k.iter().map(|s| s.to_string()).collect()),
|
|
auto_download: auto,
|
|
allow_explicit: None,
|
|
max_new_per_check: max,
|
|
blocked: vec![],
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_shared_feed_is_fetched_for_whoever_wants_the_most() {
|
|
// Nobody subscribed: the feed's own settings stand, as in a single-user install.
|
|
let p = merge_policy(&[], &feed(), 3);
|
|
assert!(p.auto_download);
|
|
assert_eq!(p.budget, 3);
|
|
assert!(p.wanted(&["anything"], &[]));
|
|
|
|
// Two filters: an item wanted by either of them is fetched, since one file serves
|
|
// both. The larger per-scan cap wins for the same reason.
|
|
let p = merge_policy(
|
|
&[sub(Some(&["rust"]), None, Some(2)), sub(Some(&["sqlite"]), None, Some(9))],
|
|
&feed(),
|
|
3,
|
|
);
|
|
assert!(p.wanted(&["rust"], &[]) && p.wanted(&["sqlite"], &[]));
|
|
assert!(!p.wanted(&["python"], &[]));
|
|
assert_eq!(p.budget, 9);
|
|
|
|
// One person taking everything removes the filter for the shared copy.
|
|
let p = merge_policy(&[sub(Some(&["rust"]), None, None), sub(Some(&[]), None, None)], &feed(), 3);
|
|
assert!(p.wanted(&["python"], &[]));
|
|
|
|
// A blocked word keeps an item from being fetched for that person, not for the rest.
|
|
let blocking = |w: &str| db::Sub { blocked: vec![w.into()], ..sub(None, None, None) };
|
|
let p = merge_policy(&[blocking("politics")], &feed(), 3);
|
|
assert!(!p.wanted(&[], &["Politics today"]));
|
|
assert!(p.wanted(&[], &["Cooking today"]));
|
|
let p = merge_policy(&[blocking("politics"), sub(None, None, None)], &feed(), 3);
|
|
assert!(p.wanted(&[], &["Politics today"]), "someone else still wants it");
|
|
|
|
// Everyone has auto-download off: nothing is fetched automatically.
|
|
let p = merge_policy(&[sub(None, Some(false), None), sub(None, Some(false), None)], &feed(), 3);
|
|
assert!(!p.auto_download);
|
|
|
|
// One of them wants it, so it is fetched.
|
|
let p = merge_policy(&[sub(None, Some(false), None), sub(None, Some(true), None)], &feed(), 3);
|
|
assert!(p.auto_download);
|
|
}
|
|
|
|
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(),
|
|
out: Emitter::terminal(),
|
|
torrents: tokio::sync::OnceCell::new(),
|
|
torrent_slots: Arc::new(tokio::sync::Semaphore::new(2)),
|
|
config_path: PathBuf::new(),
|
|
detach_torrents: false,
|
|
}
|
|
}
|
|
|
|
#[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").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");
|
|
}
|
|
}
|