rays/ipodderx-rs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Phase 2: web front end

Browse Source 
axum served from inside the daemon so it reads SQLite and the event bus
directly: browse feeds, read show notes, play with seeking, download and
delete files, mark read/flag, and edit feed settings.

Config is now hot-reloadable (Ctx.cfg behind RwLock<Arc<Config>>), so UI
edits apply without a daemon restart. Access is a shared token minted from
/dev/urandom, carried in a cookie because an <audio> element cannot send
headers. Show notes are untrusted feed HTML and are sanitized with ammonia
server-side.

read/flagged finally have a writer, which retention has needed since it
started ordering by them.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RPyeapneuXrCdojsaiXGbe
This commit is contained in:
Ray Slakinski
2026-09-10 00:55:16 +00:00
parent ed47e456d4
commit 74ec6e9281
9 changed files with 1391 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

177
src/main.rs
View File

@@ -5,11 +5,13 @@ mod feed;
mod ipc;
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 tokio::sync::{broadcast, mpsc};
#[derive(Parser)]
@@ -64,24 +66,44 @@ enum Command {
/// Write subscriptions out as OPML
Export { file: PathBuf },
/// Run the scheduler and serve the control socket
Daemon,
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.
struct Ctx {
cfg: config::Config,
db: db::Db,
client: reqwest::Client,
out: Emitter,
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.
torrents: tokio::sync::OnceCell<torrent::Torrents>,
pub torrents: tokio::sync::OnceCell<torrent::Torrents>,
}
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(&self.cfg))
.get_or_try_init(|| torrent::Torrents::new(&cfg))
.await
}
}
@@ -109,7 +131,7 @@ async fn main() -> Result<()> {
Command::Reap { dry_run } => Some(Cmd::Reap { dry_run: *dry_run }),
Command::Status => Some(Cmd::Status),
Command::List
| Command::Daemon
| Command::Daemon { .. }
| Command::Add { .. }
| Command::Rm { .. }
| Command::Import { .. }
@@ -123,7 +145,7 @@ async fn main() -> Result<()> {
}
let ctx = Ctx {
cfg,
cfg: std::sync::RwLock::new(std::sync::Arc::new(cfg)),
db,
client: reqwest::Client::builder()
.user_agent(concat!("ipx/", env!("CARGO_PKG_VERSION")))
@@ -134,7 +156,7 @@ async fn main() -> Result<()> {
match cli.command {
Command::List => list(&ctx, &config_path),
Command::Daemon => daemon(ctx).await,
Command::Daemon { web } => daemon(ctx, config_path, web).await,
Command::Add { url, folder, keywords } => {
add(ctx, &config_path, &url, folder, keywords).await
}
@@ -155,28 +177,29 @@ async fn run(ctx: &Ctx, cmd: Cmd) -> Result<()> {
Cmd::Reap { dry_run } => reap(ctx, dry_run, true),
Cmd::Status => {
let (pending, downloaded) = ctx.db.counts()?;
ctx.out.emit(Event::Status { feeds: ctx.cfg.feeds.len(), pending, downloaded });
ctx.out.emit(Event::Status { feeds: ctx.cfg().feeds.len(), pending, downloaded });
Ok(())
}
}
}
async fn daemon(ctx: Ctx) -> Result<()> {
let socket = ctx.cfg.general.socket.clone();
async fn daemon(ctx: Ctx, config_path: PathBuf, web_addr: Option<String>) -> Result<()> {
let socket = ctx.cfg().general.socket.clone();
if ipc::daemon_is_live(&socket).await {
anyhow::bail!("a daemon is already listening on {}", socket.display());
}
let (events, _) = broadcast::channel(1024);
let (tx_cmd, mut rx_cmd) = mpsc::channel::<Cmd>(64);
let ctx = Ctx { out: Emitter::socket(events.clone(), false), ..ctx };
let ctx = Arc::new(Ctx { out: Emitter::socket(events.clone(), false), ..ctx });
let server = tokio::spawn(ipc::serve(socket.clone(), events, tx_cmd));
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 = ctx.cfg.feeds.len(), "daemon started");
tracing::info!(feeds = ctx.cfg().feeds.len(), "daemon started");
loop {
tokio::select! {
@@ -197,11 +220,61 @@ async fn daemon(ctx: Ctx) -> Result<()> {
}
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()) {
@@ -216,25 +289,27 @@ async fn shutdown() {
/// Subscribes to one feed, naming it from its own title.
async fn add(
mut ctx: Ctx,
ctx: Ctx,
config_path: &std::path::Path,
url: &str,
folder: Option<String>,
keywords: Vec<String>,
) -> Result<()> {
if let Some((id, _)) = ctx.cfg.feeds.iter().find(|(_, f)| f.url == url) {
let mut cfg = (*ctx.cfg()).clone();
if let Some((id, _)) = cfg.feeds.iter().find(|(_, f)| f.url == url) {
anyhow::bail!("already subscribed as {id:?}");
}
let id = add_one(&mut ctx, url, folder, keywords).await?;
ctx.cfg.save(config_path)?;
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.
async fn add_one(
ctx: &mut Ctx,
pub async fn add_one(
ctx: &Ctx,
cfg: &mut config::Config,
url: &str,
folder: Option<String>,
keywords: Vec<String>,
@@ -262,8 +337,8 @@ async fn add_one(
}
};
let id = config::unique_slug(&title, &ctx.cfg.feeds);
ctx.cfg.feeds.insert(id.clone(), probe);
let id = config::unique_slug(&title, &cfg.feeds);
cfg.feeds.insert(id.clone(), probe);
Ok(id)
}
@@ -275,17 +350,19 @@ fn url_stem(url: &str) -> String {
.unwrap_or_else(|| url.to_owned())
}
fn rm(mut ctx: Ctx, config_path: &std::path::Path, feed: &str) -> Result<()> {
if ctx.cfg.feeds.remove(feed).is_none() {
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() {
anyhow::bail!("no feed with id {feed:?}");
}
ctx.cfg.save(config_path)?;
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(mut ctx: Ctx, config_path: &std::path::Path, file: &std::path::Path) -> Result<()> {
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}"))?;
@@ -295,12 +372,12 @@ async fn import(mut ctx: Ctx, config_path: &std::path::Path, file: &std::path::P
let mut added = 0;
for (title, url) in found {
if ctx.cfg.feeds.values().any(|f| f.url == url) {
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, &ctx.cfg.feeds);
ctx.cfg.feeds.insert(
let id = config::unique_slug(&title, &cfg.feeds);
cfg.feeds.insert(
id.clone(),
config::Feed {
url,
@@ -317,7 +394,7 @@ async fn import(mut ctx: Ctx, config_path: &std::path::Path, file: &std::path::P
println!("added {id}");
added += 1;
}
ctx.cfg.save(config_path)?;
cfg.save(config_path)?;
println!("{added} feed(s) imported");
Ok(())
}
@@ -339,7 +416,7 @@ fn export(ctx: &Ctx, file: &std::path::Path) -> Result<()> {
title: Some("ipx subscriptions".into()),
..Default::default()
});
for (id, feed) in &ctx.cfg.feeds {
for (id, feed) in &ctx.cfg().feeds {
let title = ctx
.db
.feed_summary(id)
@@ -350,16 +427,17 @@ fn export(ctx: &Ctx, file: &std::path::Path) -> Result<()> {
}
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());
println!("exported {} feed(s) to {}", ctx.cfg().feeds.len(), file.display());
Ok(())
}
fn list(ctx: &Ctx, config_path: &std::path::Path) -> Result<()> {
if ctx.cfg.feeds.is_empty() {
let cfg = ctx.cfg();
if cfg.feeds.is_empty() {
println!("No feeds configured in {}", config_path.display());
return Ok(());
}
for (id, feed) in &ctx.cfg.feeds {
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);
@@ -376,7 +454,7 @@ fn list(ctx: &Ctx, config_path: &std::path::Path) -> Result<()> {
/// 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)?;
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(),
@@ -393,15 +471,15 @@ fn reap(ctx: &Ctx, dry_run: bool, standalone: bool) -> Result<()> {
}
async fn fetch(ctx: &Ctx, only: Option<&str>, force: bool) -> Result<()> {
let cfg = ctx.cfg();
if let Some(id) = only
&& !ctx.cfg.feeds.contains_key(id)
&& !cfg.feeds.contains_key(id)
{
anyhow::bail!("no feed with id {id:?}");
}
let mut scanned = 0;
for (id, feed_cfg) in ctx
.cfg
for (id, feed_cfg) in cfg
.feeds
.iter()
.filter(|(id, _)| only.is_none_or(|o| o == *id))
@@ -410,7 +488,7 @@ async fn fetch(ctx: &Ctx, only: Option<&str>, force: bool) -> Result<()> {
// TTL: the feed's own <ttl> wins when it is longer than our poll interval.
if !force && let Some(last) = state.last_checked {
let wait = state.ttl_mins.unwrap_or(0).max(ctx.cfg.general.interval_mins) * 60;
let wait = state.ttl_mins.unwrap_or(0).max(cfg.general.interval_mins) * 60;
let due = last + wait as i64;
if due > db::now() {
ctx.out.emit(Event::FeedSkip {
@@ -507,12 +585,13 @@ async fn scan_one(
let budget = feed_cfg.max_new_per_check.unwrap_or(usize::MAX);
if feed_cfg.auto_download && budget > 0 {
let folder = download::folder_for(&ctx.cfg, id, feed_cfg, parsed.title.as_deref());
let dest_dir = ctx.cfg.general.download_dir.join(&folder);
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 {
if !ctx.cfg().torrent.enabled {
ctx.db.mark_enclosure(&item.url, "skipped", Some("torrents disabled"))?;
ctx.out.emit(Event::TorrentDeferred {
feed: id.to_string(),
@@ -603,7 +682,8 @@ async fn fetch_one(
// Throttled to whole percents, as the original's lastDLStepSize guard did.
let mut last_pct = -1i64;
let name = download::filename_for(url, None);
let got = download::download(&ctx.client, &ctx.cfg, feed_cfg, url, |done, total| {
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 {
@@ -624,7 +704,7 @@ async fn fetch_one(
// 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 {
if !ctx.cfg().torrent.enabled {
ctx.db.mark_enclosure(url, "skipped", Some("torrents disabled"))?;
anyhow::bail!("body is a torrent and torrents are disabled");
}
@@ -646,9 +726,10 @@ async fn torrent_one(
) -> Result<(PathBuf, u64)> {
let name = download::filename_for(url, None);
let mut last_pct = -1i64;
let cfg = ctx.cfg();
ctx.torrents()
.await?
.fetch(&ctx.cfg, url, dest_dir, |done, total| {
.fetch(&cfg, url, dest_dir, |done, total| {
if total > 0 {
let pct = (done * 100 / total) as i64;
if pct > last_pct {