Files
ipx/crates/ipx/src/lists.rs
rays 4f4fd4c3fe ipx's main.rs splits into leaf modules (#356)
main.rs held 3,200 lines: the command line, the shared context, the
daemon, scans, downloads, the catalogue and OPML, and their tests. The
style guide keeps a root module to declarations and a little glue, the
code in leaf modules named for what they hold:

  cli.rs        the commands, run and user_cmd, ago and duration
  ctx.rs        Ctx and CfgEdit, and Ctx::new, which main() built inline
  daemon.rs     the daemon's loop, web server, shutdown, status, reload
  scan.rs       due_at/due_after, scan_one, policy, downloads, torrents
  catalogue.rs  add/rm/leave/remove_user, subscriptions, retiring lists
  lists.rs      OPML import and syncing an OPML's or a creator's feeds
                (not opml.rs, which would shadow the opml crate)
  logging.rs    logging::init and the JSON lines with their trace
  test_util.rs  the context, feeds and feed server tests share

first_password moves to auth.rs and reap to retention.rs. Each test went
with what it tests; all 151 still run. main.rs is the module list and a
main() that hands the command on, 113 lines.

Moved verbatim, then made pub(crate) where a sibling calls it, with
calls across modules written module::function as the guide asks.
Everything the old main.rs defined is defined once in the new files.

With Ctx no longer a pub type at the crate root, Db's pub methods stop
counting as public API, and the compiler sees what nothing outside tests
calls: entries_in, count_in, unread_count and feed_description are now
#[cfg(test)], and media_feeds, called by nothing, is gone.

ipx depends on opentelemetry_sdk directly now (it was already in the
lock through ipx-directory), for logging::init's signature.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-24 01:35:24 +00:00

271 lines
12 KiB
Rust

