From 6fb470084d1813621ad3130be4e37db6d34106ca Mon Sep 17 00:00:00 2001 From: Austin Schaefer Date: Thu, 3 Sep 2026 13:20:30 +0200 Subject: [PATCH] Poll subscribed feeds for new articles on a schedule MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Feeds were already persisted, but nothing ever re-fetched them after the initial subscribe — scoring.rs had a TODO where feed polling was supposed to go. Add a feed_polling job that re-fetches every subscribed feed, relies on articles.url's unique constraint to skip ones already seen, and stamps last_fetched_at. Runs once on startup (so reopening the app catches up immediately) and every 15 minutes after, ahead of the scoring pass. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01QmwX8eafbMstJvt8XPSqft --- crates/db/src/lib.rs | 25 ++++++++++++++++ crates/web/src/server/jobs/feed_polling.rs | 29 +++++++++++++++++++ crates/web/src/server/jobs/mod.rs | 33 +++++++++++++++++++--- crates/web/src/server/jobs/scoring.rs | 5 ++-- 4 files changed, 85 insertions(+), 7 deletions(-) create mode 100644 crates/web/src/server/jobs/feed_polling.rs diff --git a/crates/db/src/lib.rs b/crates/db/src/lib.rs index 751d0b6..58c31f1 100644 --- a/crates/db/src/lib.rs +++ b/crates/db/src/lib.rs @@ -67,6 +67,31 @@ impl Db { Ok(Uuid::parse_str(&id)?) } + /// All subscribed feeds, for the polling job to iterate over. + pub async fn list_feeds(&self) -> Result> { + use schema::feeds::dsl; + let mut conn = self.pool.get().await?; + let rows: Vec<(String, String)> = dsl::feeds + .select((dsl::id, dsl::url)) + .load(&mut conn) + .await?; + rows.into_iter() + .map(|(id, url)| Ok((Uuid::parse_str(&id)?, url))) + .collect() + } + + /// Records that a feed was just polled, so the next run can be judged + /// against it (e.g. surfaced in the UI as "last checked"). + pub async fn mark_feed_fetched(&self, feed_id: Uuid) -> Result<()> { + use schema::feeds::dsl; + let mut conn = self.pool.get().await?; + diesel::update(dsl::feeds.filter(dsl::id.eq(feed_id.to_string()))) + .set(dsl::last_fetched_at.eq(Utc::now().to_rfc3339())) + .execute(&mut conn) + .await?; + Ok(()) + } + pub async fn insert_article(&self, article: &Article) -> Result<()> { use schema::articles::dsl; let mut conn = self.pool.get().await?; diff --git a/crates/web/src/server/jobs/feed_polling.rs b/crates/web/src/server/jobs/feed_polling.rs new file mode 100644 index 0000000..24bab34 --- /dev/null +++ b/crates/web/src/server/jobs/feed_polling.rs @@ -0,0 +1,29 @@ +use anyhow::Result; +use feedsignal_db::Db; + +/// Re-fetches every subscribed feed and inserts any articles published since +/// the last run. `articles.url` is unique, so re-fetching the same feed and +/// inserting its (mostly already-seen) entries is safe — `insert_article` +/// silently skips ones already in the database, leaving only genuinely new +/// articles behind. +pub async fn run(db: &Db) -> Result<()> { + let feeds = db.list_feeds().await?; + tracing::info!(count = feeds.len(), "polling subscribed feeds"); + + for (feed_id, url) in feeds { + let (_title, articles) = match feedsignal_feeds::fetch_feed(&url, feed_id).await { + Ok(result) => result, + Err(err) => { + tracing::warn!(?err, %url, "failed to poll feed, skipping"); + continue; + } + }; + + for article in &articles { + db.insert_article(article).await?; + } + db.mark_feed_fetched(feed_id).await?; + } + + Ok(()) +} diff --git a/crates/web/src/server/jobs/mod.rs b/crates/web/src/server/jobs/mod.rs index 448aaf7..8c61f6d 100644 --- a/crates/web/src/server/jobs/mod.rs +++ b/crates/web/src/server/jobs/mod.rs @@ -1,4 +1,5 @@ mod affinity_decay; +mod feed_polling; mod scoring; use anyhow::Result; @@ -7,13 +8,37 @@ use feedsignal_llm::Llm; use std::sync::Arc; use tokio_cron_scheduler::{Job, JobScheduler}; -/// Wires up the two recurring jobs described in the design discussion: -/// polling feeds + scoring new articles on a short interval, and decaying -/// topic affinities once a day so stale signals fade. Call once at server -/// startup. +/// Wires up the recurring jobs described in the design discussion: polling +/// subscribed feeds for new articles, scoring them, and decaying topic +/// affinities once a day so stale signals fade. Call once at server +/// startup, and also runs once immediately so a freshly (re)opened app +/// doesn't wait 15 minutes for its first check. pub async fn start_scheduler(db: Arc, llm: Arc) -> Result { let scheduler = JobScheduler::new().await?; + { + let db = db.clone(); + tokio::spawn(async move { + if let Err(err) = feed_polling::run(&db).await { + tracing::error!(?err, "initial feed poll failed"); + } + }); + } + + { + let db = db.clone(); + scheduler + .add(Job::new_async("0 */15 * * * *", move |_uuid, _lock| { + let db = db.clone(); + Box::pin(async move { + if let Err(err) = feed_polling::run(&db).await { + tracing::error!(?err, "feed polling run failed"); + } + }) + })?) + .await?; + } + { let db = db.clone(); let llm = llm.clone(); diff --git a/crates/web/src/server/jobs/scoring.rs b/crates/web/src/server/jobs/scoring.rs index ccf896d..96bc293 100644 --- a/crates/web/src/server/jobs/scoring.rs +++ b/crates/web/src/server/jobs/scoring.rs @@ -12,9 +12,8 @@ const EMBEDDING_SHORTLIST_THRESHOLD: f32 = 0.55; const LLM_BATCH_SIZE: i64 = 25; /// One pass of: embed the shortlist, run the LLM on it, blend into -/// `final_score`. Feed polling itself (calling `feedsignal_feeds::fetch_feed` -/// per subscribed feed and inserting new articles) is intentionally left as -/// a TODO here — wire it in once feed subscription management exists. +/// `final_score`. Assumes `feed_polling::run` has already pulled in any new +/// articles for this cycle. pub async fn run(db: &Db, llm: &Llm) -> Result<()> { let affinities = db.load_affinities().await?; let shortlist = db -- 2.45.2