Poll subscribed feeds for new articles on a schedule #5

Merged
claude-bot merged 2 commits from worktree-feed-polling into main 2026-09-03 14:07:27 +00:00
4 changed files with 86 additions and 7 deletions

View file

@ -1,6 +1,7 @@
use crate::schema; use crate::schema;
use crate::Db; use crate::Db;
use anyhow::Result; use anyhow::Result;
use chrono::Utc;
use diesel::prelude::*; use diesel::prelude::*;
use diesel_async::RunQueryDsl; use diesel_async::RunQueryDsl;
use uuid::Uuid; use uuid::Uuid;
@ -24,4 +25,29 @@ impl Db {
.await?; .await?;
Ok(Uuid::parse_str(&id)?) Ok(Uuid::parse_str(&id)?)
} }
/// All subscribed feeds, for the polling job to iterate over.
pub async fn list_feeds(&self) -> Result<Vec<(Uuid, String)>> {
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(())
}
} }

View file

@ -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(())
}

View file

@ -1,4 +1,5 @@
mod affinity_decay; mod affinity_decay;
mod feed_polling;
mod scoring; mod scoring;
use anyhow::Result; use anyhow::Result;
@ -7,13 +8,37 @@ use feedsignal_llm::Llm;
use std::sync::Arc; use std::sync::Arc;
use tokio_cron_scheduler::{Job, JobScheduler}; use tokio_cron_scheduler::{Job, JobScheduler};
/// Wires up the two recurring jobs described in the design discussion: /// Wires up the recurring jobs described in the design discussion: polling
/// polling feeds + scoring new articles on a short interval, and decaying /// subscribed feeds for new articles, scoring them, and decaying topic
/// topic affinities once a day so stale signals fade. Call once at server /// affinities once a day so stale signals fade. Call once at server
/// startup. /// 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<Db>, llm: Arc<Llm>) -> Result<JobScheduler> { pub async fn start_scheduler(db: Arc<Db>, llm: Arc<Llm>) -> Result<JobScheduler> {
let scheduler = JobScheduler::new().await?; 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 db = db.clone();
let llm = llm.clone(); let llm = llm.clone();

View file

@ -12,9 +12,8 @@ const EMBEDDING_SHORTLIST_THRESHOLD: f32 = 0.55;
const LLM_BATCH_SIZE: i64 = 25; const LLM_BATCH_SIZE: i64 = 25;
/// One pass of: embed the shortlist, run the LLM on it, blend into /// One pass of: embed the shortlist, run the LLM on it, blend into
/// `final_score`. Feed polling itself (calling `feedsignal_feeds::fetch_feed` /// `final_score`. Assumes `feed_polling::run` has already pulled in any new
/// per subscribed feed and inserting new articles) is intentionally left as /// articles for this cycle.
/// a TODO here — wire it in once feed subscription management exists.
pub async fn run(db: &Db, llm: &Llm) -> Result<()> { pub async fn run(db: &Db, llm: &Llm) -> Result<()> {
let affinities = db.load_affinities().await?; let affinities = db.load_affinities().await?;
let shortlist = db let shortlist = db