Split web crate into layered modules (api/services/jobs/view)
All checks were successful
CI / check (pull_request) Successful in 8m13s
CI / test (pull_request) Successful in 2m36s
CI / audit (pull_request) Successful in 11s

Separates controllers (api/, the #[server] endpoints) from business
services (server/services/), background scheduling (server/jobs/,
renamed from pipeline.rs), infra bootstrap/config (server/mod.rs,
server/config.rs), and view components (app/), each one file per
responsibility instead of the previous server.rs/app.rs/pipeline.rs
grab-bags.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LoXS1ERDGC1P189RqmUxAF
This commit is contained in:
Austin Schaefer 2026-09-03 12:03:15 +02:00
parent fa9876a1b4
commit 44b9724645
19 changed files with 397 additions and 332 deletions

View file

@ -0,0 +1,23 @@
use super::ArticleView;
use dioxus::prelude::*;
/// 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]
pub async fn list_ranked_articles() -> Result<Vec<ArticleView>, ServerFnError> {
let db = crate::server::db().await?;
crate::server::services::articles::list_ranked(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]
pub async fn mark_dismissed(article_id: String) -> Result<(), ServerFnError> {
let db = crate::server::db().await?;
crate::server::services::reading_events::mark_dismissed(db, article_id)
.await
.map_err(|e| ServerFnError::new(e.to_string()))
}

14
crates/web/src/api/dto.rs Normal file
View file

@ -0,0 +1,14 @@
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>,
}

View file

@ -0,0 +1,12 @@
use dioxus::prelude::*;
/// Registers a feed and fetches it immediately. Runs on the server (native,
/// has DB + network access); the `#[server]` macro generates the HTTP call
/// the browser/WASM build uses instead.
#[server]
pub async fn subscribe_feed(url: String) -> Result<usize, ServerFnError> {
let db = crate::server::db().await?;
crate::server::services::feeds::subscribe(db, url)
.await
.map_err(|e| ServerFnError::new(e.to_string()))
}

View file

@ -0,0 +1,5 @@
pub mod articles;
mod dto;
pub mod feeds;
pub use dto::ArticleView;

View file

@ -1,167 +0,0 @@
use crate::components::button::{Button, ButtonVariant};
use crate::components::input::Input;
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 mut articles = use_server_future(list_ranked_articles)?;
rsx! {
Stylesheet { href: asset!("/assets/app.css") }
Stylesheet { href: asset!("/assets/dx-components-theme.css") }
main {
h1 { "feedsignal" }
SubscribeForm { on_subscribed: move |_| { articles.restart(); } }
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..." } },
}
}
}
}
/// Form for manually subscribing to a feed by URL. Fetches the feed
/// immediately on submit (rather than waiting for the next scheduled poll)
/// so the reader isn't empty right after subscribing.
#[component]
fn SubscribeForm(on_subscribed: EventHandler<()>) -> Element {
let mut url = use_signal(String::new);
let mut status = use_signal(|| Option::<Result<String, String>>::None);
let mut submitting = use_signal(|| false);
let submit = move |_| {
let feed_url = url.read().clone();
if feed_url.trim().is_empty() {
return;
}
spawn(async move {
submitting.set(true);
status.set(None);
match subscribe_feed(feed_url).await {
Ok(count) => {
status.set(Some(Ok(format!(
"Subscribed — pulled in {count} article(s)."
))));
url.set(String::new());
on_subscribed.call(());
}
Err(err) => status.set(Some(Err(err.to_string()))),
}
submitting.set(false);
});
};
rsx! {
form {
class: "subscribe-form",
onsubmit: move |ev: FormEvent| {
ev.prevent_default();
submit(());
},
Input {
r#type: "url",
placeholder: "https://example.com/feed.xml",
value: "{url}",
required: true,
disabled: submitting(),
oninput: move |ev: FormEvent| url.set(ev.value()),
}
Button {
r#type: "submit",
variant: ButtonVariant::Primary,
disabled: submitting() || url.read().trim().is_empty(),
if submitting() {
"Subscribing..."
} else {
"Subscribe"
}
}
}
match status.read().as_ref() {
Some(Ok(msg)) => rsx! { p { class: "subscribe-status success", "{msg}" } },
Some(Err(err)) => rsx! { p { class: "subscribe-status error", "Failed to subscribe: {err}" } },
None => rsx! {},
}
}
}
/// Registers a feed and fetches it immediately. Runs on the server (native,
/// has DB + network access); the `#[server]` macro generates the HTTP call
/// the browser/WASM build uses instead.
#[server]
async fn subscribe_feed(url: String) -> Result<usize, ServerFnError> {
let db = crate::server::db().await?;
crate::server::subscribe_feed_impl(db, url)
.await
.map_err(|e| ServerFnError::new(e.to_string()))
}
#[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()))
}

View file

@ -0,0 +1,31 @@
use crate::api;
use crate::api::ArticleView;
use dioxus::prelude::*;
#[component]
pub 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 _ = api::articles::mark_dismissed(id).await;
}
},
"Not relevant"
}
}
}
}

