feat: scaffold newsfeed — user-controlled news feed

A self-hosted feed where the user owns the ranking: per-source and signed
per-interest weights, decayed by recency, via a transparent deterministic
scorer. Content is sourced algorithmically (worker RSS/Atom polling) and
agentically (per-user API tokens POSTing candidates to the ingest endpoint).
Single-user today, multi-user by construction (every row keyed on user_id).

Rust cargo workspace + Vite/React/SWC/TS SPA:
- newsfeed-entities: DTOs (ts-rs bindings -> web/src/api/bindings)
- newsfeed-core: ranking, auth primitives, ingest, data-access ports
- newsfeed-data: SQLite adapters (sqlx, runtime queries)
- newsfeed-api: axum REST/JSON daemon
- newsfeed-worker: RSS polling + rescoring loop
- web: responsive, mobile-first SPA (React Query, generated types)

Deploy (Gitea Actions, build static musl + SPA, rsync as gitea_ci):
api+worker -> slartibartfast, SPA -> oolon (nginx serves + proxies /v1).

Deliberate deviations from house conventions (documented in CLAUDE.md/readme):
- SQLite instead of Postgres; api+worker co-locate sharing one DB file.
- Runtime sqlx queries instead of query! macros (SQLite dynamic typing;
  keeps CI database-free).

Verified end-to-end: auth, token ingest, interest-weighted ranking, signals,
pagination (curl + browser); cargo fmt/clippy -D/test and pnpm build/lint pass.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016fKZzDpvjiJ9eYbPGgJvUP

crates/newsfeed-worker/Cargo.toml

@@ -0,0 +1,31 @@
+[package]
+name = "newsfeed-worker"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+authors.workspace = true
+description = "Algorithmic content-sourcing worker for newsfeed (RSS/Atom polling)."
+
+[[bin]]
+name = "newsfeed-worker"
+path = "src/main.rs"
+
+[dependencies]
+newsfeed-entities.workspace = true
+newsfeed-core.workspace = true
+newsfeed-data.workspace = true
+
+tokio.workspace = true
+reqwest.workspace = true
+feed-rs.workspace = true
+
+serde.workspace = true
+chrono.workspace = true
+uuid.workspace = true
+
+anyhow.workspace = true
+tracing.workspace = true
+tracing-subscriber.workspace = true
+figment.workspace = true
+clap.workspace = true

crates/newsfeed-worker/src/config.rs

@@ -0,0 +1,55 @@
+//! Worker configuration (figment-layered: defaults → TOML → `NEWSFEED_*` env).
+
+use std::path::{Path, PathBuf};
+
+use figment::Figment;
+use figment::providers::{Env, Format, Serialized, Toml};
+use serde::{Deserialize, Serialize};
+
+/// Worker configuration.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Config {
+    /// SQLite database file, shared with the api on the same host.
+    pub database_path: PathBuf,
+    /// SQLite pool size for the worker.
+    pub max_db_connections: u32,
+    /// Seconds to sleep between poll cycles.
+    pub tick_secs: u64,
+    /// Minimum seconds between polls of the same RSS source.
+    pub source_min_interval_secs: i64,
+    /// Max sources polled per cycle.
+    pub batch: i64,
+    /// Max items rescored per user per cycle.
+    pub rescore_limit: i64,
+    /// HTTP request timeout when fetching feeds, in seconds.
+    pub http_timeout_secs: u64,
+    /// User-Agent sent when fetching feeds.
+    pub user_agent: String,
+}
+
+impl Default for Config {
+    fn default() -> Self {
+        Self {
+            database_path: PathBuf::from("/var/lib/newsfeed/newsfeed.db"),
+            max_db_connections: 2,
+            tick_secs: 60,
+            source_min_interval_secs: 900,
+            batch: 20,
+            rescore_limit: 500,
+            http_timeout_secs: 20,
+            user_agent: concat!("newsfeed-worker/", env!("CARGO_PKG_VERSION")).to_string(),
+        }
+    }
+}
+
+impl Config {
+    /// Load configuration, layering an optional TOML file and the environment over the
+    /// defaults.
+    pub fn load(file: Option<&Path>) -> anyhow::Result<Self> {
+        let mut fig = Figment::from(Serialized::defaults(Config::default()));
+        if let Some(path) = file {
+            fig = fig.merge(Toml::file(path));
+        }
+        Ok(fig.merge(Env::prefixed("NEWSFEED_")).extract()?)
+    }
+}

