diff --git a/crates/db/src/feeds.rs b/crates/db/src/feeds.rs index 13b9d8f..06fffb1 100644 --- a/crates/db/src/feeds.rs +++ b/crates/db/src/feeds.rs @@ -1,6 +1,7 @@ use crate::schema; use crate::Db; use anyhow::Result; +use chrono::Utc; use diesel::prelude::*; use diesel_async::RunQueryDsl; use uuid::Uuid; @@ -24,4 +25,29 @@ impl Db { .await?; 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(()) + } } 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