Poll subscribed feeds for new articles on a schedule #5
4 changed files with 85 additions and 7 deletions
|
|
@ -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<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(())
|
||||
}
|
||||
|
||||
pub async fn insert_article(&self, article: &Article) -> Result<()> {
|
||||
use schema::articles::dsl;
|
||||
let mut conn = self.pool.get().await?;
|
||||
|
|
|
|||
29
crates/web/src/server/jobs/feed_polling.rs
Normal file
29
crates/web/src/server/jobs/feed_polling.rs
Normal 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(())
|
||||
}
|
||||
|
|
@ -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<Db>, llm: Arc<Llm>) -> Result<JobScheduler> {
|
||||
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();
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in a new issue