//! OPML: importing a file's feeds, and keeping a subscribed list's feeds in step with it.
use crate::ctx::Ctx;
use crate::scan::Outcome;
use crate::{catalogue, config, feed};
use anyhow::Result;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
/// The feeds an OPML document lists, up to as many as a subscribed list may bring in: each becomes
/// a feed in the catalogue, scanned for everyone, and one upload of tens of thousands held the
/// catalogue for minutes and stayed in it (#338).
pub fn opml_feeds(doc: &opml::OPML) -> Result<Vec<(String, String)>> {
let mut found = vec![];
feed::collect_outlines(&doc.body.outlines, &mut found);
anyhow::ensure!(
found.len() <= MAX_LISTED,
"it lists {} feeds, more than the {MAX_LISTED} one may have",
found.len()
);
Ok(found)
}
/// 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 and lists its feeds (`opml_feeds`), so each refuses a file 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, found: Vec<(String, String)>, user_id: i64) -> Result<(usize, usize)> {
let mut cfg = ctx.edit_cfg().await;
// By address, once: each feed looked through every subscription and the whole catalogue, and
// the catalogue's lock is held throughout (#338).
let mut known = HashMap::new();
for s in catalogue::subscriptions(ctx).await? {
// The first with an address wins, as a search in order found it.
known.entry(s.cfg.url).or_insert(s.id);
}
let mut ids = vec![];
let mut grew = false;
for (title, url) in found {
// Known already, or the same URL listed twice in one file.
let existing = known.get(&url).cloned();
let id = match existing {
Some(id) => id,
None => {
let id = config::unique_slug(&title, |s| cfg.feeds.contains_key(s));
known.insert(url.clone(), id.clone());
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,
listed: false,
private: false,
},
);
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))
}
pub(crate) async fn sync_opml(ctx: &Arc<Ctx>, parent_id: &str, 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, &listed).await
}
/// How many feeds an OPML or Patreon list may bring in.
pub(crate) const MAX_LISTED: usize = 2000;
/// 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.
pub(crate) async fn sync_group(ctx: &Arc<Ctx>, parent_id: &str, listed: &[(String, String)]) -> Result<Outcome> {
// Each is a feed scanned for everyone subscribed to the list, and the list is its author's to
// write: one of tens of thousands held the scan for hours (#337). davewiner's had 922.
anyhow::ensure!(
listed.len() <= MAX_LISTED,
"it lists {} feeds, more than the {MAX_LISTED} a list may have",
listed.len()
);
let cfg = ctx.cfg();
let existing = ctx.db.managed_feeds().await?;
let mut added = vec![];
// Read once, not once a new feed: each read is the whole feeds table.
let known = ctx.db.feed_urls().await?;
let fixed: HashSet<&str> = cfg.feeds.keys().chain(existing.iter().map(|m| &m.id)).map(String::as_str).collect();
let known_ids: HashSet<&str> = known.iter().map(|(id, _)| id.as_str()).collect();
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 its_own: Vec<&str> =
known.iter().filter(|(_, u)| feed::same_feed(u, url)).map(|(id, _)| id.as_str()).collect();
let id = config::unique_slug(title, |s| {
fixed.contains(s) || added.iter().any(|a| a == s) || (known_ids.contains(s) && !its_own.contains(&s))
});
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.
let managed = ctx.db.managed_feeds().await?;
let list_and_feeds = managed
.iter()
.filter(|m| m.group_id == parent_id)
.map(|m| m.id.clone())
.chain(std::iter::once(parent_id.to_string()));
for id in list_and_feeds {
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;
let parent = ctx.cfg().feeds.get(parent_id).cloned();
let in_group = existing.iter().filter(|m| m.group_id == parent_id);
for m in in_group {
if listed.iter().any(|(_, u)| u == &m.url) {
continue;
}
if catalogue::promote_kept(ctx, m, parent.as_ref()).await? {
kept += 1;
tracing::info!(feed = %m.id, "dropped from the OPML but someone kept it; a feed of its own now");
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() })
}
#[cfg(test)]
mod tests {
use super::*;
use crate::test_util::{feed, test_ctx};
#[tokio::test]
async fn an_opml_upload_is_capped_and_takes_a_repeated_address_once() {
let opml = |n: usize| {
let outlines: String =
(0..n).map(|i| format!(r#"<outline text="Same" xmlUrl="http://x/{}.xml"/>"#, i % 3)).collect();
feed::read_opml(&format!(r#"<opml version="2.0"><head/><body>{outlines}</body></opml>"#)).unwrap()
};
assert!(opml_feeds(&opml(MAX_LISTED + 1)).unwrap_err().to_string().contains("more than the 2000"));
let ctx = test_ctx(config::Config::default()).await;
let me = ctx.db.create_user("me", None, false).await.unwrap();
// Six outlines, three addresses, one title: three feeds, named apart.
let (added, had) = subscribe_opml(&ctx, opml_feeds(&opml(6)).unwrap(), me).await.unwrap();
let ids: std::collections::BTreeSet<String> = ctx.cfg().feeds.keys().cloned().collect();
assert_eq!(ids, ["same", "same-2", "same-3"].map(String::from).into());
assert_eq!((added, had), (3, 3));
}
#[tokio::test]
async fn a_list_of_more_feeds_than_the_limit_is_refused() {
let ctx = Arc::new(test_ctx(config::Config::default()).await);
let listed: Vec<(String, String)> =
(0..=MAX_LISTED).map(|n| (format!("Show {n}"), format!("http://x/{n}.xml"))).collect();
let err = sync_group(&ctx, "opml", &listed).await.err().expect("refused");
assert!(err.to_string().contains("more than the 2000"), "{err:#}");
assert!(ctx.db.managed_feeds().await.unwrap().is_empty(), "nothing taken from it");
// At the limit, all of them, each its own id.
sync_group(&ctx, "opml", &listed[..MAX_LISTED]).await.unwrap();
let ids: std::collections::HashSet<String> =
ctx.db.managed_feeds().await.unwrap().into_iter().map(|m| m.id).collect();
assert_eq!(ids.len(), MAX_LISTED);
}
#[tokio::test]
async fn a_kept_feed_outlives_its_opml_for_whoever_kept_it() {
let mut cfg = config::Config::default();
cfg.feeds.insert("opml".into(), config::Feed { private: true, ..feed() });
let ctx = Arc::new(test_ctx(cfg).await);
let (me, you) = (
ctx.db.create_user("me", None, false).await.unwrap(),
ctx.db.create_user("you", None, false).await.unwrap(),
);
for id in ["opml", "a", "b"] {
if id != "opml" {
ctx.db.upsert_managed(id, &format!("http://x/{id}.xml"), id, "opml").await.unwrap();
}
for u in [me, you] {
ctx.db.subscribe(u, id).await.unwrap();
}
}
assert!(ctx.db.set_kept(me, "a", true).await.unwrap());
assert!(ctx.db.set_kept(me, "b", true).await.unwrap());
// The OPML drops a: mine to keep, a feed of its own; yours, unkept, goes.
sync_group(&ctx, "opml", &[("b".into(), "http://x/b.xml".into())]).await.unwrap();
let a = ctx.cfg().feeds.get("a").cloned().expect("a is in the catalogue");
assert_eq!((a.url.as_str(), a.group, a.private), ("http://x/a.xml", None, true));
assert!(!ctx.db.managed_feeds().await.unwrap().iter().any(|m| m.id == "a"));
assert!(ctx.db.subscription(me, "a").await.unwrap().is_some());
assert!(ctx.db.subscription(you, "a").await.unwrap().is_none());
// Leaving the OPML takes what I did not keep and leaves b, still the OPML's while you read it.
catalogue::leave(&ctx, me, "opml").await.unwrap();
assert!(ctx.db.subscription(me, "b").await.unwrap().is_some());
assert!(ctx.db.managed_feeds().await.unwrap().iter().any(|m| m.id == "b"));
// Once nobody reads it, b is mine alone, and yours goes with the OPML.
catalogue::leave(&ctx, you, "opml").await.unwrap();
// Still private: the OPML is gone from config by the time b is promoted (#332).
assert!(ctx.cfg().feeds.get("b").is_some_and(|f| f.group.is_none() && f.private));
assert!(ctx.db.subscription(me, "b").await.unwrap().is_some());
assert!(ctx.db.subscription(you, "b").await.unwrap().is_none());
}
}