Scaffold feedsignal: Rust workspace for LLM-filtered RSS reader
Some checks failed
CI / check (push) Failing after 26s
Some checks failed
CI / check (push) Failing after 26s
Rust workspace with core (topic-affinity learning engine + relevance scoring), db (sqlite/sqlx schema + repo), feeds (RSS/Atom fetch), llm (rig + local Ollama embeddings/completion), and web (axum + Dioxus fullstack UI, no separate JS stack). Two-stage relevance filtering (embedding shortlist -> LLM judgment) and an engagement/surprise-based topic affinity engine with daily decay. All crates compile and core's affinity engine has passing unit tests; server and wasm client targets of feedsignal-web both check clean.
This commit is contained in:
parent
a648f4ddc9
commit
1a4a72fc27
24 changed files with 6586 additions and 0 deletions
36
.forgejo/workflows/ci.yml
Normal file
36
.forgejo/workflows/ci.yml
Normal file
|
|
@ -0,0 +1,36 @@
|
|||
name: CI
|
||||
|
||||
on:
|
||||
push:
|
||||
branches: [main]
|
||||
pull_request:
|
||||
|
||||
jobs:
|
||||
check:
|
||||
runs-on: docker
|
||||
container: docker.io/library/rust:1-bookworm
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- name: Cache cargo registry and target dir
|
||||
uses: actions/cache@v4
|
||||
with:
|
||||
path: |
|
||||
~/.cargo/registry
|
||||
target
|
||||
key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }}
|
||||
|
||||
- name: Add wasm target
|
||||
run: rustup target add wasm32-unknown-unknown
|
||||
|
||||
- name: Check native crates
|
||||
run: cargo check --workspace --exclude feedsignal-web
|
||||
|
||||
- name: Check web crate (server)
|
||||
run: cargo check -p feedsignal-web --no-default-features --features server
|
||||
|
||||
- name: Check web crate (wasm client)
|
||||
run: cargo check -p feedsignal-web --no-default-features --features web --target wasm32-unknown-unknown
|
||||
|
||||
- name: Test
|
||||
run: cargo test --workspace --exclude feedsignal-web
|
||||
5
.gitignore
vendored
Normal file
5
.gitignore
vendored
Normal file
|
|
@ -0,0 +1,5 @@
|
|||
/target
|
||||
*.db
|
||||
*.db-shm
|
||||
*.db-wal
|
||||
.env
|
||||
5417
Cargo.lock
generated
Normal file
5417
Cargo.lock
generated
Normal file
File diff suppressed because it is too large
Load diff
31
Cargo.toml
Normal file
31
Cargo.toml
Normal file
|
|
@ -0,0 +1,31 @@
|
|||
[workspace]
|
||||
resolver = "2"
|
||||
members = [
|
||||
"crates/core",
|
||||
"crates/db",
|
||||
"crates/feeds",
|
||||
"crates/llm",
|
||||
"crates/web",
|
||||
]
|
||||
|
||||
[workspace.package]
|
||||
edition = "2021"
|
||||
version = "0.1.0"
|
||||
license = "MIT"
|
||||
|
||||
[workspace.dependencies]
|
||||
tokio = { version = "1", features = ["full"] }
|
||||
serde = { version = "1", features = ["derive"] }
|
||||
serde_json = "1"
|
||||
anyhow = "1"
|
||||
thiserror = "2"
|
||||
tracing = "0.1"
|
||||
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
|
||||
chrono = { version = "0.4", features = ["serde"] }
|
||||
uuid = { version = "1", features = ["v4", "serde", "js"] }
|
||||
sqlx = { version = "0.8", features = ["runtime-tokio", "sqlite", "chrono", "uuid", "migrate"] }
|
||||
|
||||
feedsignal-core = { path = "crates/core" }
|
||||
feedsignal-db = { path = "crates/db" }
|
||||
feedsignal-feeds = { path = "crates/feeds" }
|
||||
feedsignal-llm = { path = "crates/llm" }
|
||||
104
README.md
Normal file
104
README.md
Normal file
|
|
@ -0,0 +1,104 @@
|
|||
# feedsignal
|
||||
|
||||
A self-hosted RSS reader that uses a local Ollama LLM (via [rig](https://github.com/0xPlaygrounds/rig))
|
||||
to surface the articles actually worth reading out of a large set of subscribed feeds, and learns
|
||||
from what you read (and how long, and what you dismiss) to get better at that over time.
|
||||
|
||||
## Architecture
|
||||
|
||||
```
|
||||
crates/
|
||||
core/ domain models (Article, Feed, ReadingEvent) + the topic-affinity
|
||||
learning engine + the final relevance-scoring formula. No I/O.
|
||||
db/ SQLite (sqlx) persistence: feeds, articles, the append-only
|
||||
reading_events log, topic_affinities, preferences.
|
||||
feeds/ RSS/Atom fetching + parsing (feed-rs).
|
||||
llm/ rig + local Ollama: embeddings for the cheap first-pass filter,
|
||||
chat completion for the relevance judgment on the shortlist.
|
||||
web/ axum + Dioxus fullstack UI. `server` feature = native binary
|
||||
(DB, feed polling, LLM calls, scheduler); `web` feature = the
|
||||
WASM client shipped to the browser. No separate JS/npm stack.
|
||||
```
|
||||
|
||||
### Two-stage relevance filtering
|
||||
|
||||
Running the LLM on every article from every feed doesn't scale locally. Instead:
|
||||
|
||||
1. Every new article is embedded (`nomic-embed-text` by default) and scored against your
|
||||
preference-profile embedding — cheap, runs on all of them.
|
||||
2. Only articles above `EMBEDDING_SHORTLIST_THRESHOLD` (`crates/web/src/server/pipeline.rs`)
|
||||
go to the LLM (`llama3.1` by default) for an actual relevance judgment with a rationale.
|
||||
3. `feedsignal_core::scoring::score_article` blends LLM score (dominant when present),
|
||||
embedding score (fallback / floor), and topic affinity (bounded nudge) into `final_score`,
|
||||
which is what the UI ranks by.
|
||||
|
||||
### Learning from reading behavior
|
||||
|
||||
Every interaction (impression, open, dwell time, star, dismiss) is appended to an immutable
|
||||
`reading_events` log — never mutated, so the model can be recomputed or retuned later without
|
||||
losing data (`crates/db/migrations/0001_init.sql`).
|
||||
|
||||
From that log, `feedsignal_core::affinity`:
|
||||
|
||||
- Computes an **engagement score** per article (0.0–1.0) from whether it was opened, how long
|
||||
you spent relative to its estimated read time, and explicit star/dismiss signals.
|
||||
- Compares engagement to what the pipeline *predicted* (`final_score` at ingest time) to get a
|
||||
**surprise** signal — did you engage more or less than expected?
|
||||
- Nudges the affinity of that article's topics by `learning_rate * surprise`, clamped to
|
||||
`[-1, 1]`, so "the model rated this low but I read the whole thing" measurably shifts future
|
||||
scoring, and "the model rated this high but I dismissed it unread" pulls the other way.
|
||||
- Decays all affinities toward zero once a day so stale interests fade instead of anchoring
|
||||
the model forever.
|
||||
|
||||
This is a from-scratch numeric signal engine, deliberately not an LLM-rewritten prose "taste
|
||||
profile" — the topic scores are inspectable, and the update rule is simple enough to reason
|
||||
about and tune. See `crates/core/src/affinity.rs` for the full implementation and unit tests.
|
||||
|
||||
## Status
|
||||
|
||||
This is a scaffold, not a working app yet. What's real and compiles (`cargo check`/`cargo test`
|
||||
pass for every crate, both the `server` and `wasm32-unknown-unknown` targets of `feedsignal-web`):
|
||||
|
||||
- The domain model, DB schema/migrations, and the topic-affinity engine (with tests).
|
||||
- Feed fetching/parsing.
|
||||
- Ollama embedding + chat-completion calls via rig (against real rig-core 0.42 APIs).
|
||||
- The scoring pipeline shape and a cron-scheduled job wiring (`tokio-cron-scheduler`).
|
||||
- A minimal Dioxus UI that lists ranked articles and lets you dismiss one.
|
||||
|
||||
What's *not* wired up yet (left as the natural next steps):
|
||||
|
||||
- Feed subscription management (add/remove feed URLs) — `feedsignal_feeds::fetch_feed` exists
|
||||
but nothing calls it on a schedule yet.
|
||||
- The `preferences` table (your free-text "what I care about" description) isn't read by the
|
||||
pipeline yet — `pipeline.rs` has a `TODO` where it belongs.
|
||||
- Dwell-time capture from the browser (needs a small bit of JS/visibility-API glue on the
|
||||
article view, or a "mark read" action, to backfill `dwell_seconds` on the `opened` event).
|
||||
- Full article content extraction (currently only the feed's summary/content field is used;
|
||||
no readability-style scraping of the linked page).
|
||||
|
||||
## Running it
|
||||
|
||||
Requires [Ollama](https://ollama.com) running locally with the models pulled:
|
||||
|
||||
```sh
|
||||
ollama pull nomic-embed-text
|
||||
ollama pull llama3.1
|
||||
```
|
||||
|
||||
And the [Dioxus CLI](https://dioxuslabs.com/learn/0.7/getting_started/) (`dx`) to build/serve
|
||||
the fullstack app — `dx` also has an `dx add`/component-scaffolding workflow worth using instead
|
||||
of hand-rolling UI pieces as this grows:
|
||||
|
||||
```sh
|
||||
cargo install dioxus-cli
|
||||
cd crates/web
|
||||
dx serve
|
||||
```
|
||||
|
||||
The native crates (`core`, `db`, `feeds`, `llm`) build with plain `cargo check`/`cargo test` and
|
||||
don't need `dx` at all.
|
||||
|
||||
## CI
|
||||
|
||||
`.forgejo/workflows/ci.yml` runs on the self-hosted Forgejo runner: checks the native crates,
|
||||
both feature-sets of `feedsignal-web` (server + wasm client), and runs tests.
|
||||
11
crates/core/Cargo.toml
Normal file
11
crates/core/Cargo.toml
Normal file
|
|
@ -0,0 +1,11 @@
|
|||
[package]
|
||||
name = "feedsignal-core"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
license.workspace = true
|
||||
|
||||
[dependencies]
|
||||
serde.workspace = true
|
||||
serde_json.workspace = true
|
||||
chrono.workspace = true
|
||||
uuid.workspace = true
|
||||
137
crates/core/src/affinity.rs
Normal file
137
crates/core/src/affinity.rs
Normal file
|
|
@ -0,0 +1,137 @@
|
|||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::HashMap;
|
||||
|
||||
/// Per-topic affinity scores, in `[-1.0, 1.0]`, updated from observed
|
||||
/// engagement vs. predicted relevance. Persisted as a single row per user
|
||||
/// (JSON blob) in `feedsignal-db`; the event log remains the source of
|
||||
/// truth and this can always be rebuilt by replaying it.
|
||||
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
|
||||
pub struct TopicAffinities {
|
||||
scores: HashMap<String, f64>,
|
||||
}
|
||||
|
||||
/// How much a single feedback event moves an affinity score. Kept small so
|
||||
/// no single article dominates a topic's long-run trend.
|
||||
const LEARNING_RATE: f64 = 0.15;
|
||||
|
||||
/// Fraction of every affinity pulled back toward zero on each nightly decay
|
||||
/// pass, so stale interests fade instead of anchoring the model forever.
|
||||
const DAILY_DECAY: f64 = 0.02;
|
||||
|
||||
impl TopicAffinities {
|
||||
pub fn score(&self, topic: &str) -> f64 {
|
||||
self.scores.get(topic).copied().unwrap_or(0.0)
|
||||
}
|
||||
|
||||
/// Mean affinity across an article's topics; 0.0 for an untagged
|
||||
/// article (neutral, defers entirely to the embedding/LLM stages).
|
||||
pub fn score_topics(&self, topics: &[String]) -> f64 {
|
||||
if topics.is_empty() {
|
||||
return 0.0;
|
||||
}
|
||||
topics.iter().map(|t| self.score(t)).sum::<f64>() / topics.len() as f64
|
||||
}
|
||||
|
||||
/// Update affinities for an article's topics from an observed
|
||||
/// `surprise`: `engagement_score - predicted_relevance_score`, both in
|
||||
/// `[0.0, 1.0]` (see `scoring::engagement_score`). Positive surprise
|
||||
/// (the user engaged more than the pipeline predicted) nudges those
|
||||
/// topics up; negative surprise nudges them down. This is what lets
|
||||
/// "the model thought this was irrelevant but I read the whole thing"
|
||||
/// actually change future behavior.
|
||||
pub fn apply_feedback(&mut self, topics: &[String], surprise: f64) {
|
||||
for topic in topics {
|
||||
let current = self.score(topic);
|
||||
let updated = (current + LEARNING_RATE * surprise).clamp(-1.0, 1.0);
|
||||
self.scores.insert(topic.clone(), updated);
|
||||
}
|
||||
}
|
||||
|
||||
/// Run once per day (see the scheduler in `feedsignal-web`) to let
|
||||
/// affinities the user hasn't reinforced recently drift back toward
|
||||
/// neutral rather than staying permanently pinned from a few old
|
||||
/// signals.
|
||||
pub fn decay(&mut self) {
|
||||
self.scores.retain(|_, v| v.abs() > 1e-4);
|
||||
for v in self.scores.values_mut() {
|
||||
*v -= v.signum() * DAILY_DECAY;
|
||||
}
|
||||
}
|
||||
|
||||
pub fn top_n(&self, n: usize) -> Vec<(&str, f64)> {
|
||||
let mut items: Vec<_> = self.scores.iter().map(|(k, v)| (k.as_str(), *v)).collect();
|
||||
items.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap());
|
||||
items.truncate(n);
|
||||
items
|
||||
}
|
||||
}
|
||||
|
||||
/// Converts a raw reading interaction into an engagement score in
|
||||
/// `[0.0, 1.0]`, and folds in the explicit star/dismiss signal.
|
||||
///
|
||||
/// - Never opened: 0.0
|
||||
/// - Opened: `dwell_seconds / estimated_read_seconds`, capped at 1.0, so
|
||||
/// skimming half an article scores lower than reading it fully.
|
||||
/// - Starred: +0.3 on top (capped at 1.0) — an explicit "yes" beyond dwell
|
||||
/// time alone.
|
||||
/// - Dismissed without opening: -0.3, floored at 0.0's negative counterpart
|
||||
/// handled by the caller via `surprise` (engagement itself never goes
|
||||
/// negative; the *deviation* from a predicted score can).
|
||||
pub fn engagement_score(
|
||||
opened: bool,
|
||||
dwell_seconds: Option<u32>,
|
||||
estimated_read_seconds: Option<u32>,
|
||||
starred: bool,
|
||||
dismissed: bool,
|
||||
) -> f64 {
|
||||
if dismissed && !opened {
|
||||
return 0.0;
|
||||
}
|
||||
let mut score = if !opened {
|
||||
0.0
|
||||
} else {
|
||||
match (dwell_seconds, estimated_read_seconds) {
|
||||
(Some(dwell), Some(est)) if est > 0 => (dwell as f64 / est as f64).min(1.0),
|
||||
// Opened but we don't yet know dwell time / read-time estimate:
|
||||
// credit partial engagement rather than 0 or 1.
|
||||
_ => 0.5,
|
||||
}
|
||||
};
|
||||
if starred {
|
||||
score = (score + 0.3).min(1.0);
|
||||
}
|
||||
score
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn under_predicted_relevance_boosts_topic() {
|
||||
let mut aff = TopicAffinities::default();
|
||||
let topics = vec!["rust".to_string()];
|
||||
// Model predicted 0.2 relevance, user fully read it: surprise = 0.8.
|
||||
aff.apply_feedback(&topics, 0.8);
|
||||
assert!(aff.score("rust") > 0.0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn over_predicted_relevance_lowers_topic() {
|
||||
let mut aff = TopicAffinities::default();
|
||||
let topics = vec!["crypto".to_string()];
|
||||
// Model predicted 0.9, user dismissed unread: engagement 0, surprise = -0.9.
|
||||
aff.apply_feedback(&topics, -0.9);
|
||||
assert!(aff.score("crypto") < 0.0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decay_pulls_toward_zero() {
|
||||
let mut aff = TopicAffinities::default();
|
||||
aff.apply_feedback(&["rust".to_string()], 1.0);
|
||||
let before = aff.score("rust");
|
||||
aff.decay();
|
||||
assert!(aff.score("rust") < before);
|
||||
assert!(aff.score("rust") > 0.0);
|
||||
}
|
||||
}
|
||||
7
crates/core/src/lib.rs
Normal file
7
crates/core/src/lib.rs
Normal file
|
|
@ -0,0 +1,7 @@
|
|||
pub mod models;
|
||||
pub mod affinity;
|
||||
pub mod scoring;
|
||||
|
||||
pub use affinity::TopicAffinities;
|
||||
pub use models::{Article, Feed, ReadingEvent, ReadingOutcome};
|
||||
pub use scoring::{RelevanceInputs, score_article};
|
||||
58
crates/core/src/models.rs
Normal file
58
crates/core/src/models.rs
Normal file
|
|
@ -0,0 +1,58 @@
|
|||
use chrono::{DateTime, Utc};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use uuid::Uuid;
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct Feed {
|
||||
pub id: Uuid,
|
||||
pub url: String,
|
||||
pub title: String,
|
||||
pub last_fetched_at: Option<DateTime<Utc>>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct Article {
|
||||
pub id: Uuid,
|
||||
pub feed_id: Uuid,
|
||||
pub url: String,
|
||||
pub title: String,
|
||||
pub summary: String,
|
||||
pub published_at: Option<DateTime<Utc>>,
|
||||
/// Short canonical tags assigned at ingest time (e.g. by the LLM stage),
|
||||
/// used both for display and as keys into `TopicAffinities`.
|
||||
pub topics: Vec<String>,
|
||||
/// Cosine similarity between the article embedding and the user's
|
||||
/// preference-profile embedding. Cheap first-pass filter score.
|
||||
pub embedding_score: Option<f32>,
|
||||
/// 0.0-1.0 relevance judgment from the LLM stage, only computed for
|
||||
/// articles that clear the embedding-score shortlist threshold.
|
||||
pub llm_score: Option<f32>,
|
||||
/// Final blended score actually used for ranking/surfacing, see
|
||||
/// `scoring::score_article`.
|
||||
pub final_score: Option<f32>,
|
||||
pub estimated_read_seconds: Option<u32>,
|
||||
}
|
||||
|
||||
/// A single recorded interaction between the user and an article. Every
|
||||
/// event is appended to an immutable log (see `feedsignal-db`); affinities
|
||||
/// are derived from replaying/aggregating these, never mutated in place,
|
||||
/// so the scoring model can be recomputed or tuned retroactively.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct ReadingEvent {
|
||||
pub id: Uuid,
|
||||
pub article_id: Uuid,
|
||||
pub occurred_at: DateTime<Utc>,
|
||||
pub outcome: ReadingOutcome,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub enum ReadingOutcome {
|
||||
/// Shown in the feed list but no further interaction (yet).
|
||||
Impression,
|
||||
/// User opened the article. `dwell_seconds` is filled in later via an
|
||||
/// update event (e.g. on tab close / navigation away) once known.
|
||||
Opened { dwell_seconds: Option<u32> },
|
||||
Starred,
|
||||
/// Explicitly marked not relevant, independent of whether it was opened.
|
||||
Dismissed,
|
||||
}
|
||||
34
crates/core/src/scoring.rs
Normal file
34
crates/core/src/scoring.rs
Normal file
|
|
@ -0,0 +1,34 @@
|
|||
use crate::affinity::TopicAffinities;
|
||||
|
||||
/// Inputs to the final blended relevance score for one article.
|
||||
pub struct RelevanceInputs<'a> {
|
||||
/// Cosine similarity (0.0-1.0, already renormalized from [-1,1] if
|
||||
/// needed) between article and preference-profile embeddings.
|
||||
pub embedding_score: f32,
|
||||
/// LLM judgment (0.0-1.0), `None` if the article didn't clear the
|
||||
/// embedding shortlist threshold and so was never sent to the LLM.
|
||||
pub llm_score: Option<f32>,
|
||||
pub topics: &'a [String],
|
||||
pub affinities: &'a TopicAffinities,
|
||||
}
|
||||
|
||||
/// Weights are deliberately conservative: the LLM judgment dominates when
|
||||
/// present (it has read the actual content), the embedding score is a
|
||||
/// fallback when the LLM stage was skipped, and topic affinity acts as a
|
||||
/// bounded nudge rather than a veto — a strong LLM match should still
|
||||
/// surface even for a topic the user historically skims.
|
||||
const W_LLM: f32 = 0.6;
|
||||
const W_EMBEDDING: f32 = 0.25;
|
||||
const W_AFFINITY: f32 = 0.15;
|
||||
|
||||
pub fn score_article(inputs: RelevanceInputs) -> f32 {
|
||||
let affinity = inputs.affinities.score_topics(inputs.topics) as f32; // [-1, 1]
|
||||
let affinity_component = (affinity + 1.0) / 2.0; // renormalize to [0, 1]
|
||||
|
||||
match inputs.llm_score {
|
||||
Some(llm) => W_LLM * llm + W_EMBEDDING * inputs.embedding_score + W_AFFINITY * affinity_component,
|
||||
// No LLM score yet: redistribute its weight onto the embedding score.
|
||||
None => (W_LLM + W_EMBEDDING) * inputs.embedding_score + W_AFFINITY * affinity_component,
|
||||
}
|
||||
.clamp(0.0, 1.0)
|
||||
}
|
||||
15
crates/db/Cargo.toml
Normal file
15
crates/db/Cargo.toml
Normal file
|
|
@ -0,0 +1,15 @@
|
|||
[package]
|
||||
name = "feedsignal-db"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
license.workspace = true
|
||||
|
||||
[dependencies]
|
||||
feedsignal-core.workspace = true
|
||||
sqlx.workspace = true
|
||||
tokio.workspace = true
|
||||
serde.workspace = true
|
||||
serde_json.workspace = true
|
||||
chrono.workspace = true
|
||||
uuid.workspace = true
|
||||
anyhow.workspace = true
|
||||
60
crates/db/migrations/0001_init.sql
Normal file
60
crates/db/migrations/0001_init.sql
Normal file
|
|
@ -0,0 +1,60 @@
|
|||
CREATE TABLE feeds (
|
||||
id TEXT PRIMARY KEY,
|
||||
url TEXT NOT NULL UNIQUE,
|
||||
title TEXT NOT NULL,
|
||||
last_fetched_at TEXT
|
||||
);
|
||||
|
||||
CREATE TABLE articles (
|
||||
id TEXT PRIMARY KEY,
|
||||
feed_id TEXT NOT NULL REFERENCES feeds(id),
|
||||
url TEXT NOT NULL UNIQUE,
|
||||
title TEXT NOT NULL,
|
||||
summary TEXT NOT NULL,
|
||||
content TEXT,
|
||||
published_at TEXT,
|
||||
fetched_at TEXT NOT NULL,
|
||||
topics TEXT NOT NULL DEFAULT '[]', -- JSON array of strings
|
||||
embedding TEXT, -- JSON array of floats
|
||||
embedding_score REAL,
|
||||
llm_score REAL,
|
||||
llm_rationale TEXT,
|
||||
final_score REAL,
|
||||
estimated_read_seconds INTEGER
|
||||
);
|
||||
|
||||
CREATE INDEX idx_articles_feed_id ON articles(feed_id);
|
||||
CREATE INDEX idx_articles_final_score ON articles(final_score);
|
||||
|
||||
-- Append-only interaction log. Affinities and any future scoring model are
|
||||
-- derived from this; it is never mutated except to backfill dwell_seconds
|
||||
-- once known.
|
||||
CREATE TABLE reading_events (
|
||||
id TEXT PRIMARY KEY,
|
||||
article_id TEXT NOT NULL REFERENCES articles(id),
|
||||
occurred_at TEXT NOT NULL,
|
||||
outcome TEXT NOT NULL, -- 'impression' | 'opened' | 'starred' | 'dismissed'
|
||||
dwell_seconds INTEGER
|
||||
);
|
||||
|
||||
CREATE INDEX idx_reading_events_article_id ON reading_events(article_id);
|
||||
|
||||
-- Single-row-per-user table (multi-user support can add a user_id column
|
||||
-- later); holds the current TopicAffinities snapshot as JSON so it doesn't
|
||||
-- need to be recomputed from the full event log on every request. Rebuild
|
||||
-- from reading_events at any time if the scoring model changes.
|
||||
CREATE TABLE topic_affinities (
|
||||
user_id TEXT PRIMARY KEY DEFAULT 'default',
|
||||
affinities TEXT NOT NULL DEFAULT '{}',
|
||||
updated_at TEXT NOT NULL
|
||||
);
|
||||
|
||||
-- Free-text preferences the user writes explicitly (e.g. "I care about
|
||||
-- distributed systems and Rust internals, not general AI news"), embedded
|
||||
-- once and used as the anchor for the embedding-similarity stage.
|
||||
CREATE TABLE preferences (
|
||||
user_id TEXT PRIMARY KEY DEFAULT 'default',
|
||||
description TEXT NOT NULL DEFAULT '',
|
||||
embedding TEXT,
|
||||
updated_at TEXT NOT NULL
|
||||
);
|
||||
122
crates/db/src/lib.rs
Normal file
122
crates/db/src/lib.rs
Normal file
|
|
@ -0,0 +1,122 @@
|
|||
use anyhow::Result;
|
||||
use feedsignal_core::{Article, ReadingEvent, ReadingOutcome, TopicAffinities};
|
||||
use sqlx::sqlite::{SqlitePool, SqlitePoolOptions};
|
||||
use uuid::Uuid;
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct Db {
|
||||
pool: SqlitePool,
|
||||
}
|
||||
|
||||
impl Db {
|
||||
/// `path` e.g. "sqlite://feedsignal.db?mode=rwc" — everything lives in
|
||||
/// one file, no separate database server to run.
|
||||
pub async fn connect(url: &str) -> Result<Self> {
|
||||
let pool = SqlitePoolOptions::new().max_connections(5).connect(url).await?;
|
||||
sqlx::migrate!("./migrations").run(&pool).await?;
|
||||
Ok(Self { pool })
|
||||
}
|
||||
|
||||
pub async fn upsert_feed(&self, url: &str, title: &str) -> Result<Uuid> {
|
||||
let id = Uuid::new_v4();
|
||||
sqlx::query(
|
||||
"INSERT INTO feeds (id, url, title) VALUES (?, ?, ?)
|
||||
ON CONFLICT(url) DO UPDATE SET title = excluded.title",
|
||||
)
|
||||
.bind(id.to_string())
|
||||
.bind(url)
|
||||
.bind(title)
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
let row: (String,) = sqlx::query_as("SELECT id FROM feeds WHERE url = ?")
|
||||
.bind(url)
|
||||
.fetch_one(&self.pool)
|
||||
.await?;
|
||||
Ok(Uuid::parse_str(&row.0)?)
|
||||
}
|
||||
|
||||
pub async fn insert_article(&self, article: &Article) -> Result<()> {
|
||||
let topics = serde_json::to_string(&article.topics)?;
|
||||
sqlx::query(
|
||||
"INSERT OR IGNORE INTO articles
|
||||
(id, feed_id, url, title, summary, fetched_at, topics, embedding_score, llm_score, final_score, estimated_read_seconds)
|
||||
VALUES (?, ?, ?, ?, ?, datetime('now'), ?, ?, ?, ?, ?)",
|
||||
)
|
||||
.bind(article.id.to_string())
|
||||
.bind(article.feed_id.to_string())
|
||||
.bind(&article.url)
|
||||
.bind(&article.title)
|
||||
.bind(&article.summary)
|
||||
.bind(topics)
|
||||
.bind(article.embedding_score)
|
||||
.bind(article.llm_score)
|
||||
.bind(article.final_score)
|
||||
.bind(article.estimated_read_seconds)
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Articles above the embedding-similarity threshold that haven't been
|
||||
/// through the (slower) LLM scoring stage yet.
|
||||
pub async fn shortlist_for_llm_scoring(&self, threshold: f32, limit: i64) -> Result<Vec<Uuid>> {
|
||||
let rows: Vec<(String,)> = sqlx::query_as(
|
||||
"SELECT id FROM articles
|
||||
WHERE embedding_score >= ? AND llm_score IS NULL
|
||||
ORDER BY embedding_score DESC
|
||||
LIMIT ?",
|
||||
)
|
||||
.bind(threshold)
|
||||
.bind(limit)
|
||||
.fetch_all(&self.pool)
|
||||
.await?;
|
||||
rows.into_iter().map(|(s,)| Ok(Uuid::parse_str(&s)?)).collect()
|
||||
}
|
||||
|
||||
pub async fn record_event(&self, event: &ReadingEvent) -> Result<()> {
|
||||
let (outcome, dwell) = match &event.outcome {
|
||||
ReadingOutcome::Impression => ("impression", None),
|
||||
ReadingOutcome::Opened { dwell_seconds } => ("opened", *dwell_seconds),
|
||||
ReadingOutcome::Starred => ("starred", None),
|
||||
ReadingOutcome::Dismissed => ("dismissed", None),
|
||||
};
|
||||
sqlx::query(
|
||||
"INSERT INTO reading_events (id, article_id, occurred_at, outcome, dwell_seconds)
|
||||
VALUES (?, ?, ?, ?, ?)",
|
||||
)
|
||||
.bind(event.id.to_string())
|
||||
.bind(event.article_id.to_string())
|
||||
.bind(event.occurred_at.to_rfc3339())
|
||||
.bind(outcome)
|
||||
.bind(dwell)
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn load_affinities(&self) -> Result<TopicAffinities> {
|
||||
let row: Option<(String,)> = sqlx::query_as("SELECT affinities FROM topic_affinities WHERE user_id = 'default'")
|
||||
.fetch_optional(&self.pool)
|
||||
.await?;
|
||||
Ok(match row {
|
||||
Some((json,)) => serde_json::from_str(&json)?,
|
||||
None => TopicAffinities::default(),
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn save_affinities(&self, affinities: &TopicAffinities) -> Result<()> {
|
||||
let json = serde_json::to_string(affinities)?;
|
||||
sqlx::query(
|
||||
"INSERT INTO topic_affinities (user_id, affinities, updated_at) VALUES ('default', ?, datetime('now'))
|
||||
ON CONFLICT(user_id) DO UPDATE SET affinities = excluded.affinities, updated_at = excluded.updated_at",
|
||||
)
|
||||
.bind(json)
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn pool(&self) -> &SqlitePool {
|
||||
&self.pool
|
||||
}
|
||||
}
|
||||
13
crates/feeds/Cargo.toml
Normal file
13
crates/feeds/Cargo.toml
Normal file
|
|
@ -0,0 +1,13 @@
|
|||
[package]
|
||||
name = "feedsignal-feeds"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
license.workspace = true
|
||||
|
||||
[dependencies]
|
||||
feedsignal-core.workspace = true
|
||||
anyhow.workspace = true
|
||||
chrono.workspace = true
|
||||
uuid.workspace = true
|
||||
reqwest = { version = "0.12", default-features = false, features = ["rustls-tls"] }
|
||||
feed-rs = "2"
|
||||
53
crates/feeds/src/lib.rs
Normal file
53
crates/feeds/src/lib.rs
Normal file
|
|
@ -0,0 +1,53 @@
|
|||
use anyhow::Result;
|
||||
use chrono::{DateTime, Utc};
|
||||
use feedsignal_core::Article;
|
||||
use uuid::Uuid;
|
||||
|
||||
/// Fetches and parses one RSS/Atom feed. `feed-rs` normalizes both formats
|
||||
/// into a common model so callers don't need to branch on feed type.
|
||||
pub async fn fetch_feed(feed_url: &str, feed_id: Uuid) -> Result<Vec<Article>> {
|
||||
let bytes = reqwest::get(feed_url).await?.bytes().await?;
|
||||
let parsed = feed_rs::parser::parse(&bytes[..])?;
|
||||
|
||||
let articles = parsed
|
||||
.entries
|
||||
.into_iter()
|
||||
.filter_map(|entry| {
|
||||
let url = entry.links.first()?.href.clone();
|
||||
let title = entry.title.map(|t| t.content).unwrap_or_default();
|
||||
let summary = entry
|
||||
.summary
|
||||
.map(|s| s.content)
|
||||
.or_else(|| entry.content.and_then(|c| c.body))
|
||||
.unwrap_or_default();
|
||||
let published_at: Option<DateTime<Utc>> = entry.published.or(entry.updated);
|
||||
let estimated_read_seconds = Some(estimate_read_seconds(&summary));
|
||||
|
||||
Some(Article {
|
||||
id: Uuid::new_v4(),
|
||||
feed_id,
|
||||
url,
|
||||
title,
|
||||
summary,
|
||||
published_at,
|
||||
topics: Vec::new(),
|
||||
embedding_score: None,
|
||||
llm_score: None,
|
||||
final_score: None,
|
||||
estimated_read_seconds,
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
|
||||
Ok(articles)
|
||||
}
|
||||
|
||||
/// Rough estimate from word count at ~200 wpm, used as the denominator for
|
||||
/// dwell-time-based engagement scoring (see `feedsignal_core::affinity`).
|
||||
/// This is a placeholder — the full article body isn't fetched here, so
|
||||
/// it's a floor rather than an accurate estimate until content extraction
|
||||
/// is added.
|
||||
fn estimate_read_seconds(text: &str) -> u32 {
|
||||
let words = text.split_whitespace().count() as u32;
|
||||
((words.max(1) as f32 / 200.0) * 60.0) as u32
|
||||
}
|
||||
12
crates/llm/Cargo.toml
Normal file
12
crates/llm/Cargo.toml
Normal file
|
|
@ -0,0 +1,12 @@
|
|||
[package]
|
||||
name = "feedsignal-llm"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
license.workspace = true
|
||||
|
||||
[dependencies]
|
||||
anyhow.workspace = true
|
||||
serde.workspace = true
|
||||
serde_json.workspace = true
|
||||
tokio.workspace = true
|
||||
rig-core = "0.42"
|
||||
102
crates/llm/src/lib.rs
Normal file
102
crates/llm/src/lib.rs
Normal file
|
|
@ -0,0 +1,102 @@
|
|||
use anyhow::{Context, Result};
|
||||
use rig_core::client::{CompletionClient, EmbeddingsClient, Nothing};
|
||||
use rig_core::completion::CompletionModel;
|
||||
use rig_core::embeddings::EmbeddingModel as _;
|
||||
use rig_core::providers::ollama;
|
||||
use serde::Deserialize;
|
||||
|
||||
/// Thin wrapper around a local Ollama instance via rig. Two models are used
|
||||
/// deliberately: a small/cheap embedding model for the first-pass filter
|
||||
/// over every article, and a larger chat model reserved for the shortlist
|
||||
/// that clears the embedding threshold (see `feedsignal_core::scoring`).
|
||||
pub struct Llm {
|
||||
client: ollama::Client,
|
||||
embedding_model: String,
|
||||
embedding_dims: usize,
|
||||
chat_model: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub struct RelevanceJudgment {
|
||||
pub score: f32,
|
||||
pub rationale: String,
|
||||
}
|
||||
|
||||
impl Llm {
|
||||
/// `base_url` e.g. "http://localhost:11434". Model names must already
|
||||
/// be pulled locally, e.g. `ollama pull nomic-embed-text` and
|
||||
/// `ollama pull llama3.1`. `embedding_dims` must match the pulled
|
||||
/// embedding model (768 for nomic-embed-text, 384 for all-minilm).
|
||||
pub fn new(base_url: &str, embedding_model: impl Into<String>, embedding_dims: usize, chat_model: impl Into<String>) -> Result<Self> {
|
||||
let client = ollama::Client::builder()
|
||||
.api_key(Nothing)
|
||||
.base_url(base_url)
|
||||
.build()
|
||||
.context("failed to build ollama client")?;
|
||||
Ok(Self {
|
||||
client,
|
||||
embedding_model: embedding_model.into(),
|
||||
embedding_dims,
|
||||
chat_model: chat_model.into(),
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn embed(&self, text: &str) -> Result<Vec<f32>> {
|
||||
let model = self.client.embedding_model_with_ndims(&self.embedding_model, self.embedding_dims);
|
||||
let embedding = model.embed_text(text).await.context("ollama embedding request failed")?;
|
||||
Ok(embedding.vec.into_iter().map(|v| v as f32).collect())
|
||||
}
|
||||
|
||||
/// Runs the second-stage relevance judgment for one article. Called
|
||||
/// only for the embedding-similarity shortlist, not every article —
|
||||
/// see `feedsignal-db::shortlist_for_llm_scoring`.
|
||||
pub async fn judge_relevance(
|
||||
&self,
|
||||
preferences_description: &str,
|
||||
topic_affinities_summary: &str,
|
||||
article_title: &str,
|
||||
article_summary: &str,
|
||||
) -> Result<RelevanceJudgment> {
|
||||
let model = self.client.completion_model(&self.chat_model);
|
||||
|
||||
let prompt = format!(
|
||||
"Preferences: {preferences_description}\n\
|
||||
Historical topic affinities: {topic_affinities_summary}\n\n\
|
||||
Article title: {article_title}\n\
|
||||
Article summary: {article_summary}"
|
||||
);
|
||||
|
||||
let request = model
|
||||
.completion_request(prompt)
|
||||
.preamble(
|
||||
"You screen articles for a personal RSS reader. Given the \
|
||||
reader's stated preferences, their historical topic \
|
||||
affinities, and one article, respond with ONLY a JSON \
|
||||
object: {\"score\": <0.0-1.0>, \"rationale\": \"<one \
|
||||
sentence>\"}. score is how relevant this specific article \
|
||||
is to this specific reader right now."
|
||||
.to_string(),
|
||||
)
|
||||
.build();
|
||||
|
||||
let response = model.completion(request).await.context("ollama chat request failed")?;
|
||||
let text: String = response
|
||||
.choice
|
||||
.into_iter()
|
||||
.filter_map(|content| match content {
|
||||
rig_core::completion::AssistantContent::Text(t) => Some(t.text),
|
||||
_ => None,
|
||||
})
|
||||
.collect();
|
||||
parse_judgment(&text)
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_judgment(raw: &str) -> Result<RelevanceJudgment> {
|
||||
// Models sometimes wrap JSON in prose or code fences despite instructions;
|
||||
// take the outermost {...} span rather than trusting the whole response.
|
||||
let start = raw.find('{').context("no JSON object in LLM response")?;
|
||||
let end = raw.rfind('}').context("no JSON object in LLM response")?;
|
||||
let json = &raw[start..=end];
|
||||
Ok(serde_json::from_str(json)?)
|
||||
}
|
||||
45
crates/web/Cargo.toml
Normal file
45
crates/web/Cargo.toml
Normal file
|
|
@ -0,0 +1,45 @@
|
|||
[package]
|
||||
name = "feedsignal-web"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
license.workspace = true
|
||||
|
||||
[dependencies]
|
||||
feedsignal-core.workspace = true
|
||||
serde.workspace = true
|
||||
dioxus = { version = "0.7", features = ["router", "fullstack"] }
|
||||
|
||||
# Server-only dependencies: feed polling, the DB/LLM pipeline, and the
|
||||
# scheduler only need to exist in the native server binary, never in the
|
||||
# WASM bundle shipped to the browser.
|
||||
feedsignal-db = { workspace = true, optional = true }
|
||||
feedsignal-feeds = { workspace = true, optional = true }
|
||||
feedsignal-llm = { workspace = true, optional = true }
|
||||
tokio = { workspace = true, optional = true }
|
||||
tracing = { workspace = true, optional = true }
|
||||
tracing-subscriber = { workspace = true, optional = true }
|
||||
anyhow = { workspace = true, optional = true }
|
||||
sqlx = { workspace = true, optional = true }
|
||||
uuid = { workspace = true, optional = true }
|
||||
chrono = { workspace = true, optional = true }
|
||||
serde_json = { workspace = true, optional = true }
|
||||
tokio-cron-scheduler = { version = "0.13", optional = true }
|
||||
|
||||
[features]
|
||||
default = ["web"]
|
||||
web = ["dioxus/web"]
|
||||
server = [
|
||||
"dioxus/server",
|
||||
"dep:feedsignal-db",
|
||||
"dep:feedsignal-feeds",
|
||||
"dep:feedsignal-llm",
|
||||
"dep:tokio",
|
||||
"dep:tracing",
|
||||
"dep:tracing-subscriber",
|
||||
"dep:anyhow",
|
||||
"dep:sqlx",
|
||||
"dep:uuid",
|
||||
"dep:chrono",
|
||||
"dep:serde_json",
|
||||
"dep:tokio-cron-scheduler",
|
||||
]
|
||||
9
crates/web/Dioxus.toml
Normal file
9
crates/web/Dioxus.toml
Normal file
|
|
@ -0,0 +1,9 @@
|
|||
[application]
|
||||
name = "feedsignal"
|
||||
|
||||
[web.app]
|
||||
title = "feedsignal"
|
||||
|
||||
[web.watcher]
|
||||
reload_html = true
|
||||
watch_path = ["src"]
|
||||
8
crates/web/assets/app.css
Normal file
8
crates/web/assets/app.css
Normal file
|
|
@ -0,0 +1,8 @@
|
|||
body { font-family: system-ui, sans-serif; max-width: 720px; margin: 2rem auto; color: #1a1a1a; }
|
||||
.article-list { list-style: none; padding: 0; }
|
||||
.article-row { border-bottom: 1px solid #ddd; padding: 1rem 0; }
|
||||
.article-row a { font-weight: 600; text-decoration: none; color: #0b5fff; }
|
||||
.score { float: right; font-variant-numeric: tabular-nums; color: #555; }
|
||||
.summary { color: #444; margin: 0.4rem 0; }
|
||||
.topic-tag { display: inline-block; font-size: 0.75rem; background: #eee; border-radius: 4px; padding: 0.1rem 0.4rem; margin-right: 0.3rem; }
|
||||
.error { color: #b00020; }
|
||||
87
crates/web/src/app.rs
Normal file
87
crates/web/src/app.rs
Normal file
|
|
@ -0,0 +1,87 @@
|
|||
use dioxus::prelude::*;
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
/// Article shape sent to the browser — a trimmed view of
|
||||
/// `feedsignal_core::Article`, kept separate so the wire format can evolve
|
||||
/// independently of the storage model.
|
||||
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
||||
pub struct ArticleView {
|
||||
pub id: String,
|
||||
pub title: String,
|
||||
pub url: String,
|
||||
pub summary: String,
|
||||
pub topics: Vec<String>,
|
||||
pub final_score: Option<f32>,
|
||||
}
|
||||
|
||||
#[component]
|
||||
pub fn App() -> Element {
|
||||
let articles = use_server_future(list_ranked_articles)?;
|
||||
|
||||
rsx! {
|
||||
style { {include_str!("../assets/app.css")} }
|
||||
main {
|
||||
h1 { "feedsignal" }
|
||||
match articles.read().as_ref() {
|
||||
Some(Ok(articles)) => rsx! {
|
||||
ul { class: "article-list",
|
||||
for article in articles.iter() {
|
||||
ArticleRow { article: article.clone() }
|
||||
}
|
||||
}
|
||||
},
|
||||
Some(Err(err)) => rsx! { p { class: "error", "Failed to load articles: {err}" } },
|
||||
None => rsx! { p { "Loading..." } },
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[component]
|
||||
fn ArticleRow(article: ArticleView) -> Element {
|
||||
let score_pct = article.final_score.map(|s| (s * 100.0).round() as i32);
|
||||
rsx! {
|
||||
li { class: "article-row",
|
||||
a { href: "{article.url}", target: "_blank", "{article.title}" }
|
||||
if let Some(pct) = score_pct {
|
||||
span { class: "score", "{pct}%" }
|
||||
}
|
||||
p { class: "summary", "{article.summary}" }
|
||||
div { class: "topics",
|
||||
for topic in article.topics.iter() {
|
||||
span { class: "topic-tag", "{topic}" }
|
||||
}
|
||||
}
|
||||
button {
|
||||
onclick: move |_| {
|
||||
let id = article.id.clone();
|
||||
async move {
|
||||
let _ = mark_dismissed(id).await;
|
||||
}
|
||||
},
|
||||
"Not relevant"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns articles ranked by `final_score` descending, highest-relevance
|
||||
/// first. Runs on the server (native, has DB access); the `#[server]`
|
||||
/// macro generates the HTTP call the browser/WASM build uses instead.
|
||||
#[server]
|
||||
async fn list_ranked_articles() -> Result<Vec<ArticleView>, ServerFnError> {
|
||||
let db = crate::server::db().await?;
|
||||
crate::server::list_ranked_articles_impl(db)
|
||||
.await
|
||||
.map_err(|e| ServerFnError::new(e.to_string()))
|
||||
}
|
||||
|
||||
/// Records an explicit "not relevant" signal, which feeds directly into the
|
||||
/// topic-affinity update (see `feedsignal_core::affinity::apply_feedback`).
|
||||
#[server]
|
||||
async fn mark_dismissed(article_id: String) -> Result<(), ServerFnError> {
|
||||
let db = crate::server::db().await?;
|
||||
crate::server::mark_dismissed_impl(db, article_id)
|
||||
.await
|
||||
.map_err(|e| ServerFnError::new(e.to_string()))
|
||||
}
|
||||
11
crates/web/src/main.rs
Normal file
11
crates/web/src/main.rs
Normal file
|
|
@ -0,0 +1,11 @@
|
|||
mod app;
|
||||
|
||||
#[cfg(feature = "server")]
|
||||
mod server;
|
||||
|
||||
fn main() {
|
||||
#[cfg(feature = "server")]
|
||||
server::init_tracing();
|
||||
|
||||
dioxus::launch(app::App);
|
||||
}
|
||||
81
crates/web/src/server.rs
Normal file
81
crates/web/src/server.rs
Normal file
|
|
@ -0,0 +1,81 @@
|
|||
use crate::app::ArticleView;
|
||||
use anyhow::Result;
|
||||
use feedsignal_core::{ReadingEvent, ReadingOutcome};
|
||||
use feedsignal_db::Db;
|
||||
use feedsignal_llm::Llm;
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::OnceCell;
|
||||
use uuid::Uuid;
|
||||
|
||||
pub mod pipeline;
|
||||
|
||||
pub fn init_tracing() {
|
||||
tracing_subscriber::fmt::init();
|
||||
}
|
||||
|
||||
static BACKGROUND_JOBS: OnceCell<()> = OnceCell::const_new();
|
||||
|
||||
/// Lazily starts the feed-polling/scoring scheduler on first use. Deferred
|
||||
/// rather than started at process boot because `dioxus_server::launch_cfg`
|
||||
/// owns the tokio runtime construction — this runs the first time a server
|
||||
/// function executes, which is guaranteed to already be inside that runtime.
|
||||
async fn ensure_background_jobs_started(db: Arc<Db>) {
|
||||
BACKGROUND_JOBS
|
||||
.get_or_init(|| async {
|
||||
// TODO: move base_url/model names to config/env once there's a
|
||||
// settings story; hardcoded to the common local Ollama defaults.
|
||||
let llm = Arc::new(Llm::new("http://localhost:11434", "nomic-embed-text", 768, "llama3.1").expect("failed to construct ollama client"));
|
||||
if let Err(err) = pipeline::start_scheduler(db, llm).await {
|
||||
tracing::error!(?err, "failed to start background job scheduler");
|
||||
}
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
pub async fn db() -> Result<Db, dioxus::prelude::ServerFnError> {
|
||||
// TODO: hold this in a `OnceCell`/app-wide state instead of reconnecting
|
||||
// per request once the server-state story is wired up.
|
||||
let db = Db::connect("sqlite://feedsignal.db?mode=rwc")
|
||||
.await
|
||||
.map_err(|e| dioxus::prelude::ServerFnError::new(e.to_string()))?;
|
||||
ensure_background_jobs_started(Arc::new(db.clone())).await;
|
||||
Ok(db)
|
||||
}
|
||||
|
||||
pub async fn list_ranked_articles_impl(db: Db) -> Result<Vec<ArticleView>> {
|
||||
let rows: Vec<(String, String, String, String, String, Option<f32>)> = sqlx::query_as(
|
||||
"SELECT id, title, url, summary, topics, final_score
|
||||
FROM articles
|
||||
ORDER BY final_score DESC NULLS LAST
|
||||
LIMIT 100",
|
||||
)
|
||||
.fetch_all(db.pool())
|
||||
.await?;
|
||||
|
||||
Ok(rows
|
||||
.into_iter()
|
||||
.map(|(id, title, url, summary, topics_json, final_score)| ArticleView {
|
||||
id,
|
||||
title,
|
||||
url,
|
||||
summary,
|
||||
topics: serde_json::from_str(&topics_json).unwrap_or_default(),
|
||||
final_score,
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn mark_dismissed_impl(db: Db, article_id: String) -> Result<()> {
|
||||
let event = ReadingEvent {
|
||||
id: Uuid::new_v4(),
|
||||
article_id: Uuid::parse_str(&article_id)?,
|
||||
occurred_at: chrono::Utc::now(),
|
||||
outcome: ReadingOutcome::Dismissed,
|
||||
};
|
||||
db.record_event(&event).await?;
|
||||
// Affinity re-weighting from this event happens in the batched pipeline
|
||||
// job (see `pipeline::apply_pending_feedback`) rather than inline here,
|
||||
// so a burst of dismissals doesn't serialize on read-modify-write of
|
||||
// the single affinities row.
|
||||
Ok(())
|
||||
}
|
||||
128
crates/web/src/server/pipeline.rs
Normal file
128
crates/web/src/server/pipeline.rs
Normal file
|
|
@ -0,0 +1,128 @@
|
|||
use anyhow::Result;
|
||||
use feedsignal_core::{score_article, RelevanceInputs};
|
||||
use feedsignal_db::Db;
|
||||
use feedsignal_llm::Llm;
|
||||
use std::sync::Arc;
|
||||
use tokio_cron_scheduler::{Job, JobScheduler};
|
||||
|
||||
/// Below this embedding-similarity score, an article is never sent to the
|
||||
/// LLM at all — it's assumed irrelevant cheaply. Tune once real usage data
|
||||
/// exists; too low wastes LLM time, too high risks silently dropping
|
||||
/// things the LLM would have caught (e.g. a term-mismatch the embedding
|
||||
/// space doesn't capture well).
|
||||
const EMBEDDING_SHORTLIST_THRESHOLD: f32 = 0.55;
|
||||
const LLM_BATCH_SIZE: i64 = 25;
|
||||
|
||||
/// 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.
|
||||
pub async fn start_scheduler(db: Arc<Db>, llm: Arc<Llm>) -> Result<JobScheduler> {
|
||||
let scheduler = JobScheduler::new().await?;
|
||||
|
||||
{
|
||||
let db = db.clone();
|
||||
let llm = llm.clone();
|
||||
scheduler
|
||||
.add(Job::new_async("0 */15 * * * *", move |_uuid, _lock| {
|
||||
let db = db.clone();
|
||||
let llm = llm.clone();
|
||||
Box::pin(async move {
|
||||
if let Err(err) = run_scoring_pipeline(&db, &llm).await {
|
||||
tracing::error!(?err, "scoring pipeline run failed");
|
||||
}
|
||||
})
|
||||
})?)
|
||||
.await?;
|
||||
}
|
||||
|
||||
{
|
||||
let db = db.clone();
|
||||
scheduler
|
||||
.add(Job::new_async("0 0 4 * * *", move |_uuid, _lock| {
|
||||
let db = db.clone();
|
||||
Box::pin(async move {
|
||||
if let Err(err) = decay_affinities(&db).await {
|
||||
tracing::error!(?err, "affinity decay run failed");
|
||||
}
|
||||
})
|
||||
})?)
|
||||
.await?;
|
||||
}
|
||||
|
||||
scheduler.start().await?;
|
||||
Ok(scheduler)
|
||||
}
|
||||
|
||||
/// 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.
|
||||
async fn run_scoring_pipeline(db: &Db, llm: &Llm) -> Result<()> {
|
||||
let affinities = db.load_affinities().await?;
|
||||
let shortlist = db.shortlist_for_llm_scoring(EMBEDDING_SHORTLIST_THRESHOLD, LLM_BATCH_SIZE).await?;
|
||||
|
||||
tracing::info!(count = shortlist.len(), "scoring shortlist with LLM");
|
||||
|
||||
let affinity_summary = {
|
||||
let top = affinities.top_n(10);
|
||||
serde_json::to_string(&top)?
|
||||
};
|
||||
|
||||
for article_id in shortlist {
|
||||
// Fetch just the fields needed for the prompt; a real implementation
|
||||
// would batch this rather than one query per article.
|
||||
let Some((title, summary, topics_json, embedding_score)) = fetch_article_fields(db, article_id).await? else {
|
||||
continue;
|
||||
};
|
||||
let topics: Vec<String> = serde_json::from_str(&topics_json).unwrap_or_default();
|
||||
|
||||
let judgment = llm
|
||||
.judge_relevance(
|
||||
"TODO: load from `preferences` table",
|
||||
&affinity_summary,
|
||||
&title,
|
||||
&summary,
|
||||
)
|
||||
.await?;
|
||||
|
||||
let final_score = score_article(RelevanceInputs {
|
||||
embedding_score,
|
||||
llm_score: Some(judgment.score),
|
||||
topics: &topics,
|
||||
affinities: &affinities,
|
||||
});
|
||||
|
||||
store_llm_result(db, article_id, judgment.score, &judgment.rationale, final_score).await?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn fetch_article_fields(db: &Db, article_id: uuid::Uuid) -> Result<Option<(String, String, String, f32)>> {
|
||||
let row: Option<(String, String, String, Option<f32>)> = sqlx::query_as(
|
||||
"SELECT title, summary, topics, embedding_score FROM articles WHERE id = ?",
|
||||
)
|
||||
.bind(article_id.to_string())
|
||||
.fetch_optional(db.pool())
|
||||
.await?;
|
||||
Ok(row.map(|(t, s, topics, score)| (t, s, topics, score.unwrap_or(0.0))))
|
||||
}
|
||||
|
||||
async fn store_llm_result(db: &Db, article_id: uuid::Uuid, llm_score: f32, rationale: &str, final_score: f32) -> Result<()> {
|
||||
sqlx::query("UPDATE articles SET llm_score = ?, llm_rationale = ?, final_score = ? WHERE id = ?")
|
||||
.bind(llm_score)
|
||||
.bind(rationale)
|
||||
.bind(final_score)
|
||||
.bind(article_id.to_string())
|
||||
.execute(db.pool())
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn decay_affinities(db: &Db) -> Result<()> {
|
||||
let mut affinities = db.load_affinities().await?;
|
||||
affinities.decay();
|
||||
db.save_affinities(&affinities).await?;
|
||||
Ok(())
|
||||
}
|
||||
Loading…
Reference in a new issue