32
crates/web/src/app/mod.rs Normal file
View file

@ -0,0 +1,32 @@
mod article_row;
mod subscribe_form;
use crate::api;
use article_row::ArticleRow;
use dioxus::prelude::*;
use subscribe_form::SubscribeForm;
#[component]
pub fn App() -> Element {
let mut articles = use_server_future(api::articles::list_ranked_articles)?;
rsx! {
Stylesheet { href: asset!("/assets/app.css") }
Stylesheet { href: asset!("/assets/dx-components-theme.css") }
main {
h1 { "feedsignal" }
SubscribeForm { on_subscribed: move |_| { articles.restart(); } }
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..." } },
}
}
}
}

View file

@ -0,0 +1,69 @@
use crate::api;
use crate::components::button::{Button, ButtonVariant};
use crate::components::input::Input;
use dioxus::prelude::*;
/// Form for manually subscribing to a feed by URL. Fetches the feed
/// immediately on submit (rather than waiting for the next scheduled poll)
/// so the reader isn't empty right after subscribing.
#[component]
pub fn SubscribeForm(on_subscribed: EventHandler<()>) -> Element {
let mut url = use_signal(String::new);
let mut status = use_signal(|| Option::<Result<String, String>>::None);
let mut submitting = use_signal(|| false);
let submit = move |_| {
let feed_url = url.read().clone();
if feed_url.trim().is_empty() {
return;
}
spawn(async move {
submitting.set(true);
status.set(None);
match api::feeds::subscribe_feed(feed_url).await {
Ok(count) => {
status.set(Some(Ok(format!(
"Subscribed — pulled in {count} article(s)."
))));
url.set(String::new());
on_subscribed.call(());
}
Err(err) => status.set(Some(Err(err.to_string()))),
}
submitting.set(false);
});
};
rsx! {
form {
class: "subscribe-form",
onsubmit: move |ev: FormEvent| {
ev.prevent_default();
submit(());
},
Input {
r#type: "url",
placeholder: "https://example.com/feed.xml",
value: "{url}",
required: true,
disabled: submitting(),
oninput: move |ev: FormEvent| url.set(ev.value()),
}
Button {
r#type: "submit",
variant: ButtonVariant::Primary,
disabled: submitting() || url.read().trim().is_empty(),
if submitting() {
"Subscribing..."
} else {
"Subscribe"
}
}
}
match status.read().as_ref() {
Some(Ok(msg)) => rsx! { p { class: "subscribe-status success", "{msg}" } },
Some(Err(err)) => rsx! { p { class: "subscribe-status error", "Failed to subscribe: {err}" } },
None => rsx! {},
}
}
}

View file

@ -1,3 +1,4 @@
mod api;
mod app;
mod components;

View file

@ -1,114 +0,0 @@
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 models already pulled locally.
let llm = Arc::new(
Llm::new(
"http://localhost:11434",
"nomic-embed-text",
768,
"gemma4-e4b",
)
.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("feedsignal.db")
.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>> {
Ok(db
.list_ranked_articles(100)
.await?
.into_iter()
.map(
|(id, title, url, summary, topics, final_score)| ArticleView {
id,
title,
url,
summary,
topics,
final_score,
},
)
.collect())
}
/// Subscribes to a feed by URL: registers it (or updates its title if
/// already subscribed) and does an immediate first fetch so the reader
/// isn't empty until the next scheduled poll. Returns the number of
/// articles pulled in on this fetch.
///
/// Fetches under a throwaway id before writing anything, so there's exactly
/// one write to the `feeds` table (`upsert_feed` is the single source of
/// truth for the real id — the existing one on a re-subscribe, a fresh one
/// otherwise) and no half-registered row left behind if the fetch fails
/// (e.g. the URL is well-formed but points at nothing, or the feed no
/// longer exists — `fetch_feed` surfaces both as an `Err` via
/// `error_for_status`/feed-rs parse failure, which becomes the error
/// message shown in the subscribe form).
pub async fn subscribe_feed_impl(db: Db, url: String) -> Result<usize> {
anyhow::ensure!(!url.trim().is_empty(), "feed URL is required");
let url = url.trim();
let (title, mut articles) = feedsignal_feeds::fetch_feed(url, Uuid::new_v4()).await?;
let feed_id = db.upsert_feed(url, &title).await?;
for article in &mut articles {
article.feed_id = feed_id;
}
let count = articles.len();
for article in &articles {
db.insert_article(article).await?;
}
Ok(count)
}
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(())
}

View file

@ -0,0 +1,21 @@
/// Ollama connection settings for the embedding + judgment models.
///
/// TODO: move to config/env once there's a settings story; hardcoded to
/// models already pulled locally.
pub struct LlmConfig {
pub base_url: &'static str,
pub embedding_model: &'static str,
pub embedding_dims: usize,
pub judge_model: &'static str,
}
impl Default for LlmConfig {
fn default() -> Self {
Self {
base_url: "http://localhost:11434",
embedding_model: "nomic-embed-text",
embedding_dims: 768,
judge_model: "gemma4-e4b",
}
}
}

