Step 3: feed fetch and parse

Conditional GET plus an RSS-first, Atom-fallback parser normalising both
into one entry model, with iTunes explicit and ttl handling carried over
from the Python. Enclosures are recorded as pending; nothing downloads yet.

GUID falls back guid -> link -> enclosure url -> title rather than hashing
the description as the original did.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RPyeapneuXrCdojsaiXGbe

src/db.rs

@@ -121,6 +121,128 @@ impl Db {
     }
 }
 
+
+/// Everything `fetch` needs to decide whether to poll a feed, and how.
+#[derive(Debug, Default)]
+pub struct HttpState {
+    pub etag: Option<String>,
+    pub last_modified: Option<String>,
+    pub last_checked: Option<i64>,
+    pub ttl_mins: Option<u64>,
+}
+
+impl Db {
+    pub fn http_state(&self, feed_id: &str) -> Result<HttpState> {
+        let conn = self.conn.lock().unwrap();
+        Ok(conn
+            .query_row(
+                "SELECT etag, last_modified, last_checked, ttl_mins FROM feeds WHERE id = ?1",
+                [feed_id],
+                |r| {
+                    Ok(HttpState {
+                        etag: r.get(0)?,
+                        last_modified: r.get(1)?,
+                        last_checked: r.get(2)?,
+                        ttl_mins: r.get::<_, Option<i64>>(3)?.map(|t| t.max(0) as u64),
+                    })
+                },
+            )
+            .optional()?
+            .unwrap_or_default())
+    }
+
+    /// Upsert after a successful poll. Clears any previous error.
+    pub fn record_feed(
+        &self,
+        feed_id: &str,
+        url: &str,
+        title: Option<&str>,
+        etag: Option<&str>,
+        last_modified: Option<&str>,
+        ttl_mins: Option<u64>,
+    ) -> Result<()> {
+        let conn = self.conn.lock().unwrap();
+        conn.execute(
+            "INSERT INTO feeds (id, url, title, etag, last_modified, last_checked, ttl_mins, last_error)
+             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL)
+             ON CONFLICT(id) DO UPDATE SET
+                url = excluded.url,
+                title = coalesce(excluded.title, feeds.title),
+                etag = excluded.etag,
+                last_modified = excluded.last_modified,
+                last_checked = excluded.last_checked,
+                ttl_mins = excluded.ttl_mins,
+                last_error = NULL",
+            rusqlite::params![feed_id, url, title, etag, last_modified, now(), ttl_mins.map(|t| t as i64)],
+        )?;
+        Ok(())
+    }
+
+    /// 304, or any other poll that produced no new data: only the clock moves.
+    pub fn touch_feed(&self, feed_id: &str, url: &str) -> Result<()> {
+        let conn = self.conn.lock().unwrap();
+        conn.execute(
+            "INSERT INTO feeds (id, url, last_checked) VALUES (?1, ?2, ?3)
+             ON CONFLICT(id) DO UPDATE SET last_checked = excluded.last_checked, last_error = NULL",
+            rusqlite::params![feed_id, url, now()],
+        )?;
+        Ok(())
+    }
+
+    pub fn set_feed_error(&self, feed_id: &str, url: &str, msg: &str) -> Result<()> {
+        let conn = self.conn.lock().unwrap();
+        conn.execute(
+            "INSERT INTO feeds (id, url, last_checked, last_error) VALUES (?1, ?2, ?3, ?4)
+             ON CONFLICT(id) DO UPDATE SET last_checked = excluded.last_checked, last_error = excluded.last_error",
+            rusqlite::params![feed_id, url, now(), msg],
+        )?;
+        Ok(())
+    }
+
+    /// Returns true when this entry had not been seen before.
+    ///
+    /// A changed description or title flips `read` back to 0, which is what the original's
+    /// textDiff dance was ultimately for -- minus the diff markup, which the UI can do.
+    pub fn record_entry(&self, feed_id: &str, e: &crate::feed::Entry) -> Result<bool> {
+        let conn = self.conn.lock().unwrap();
+        let inserted = conn.execute(
+            "INSERT OR IGNORE INTO entries
+                (feed_id, guid, title, link, published, description, first_seen, read, flagged)
+             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 0, 0)",
+            rusqlite::params![feed_id, e.guid, e.title, e.link, e.published, e.description, now()],
+        )?;
+        if inserted == 0 {
+            // The SET expressions see the pre-update row, so this compares old vs new.
+            conn.execute(
+                "UPDATE entries SET
+                    title = coalesce(?3, title),
+                    description = coalesce(?4, description),
+                    read = CASE WHEN description IS NOT ?4 OR title IS NOT ?3 THEN 0 ELSE read END
+                 WHERE feed_id = ?1 AND guid = ?2",
+                rusqlite::params![feed_id, e.guid, e.title, e.description],
+            )?;
+        }
+        Ok(inserted == 1)
+    }
+
+    /// Returns true when this enclosure URL is new. False means we have downloaded it
+    /// before, or deliberately reaped it -- either way it is not fetched again.
+    pub fn record_enclosure(
+        &self,
+        feed_id: &str,
+        guid: &str,
+        enc: &crate::feed::Enclosure,
+    ) -> Result<bool> {
+        let conn = self.conn.lock().unwrap();
+        let inserted = conn.execute(
+            "INSERT OR IGNORE INTO enclosures (feed_id, guid, url, mime, length, state)
+             VALUES (?1, ?2, ?3, ?4, ?5, 'pending')",
+            rusqlite::params![feed_id, guid, enc.url, enc.mime, enc.length],
+        )?;
+        Ok(inserted == 1)
+    }
+}
+
 /// Unix seconds. Everything time-shaped in the DB is stored this way.
 pub fn now() -> i64 {
     std::time::SystemTime::now()

src/feed.rs

@@ -0,0 +1,336 @@
+//! Feed fetching and parsing. Replaces FeedData.__getFeed / __getEntries.
+
+use anyhow::{Context, Result, anyhow};
+use reqwest::StatusCode;
+use reqwest::header::{ETAG, IF_MODIFIED_SINCE, IF_NONE_MATCH, LAST_MODIFIED};
+
+use crate::config::Feed as FeedCfg;
+
+#[derive(Debug, Default)]
+pub struct ParsedFeed {
+    pub title: Option<String>,
+    pub link: Option<String>,
+    pub ttl_mins: Option<u64>,
+    /// Feed-level explicit flag; per the original, it overrides the entry level.
+    pub explicit: bool,
+    pub entries: Vec<Entry>,
+}
+
+#[derive(Debug, Default)]
+pub struct Entry {
+    pub guid: String,
+    pub title: Option<String>,
+    pub link: Option<String>,
+    pub published: Option<i64>,
+    pub description: Option<String>,
+    pub categories: Vec<String>,
+    pub explicit: bool,
+    pub enclosures: Vec<Enclosure>,
+}
+
+#[derive(Debug, Default, PartialEq)]
+pub struct Enclosure {
+    pub url: String,
+    pub mime: Option<String>,
+    pub length: Option<i64>,
+}
+
+pub enum Fetched {
+    /// Server said 304, or returned a body we already have.
+    NotModified,
+    Body {
+        bytes: Vec<u8>,
+        etag: Option<String>,
+        last_modified: Option<String>,
+    },
+}
+
+/// Conditional GET. reqwest handles gzip and redirects; the original's hand-rolled
+/// CONNECT/socket.ssl proxy path is gone -- `system-proxy` reads http_proxy/https_proxy.
+pub async fn fetch(
+    client: &reqwest::Client,
+    cfg: &FeedCfg,
+    etag: Option<&str>,
+    last_modified: Option<&str>,
+) -> Result<Fetched> {
+    let mut req = client.get(&cfg.url);
+    if let Some(tag) = etag {
+        req = req.header(IF_NONE_MATCH, tag);
+    }
+    if let Some(lm) = last_modified {
+        req = req.header(IF_MODIFIED_SINCE, lm);
+    }
+    if let Some(user) = &cfg.username {
+        req = req.basic_auth(user, cfg.password());
+    }
+
+    let resp = req.send().await.context("connecting")?;
+    if resp.status() == StatusCode::NOT_MODIFIED {
+        return Ok(Fetched::NotModified);
+    }
+    let status = resp.status();
+    if !status.is_success() {
+        // The original surfaced 401/407 specially; the code is enough for a UI to switch on.
+        return Err(anyhow!("HTTP {status}"));
+    }
+
+    let header = |h: reqwest::header::HeaderName| {
+        resp.headers().get(&h).and_then(|v| v.to_str().ok()).map(str::to_owned)
+    };
+    let etag = header(ETAG);
+    let last_modified = header(LAST_MODIFIED);
+    let bytes = resp.bytes().await.context("reading body")?.to_vec();
+    Ok(Fetched::Body { bytes, etag, last_modified })
+}
+
+/// RSS first, then Atom -- the same split the original made on `parsedFeed.version`.
+pub fn parse(bytes: &[u8]) -> Result<ParsedFeed> {
+    match rss::Channel::read_from(bytes) {
+        Ok(ch) => Ok(from_rss(ch)),
+        Err(rss_err) => match atom_syndication::Feed::read_from(bytes) {
+            Ok(feed) => Ok(from_atom(feed)),
+            Err(atom_err) => Err(anyhow!("not RSS ({rss_err}) and not Atom ({atom_err})")),
+        },
+    }
+}
+
+fn from_rss(ch: rss::Channel) -> ParsedFeed {
+    let explicit = ch
+        .itunes_ext()
+        .and_then(|it| it.explicit())
+        .is_some_and(is_yes);
+
+    let entries = ch
+        .items()
+        .iter()
+        .filter_map(|item| {
+            let enclosures: Vec<Enclosure> = item
+                .enclosure()
+                .into_iter()
+                .map(|e| Enclosure {
+                    url: e.url().trim().to_owned(),
+                    mime: non_empty(Some(e.mime_type())),
+                    length: e.length().parse().ok(),
+                })
+                .filter(|e| !e.url.is_empty())
+                .collect();
+
+            let guid = pick_guid(
+                item.guid().map(|g| g.value()),
+                item.link(),
+                enclosures.first().map(|e| e.url.as_str()),
+                item.title(),
+            )?;
+
+            let entry_explicit = item
+                .itunes_ext()
+                .and_then(|it| it.explicit())
+                .is_some_and(is_yes);
+
+            Some(Entry {
+                guid,
+                title: non_empty(item.title()),
+                link: non_empty(item.link()),
+                published: item.pub_date().and_then(parse_date),
+                // Content wins over description, as __getEntries preferred entry.content.
+                description: non_empty(item.content()).or_else(|| non_empty(item.description())),
+                categories: item
+                    .categories()
+                    .iter()
+                    .map(|c| c.name().to_owned())
+                    .filter(|c| !c.is_empty() && !c.starts_with("http"))
+                    .collect(),
+                explicit: explicit || entry_explicit,
+                enclosures,
+            })
+        })
+        .collect();
+
+    ParsedFeed {
+        title: non_empty(Some(ch.title())),
+        link: non_empty(Some(ch.link())),
+        ttl_mins: ch.ttl().and_then(|t| t.trim().parse().ok()),
+        explicit,
+        entries,
+    }
+}
+
+fn from_atom(feed: atom_syndication::Feed) -> ParsedFeed {
+    let entries = feed
+        .entries()
+        .iter()
+        .filter_map(|e| {
+            // Atom carries enclosures as <link rel="enclosure">.
+            let enclosures: Vec<Enclosure> = e
+                .links()
+                .iter()
+                .filter(|l| l.rel() == "enclosure")
+                .map(|l| Enclosure {
+                    url: l.href().trim().to_owned(),
+                    mime: non_empty(l.mime_type()),
+                    length: l.length().and_then(|s| s.parse().ok()),
+                })
+                .filter(|e| !e.url.is_empty())
+                .collect();
+
+            let alt = e
+                .links()
+                .iter()
+                .find(|l| l.rel() == "alternate" || l.rel().is_empty())
+                .map(|l| l.href());
+
+            let guid = pick_guid(
+                Some(e.id()),
+                alt,
+                enclosures.first().map(|x| x.url.as_str()),
+                Some(e.title().as_str()),
+            )?;
+
+            Some(Entry {
+                guid,
+                title: non_empty(Some(e.title().as_str())),
+                link: alt.map(str::to_owned),
+                published: e.published().or(Some(e.updated())).map(|d| d.timestamp()),
+                description: e
+                    .content()
+                    .and_then(|c| c.value())
+                    .or_else(|| e.summary().map(|s| s.as_str()))
+                    .map(str::to_owned),
+                categories: e.categories().iter().map(|c| c.term().to_owned()).collect(),
+                explicit: false,
+                enclosures,
+            })
+        })
+        .collect();
+
+    ParsedFeed {
+        title: non_empty(Some(feed.title().as_str())),
+        link: feed
+            .links()
+            .iter()
+            .find(|l| l.rel() == "alternate")
+            .map(|l| l.href().to_owned()),
+        ttl_mins: None,
+        explicit: false,
+        entries,
+    }
+}
+
+/// The original fell back to hashing the title or description. A guid, permalink or
+/// enclosure URL is a stable identifier already, so no hashing is needed; an entry with
+/// none of them has nothing to download and is dropped.
+fn pick_guid(
+    guid: Option<&str>,
+    link: Option<&str>,
+    enclosure: Option<&str>,
+    title: Option<&str>,
+) -> Option<String> {
+    [guid, link, enclosure, title]
+        .into_iter()
+        .flatten()
+        .map(str::trim)
+        .find(|s| !s.is_empty())
+        .map(str::to_owned)
+}
+
+fn is_yes(s: &str) -> bool {
+    matches!(s.trim().to_ascii_lowercase().as_str(), "yes" | "true" | "explicit")
+}
+
+fn non_empty(s: Option<&str>) -> Option<String> {
+    s.map(str::trim).filter(|s| !s.is_empty()).map(str::to_owned)
+}
+
+/// RSS pubDate is RFC 2822; some feeds ship RFC 3339 instead.
+fn parse_date(s: &str) -> Option<i64> {
+    let s = s.trim();
+    chrono::DateTime::parse_from_rfc2822(s)
+        .or_else(|_| chrono::DateTime::parse_from_rfc3339(s))
+        .ok()
+        .map(|d| d.timestamp())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parses_rss_with_itunes_extensions() {
+        let bytes = include_bytes!("../tests/data/rss2.xml");
+        let feed = parse(bytes).unwrap();
+
+        assert_eq!(feed.title.as_deref(), Some("Test Cast"));
+        assert_eq!(feed.ttl_mins, Some(45));
+        assert!(!feed.explicit, "feed-level explicit is 'no'");
+        assert_eq!(feed.entries.len(), 3);
+
+        let ep = &feed.entries[0];
+        assert_eq!(ep.guid, "https://example.com/ep/1");
+        assert_eq!(ep.title.as_deref(), Some("Episode One"));
+        assert_eq!(ep.published, Some(1_078_016_400));
+        assert_eq!(ep.categories, vec!["Tech"]);
+        assert!(!ep.explicit);
+        assert_eq!(
+            ep.enclosures,
+            vec![Enclosure {
+                url: "https://example.com/ep1.mp3".into(),
+                mime: Some("audio/mpeg".into()),
+                length: Some(12_345_678),
+            }]
+        );
+
+        assert!(feed.entries[1].explicit, "entry-level itunes:explicit=yes");
+        assert_eq!(
+            feed.entries[2].enclosures[0].mime.as_deref(),
+            Some("application/x-bittorrent")
+        );
+    }
+
+    #[test]
+    fn feed_level_explicit_overrides_entries() {
+        let xml = br#"<?xml version="1.0"?>
+        <rss version="2.0" xmlns:itunes="http://www.itunes.com/dtds/podcast-1.0.dtd">
+        <channel><title>X</title><link>https://x</link><description>d</description>
+        <itunes:explicit>yes</itunes:explicit>
+        <item><title>a</title><guid>g1</guid>
+        <enclosure url="https://x/a.mp3" length="1" type="audio/mpeg"/></item>
+        </channel></rss>"#;
+        let feed = parse(xml).unwrap();
+        assert!(feed.explicit);
+        assert!(feed.entries[0].explicit, "feed level must win");
+    }
+
+    #[test]
+    fn parses_atom_enclosure_links() {
+        let bytes = include_bytes!("../tests/data/atom.xml");
+        let feed = parse(bytes).unwrap();
+
+        assert_eq!(feed.title.as_deref(), Some("Atom Cast"));
+        assert_eq!(feed.entries.len(), 1);
+        let ep = &feed.entries[0];
+        assert_eq!(ep.guid, "urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a");
+        assert_eq!(ep.link.as_deref(), Some("https://example.org/ep/1"));
+        assert_eq!(ep.published, Some(1_078_016_400));
+        assert_eq!(
+            ep.enclosures,
+            vec![Enclosure {
+                url: "https://example.org/ep1.m4a".into(),
+                mime: Some("audio/mp4".into()),
+                length: Some(9_876_543),
+            }],
+            "rel=enclosure only; the alternate link must not become an enclosure"
+        );
+    }
+
+    #[test]
+    fn rejects_html_masquerading_as_a_feed() {
+        assert!(parse(b"<html><body>nope</body></html>").is_err());
+    }
+
+    #[test]
+    fn guid_falls_back_through_link_then_enclosure() {
+        assert_eq!(pick_guid(Some(" "), Some("l"), Some("e"), None).as_deref(), Some("l"));
+        assert_eq!(pick_guid(None, None, Some("e"), Some("t")).as_deref(), Some("e"));
+        assert_eq!(pick_guid(None, None, None, None), None);
+    }
+}

src/main.rs

@@ -1,15 +1,17 @@
 mod config;
 mod db;
+mod feed;
 
 use anyhow::Result;
 use clap::{Parser, Subcommand};
+use std::path::PathBuf;
 
 #[derive(Parser)]
 #[command(name = "ipx", version, about = "A headless podcatcher")]
 struct Cli {
     /// Config file (default: $XDG_CONFIG_HOME/ipx/config.toml)
     #[arg(long, global = true)]
-    config: Option<std::path::PathBuf>,
+    config: Option<PathBuf>,
 
     #[command(subcommand)]
     command: Command,
@@ -19,28 +21,37 @@ struct Cli {
 enum Command {
     /// Show configured feeds and their state
     List,
+    /// Scan feeds for new entries
+    Fetch {
+        /// Only this feed id
+        feed: Option<String>,
+        /// Poll even when the feed is not due yet
+        #[arg(long)]
+        force: bool,
+    },
 }
 
-fn main() -> Result<()> {
+#[tokio::main]
+async fn main() -> Result<()> {
     let cli = Cli::parse();
-    let config_path = cli.config.unwrap_or_else(config::config_path);
+    let config_path = cli.config.clone().unwrap_or_else(config::config_path);
     let cfg = config::Config::load(&config_path)?;
     let db = db::Db::open(&config::data_dir().join("state.db"))?;
 
     match cli.command {
-        Command::List => list(&cfg, &db),
+        Command::List => list(&cfg, &db, &config_path),
+        Command::Fetch { feed, force } => fetch(&cfg, &db, feed.as_deref(), force).await,
     }
 }
 
-fn list(cfg: &config::Config, db: &db::Db) -> Result<()> {
+fn list(cfg: &config::Config, db: &db::Db, config_path: &std::path::Path) -> Result<()> {
     if cfg.feeds.is_empty() {
-        println!("No feeds configured in {}", config::config_path().display());
+        println!("No feeds configured in {}", config_path.display());
         return Ok(());
     }
     for (id, feed) in &cfg.feeds {
         let s = db.feed_summary(id)?;
-        let title = s.title.as_deref().unwrap_or("-");
-        println!("{id}  {title}");
+        println!("{id}  {}", s.title.as_deref().unwrap_or("-"));
         println!("    url          {}", feed.url);
         println!("    last checked {}", ago(s.last_checked));
         println!("    entries      {} ({} downloaded)", s.entries, s.downloaded);
@@ -51,13 +62,110 @@ fn list(cfg: &config::Config, db: &db::Db) -> Result<()> {
     Ok(())
 }
 
+async fn fetch(
+    cfg: &config::Config,
+    db: &db::Db,
+    only: Option<&str>,
+    force: bool,
+) -> Result<()> {
+    let client = reqwest::Client::builder()
+        .user_agent(concat!("ipx/", env!("CARGO_PKG_VERSION")))
+        .build()?;
+
+    if let Some(id) = only
+        && !cfg.feeds.contains_key(id)
+    {
+        anyhow::bail!("no feed with id {id:?}");
+    }
+
+    for (id, feed_cfg) in cfg.feeds.iter().filter(|(id, _)| only.is_none_or(|o| o == *id)) {
+        let state = db.http_state(id)?;
+
+        // TTL: the feed's own <ttl> wins when it is longer than our poll interval.
+        if !force && let Some(last) = state.last_checked {
+            let wait = state.ttl_mins.unwrap_or(0).max(cfg.general.interval_mins) * 60;
+            let due = last + wait as i64;
+            if due > db::now() {
+                println!("{id}: not due for {}", duration((due - db::now()) as u64));
+                continue;
+            }
+        }
+
+        match scan_one(&client, db, id, feed_cfg, &state).await {
+            Ok(Some((new_entries, new_encs))) => {
+                println!("{id}: {new_entries} new entries, {new_encs} new enclosures")
+            }
+            Ok(None) => println!("{id}: not modified"),
+            Err(e) => {
+                // One bad feed must not end the scan.
+                let msg = format!("{e:#}");
+                println!("{id}: error: {msg}");
+                db.set_feed_error(id, &feed_cfg.url, &msg)?;
+            }
+        }
+    }
+    Ok(())
+}
+
+/// Ok(None) means 304.
+async fn scan_one(
+    client: &reqwest::Client,
+    db: &db::Db,
+    id: &str,
+    feed_cfg: &config::Feed,
+    state: &db::HttpState,
+) -> Result<Option<(usize, usize)>> {
+    let fetched = feed::fetch(
+        client,
+        feed_cfg,
+        state.etag.as_deref(),
+        state.last_modified.as_deref(),
+    )
+    .await?;
+
+    let (bytes, etag, last_modified) = match fetched {
+        feed::Fetched::NotModified => {
+            db.touch_feed(id, &feed_cfg.url)?;
+            return Ok(None);
+        }
+        feed::Fetched::Body { bytes, etag, last_modified } => (bytes, etag, last_modified),
+    };
+
+    let parsed = feed::parse(&bytes)?;
+    db.record_feed(
+        id,
+        &feed_cfg.url,
+        parsed.title.as_deref(),
+        etag.as_deref(),
+        last_modified.as_deref(),
+        parsed.ttl_mins,
+    )?;
+
+    let mut new_entries = 0;
+    let mut new_encs = 0;
+    for entry in &parsed.entries {
+        if db.record_entry(id, entry)? {
+            new_entries += 1;
+        }
+        for enc in &entry.enclosures {
+            if db.record_enclosure(id, &entry.guid, enc)? {
+                new_encs += 1;
+            }
+        }
+    }
+    Ok(Some((new_entries, new_encs)))
+}
+
 fn ago(t: Option<i64>) -> String {
     let Some(t) = t else { return "never".into() };
-    let secs = (db::now() - t).max(0);
+    format!("{} ago", duration((db::now() - t).max(0) as u64))
+}
+
+fn duration(secs: u64) -> String {
     match secs {
-        s if s < 90 => format!("{s}s ago"),
-        s if s < 5400 => format!("{}m ago", s / 60),
-        s if s < 172_800 => format!("{}h ago", s / 3600),
-        s => format!("{}d ago", s / 86_400),
+        s if s < 90 => format!("{s}s"),
+        s if s < 5400 => format!("{}m", s / 60),
+        s if s < 172_800 => format!("{}h", s / 3600),
+        s => format!("{}d", s / 86_400),
     }
 }