crates/newsfeed-worker/src/main.rs

@@ -0,0 +1,168 @@
+//! newsfeed algorithmic-sourcing worker.
+//!
+//! Co-located with the api on the same host (they share one SQLite file). Each cycle it
+//! polls due RSS/Atom sources, normalises and scores new entries into their owners'
+//! feeds, then rescores existing items so weight changes re-rank the feed. Idempotent:
+//! re-seeing an item's `external_id` is a no-op.
+
+mod config;
+mod sourcing;
+
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::time::Duration;
+
+use anyhow::Context;
+use chrono::Utc;
+use clap::Parser;
+use reqwest::Client;
+use tokio::signal;
+use uuid::Uuid;
+
+use newsfeed_core::ports::{InterestStore, ItemStore, SourceStore};
+use newsfeed_core::{ingest, service};
+use newsfeed_data::SqliteStore;
+use newsfeed_entities::interest::Interest;
+
+use crate::config::Config;
+use crate::sourcing::rss;
+
+#[derive(Parser)]
+#[command(
+    name = "newsfeed-worker",
+    version,
+    about = "newsfeed content-sourcing worker"
+)]
+struct Cli {
+    /// Path to a TOML config file. Env (`NEWSFEED_*`) overrides file values.
+    #[arg(long, default_value = "/etc/newsfeed/worker.toml")]
+    config: PathBuf,
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    init_tracing();
+
+    let cli = Cli::parse();
+    let config_file = cli.config.exists().then_some(cli.config.as_path());
+    let cfg = Config::load(config_file).context("loading configuration")?;
+    tracing::info!(db = %cfg.database_path.display(), tick_secs = cfg.tick_secs, "starting newsfeed-worker");
+
+    let store = newsfeed_data::connect(&cfg.database_path, cfg.max_db_connections)
+        .await
+        .context("opening database")?;
+    let client = Client::builder()
+        .user_agent(&cfg.user_agent)
+        .timeout(Duration::from_secs(cfg.http_timeout_secs))
+        .build()
+        .context("building HTTP client")?;
+
+    let mut ticker = tokio::time::interval(Duration::from_secs(cfg.tick_secs));
+    loop {
+        tokio::select! {
+            _ = ticker.tick() => {
+                if let Err(e) = run_cycle(&store, &client, &cfg).await {
+                    tracing::error!(error = %e, "poll cycle failed");
+                }
+            }
+            _ = shutdown_signal() => {
+                tracing::info!("shutdown signal received, exiting");
+                break;
+            }
+        }
+    }
+    Ok(())
+}
+
+/// One poll + rescore cycle.
+async fn run_cycle(store: &SqliteStore, client: &Client, cfg: &Config) -> anyhow::Result<()> {
+    let now = Utc::now();
+    let due = store
+        .due_rss_sources(now, cfg.source_min_interval_secs, cfg.batch)
+        .await?;
+    tracing::debug!(count = due.len(), "polling due sources");
+
+    let mut interests_cache: HashMap<Uuid, Vec<Interest>> = HashMap::new();
+
+    for source in due {
+        let Some(url) = source.url.clone() else {
+            store.mark_polled(source.id, now).await?;
+            continue;
+        };
+
+        match rss::fetch(client, &url).await {
+            Ok(entries) => {
+                let interests = if let Some(cached) = interests_cache.get(&source.user_id) {
+                    cached
+                } else {
+                    let loaded = store.list_interests(source.user_id).await?;
+                    interests_cache.entry(source.user_id).or_insert(loaded)
+                };
+
+                let mut added = 0usize;
+                for entry in entries {
+                    let candidate =
+                        ingest::candidate_from_feed_entry(source.user_id, source.id, entry);
+                    match ingest::persist_and_rank(store, candidate, source.weight, interests).await
+                    {
+                        Ok(item) if item.ingested_at >= now => added += 1,
+                        Ok(_) => {}
+                        Err(e) => {
+                            tracing::warn!(source = %source.name, error = %e, "failed to persist candidate")
+                        }
+                    }
+                }
+                tracing::info!(source = %source.name, added, "polled source");
+            }
+            Err(e) => {
+                tracing::warn!(source = %source.name, %url, error = %e, "failed to fetch feed")
+            }
+        }
+
+        // Mark polled regardless so a persistently-failing feed doesn't monopolise cycles.
+        store.mark_polled(source.id, now).await?;
+    }
+
+    // Rescore each active user's feed against current weights.
+    for user_id in store.user_ids_with_sources().await? {
+        match service::rescore_user(store, user_id, cfg.rescore_limit).await {
+            Ok(n) => tracing::debug!(%user_id, rescored = n, "rescored feed"),
+            Err(e) => tracing::warn!(%user_id, error = %e, "rescore failed"),
+        }
+    }
+
+    Ok(())
+}
+
+fn init_tracing() {
+    use tracing_subscriber::{EnvFilter, fmt, prelude::*};
+    let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
+    let registry = tracing_subscriber::registry().with(filter);
+    if std::env::var_os("JOURNAL_STREAM").is_some() {
+        registry
+            .with(fmt::layer().json().flatten_event(true))
+            .init();
+    } else {
+        registry.with(fmt::layer()).init();
+    }
+}
+
+async fn shutdown_signal() {
+    let ctrl_c = async {
+        signal::ctrl_c().await.expect("install Ctrl-C handler");
+    };
+    #[cfg(unix)]
+    let terminate = async {
+        signal::unix::signal(signal::unix::SignalKind::terminate())
+            .expect("install SIGTERM handler")
+            .recv()
+            .await;
+    };
+    #[cfg(not(unix))]
+    let terminate = std::future::pending::<()>();
+
+    tokio::select! {
+        _ = ctrl_c => {},
+        _ = terminate => {},
+    }
+}