View file

@ -0,0 +1,9 @@
use anyhow::Result;
use feedsignal_db::Db;
pub async fn run(db: &Db) -> Result<()> {
let mut affinities = db.load_affinities().await?;
affinities.decay();
db.save_affinities(&affinities).await?;
Ok(())
}

View file

@ -0,0 +1,49 @@
mod affinity_decay;
mod scoring;
use anyhow::Result;
use feedsignal_db::Db;
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.
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) = scoring::run(&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) = affinity_decay::run(&db).await {
tracing::error!(?err, "affinity decay run failed");
}
})
})?)
.await?;
}
scheduler.start().await?;
Ok(scheduler)
}

View file

@ -2,8 +2,6 @@ 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
@ -13,52 +11,11 @@ use tokio_cron_scheduler::{Job, JobScheduler};
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<()> {
pub async fn run(db: &Db, llm: &Llm) -> Result<()> {
let affinities = db.load_affinities().await?;
let shortlist = db
.shortlist_for_llm_scoring(EMBEDDING_SHORTLIST_THRESHOLD, LLM_BATCH_SIZE)
@ -102,10 +59,3 @@ async fn run_scoring_pipeline(db: &Db, llm: &Llm) -> Result<()> {
Ok(())
}
async fn decay_affinities(db: &Db) -> Result<()> {
let mut affinities = db.load_affinities().await?;
affinities.decay();
db.save_affinities(&affinities).await?;
Ok(())
}

View file

@ -0,0 +1,51 @@
use anyhow::Result;
use feedsignal_db::Db;
use feedsignal_llm::Llm;
use std::sync::Arc;
use tokio::sync::OnceCell;
mod config;
pub mod jobs;
pub mod services;
use config::LlmConfig;
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 {
let cfg = LlmConfig::default();
let llm = Arc::new(
Llm::new(
cfg.base_url,
cfg.embedding_model,
cfg.embedding_dims,
cfg.judge_model,
)
.expect("failed to construct ollama client"),
);
if let Err(err) = jobs::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("feedsignal.db")
.await
.map_err(|e| dioxus::prelude::ServerFnError::new(e.to_string()))?;
ensure_background_jobs_started(Arc::new(db.clone())).await;
Ok(db)
}

View file

@ -0,0 +1,22 @@
use crate::api::ArticleView;
use anyhow::Result;
use feedsignal_db::Db;
/// Returns articles ranked by `final_score` descending, highest-relevance first.
pub async fn list_ranked(db: Db) -> Result<Vec<ArticleView>> {
Ok(db
.list_ranked_articles(100)
.await?
.into_iter()
.map(
|(id, title, url, summary, topics, final_score)| ArticleView {
id,
title,
url,
summary,
topics,
final_score,
},
)
.collect())
}

View file

@ -0,0 +1,33 @@
use anyhow::Result;
use feedsignal_db::Db;
use uuid::Uuid;
/// Subscribes to a feed by URL: registers it (or updates its title if
/// already subscribed) and does an immediate first fetch so the reader
/// isn't empty until the next scheduled poll. Returns the number of
/// articles pulled in on this fetch.
///
/// Fetches under a throwaway id before writing anything, so there's exactly
/// one write to the `feeds` table (`upsert_feed` is the single source of
/// truth for the real id — the existing one on a re-subscribe, a fresh one
/// otherwise) and no half-registered row left behind if the fetch fails
/// (e.g. the URL is well-formed but points at nothing, or the feed no
/// longer exists — `fetch_feed` surfaces both as an `Err` via
/// `error_for_status`/feed-rs parse failure, which becomes the error
/// message shown in the subscribe form).
pub async fn subscribe(db: Db, url: String) -> Result<usize> {
anyhow::ensure!(!url.trim().is_empty(), "feed URL is required");
let url = url.trim();
let (title, mut articles) = feedsignal_feeds::fetch_feed(url, Uuid::new_v4()).await?;
let feed_id = db.upsert_feed(url, &title).await?;
for article in &mut articles {
article.feed_id = feed_id;
}
let count = articles.len();
for article in &articles {
db.insert_article(article).await?;
}
Ok(count)
}

View file

@ -0,0 +1,3 @@
pub mod articles;
pub mod feeds;
pub mod reading_events;

View file

@ -0,0 +1,21 @@
use anyhow::Result;
use feedsignal_core::{ReadingEvent, ReadingOutcome};
use feedsignal_db::Db;
use uuid::Uuid;
/// Records an explicit "not relevant" signal, which feeds directly into the
/// topic-affinity update (see `feedsignal_core::affinity::apply_feedback`).
pub async fn mark_dismissed(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 `jobs::scoring::run`) rather than inline here, so a burst of
// dismissals doesn't serialize on read-modify-write of the single
// affinities row.
Ok(())
}