diff --git a/migrations/0010_feed_discovery.sql b/migrations/0010_feed_discovery.sql new file mode 100644 index 0000000..87bab46 --- /dev/null +++ b/migrations/0010_feed_discovery.sql @@ -0,0 +1,27 @@ +-- Feed discovery: proposed Miniflux subscriptions found behind aggregator-only +-- articles (feed discovery plan §4 step 1). + +CREATE TABLE feed_candidates ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + feed_url TEXT NOT NULL UNIQUE, + host TEXT NOT NULL, + title TEXT, + status TEXT NOT NULL CHECK (status IN ('candidate', 'added', 'dismissed')), + first_seen TEXT NOT NULL, + last_seen TEXT NOT NULL, + miniflux_feed_id INTEGER, + decided_at TEXT +); +CREATE INDEX idx_feed_candidates_status_host ON feed_candidates(status, host); + +CREATE TABLE feed_candidate_articles ( + candidate_id INTEGER NOT NULL REFERENCES feed_candidates(id) ON DELETE CASCADE, + article_id INTEGER NOT NULL REFERENCES articles(id) ON DELETE CASCADE, + PRIMARY KEY (candidate_id, article_id) +); + +CREATE TABLE feed_discovery_hosts ( + host TEXT PRIMARY KEY, + checked_at TEXT NOT NULL, + candidates INTEGER NOT NULL +); diff --git a/src/discovery.rs b/src/discovery.rs new file mode 100644 index 0000000..ca1be69 --- /dev/null +++ b/src/discovery.rs @@ -0,0 +1,1577 @@ +//! Feed discovery: propose new Miniflux subscriptions from aggregator-only +//! articles (feed discovery plan, 2026-09-07). +//! +//! Most of the paper's stories arrive through an aggregator (HN, Lobsters, +//! Reddit, Scour) rather than the author's own feed, and every one of those is a +//! lead on a feed the operator does not subscribe to yet. [`run`] is a +//! best-effort pipeline stage (and the same function the `feeds discover` CLI +//! backfill calls) that asks Miniflux what feeds sit behind those articles, +//! **validates every answer itself**, and records the survivors in +//! `feed_candidates` for the dashboard to accept or dismiss. +//! +//! Miniflux's `POST /v1/discover` falls back to probing well-known paths and +//! returns every path that answered 200 without checking the body is a feed, so +//! a site that serves its SPA shell for any path yields up to nine bogus hits +//! (plan §2). Discover results are leads, not facts: [`sniff_feed`] decides. +//! +//! Like `imports`, this module owns its own SQL rather than growing `db.rs`. + +use std::collections::{BTreeMap, HashMap, HashSet}; +use std::time::Duration; + +use anyhow::{Context, Result}; +use futures::StreamExt as _; +use jiff::Timestamp; +use sqlx::Row as _; + +use crate::config::DiscoveryConfig; +use crate::curate::telemetry::SignalsJson; +use crate::db::{Db, fmt_ts, parse_ts}; +use crate::miniflux::{Discovered, FeedMeta, MinifluxClient, MinifluxError}; +use crate::types::{Article, ArticleId, FeedId, SourceKind}; + +/// How long a checked host is left alone before it is looked up again (§3). +pub const RECHECK_DAYS: i64 = 90; +/// Most candidates kept from one host's discover results (§3). +const MAX_PER_HOST: usize = 3; +/// Validation fetches in flight across hosts (§3). +const CONCURRENCY: usize = 4; +/// Per-request timeout for a validation fetch (§3). +const FETCH_TIMEOUT: Duration = Duration::from_secs(10); +/// Never buffer more than this much of a body before sniffing it (§3). +const MAX_BODY_BYTES: usize = 256 * 1024; + +// --------------------------------------------------------------------------- +// Rows +// --------------------------------------------------------------------------- + +/// One row of `feed_candidates`, plus the articles that led us to it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Candidate { + pub id: i64, + pub feed_url: String, + pub host: String, + pub title: Option, + /// `candidate` | `added` | `dismissed`. + pub status: String, + pub first_seen: String, + pub last_seen: String, + pub miniflux_feed_id: Option, + pub decided_at: Option, + /// Articles linked to this candidate, newest article id first. + pub article_ids: Vec, +} + +const CANDIDATE_COLUMNS: &str = + "id, feed_url, host, title, status, first_seen, last_seen, miniflux_feed_id, decided_at"; + +fn row_from(row: &sqlx::sqlite::SqliteRow) -> Candidate { + Candidate { + id: row.get("id"), + feed_url: row.get("feed_url"), + host: row.get("host"), + title: row.get("title"), + status: row.get("status"), + first_seen: row.get("first_seen"), + last_seen: row.get("last_seen"), + miniflux_feed_id: row.get("miniflux_feed_id"), + decided_at: row.get("decided_at"), + article_ids: Vec::new(), + } +} + +// --------------------------------------------------------------------------- +// Database helpers (plan §4 step 1) +// --------------------------------------------------------------------------- + +/// When each of `hosts` was last looked up, and how many candidates it yielded. +pub async fn hosts_checked(db: &Db, hosts: &[String]) -> Result> { + let mut out = HashMap::new(); + for chunk in hosts.chunks(500) { + let placeholders = vec!["?"; chunk.len()].join(", "); + let mut query = sqlx::query(sqlx::AssertSqlSafe(format!( + "SELECT host, checked_at, candidates FROM feed_discovery_hosts + WHERE host IN ({placeholders})" + ))); + for host in chunk { + query = query.bind(host); + } + for row in query.fetch_all(db.pool()).await? { + let checked_at = parse_ts( + "feed_discovery_hosts.checked_at", + &row.get::("checked_at"), + )?; + out.insert( + row.get::("host"), + (checked_at, row.get::("candidates")), + ); + } + } + Ok(out) +} + +/// Memo that `host` was looked up, so it is not looked up again for +/// [`RECHECK_DAYS`]. +pub async fn mark_host_checked(db: &Db, host: &str, candidates: i64, now: Timestamp) -> Result<()> { + sqlx::query( + "INSERT INTO feed_discovery_hosts (host, checked_at, candidates) VALUES (?, ?, ?) + ON CONFLICT(host) DO UPDATE SET checked_at = excluded.checked_at, + candidates = excluded.candidates", + ) + .bind(host) + .bind(fmt_ts(now)) + .bind(candidates) + .execute(db.pool()) + .await?; + Ok(()) +} + +async fn existing_candidate_id(db: &Db, feed_url: &str) -> Result> { + let row = sqlx::query("SELECT id FROM feed_candidates WHERE feed_url = ?") + .bind(feed_url) + .fetch_optional(db.pool()) + .await?; + Ok(row.map(|row| row.get::("id"))) +} + +/// Insert `feed_url` as a candidate or bump its `last_seen`, returning its id. +/// +/// A decision already taken is never undone: the `status` of an `added` or +/// `dismissed` row is left alone, and an existing title wins over a new one. +pub async fn upsert_candidate( + db: &Db, + feed_url: &str, + host: &str, + title: Option<&str>, + now: Timestamp, +) -> Result { + let now = fmt_ts(now); + let row = sqlx::query( + "INSERT INTO feed_candidates (feed_url, host, title, status, first_seen, last_seen) + VALUES (?, ?, ?, 'candidate', ?, ?) + ON CONFLICT(feed_url) DO UPDATE + SET last_seen = excluded.last_seen, + title = COALESCE(feed_candidates.title, excluded.title) + RETURNING id", + ) + .bind(feed_url) + .bind(host) + .bind(title) + .bind(&now) + .bind(&now) + .fetch_one(db.pool()) + .await?; + Ok(row.get::("id")) +} + +/// Record that `article_id` is evidence for `candidate_id`. +pub async fn link_article(db: &Db, candidate_id: i64, article_id: ArticleId) -> Result<()> { + sqlx::query( + "INSERT OR IGNORE INTO feed_candidate_articles (candidate_id, article_id) VALUES (?, ?)", + ) + .bind(candidate_id) + .bind(article_id) + .execute(db.pool()) + .await?; + Ok(()) +} + +/// Ids of the still-undecided candidates on `host`. +pub async fn candidates_for_host(db: &Db, host: &str) -> Result> { + let rows = sqlx::query( + "SELECT id FROM feed_candidates WHERE host = ? AND status = 'candidate' ORDER BY id", + ) + .bind(host) + .fetch_all(db.pool()) + .await?; + Ok(rows.iter().map(|row| row.get::("id")).collect()) +} + +/// Attach `article_ids` to every candidate in one statement. +async fn attach_article_ids(db: &Db, candidates: &mut [Candidate]) -> Result<()> { + if candidates.is_empty() { + return Ok(()); + } + let mut by_id: HashMap = HashMap::new(); + for (index, candidate) in candidates.iter().enumerate() { + by_id.insert(candidate.id, index); + } + let ids: Vec = candidates.iter().map(|c| c.id).collect(); + for chunk in ids.chunks(500) { + let placeholders = vec!["?"; chunk.len()].join(", "); + let mut query = sqlx::query(sqlx::AssertSqlSafe(format!( + "SELECT candidate_id, article_id FROM feed_candidate_articles + WHERE candidate_id IN ({placeholders}) ORDER BY article_id DESC" + ))); + for id in chunk { + query = query.bind(*id); + } + for row in query.fetch_all(db.pool()).await? { + let candidate_id: i64 = row.get("candidate_id"); + if let Some(index) = by_id.get(&candidate_id) { + candidates[*index].article_ids.push(row.get("article_id")); + } + } + } + Ok(()) +} + +/// One page of candidates with `status`, newest first, and the total count. +pub async fn list( + db: &Db, + status: &str, + page: u32, + per_page: u32, +) -> Result<(Vec, i64)> { + let total = count(db, status).await?; + let per_page = i64::from(per_page.max(1)); + let offset = i64::from(page.saturating_sub(1)) * per_page; + let rows = sqlx::query(sqlx::AssertSqlSafe(format!( + "SELECT {CANDIDATE_COLUMNS} FROM feed_candidates WHERE status = ? + ORDER BY last_seen DESC, id DESC LIMIT ? OFFSET ?" + ))) + .bind(status) + .bind(per_page) + .bind(offset) + .fetch_all(db.pool()) + .await?; + let mut candidates: Vec = rows.iter().map(row_from).collect(); + attach_article_ids(db, &mut candidates).await?; + Ok((candidates, total)) +} + +/// Every undecided candidate — ranking sorts globally, so the page loads them +/// all and paginates in Rust (plan §4 step 4). +pub async fn all_candidates(db: &Db) -> Result> { + let rows = sqlx::query(sqlx::AssertSqlSafe(format!( + "SELECT {CANDIDATE_COLUMNS} FROM feed_candidates WHERE status = 'candidate' + ORDER BY last_seen DESC, id DESC" + ))) + .fetch_all(db.pool()) + .await?; + let mut candidates: Vec = rows.iter().map(row_from).collect(); + attach_article_ids(db, &mut candidates).await?; + Ok(candidates) +} + +/// One candidate by id, with its linked articles. +pub async fn candidate(db: &Db, id: i64) -> Result> { + let row = sqlx::query(sqlx::AssertSqlSafe(format!( + "SELECT {CANDIDATE_COLUMNS} FROM feed_candidates WHERE id = ?" + ))) + .bind(id) + .fetch_optional(db.pool()) + .await?; + let Some(row) = row else { return Ok(None) }; + let mut found = vec![row_from(&row)]; + attach_article_ids(db, &mut found).await?; + Ok(found.pop()) +} + +/// Record the operator's (or reconciliation's) decision on one candidate. +pub async fn set_status( + db: &Db, + id: i64, + status: &str, + miniflux_feed_id: Option, + now: Timestamp, +) -> Result<()> { + sqlx::query( + "UPDATE feed_candidates + SET status = ?, + miniflux_feed_id = COALESCE(?, miniflux_feed_id), + decided_at = ? + WHERE id = ?", + ) + .bind(status) + .bind(miniflux_feed_id) + .bind(fmt_ts(now)) + .bind(id) + .execute(db.pool()) + .await?; + Ok(()) +} + +/// How many candidates have `status`. +pub async fn count(db: &Db, status: &str) -> Result { + Ok( + sqlx::query_scalar("SELECT COUNT(*) FROM feed_candidates WHERE status = ?") + .bind(status) + .fetch_one(db.pool()) + .await?, + ) +} + +/// Flip candidates the operator subscribed to by other means to `added`. +/// +/// A candidate matches a subscription when its feed URL is that feed's +/// `feed_url`, or when its host is the host of that feed's `feed_url` or +/// `site_url`. Returns how many rows changed. +pub async fn reconcile_added( + db: &Db, + subscribed: &HashMap, + now: Timestamp, +) -> Result { + let mut by_url: HashMap = HashMap::new(); + let mut by_host: HashMap = HashMap::new(); + for (id, meta) in subscribed { + by_url.insert(norm_url(&meta.feed_url), *id); + for url in [&meta.feed_url, &meta.site_url] { + if let Some(host) = host_of(url) { + by_host.entry(host).or_insert(*id); + } + } + } + let mut reconciled = 0usize; + for candidate in all_candidates(db).await? { + let feed_id = by_url + .get(&norm_url(&candidate.feed_url)) + .or_else(|| by_host.get(&candidate.host)); + if let Some(feed_id) = feed_id { + set_status(db, candidate.id, "added", Some(*feed_id), now).await?; + reconciled += 1; + tracing::info!( + candidate = candidate.id, + feed_url = %candidate.feed_url, + miniflux_feed_id = feed_id, + "feed candidate is already subscribed; marking added" + ); + } + } + Ok(reconciled) +} + +// --------------------------------------------------------------------------- +// Pure helpers (plan §4 step 3) +// --------------------------------------------------------------------------- + +/// True when no feed of the operator's carried this story — the only articles +/// worth a lookup (§3). +pub fn aggregator_only(article: &Article) -> bool { + !article.came_via(SourceKind::Feed) +} + +/// `https://WWW.Example.com/x` → `example.com`. +pub fn host_of(url: &str) -> Option { + let parsed = url::Url::parse(url.trim()).ok()?; + let host = parsed.host_str()?.to_ascii_lowercase(); + let host = host.strip_prefix("www.").unwrap_or(&host); + (!host.is_empty()).then(|| host.to_string()) +} + +/// Comparison form of a feed URL: trimmed, lowercased, no trailing slash. +fn norm_url(url: &str) -> String { + url.trim().trim_end_matches('/').to_ascii_lowercase() +} + +/// True when `host` is `entry` or a subdomain of it. +fn host_matches(host: &str, entry: &str) -> bool { + let entry = entry.trim().to_ascii_lowercase(); + !entry.is_empty() && (host == entry || host.ends_with(&format!(".{entry}"))) +} + +/// True when the operator asked never to look this host up (`skip_hosts`). +fn skipped(host: &str, skip_hosts: &[String]) -> bool { + skip_hosts.iter().any(|entry| host_matches(host, entry)) +} + +/// Worth fetching at all? Comment feeds are noise (§3). +pub fn keep_result(result: &Discovered) -> bool { + let url = result.url.to_ascii_lowercase(); + let title = result.title.to_ascii_lowercase(); + !url.is_empty() && !url.contains("comment") && !title.contains("comment") +} + +/// What a validated body turned out to be. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum FeedKind { + Rss, + Atom, + Rdf, + Json, +} + +/// Decide whether `body` really is a feed (§3). +/// +/// Content type is useless here — SPAs answer `text/html` for every path and +/// real feeds turn up as `text/xml` or `application/octet-stream` — so the root +/// element decides, after an optional BOM, whitespace and XML declaration. +pub fn sniff_feed(body: &[u8]) -> Option { + let body = body.strip_prefix(&[0xEF, 0xBB, 0xBF]).unwrap_or(body); + let mut rest = trim_start(body); + if rest.starts_with(b"")?; + rest = trim_start(&rest[end + 2..]); + } + let head: String = String::from_utf8_lossy(&rest[..rest.len().min(512)]).to_ascii_lowercase(); + if head.starts_with(" &[u8] { + let start = body + .iter() + .position(|b| !b.is_ascii_whitespace()) + .unwrap_or(body.len()); + &body[start..] +} + +/// The feed's own ``, used when Miniflux had no title to give (§3). +pub fn feed_title(body: &[u8]) -> Option<String> { + let text = String::from_utf8_lossy(&body[..body.len().min(64 * 1024)]); + let lower = text.to_ascii_lowercase(); + let open = lower.find("<title")?; + let start = open + lower[open..].find('>')? + 1; + let end = start + lower[start..].find("")?; + let raw = text[start..end] + .trim() + .trim_start_matches("") + .trim(); + let title = unescape(raw); + (!title.is_empty()).then_some(title) +} + +/// The five entities a feed title realistically carries; a whole HTML entity +/// table is not worth a dependency here. +fn unescape(raw: &str) -> String { + raw.replace("<", "<") + .replace(">", ">") + .replace(""", "\"") + .replace("'", "'") + .replace("'", "'") + .replace("&", "&") +} + +/// A discover result whose body sniffed as a feed. +#[derive(Debug, Clone, PartialEq)] +pub struct Validated { + pub url: String, + /// Miniflux's title for the result; `== url` when it was a well-known-path + /// guess rather than a `` hit. + pub raw_title: String, + /// `rss` | `atom` | `json`, as Miniflux reported it. + pub kind: String, + /// The `` read out of the body, when we found one. + pub sniffed_title: Option<String>, +} + +impl Validated { + /// True when this came from probing a well-known path, not a link tag (§2). + pub fn is_guess(&self) -> bool { + self.raw_title == self.url + } + + /// What to store as the candidate's title. + pub fn title(&self) -> Option<String> { + if self.is_guess() { + self.sniffed_title.clone() + } else { + Some(self.raw_title.clone()) + } + } +} + +/// Collapse one host's validated results into the few worth proposing (§3). +/// +/// Link-tag hits are grouped by title (a site commonly offers the same feed as +/// Atom and JSON Feed under one title), preferring XML over JSON; at most one +/// well-known-path guess survives, since WordPress answers both `/feed/` and +/// `/rss/` with the same content. At most [`MAX_PER_HOST`] overall. +pub fn dedupe_guesses(results: Vec<Validated>) -> Vec<Validated> { + let mut kept: Vec<Validated> = Vec::new(); + let mut titles: Vec<String> = Vec::new(); + for result in results.iter().filter(|r| !r.is_guess()) { + match titles.iter().position(|t| t == &result.raw_title) { + Some(index) => { + if kept[index].kind.eq_ignore_ascii_case("json") + && !result.kind.eq_ignore_ascii_case("json") + { + kept[index] = result.clone(); + } + } + None => { + titles.push(result.raw_title.clone()); + kept.push(result.clone()); + } + } + } + if let Some(guess) = results.into_iter().find(Validated::is_guess) { + kept.push(guess); + } + kept.truncate(MAX_PER_HOST); + kept +} + +// --------------------------------------------------------------------------- +// The stage (plan §4 step 3) +// --------------------------------------------------------------------------- + +/// What one discovery pass did, for the report line and the CLI. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct Summary { + /// Hosts looked up in Miniflux this pass (an error response counts: the + /// host is memoized either way). + pub hosts_checked: usize, + /// Hosts skipped because the operator already subscribes to them. + pub hosts_skipped_subscribed: usize, + /// Hosts whose memo was still fresh, so their articles were merely linked. + pub hosts_reused: usize, + /// Discover results that did not sniff as a feed. + pub results_rejected: usize, + /// Candidates inserted for the first time. + pub candidates_new: usize, + /// Article ↔ candidate links written. + pub articles_linked: usize, +} + +impl std::fmt::Display for Summary { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "{} hosts checked, {} reused, {} already subscribed · {} results rejected · \ + {} new candidates, {} articles linked", + self.hosts_checked, + self.hosts_reused, + self.hosts_skipped_subscribed, + self.results_rejected, + self.candidates_new, + self.articles_linked, + ) + } +} + +/// One host queued for a Miniflux lookup. +#[derive(Debug, Clone)] +struct Lookup { + host: String, + /// The article URL to discover from — link tags only exist on real pages. + url: String, + article_ids: Vec<ArticleId>, + /// Best social score among the host's articles; the queue's sort key. + score: f64, +} + +/// What one host's lookup produced, before anything is written. +struct Outcome { + host: String, + article_ids: Vec<ArticleId>, + validated: Vec<Validated>, + rejected: usize, + /// False only for a transport failure: the host is left unmemoized so the + /// next run tries again. + checked: bool, +} + +/// Run one discovery pass over `articles` (plan §4 step 3). +/// +/// Best effort by construction: a failing host is logged and skipped, and only +/// a database failure produces an `Err`. Safe to call with no articles and no +/// subscriptions. +pub async fn run( + db: &Db, + miniflux: &MinifluxClient, + http: &reqwest::Client, + cfg: &DiscoveryConfig, + articles: &[Article], + subscribed: &HashMap<FeedId, FeedMeta>, + now: Timestamp, +) -> Result<Summary> { + let mut summary = Summary::default(); + + let reconciled = reconcile_added(db, subscribed, now) + .await + .context("reconciling feed candidates against subscriptions")?; + + let subscribed_hosts: HashSet<String> = subscribed + .values() + .flat_map(|meta| [host_of(&meta.feed_url), host_of(&meta.site_url)]) + .flatten() + .collect(); + let subscribed_urls: HashSet<String> = subscribed + .values() + .map(|meta| norm_url(&meta.feed_url)) + .collect(); + + // Group the aggregator-only articles by host, keeping the best social score + // as the queue's priority and the best-scoring article as the lookup URL. + let mut by_host: BTreeMap<String, Lookup> = BTreeMap::new(); + let mut skipped_hosts: HashSet<String> = HashSet::new(); + for article in articles.iter().filter(|a| aggregator_only(a)) { + let Some(host) = host_of(&article.canonical_url) else { + continue; + }; + if skipped(&host, &cfg.skip_hosts) { + continue; + } + if subscribed_hosts.contains(&host) { + skipped_hosts.insert(host); + continue; + } + let score = article.social_score(); + let entry = by_host.entry(host.clone()).or_insert_with(|| Lookup { + host, + url: article.canonical_url.clone(), + article_ids: Vec::new(), + score: f64::MIN, + }); + entry.article_ids.push(article.id); + if score > entry.score { + entry.score = score; + entry.url = article.canonical_url.clone(); + } + } + summary.hosts_skipped_subscribed = skipped_hosts.len(); + + // Hosts checked recently just collect links; the rest queue for a lookup. + let hosts: Vec<String> = by_host.keys().cloned().collect(); + let memo = hosts_checked(db, &hosts) + .await + .context("loading the discovery host memo")?; + let recheck_before = now + .checked_sub(jiff::Span::new().hours(RECHECK_DAYS * 24)) + .unwrap_or(Timestamp::UNIX_EPOCH); + let mut queue: Vec<Lookup> = Vec::new(); + for (host, lookup) in by_host { + match memo.get(&host) { + Some((checked_at, _)) if *checked_at >= recheck_before => { + summary.hosts_reused += 1; + for candidate_id in candidates_for_host(db, &host).await? { + for article_id in &lookup.article_ids { + link_article(db, candidate_id, *article_id).await?; + summary.articles_linked += 1; + } + } + } + _ => queue.push(lookup), + } + } + queue.sort_by(|a, b| b.score.total_cmp(&a.score)); + queue.truncate(cfg.max_lookups_per_run); + + tracing::info!( + queued = queue.len(), + reused = summary.hosts_reused, + subscribed_hosts = summary.hosts_skipped_subscribed, + reconciled, + "feed discovery: looking up hosts" + ); + if queue.is_empty() { + return Ok(summary); + } + + // Network only: every write happens below, in one sequential pass. + let outcomes: Vec<Outcome> = futures::stream::iter(queue) + .map(|lookup| look_up(miniflux, http, &subscribed_urls, lookup)) + .buffer_unordered(CONCURRENCY) + .collect() + .await; + + for outcome in outcomes { + if !outcome.checked { + continue; + } + summary.hosts_checked += 1; + summary.results_rejected += outcome.rejected; + let kept = dedupe_guesses(outcome.validated); + for validated in &kept { + let is_new = existing_candidate_id(db, &validated.url).await?.is_none(); + let id = upsert_candidate( + db, + &validated.url, + &outcome.host, + validated.title().as_deref(), + now, + ) + .await + .context("recording a feed candidate")?; + if is_new { + summary.candidates_new += 1; + tracing::info!( + candidate = id, + feed_url = %validated.url, + title = ?validated.title(), + "new feed candidate" + ); + } + for article_id in &outcome.article_ids { + link_article(db, id, *article_id).await?; + summary.articles_linked += 1; + } + } + mark_host_checked(db, &outcome.host, kept.len() as i64, now).await?; + } + Ok(summary) +} + +/// Discover and validate one host's feeds. Never fails: a bad host is an +/// outcome, not an error. +async fn look_up( + miniflux: &MinifluxClient, + http: &reqwest::Client, + subscribed_urls: &HashSet<String>, + lookup: Lookup, +) -> Outcome { + let mut outcome = Outcome { + host: lookup.host.clone(), + article_ids: lookup.article_ids, + validated: Vec::new(), + rejected: 0, + checked: true, + }; + let results = match miniflux.discover(&lookup.url).await { + Ok(results) => results, + // A `fetcher: …` / `resource not found` response is Miniflux's verdict + // on the site: nothing found, and nothing to retry this run. + Err(error @ MinifluxError::Status { .. }) => { + tracing::info!(host = %lookup.host, %error, "discover found nothing"); + return outcome; + } + // A transport failure (or a response we could not parse) says nothing + // about the host: leave it unmemoized so the next run tries again. + Err(error) => { + tracing::warn!(host = %lookup.host, %error, "discover failed; host left unchecked"); + outcome.checked = false; + return outcome; + } + }; + + for result in results.into_iter().filter(keep_result) { + if subscribed_urls.contains(&norm_url(&result.url)) { + continue; + } + match fetch_head(http, &result.url).await { + Ok(body) if sniff_feed(&body).is_some() => outcome.validated.push(Validated { + url: result.url, + raw_title: result.title, + kind: result.kind, + sniffed_title: feed_title(&body), + }), + Ok(_) => { + outcome.rejected += 1; + tracing::debug!(url = %result.url, "discover result is not a feed"); + } + Err(error) => { + outcome.rejected += 1; + tracing::debug!(url = %result.url, %error, "could not validate discover result"); + } + } + } + outcome +} + +/// GET at most [`MAX_BODY_BYTES`] of `url`, so a stray 4 GB "feed" cannot eat +/// the run's memory. +async fn fetch_head(http: &reqwest::Client, url: &str) -> Result<Vec<u8>> { + let mut resp = http.get(url).timeout(FETCH_TIMEOUT).send().await?; + let status = resp.status(); + if !status.is_success() { + anyhow::bail!("{status}"); + } + let mut body: Vec<u8> = Vec::new(); + while let Some(chunk) = resp.chunk().await? { + body.extend_from_slice(&chunk); + if body.len() >= MAX_BODY_BYTES { + body.truncate(MAX_BODY_BYTES); + break; + } + } + Ok(body) +} + +/// Aggregator-only articles first seen at or after `since`, hydrated with their +/// sources and social scores (the CLI backfill's input, plan §4 step 6). +pub async fn articles_since(db: &Db, since: Timestamp) -> Result<Vec<Article>> { + let ids = db.article_ids_since(since).await?; + let mut articles: Vec<Article> = db.get_articles(&ids).await?.into_values().collect(); + articles.sort_by_key(|a| a.id); + Ok(articles) +} + +// --------------------------------------------------------------------------- +// Ranking (plan §4 step 4) +// --------------------------------------------------------------------------- + +/// The score a feed with no evidence at all lands on, on the 0–1 scale. +pub const PRIOR: f64 = 0.3; +/// Shrinkage strength: how many prior-valued articles the mean is padded with. +/// One 95-utility article should not outrank three 80s. +pub const K: f64 = 2.0; + +/// What one linked article says about the feed behind it. +#[derive(Debug, Clone, PartialEq)] +pub struct ArticleEvidence { + pub article_id: ArticleId, + pub title: String, + /// −1…1: the explicit verdict if there is one, else how far the article got. + pub value: f64, + /// Interest names this article matched, from `signals_json`. + pub interests: Vec<String>, +} + +/// Shrunk mean of the evidence, 0–100 (plan §3 "Ranking"). +pub fn score(evidence: &[ArticleEvidence]) -> f64 { + let sum: f64 = evidence.iter().map(|e| e.value).sum(); + let mean = (sum + PRIOR * K) / (evidence.len() as f64 + K); + (mean * 100.0).clamp(0.0, 100.0) +} + +/// The three interests that come up most often across the linked articles. +pub fn why(evidence: &[ArticleEvidence]) -> Vec<String> { + let mut counts: BTreeMap<&str, usize> = BTreeMap::new(); + for name in evidence.iter().flat_map(|e| e.interests.iter()) { + *counts.entry(name.as_str()).or_insert(0) += 1; + } + let mut ranked: Vec<(&str, usize)> = counts.into_iter().collect(); + // Frequency first; the BTreeMap already has ties in name order. + ranked.sort_by_key(|a| std::cmp::Reverse(a.1)); + ranked + .into_iter() + .take(3) + .map(|(name, _)| name.to_string()) + .collect() +} + +/// Load the ranking evidence for `article_ids`: the newest `candidate_runs` row +/// per article, overridden by an explicit rating where one exists. +/// +/// Articles the pipeline never scored and nobody rated simply have no entry; +/// they contribute nothing but the prior. +pub async fn load_evidence( + db: &Db, + article_ids: &[ArticleId], + rating_lookback_days: i64, +) -> Result<HashMap<ArticleId, ArticleEvidence>> { + let mut evidence: HashMap<ArticleId, ArticleEvidence> = HashMap::new(); + if article_ids.is_empty() { + return Ok(evidence); + } + let wanted: HashSet<ArticleId> = article_ids.iter().copied().collect(); + let ids: Vec<ArticleId> = wanted.iter().copied().collect(); + + let mut titles: HashMap<ArticleId, String> = HashMap::new(); + for chunk in ids.chunks(500) { + let placeholders = vec!["?"; chunk.len()].join(", "); + let mut query = sqlx::query(sqlx::AssertSqlSafe(format!( + "SELECT id, COALESCE(title, '') AS title FROM articles WHERE id IN ({placeholders})" + ))); + for id in chunk { + query = query.bind(*id); + } + for row in query.fetch_all(db.pool()).await? { + titles.insert(row.get("id"), row.get("title")); + } + + let mut query = sqlx::query(sqlx::AssertSqlSafe(format!( + "SELECT article_id, stage, utility, signals_json FROM candidate_runs + WHERE article_id IN ({placeholders}) ORDER BY article_id, run_id DESC" + ))); + for id in chunk { + query = query.bind(*id); + } + for row in query.fetch_all(db.pool()).await? { + let article_id: ArticleId = row.get("article_id"); + // Rows come newest-run first per article; keep the first. + if evidence.contains_key(&article_id) { + continue; + } + let signals: SignalsJson = + serde_json::from_str(&row.get::<String, _>("signals_json")).unwrap_or_default(); + let stage: String = row.get("stage"); + let utility: Option<f64> = row.get("utility"); + let Some(value) = stage_value(&stage, utility, &signals) else { + continue; + }; + evidence.insert( + article_id, + ArticleEvidence { + article_id, + title: titles.get(&article_id).cloned().unwrap_or_default(), + value, + interests: signals + .top_interests + .iter() + .map(|i| i.name.clone()) + .collect(), + }, + ); + } + } + + // An explicit verdict is the strongest evidence there is; it overrides. + for rating in db.current_ratings(rating_lookback_days).await? { + if !wanted.contains(&rating.article_id) { + continue; + } + evidence + .entry(rating.article_id) + .or_insert_with(|| ArticleEvidence { + article_id: rating.article_id, + title: rating.title.clone(), + value: 0.0, + interests: Vec::new(), + }) + .value = rating.value.clamp(-1.0, 1.0); + } + Ok(evidence) +} + +/// How far the article got, on the −1…1 evidence scale. +fn stage_value(stage: &str, utility: Option<f64>, signals: &SignalsJson) -> Option<f64> { + if stage == "selected" { + return Some(0.85); + } + utility + .or_else(|| signals.blend()) + .map(|value| (value / 100.0).clamp(-1.0, 1.0)) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::curate::signals::TopInterest; + use crate::types::{ExtractMethod, SourceRef}; + + async fn test_db() -> (tempfile::TempDir, Db) { + let dir = tempfile::tempdir().unwrap(); + let db = Db::open_and_migrate(&dir.path().join("db.sqlite")) + .await + .unwrap(); + (dir, db) + } + + fn ts(s: &str) -> Timestamp { + s.parse().unwrap() + } + + /// A client pointed at `base_url`; the tests that pass one use an + /// unroutable port, so any lookup at all would fail the test. + fn miniflux(http: reqwest::Client, base_url: &str) -> MinifluxClient { + MinifluxClient::new( + &crate::config::MinifluxConfig { + api_key: Some("token".into()), + base_url: base_url.into(), + ..Default::default() + }, + http, + ) + .unwrap() + } + + fn discovered(title: &str, url: &str, kind: &str) -> Discovered { + Discovered { + url: url.into(), + title: title.into(), + kind: kind.into(), + } + } + + fn validated(result: &Discovered, sniffed_title: Option<&str>) -> Validated { + Validated { + url: result.url.clone(), + raw_title: result.title.clone(), + kind: result.kind.clone(), + sniffed_title: sniffed_title.map(str::to_string), + } + } + + /// The two `<link rel=alternate>` hits observed for blog.philz.dev (plan §2). + fn philz_results() -> Vec<Discovered> { + vec![ + discovered( + "blog.philz.dev", + "https://blog.philz.dev/feed/feed.xml", + "atom", + ), + discovered( + "blog.philz.dev", + "https://blog.philz.dev/feed/feed.json", + "json", + ), + ] + } + + /// The nine well-known-path guesses observed for zombo.com (plan §2). + fn zombo_results() -> Vec<Discovered> { + [ + ("https://zombo.com/atom.xml", "atom"), + ("https://zombo.com/feed.atom", "atom"), + ("https://zombo.com/feed.xml", "atom"), + ("https://zombo.com/feed/", "atom"), + ("https://zombo.com/index.rss", "rss"), + ("https://zombo.com/index.xml", "rss"), + ("https://zombo.com/rss.xml", "rss"), + ("https://zombo.com/rss/", "rss"), + ("https://zombo.com/rss/feed.xml", "rss"), + ] + .into_iter() + .map(|(url, kind)| discovered(url, url, kind)) + .collect() + } + + const SPA_SHELL: &[u8] = br#"<!doctype html><html><head><title>Zombo +
"#; + + fn article(id: ArticleId, url: &str, kinds: &[SourceKind]) -> Article { + Article { + id, + canonical_url: url.into(), + title: format!("Article {id}"), + best_entry_id: id, + content_html: String::new(), + word_count: 100, + excerpt_only: false, + image_count: 0, + sources: kinds + .iter() + .map(|kind| SourceRef { + entry_id: id, + feed_id: 1, + feed_title: "Feed".into(), + category: None, + kind: *kind, + }) + .collect(), + first_seen: ts("2026-09-07T00:00:00Z"), + url: url.into(), + author: None, + feed_id: 1, + feed_title: "Feed".into(), + category: None, + published_at: None, + comments_url: None, + image_urls: Vec::new(), + social: Vec::new(), + extract_method: ExtractMethod::Miniflux, + } + } + + #[test] + fn only_articles_without_a_direct_feed_are_leads() { + assert!(aggregator_only(&article( + 1, + "https://example.com/a", + &[SourceKind::HnFrontpage] + ))); + assert!(aggregator_only(&article(2, "https://example.com/b", &[]))); + assert!(!aggregator_only(&article( + 3, + "https://example.com/c", + &[SourceKind::HnFrontpage, SourceKind::Feed] + ))); + } + + #[test] + fn hosts_are_normalized_and_matched_by_suffix() { + assert_eq!( + host_of("https://WWW.Example.com/post?x=1"), + Some("example.com".into()) + ); + assert_eq!( + host_of("http://blog.philz.dev/"), + Some("blog.philz.dev".into()) + ); + assert_eq!(host_of("not a url"), None); + assert_eq!(host_of("mailto:a@b.com"), None); + + assert!(skipped("reddit.com", &["reddit.com".to_string()])); + assert!(skipped("old.reddit.com", &["reddit.com".to_string()])); + assert!(!skipped("notreddit.com", &["reddit.com".to_string()])); + assert!(!skipped("example.com", &[])); + } + + #[test] + fn comment_feeds_are_dropped_before_they_are_fetched() { + assert!(keep_result(&discovered( + "Blog", + "https://a.dev/feed.xml", + "atom" + ))); + assert!(!keep_result(&discovered( + "Blog", + "https://a.dev/comments/feed/", + "rss" + ))); + assert!(!keep_result(&discovered( + "Comments for Blog", + "https://a.dev/feed2", + "rss" + ))); + assert!(!keep_result(&discovered("Blog", "", "rss"))); + } + + #[test] + fn sniff_accepts_real_feeds_and_rejects_app_shells() { + // BOM + XML declaration + leading whitespace before the root element. + let mut rss = vec![0xEF, 0xBB, 0xBF]; + rss.extend_from_slice(b"\n\n A Blog"); + assert_eq!(sniff_feed(&rss), Some(FeedKind::Rss)); + + assert_eq!( + sniff_feed(b"X"), + Some(FeedKind::Atom) + ); + assert_eq!( + sniff_feed(b""), + Some(FeedKind::Rdf) + ); + assert_eq!( + sniff_feed( + br#"{"version": "https://jsonfeed.org/version/1.1", "title": "A Blog", + "items": []}"# + ), + Some(FeedKind::Json) + ); + + assert_eq!(sniff_feed(SPA_SHELL), None); + assert_eq!(sniff_feed(b""), None); + assert_eq!(sniff_feed(b"{\"items\": []}"), None); + assert_eq!(sniff_feed(b"hi"), None); + } + + #[test] + fn feed_title_is_unescaped_and_trimmed() { + assert_eq!( + feed_title( + br#" + Ben & Dave's <Weekly> "# + ) + .as_deref(), + Some("Ben & Dave's ") + ); + assert_eq!( + feed_title(b"<![CDATA[ Phil's blog ]]>").as_deref(), + Some("Phil's blog") + ); + assert_eq!(feed_title(b"x"), None); + assert_eq!(feed_title(b" "), None); + } + + #[test] + fn one_link_tag_hit_per_title_preferring_xml() { + let results = philz_results(); + let kept = dedupe_guesses(results.iter().map(|r| validated(r, None)).collect()); + assert_eq!(kept.len(), 1); + assert_eq!(kept[0].url, "https://blog.philz.dev/feed/feed.xml"); + assert_eq!(kept[0].title().as_deref(), Some("blog.philz.dev")); + + // JSON first still yields the XML flavour. + let mut reversed = results; + reversed.reverse(); + let kept = dedupe_guesses(reversed.iter().map(|r| validated(r, None)).collect()); + assert_eq!(kept.len(), 1); + assert_eq!(kept[0].kind, "atom"); + } + + #[test] + fn at_most_one_guess_survives_and_it_takes_the_sniffed_title() { + let kept = dedupe_guesses( + zombo_results() + .iter() + .map(|r| validated(r, Some("Zombo Feed"))) + .collect(), + ); + assert_eq!(kept.len(), 1); + assert_eq!(kept[0].url, "https://zombo.com/atom.xml"); + assert_eq!(kept[0].title().as_deref(), Some("Zombo Feed")); + + // Three link-tag hits plus a guess still stop at MAX_PER_HOST. + let mut many: Vec = ["Main", "Rust tag", "Notes", "Extra"] + .iter() + .enumerate() + .map(|(i, title)| { + validated( + &discovered(title, &format!("https://a.dev/{i}.xml"), "atom"), + None, + ) + }) + .collect(); + many.push(validated( + &discovered("https://a.dev/feed/", "https://a.dev/feed/", "rss"), + None, + )); + let kept = dedupe_guesses(many); + assert_eq!(kept.len(), MAX_PER_HOST); + assert!(kept.iter().all(|k| !k.is_guess())); + } + + #[test] + fn the_zombo_response_yields_no_candidates_because_nothing_sniffs() { + // Every one of the nine guessed paths answers with the SPA shell. + let survivors: Vec = zombo_results() + .iter() + .filter(|r| keep_result(r) && sniff_feed(SPA_SHELL).is_some()) + .map(|r| validated(r, None)) + .collect(); + assert!(survivors.is_empty()); + assert!(dedupe_guesses(survivors).is_empty()); + } + + // --- ranking ------------------------------------------------------- + + fn evidence(value: f64, interests: &[&str]) -> ArticleEvidence { + ArticleEvidence { + article_id: 1, + title: "T".into(), + value, + interests: interests.iter().map(|s| s.to_string()).collect(), + } + } + + #[test] + fn no_evidence_scores_the_prior() { + assert!((score(&[]) - PRIOR * 100.0).abs() < 1e-9); + } + + #[test] + fn a_rating_dominates_and_shrinkage_favours_consistency() { + let loved = score(&[evidence(1.0, &[])]); + let disliked = score(&[evidence(-1.0, &[])]); + assert!(loved > 50.0 && disliked < 5.0); + + let one_great = score(&[evidence(0.95, &[])]); + let three_good = score(&[evidence(0.8, &[]), evidence(0.8, &[]), evidence(0.8, &[])]); + assert!(three_good > one_great, "{three_good} vs {one_great}"); + assert!((0.0..=100.0).contains(&three_good)); + } + + #[test] + fn why_names_the_modal_interests() { + let names = why(&[ + evidence(0.5, &["rust", "databases"]), + evidence(0.5, &["rust", "typography"]), + evidence(0.5, &["rust", "databases", "urbanism"]), + ]); + assert_eq!(names[0], "rust"); + assert_eq!(names[1], "databases"); + assert_eq!(names.len(), 3); + assert!(why(&[]).is_empty()); + } + + #[test] + fn stage_value_prefers_verdict_then_selection_then_utility_then_blend() { + let mut signals = SignalsJson { + v: 1, + ..Default::default() + }; + signals.weights.insert("interest".into(), 1.0); + signals.norm.insert("interest".into(), 0.4); + signals.top_interests.push(TopInterest { + name: "rust".into(), + z: 1.0, + cos: 0.5, + }); + + assert_eq!(stage_value("selected", None, &signals), Some(0.85)); + assert_eq!(stage_value("assessed", Some(72.0), &signals), Some(0.72)); + let blend = stage_value("triaged", None, &signals).unwrap(); + assert!((blend - 0.4).abs() < 1e-9, "{blend}"); + assert_eq!(stage_value("excluded", None, &SignalsJson::default()), None); + } + + // --- database ------------------------------------------------------ + + async fn seed_article(db: &Db, url: &str) -> ArticleId { + db.upsert_article(&article(0, url, &[SourceKind::HnFrontpage])) + .await + .unwrap() + } + + #[tokio::test] + async fn upsert_bumps_last_seen_and_never_undoes_a_decision() { + let (_dir, db) = test_db().await; + let first = ts("2026-09-01T00:00:00Z"); + let id = upsert_candidate(&db, "https://a.dev/feed.xml", "a.dev", None, first) + .await + .unwrap(); + + set_status(&db, id, "dismissed", None, first).await.unwrap(); + let later = ts("2026-09-08T00:00:00Z"); + let again = upsert_candidate( + &db, + "https://a.dev/feed.xml", + "a.dev", + Some("A Blog"), + later, + ) + .await + .unwrap(); + assert_eq!(again, id); + + let row = candidate(&db, id).await.unwrap().unwrap(); + assert_eq!(row.status, "dismissed"); + assert_eq!(row.first_seen, fmt_ts(first)); + assert_eq!(row.last_seen, fmt_ts(later)); + assert_eq!(row.title.as_deref(), Some("A Blog")); + assert_eq!(count(&db, "dismissed").await.unwrap(), 1); + assert_eq!(count(&db, "candidate").await.unwrap(), 0); + // A dismissed host's candidates are not offered for re-linking. + assert!(candidates_for_host(&db, "a.dev").await.unwrap().is_empty()); + } + + #[tokio::test] + async fn the_host_memo_records_what_was_checked() { + let (_dir, db) = test_db().await; + let now = ts("2026-09-07T12:00:00Z"); + assert!( + hosts_checked(&db, &["a.dev".into()]) + .await + .unwrap() + .is_empty() + ); + mark_host_checked(&db, "a.dev", 2, now).await.unwrap(); + mark_host_checked(&db, "a.dev", 3, now).await.unwrap(); + let memo = hosts_checked(&db, &["a.dev".into(), "b.dev".into()]) + .await + .unwrap(); + assert_eq!(memo.len(), 1); + assert_eq!(memo["a.dev"], (now, 3)); + } + + #[tokio::test] + async fn reconcile_marks_candidates_the_operator_already_subscribes_to() { + let (_dir, db) = test_db().await; + let now = ts("2026-09-07T12:00:00Z"); + let by_url = upsert_candidate(&db, "https://a.dev/feed.xml", "a.dev", None, now) + .await + .unwrap(); + let by_host = upsert_candidate(&db, "https://b.dev/atom.xml", "b.dev", None, now) + .await + .unwrap(); + let untouched = upsert_candidate(&db, "https://c.dev/feed.xml", "c.dev", None, now) + .await + .unwrap(); + + let mut subscribed = HashMap::new(); + subscribed.insert( + 7, + FeedMeta { + id: 7, + title: "A".into(), + site_url: "https://a.dev".into(), + feed_url: "https://a.dev/feed.xml/".into(), + category: None, + }, + ); + subscribed.insert( + 9, + FeedMeta { + id: 9, + title: "B".into(), + site_url: "https://www.b.dev/".into(), + feed_url: "https://b.dev/other.xml".into(), + category: None, + }, + ); + assert_eq!(reconcile_added(&db, &subscribed, now).await.unwrap(), 2); + + assert_eq!( + candidate(&db, by_url).await.unwrap().unwrap().status, + "added" + ); + let b = candidate(&db, by_host).await.unwrap().unwrap(); + assert_eq!(b.status, "added"); + assert_eq!(b.miniflux_feed_id, Some(9)); + assert_eq!( + candidate(&db, untouched).await.unwrap().unwrap().status, + "candidate" + ); + // Nothing left to reconcile on a second pass. + assert_eq!(reconcile_added(&db, &subscribed, now).await.unwrap(), 0); + } + + #[tokio::test] + async fn listing_paginates_and_carries_the_linked_articles() { + let (_dir, db) = test_db().await; + let article_id = seed_article(&db, "https://a.dev/post").await; + for n in 0..5 { + let now = ts("2026-09-07T12:00:00Z") + .checked_add(jiff::Span::new().minutes(n)) + .unwrap(); + let id = upsert_candidate(&db, &format!("https://a.dev/{n}.xml"), "a.dev", None, now) + .await + .unwrap(); + link_article(&db, id, article_id).await.unwrap(); + link_article(&db, id, article_id).await.unwrap(); + } + let (first, total) = list(&db, "candidate", 1, 2).await.unwrap(); + assert_eq!(total, 5); + assert_eq!(first.len(), 2); + assert_eq!(first[0].feed_url, "https://a.dev/4.xml"); + assert_eq!(first[0].article_ids, vec![article_id]); + + let (last, _) = list(&db, "candidate", 3, 2).await.unwrap(); + assert_eq!(last.len(), 1); + assert_eq!(last[0].feed_url, "https://a.dev/0.xml"); + assert_eq!(all_candidates(&db).await.unwrap().len(), 5); + assert!(list(&db, "added", 1, 2).await.unwrap().0.is_empty()); + } + + #[tokio::test] + async fn evidence_prefers_a_rating_over_the_run_telemetry() { + let (_dir, db) = test_db().await; + let rated = seed_article(&db, "https://a.dev/rated").await; + let scored = seed_article(&db, "https://a.dev/scored").await; + let unknown = seed_article(&db, "https://a.dev/unknown").await; + let run_id = db + .start_run("2026-09-07".parse().unwrap(), ts("2026-09-07T00:00:00Z")) + .await + .unwrap(); + for (article_id, stage, utility) in + [(rated, "excluded", None), (scored, "selected", Some(90.0))] + { + sqlx::query( + "INSERT INTO candidate_runs (run_id, article_id, stage, signals_json, utility) + VALUES (?, ?, ?, ?, ?)", + ) + .bind(run_id) + .bind(article_id) + .bind(stage) + .bind(r#"{"v":1,"top_interests":[{"name":"rust","z":1.0,"cos":0.5}]}"#) + .bind(utility) + .execute(db.pool()) + .await + .unwrap(); + } + crate::rate::record_explicit( + &crate::config::Config::default(), + &db, + rated, + Some(crate::types::Vote::Loved), + "test", + None, + None, + ) + .await + .unwrap(); + + let evidence = load_evidence(&db, &[rated, scored, unknown], 180) + .await + .unwrap(); + assert_eq!(evidence.len(), 2); + assert_eq!(evidence[&rated].value, 1.0); + assert_eq!(evidence[&scored].value, 0.85); + assert_eq!(evidence[&scored].interests, vec!["rust".to_string()]); + assert_eq!(evidence[&scored].title, "Article 0"); + assert!(!evidence.contains_key(&unknown)); + assert!(load_evidence(&db, &[], 180).await.unwrap().is_empty()); + } + + #[tokio::test] + async fn a_run_with_nothing_to_do_is_a_no_op() { + let (_dir, db) = test_db().await; + let http = crate::http::build_client(crate::http::DEFAULT_TIMEOUT).unwrap(); + let client = miniflux(http.clone(), "http://127.0.0.1:1/"); + let summary = run( + &db, + &client, + &http, + &DiscoveryConfig::default(), + &[], + &HashMap::new(), + ts("2026-09-07T12:00:00Z"), + ) + .await + .unwrap(); + assert_eq!(summary, Summary::default()); + assert_eq!( + summary.to_string(), + "0 hosts checked, 0 reused, 0 already subscribed · 0 results rejected · 0 new candidates, 0 articles linked" + ); + } + + #[tokio::test] + async fn subscribed_and_skipped_hosts_are_never_looked_up() { + let (_dir, db) = test_db().await; + let http = crate::http::build_client(crate::http::DEFAULT_TIMEOUT).unwrap(); + let client = miniflux(http.clone(), "http://127.0.0.1:1/"); + + let mut subscribed = HashMap::new(); + subscribed.insert( + 7, + FeedMeta { + id: 7, + title: "A".into(), + site_url: "https://www.a.dev/".into(), + feed_url: "https://a.dev/feed.xml".into(), + category: None, + }, + ); + let articles = vec![ + article(1, "https://a.dev/post", &[SourceKind::HnFrontpage]), + article( + 2, + "https://news.ycombinator.com/item?id=1", + &[SourceKind::HnFrontpage], + ), + article(3, "https://b.dev/post", &[SourceKind::Feed]), + ]; + let summary = run( + &db, + &client, + &http, + &DiscoveryConfig::default(), + &articles, + &subscribed, + ts("2026-09-07T12:00:00Z"), + ) + .await + .unwrap(); + assert_eq!(summary.hosts_skipped_subscribed, 1); + assert_eq!(summary.hosts_checked, 0); + assert_eq!(summary.candidates_new, 0); + } + + #[tokio::test] + async fn a_fresh_memo_links_new_articles_without_a_lookup() { + let (_dir, db) = test_db().await; + let http = crate::http::build_client(crate::http::DEFAULT_TIMEOUT).unwrap(); + let client = miniflux(http.clone(), "http://127.0.0.1:1/"); + + let now = ts("2026-09-07T12:00:00Z"); + let article_id = seed_article(&db, "https://b.dev/post").await; + let candidate_id = upsert_candidate(&db, "https://b.dev/feed.xml", "b.dev", None, now) + .await + .unwrap(); + mark_host_checked(&db, "b.dev", 1, now).await.unwrap(); + + let mut lead = article(article_id, "https://b.dev/post", &[SourceKind::Lobsters]); + lead.id = article_id; + let summary = run( + &db, + &client, + &http, + &DiscoveryConfig::default(), + &[lead], + &HashMap::new(), + now.checked_add(jiff::Span::new().hours(24)).unwrap(), + ) + .await + .unwrap(); + assert_eq!(summary.hosts_reused, 1); + assert_eq!(summary.hosts_checked, 0); + assert_eq!(summary.articles_linked, 1); + assert_eq!( + candidate(&db, candidate_id) + .await + .unwrap() + .unwrap() + .article_ids, + vec![article_id] + ); + } +} diff --git a/src/lib.rs b/src/lib.rs index bb7b1d9..c7fb7d9 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -19,6 +19,7 @@ pub mod config; pub mod curate; pub mod db; pub mod dedupe; +pub mod discovery; pub mod epub; pub mod extract; pub mod html; diff --git a/src/main.rs b/src/main.rs index dd32b97..5dce137 100644 --- a/src/main.rs +++ b/src/main.rs @@ -17,7 +17,7 @@ use daily_epub::db::Db; use daily_epub::pipeline::{self, GenerateOptions, GenerateOutcome}; use daily_epub::report::{RunReport, VOYAGE_PROVIDER}; use daily_epub::types::{ArticleId, Vote}; -use daily_epub::{curate, http, imports, jobs, lock, rate, server, social}; +use daily_epub::{curate, discovery, http, imports, jobs, lock, rate, server, social}; /// A personalized daily newspaper, delivered as an EPUB. #[derive(Debug, Parser)] @@ -61,6 +61,9 @@ enum Command { /// Manage dashboard users without taking the pipeline run lock. #[command(subcommand)] Users(UsersCommand), + /// Feed subscription candidates (feed discovery plan §4 step 6). + #[command(subcommand)] + Feeds(FeedsCommand), /// Operator jobs (what `daily-epub-job@.service` runs). #[command(subcommand)] Job(JobCommand), @@ -76,6 +79,23 @@ enum JobCommand { }, } +#[derive(Debug, Subcommand)] +enum FeedsCommand { + /// Look for feeds behind recent aggregator-only articles and record the + /// ones not already subscribed as candidates for `/dashboard/feeds`. + Discover(FeedsDiscoverArgs), +} + +#[derive(Debug, clap::Args)] +struct FeedsDiscoverArgs { + /// How far back to look for articles. + #[arg(long, default_value_t = 14)] + days: i64, + /// How many not-yet-checked hosts to look up. + #[arg(long, default_value_t = 200)] + limit: usize, +} + #[derive(Debug, Subcommand)] enum UsersCommand { /// Add a user. @@ -381,6 +401,10 @@ async fn main() -> Result<()> { let db = Db::open_and_migrate(&config.database_path).await?; cmd_users(&db, command).await?; } + Command::Feeds(FeedsCommand::Discover(args)) => { + let db = Db::open_and_migrate(&config.database_path).await?; + println!("{}", cmd_feeds_discover(&config, &db, args).await?); + } Command::Job(JobCommand::Run { name }) => { let Some(job) = jobs::Job::parse(&name) else { eprintln!("unknown job {name:?}; the catalogue is:"); @@ -411,6 +435,8 @@ async fn main() -> Result<()> { fn lock_holder(command: &Command) -> Option<&'static str> { match command { Command::Generate(_) => Some("generate"), + // Writes the same tables the discovery stage of a run writes. + Command::Feeds(FeedsCommand::Discover(_)) => Some("feeds discover"), Command::Profile(ProfileCommand::Rebuild) => Some("profile rebuild"), Command::Features(FeaturesCommand::Backfill(_)) => Some("features backfill"), Command::BackfillSocial(_) => Some("backfill-social"), @@ -430,6 +456,39 @@ fn lock_holder(command: &Command) -> Option<&'static str> { } } +/// `daily-epub feeds discover` — the same stage `generate` runs, over the last +/// `--days` of articles, so the dashboard page is useful before tomorrow's run. +async fn cmd_feeds_discover(config: &Config, db: &Db, args: FeedsDiscoverArgs) -> Result { + let since = jiff::Timestamp::now() + .checked_sub(jiff::Span::new().hours(args.days.max(0).saturating_mul(24))) + .unwrap_or(jiff::Timestamp::UNIX_EPOCH); + let articles = discovery::articles_since(db, since).await?; + let http = http::build_client(http::DEFAULT_TIMEOUT).context("building http client")?; + let client = daily_epub::miniflux::MinifluxClient::new(&config.miniflux, http.clone()) + .context("constructing the miniflux client")?; + let feeds = client + .feed_map() + .await + .context("loading the miniflux feed list")?; + let mut cfg = config.discovery.clone(); + cfg.max_lookups_per_run = args.limit; + let summary = discovery::run( + db, + &client, + &http, + &cfg, + &articles, + &feeds, + jiff::Timestamp::now(), + ) + .await?; + Ok(format!( + "{} articles since {}: {summary}", + articles.len(), + since + )) +} + async fn cmd_users(db: &Db, command: UsersCommand) -> Result<()> { use daily_epub::web::users; match command { diff --git a/src/miniflux.rs b/src/miniflux.rs index 3c3b908..2e74c03 100644 --- a/src/miniflux.rs +++ b/src/miniflux.rs @@ -1,13 +1,16 @@ //! Miniflux API client (spec §3.1). //! -//! Reads only: entries are fetched with `published_after` inside the lookback -//! window **regardless of read/unread status**, and read state is never mutated -//! so normal reader usage is undisturbed. +//! Almost reads only: entries are fetched with `published_after` inside the +//! lookback window **regardless of read/unread status**, and read state is never +//! mutated so normal reader usage is undisturbed. The single writing call is +//! [`MinifluxClient::create_feed`], which the dashboard's feed-discovery page +//! uses to subscribe to a candidate the operator picked (feed discovery plan +//! §4 step 2); the pipeline never calls it. use std::collections::HashMap; use jiff::Timestamp; -use serde::{Deserialize, Serialize}; +use serde::{Deserialize, Serialize, de::DeserializeOwned}; use crate::config::MinifluxConfig; use crate::http::{RetryPolicy, is_retryable}; @@ -24,11 +27,13 @@ pub enum MinifluxError { MissingApiKey, #[error("miniflux request failed: {0}")] Http(#[from] reqwest::Error), - #[error("miniflux returned {status} for {path}: {body}")] + #[error("miniflux returned {status} for {path}: {message}")] Status { status: u16, path: String, - body: String, + /// `error_message` when the body was Miniflux's JSON error envelope, + /// else the truncated body — what a dashboard flash should show. + message: String, }, #[error("could not parse miniflux response for {path}: {source}")] Decode { @@ -40,6 +45,46 @@ pub enum MinifluxError { type Result = std::result::Result; +impl MinifluxError { + /// Build a [`MinifluxError::Status`], pulling `error_message` out of + /// Miniflux's JSON error envelope when the body has one (§4 step 2). + fn status(status: u16, path: &str, body: &str) -> Self { + #[derive(Deserialize)] + struct ErrorBody { + error_message: String, + } + let body: String = body.chars().take(300).collect(); + let message = serde_json::from_str::(&body) + .map(|e| e.error_message) + .unwrap_or_else(|_| body.clone()); + MinifluxError::Status { + status, + path: path.to_string(), + message, + } + } + + /// True for the `400 This feed already exists.` a duplicate `POST /v1/feeds` + /// returns — the dashboard flips the row to `added` on it instead of + /// treating it as a failure. + pub fn is_duplicate_feed(&self) -> bool { + matches!( + self, + MinifluxError::Status { status: 400, message, .. } + if message.trim().eq_ignore_ascii_case("This feed already exists.") + ) + } +} + +/// Network failures and 5xx/429 responses are worth another attempt (§3.1). +fn retryable(error: &MinifluxError) -> bool { + match error { + MinifluxError::Http(e) => is_retryable(e), + MinifluxError::Status { status, .. } => *status >= 500 || *status == 429, + _ => false, + } +} + // --------------------------------------------------------------------------- // Wire types (only the fields §3.1 lists as used) // --------------------------------------------------------------------------- @@ -106,6 +151,27 @@ pub struct EntriesResponse { pub entries: Vec, } +/// `POST /v1/discover` element — a *lead*, not a verified feed: Miniflux +/// returns well-known paths that merely answered 200 (plan §2), so every one is +/// validated before it becomes a candidate. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct Discovered { + #[serde(default)] + pub url: String, + /// Miniflux substitutes the URL when the `` tag has no title, which is + /// how a well-known-path guess is told apart from a real link-tag hit. + #[serde(default)] + pub title: String, + #[serde(default, rename = "type")] + pub kind: String, +} + +/// `POST /v1/feeds` response. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct CreatedFeed { + pub feed_id: i64, +} + /// Feed metadata joined onto every entry we persist (§3.1). #[derive(Debug, Clone, PartialEq)] pub struct FeedMeta { @@ -247,34 +313,22 @@ impl MinifluxClient { let url = self.url(path); let body = self .retry - .run( - &format!("GET {path}"), - |e: &MinifluxError| match e { - MinifluxError::Http(e) => is_retryable(e), - MinifluxError::Status { status, .. } => *status >= 500 || *status == 429, - _ => false, - }, - || async { - let resp = self - .http - .get(&url) - .header("X-Auth-Token", &self.api_key) - .header("Accept", "application/json") - .query(query) - .send() - .await?; - let status = resp.status(); - let text = resp.text().await?; - if !status.is_success() { - return Err(MinifluxError::Status { - status: status.as_u16(), - path: path.to_string(), - body: text.chars().take(300).collect(), - }); - } - Ok(text) - }, - ) + .run(&format!("GET {path}"), retryable, || async { + let resp = self + .http + .get(&url) + .header("X-Auth-Token", &self.api_key) + .header("Accept", "application/json") + .query(query) + .send() + .await?; + let status = resp.status(); + let text = resp.text().await?; + if !status.is_success() { + return Err(MinifluxError::status(status.as_u16(), path, &text)); + } + Ok(text) + }) .await?; serde_json::from_str(&body).map_err(|source| MinifluxError::Decode { path: path.to_string(), @@ -282,6 +336,74 @@ impl MinifluxClient { }) } + /// One `POST path` with a JSON body, no retry. + async fn post_once(&self, path: &str, body: &B) -> Result { + let resp = self + .http + .post(self.url(path)) + .header("X-Auth-Token", &self.api_key) + .header("Accept", "application/json") + .json(body) + .send() + .await?; + let status = resp.status(); + let text = resp.text().await?; + if !status.is_success() { + return Err(MinifluxError::status(status.as_u16(), path, &text)); + } + Ok(text) + } + + /// POST `path` with the auth header, mirroring [`Self::get_json`]. + /// + /// `retry` is false for calls that are not safe to repeat: a retried + /// `POST /v1/feeds` whose response was merely lost would subscribe twice. + async fn post_json( + &self, + path: &str, + body: &B, + retry: bool, + ) -> Result { + let text = if retry { + self.retry + .run(&format!("POST {path}"), retryable, || { + self.post_once(path, body) + }) + .await? + } else { + self.post_once(path, body).await? + }; + serde_json::from_str(&text).map_err(|source| MinifluxError::Decode { + path: path.to_string(), + source, + }) + } + + /// `POST /v1/discover` — the feed(s) Miniflux can find behind a page URL + /// (feed discovery plan §4 step 2). Retried: it changes nothing. + pub async fn discover(&self, url: &str) -> Result> { + self.post_json("/discover", &serde_json::json!({ "url": url }), true) + .await + } + + /// `GET /v1/categories` — the Add form's category picker. + pub async fn categories(&self) -> Result> { + self.get_json("/categories", &[]).await + } + + /// `POST /v1/feeds` — subscribe to `feed_url` in `category_id`, returning + /// the new feed id. Never retried (a repeat can double-subscribe). + pub async fn create_feed(&self, feed_url: &str, category_id: i64) -> Result { + let created: CreatedFeed = self + .post_json( + "/feeds", + &serde_json::json!({ "feed_url": feed_url, "category_id": category_id }), + false, + ) + .await?; + Ok(created.feed_id) + } + /// `GET /v1/feeds` — once per run (§3.1). pub async fn feeds(&self) -> Result> { self.get_json("/feeds", &[]).await @@ -483,6 +605,93 @@ mod tests { assert_eq!(entries[1].category, None); } + /// Observed live for an article URL whose page carries `` + /// tags: the same feed offered twice under one title (plan §2). + const DISCOVER_LINK_TAGS_JSON: &str = r#"[ + {"title":"blog.philz.dev","url":"https://blog.philz.dev/feed/feed.xml","type":"atom"}, + {"title":"blog.philz.dev","url":"https://blog.philz.dev/feed/feed.json","type":"json"} + ]"#; + + /// Observed live for a site that answers 200 for any path: nine unverified + /// well-known-path guesses, each with `title == url` (plan §2). + const DISCOVER_GUESSES_JSON: &str = r#"[ + {"title":"https://zombo.com/atom.xml","url":"https://zombo.com/atom.xml","type":"atom"}, + {"title":"https://zombo.com/feed.atom","url":"https://zombo.com/feed.atom","type":"atom"}, + {"title":"https://zombo.com/feed.xml","url":"https://zombo.com/feed.xml","type":"atom"}, + {"title":"https://zombo.com/feed/","url":"https://zombo.com/feed/","type":"atom"}, + {"title":"https://zombo.com/index.rss","url":"https://zombo.com/index.rss","type":"rss"}, + {"title":"https://zombo.com/index.xml","url":"https://zombo.com/index.xml","type":"rss"}, + {"title":"https://zombo.com/rss.xml","url":"https://zombo.com/rss.xml","type":"rss"}, + {"title":"https://zombo.com/rss/","url":"https://zombo.com/rss/","type":"rss"}, + {"title":"https://zombo.com/rss/feed.xml","url":"https://zombo.com/rss/feed.xml","type":"rss"} + ]"#; + + const CATEGORIES_JSON: &str = r#"[ + {"id": 1, "user_id": 1, "title": "Tech", "hide_globally": false}, + {"id": 4, "user_id": 1, "title": "Long reads", "hide_globally": true} + ]"#; + + #[test] + fn deserializes_discover_results() { + let hits: Vec = serde_json::from_str(DISCOVER_LINK_TAGS_JSON).unwrap(); + assert_eq!(hits.len(), 2); + assert_eq!(hits[0].title, "blog.philz.dev"); + assert_eq!(hits[0].url, "https://blog.philz.dev/feed/feed.xml"); + assert_eq!(hits[0].kind, "atom"); + assert_eq!(hits[1].kind, "json"); + + let guesses: Vec = serde_json::from_str(DISCOVER_GUESSES_JSON).unwrap(); + assert_eq!(guesses.len(), 9); + assert!(guesses.iter().all(|g| g.title == g.url)); + } + + #[test] + fn deserializes_categories_and_created_feed() { + let categories: Vec = serde_json::from_str(CATEGORIES_JSON).unwrap(); + assert_eq!(categories.len(), 2); + assert_eq!(categories[1].id, 4); + assert_eq!(categories[1].title, "Long reads"); + + let created: CreatedFeed = serde_json::from_str(r#"{"feed_id": 123}"#).unwrap(); + assert_eq!(created.feed_id, 123); + } + + #[test] + fn status_errors_carry_the_error_message() { + let duplicate = MinifluxError::status( + 400, + "/feeds", + r#"{"error_message":"This feed already exists."}"#, + ); + assert!(duplicate.is_duplicate_feed()); + assert_eq!( + duplicate.to_string(), + "miniflux returned 400 for /feeds: This feed already exists." + ); + + let fetcher = MinifluxError::status( + 502, + "/discover", + r#"{"error_message":"fetcher: bad gateway (502 status code)"}"#, + ); + assert!(!fetcher.is_duplicate_feed()); + assert!(retryable(&fetcher)); + match fetcher { + MinifluxError::Status { message, .. } => { + assert_eq!(message, "fetcher: bad gateway (502 status code)"); + } + other => panic!("unexpected error {other:?}"), + } + + // A non-JSON body falls back to the truncated body itself. + let html = MinifluxError::status(404, "/discover", "nope"); + assert!(!retryable(&html)); + assert_eq!( + html.to_string(), + "miniflux returned 404 for /discover: nope" + ); + } + #[test] fn requires_an_api_key() { let http = crate::http::build_client(crate::http::DEFAULT_TIMEOUT).unwrap(); diff --git a/src/pipeline.rs b/src/pipeline.rs index 297fe72..cb456b5 100644 --- a/src/pipeline.rs +++ b/src/pipeline.rs @@ -46,7 +46,7 @@ use crate::types::{ Article, ArticleId, Artifact, BehindThePaper, Candidate, Colophon, Edition, Issue, IssueMeta, Lineup, Models, TokenUsage, reading_minutes, }; -use crate::{comments, dedupe, epub, http, miniflux, publish, social, world}; +use crate::{comments, dedupe, discovery, epub, http, miniflux, publish, social, world}; /// One `generate` invocation's inputs — the CLI flags, already parsed (§2). #[derive(Debug, Clone, Default)] @@ -449,6 +449,32 @@ async fn run_stages( report.counts.social_hits = enricher.enrich_all(&mut articles).await as i64; report.timings.record("social", elapsed_ms(stage)); + // --- Stage 5b: feed discovery (feed discovery plan §4) — best effort --- + // Runs here because it needs the real article ids stage 4 minted and the + // subscription map stage 1 already loaded, and because `articles` is moved + // into hygiene next. + if config.discovery.enabled { + let stage = Timestamp::now(); + match discovery::run( + db, + &client, + &http, + &config.discovery, + &articles, + &feeds, + Timestamp::now(), + ) + .await + { + Ok(summary) => { + tracing::info!(%summary, "feed discovery"); + report.counts.feed_candidates_new = summary.candidates_new as i64; + } + Err(error) => report.warn(format!("feed discovery failed: {error:#}")), + } + report.timings.record("discovery", elapsed_ms(stage)); + } + // --- Stage 6: hygiene, embeddings, and cheap signals (§8.1, §9) --- let stage = Timestamp::now(); let mut personalized = admit::hygiene( diff --git a/src/report.rs b/src/report.rs index e2827b4..23d9be4 100644 --- a/src/report.rs +++ b/src/report.rs @@ -67,6 +67,10 @@ pub struct StageCounts { pub excerpt_only: i64, /// Social lookups that returned a hit (§3.4). pub social_hits: i64, + /// Feed candidates recorded for the first time by the discovery stage + /// (feed discovery plan §4 step 3). + #[serde(default)] + pub feed_candidates_new: i64, /// Articles passing hygiene and eligible for personalized signals. pub eligible: i64, /// Eligible articles with a valid embedding.