crates/newsfeed-worker/src/sourcing/mod.rs

@@ -0,0 +1,3 @@
+//! Content sourcing. Currently RSS/Atom polling; other algorithmic sources plug in here.
+
+pub mod rss;

crates/newsfeed-worker/src/sourcing/rss.rs

@@ -0,0 +1,80 @@
+//! Fetch and parse an RSS/Atom feed into backend-agnostic [`FeedEntry`] values.
+
+use anyhow::Context;
+use reqwest::Client;
+
+use newsfeed_core::ingest::FeedEntry;
+use newsfeed_entities::item::Media;
+
+/// Fetch `url` and parse it, returning one [`FeedEntry`] per feed item.
+pub async fn fetch(client: &Client, url: &str) -> anyhow::Result<Vec<FeedEntry>> {
+    let bytes = client
+        .get(url)
+        .send()
+        .await
+        .with_context(|| format!("fetching {url}"))?
+        .error_for_status()
+        .with_context(|| format!("bad status from {url}"))?
+        .bytes()
+        .await
+        .with_context(|| format!("reading body from {url}"))?;
+
+    let feed = feed_rs::parser::parse(&bytes[..]).with_context(|| format!("parsing feed {url}"))?;
+
+    Ok(feed.entries.into_iter().map(entry_to_feed_entry).collect())
+}
+
+fn entry_to_feed_entry(entry: feed_rs::model::Entry) -> FeedEntry {
+    let url = entry.links.first().map(|l| l.href.clone());
+    // Prefer the entry's own id; fall back to the link so de-dup still works.
+    let external_id = if entry.id.is_empty() {
+        url.clone().unwrap_or_else(|| {
+            entry
+                .title
+                .as_ref()
+                .map(|t| t.content.clone())
+                .unwrap_or_default()
+        })
+    } else {
+        entry.id.clone()
+    };
+    let title = entry
+        .title
+        .map(|t| t.content)
+        .unwrap_or_else(|| "(untitled)".to_string());
+    let summary = entry
+        .summary
+        .map(|t| t.content)
+        .or_else(|| entry.content.and_then(|c| c.body));
+    let author = entry.authors.first().map(|p| p.name.clone());
+    let tags = entry.categories.into_iter().map(|c| c.term).collect();
+    let published_at = entry.published.or(entry.updated);
+
+    let media = entry
+        .media
+        .into_iter()
+        .flat_map(|m| m.content)
+        .filter_map(|c| {
+            c.url.map(|u| Media {
+                kind: c
+                    .content_type
+                    .map(|m| m.to_string())
+                    .unwrap_or_else(|| "media".to_string()),
+                url: u.to_string(),
+                width: c.width,
+                height: c.height,
+            })
+        })
+        .collect();
+
+    FeedEntry {
+        external_id,
+        url,
+        title,
+        summary,
+        author,
+        tags,
+        media,
+        published_at,
+    }
+}