Merge branch 'fd-core' into feed-discovery

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
+);

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;

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@<name>.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<String> {
+    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 {

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<T> = std::result::Result<T, MinifluxError>;
 
+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::<ErrorBody>(&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<MinifluxEntry>,
 }
 
+/// `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 `<link>` 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<B: Serialize>(&self, path: &str, body: &B) -> Result<String> {
+        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<B: Serialize, T: DeserializeOwned>(
+        &self,
+        path: &str,
+        body: &B,
+        retry: bool,
+    ) -> Result<T> {
+        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<Vec<Discovered>> {
+        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<Vec<MinifluxCategory>> {
+        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<i64> {
+        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<Vec<MinifluxFeed>> {
         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 `<link rel=alternate>`
+    /// 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<Discovered> = 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<Discovered> = 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<MinifluxCategory> = 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", "<html>nope</html>");
+        assert!(!retryable(&html));
+        assert_eq!(
+            html.to_string(),
+            "miniflux returned 404 for /discover: <html>nope</html>"
+        );
+    }
+
     #[test]
     fn requires_an_api_key() {
         let http = crate::http::build_client(crate::http::DEFAULT_TIMEOUT).unwrap();

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(

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.