Replace sqlx with Diesel + diesel-async for compile-time-checked queries

sqlx's query!/query_as! macros only check raw SQL strings against the live
schema; Diesel's table!-derived DSL type-checks query structure itself at
compile time. SQLite has no native async driver, so diesel-async wraps a
blocking SqliteConnection via SyncConnectionWrapper, pooled with bb8.

Migrations move from sqlx's single-file-per-migration format to Diesel's
up.sql/down.sql pairs, run transactionally via diesel_migrations against a
throwaway sync connection at boot (MigrationHarness needs a sync
Connection), giving revertable migrations that sqlx::migrate! doesn't
support.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Cargo.toml

@@ -23,7 +23,11 @@ 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"] }
+diesel = { version = "2.3", features = ["sqlite", "chrono"] }
+diesel-async = { version = "0.9", features = ["sqlite", "bb8"] }
+diesel_migrations = { version = "2.3", features = ["sqlite"] }
+bb8 = "0.9"
+libsqlite3-sys = { version = "0.30", features = ["bundled"] }
 
 feedsignal-core = { path = "crates/core" }
 feedsignal-db = { path = "crates/db" }

crates/db/Cargo.toml

@@ -6,7 +6,11 @@ license.workspace = true
 
 [dependencies]
 feedsignal-core.workspace = true
-sqlx.workspace = true
+diesel.workspace = true
+diesel-async.workspace = true
+diesel_migrations.workspace = true
+bb8.workspace = true
+libsqlite3-sys.workspace = true
 tokio.workspace = true
 serde.workspace = true
 serde_json.workspace = true

crates/db/migrations/2026-08-21-000000_init/down.sql

@@ -0,0 +1,8 @@
+DROP TABLE preferences;
+DROP TABLE topic_affinities;
+DROP INDEX idx_reading_events_article_id;
+DROP TABLE reading_events;
+DROP INDEX idx_articles_final_score;
+DROP INDEX idx_articles_feed_id;
+DROP TABLE articles;
+DROP TABLE feeds;

crates/db/src/lib.rs

@@ -1,58 +1,87 @@
+mod schema;
+
 use anyhow::Result;
+use chrono::Utc;
+use diesel::prelude::*;
+use diesel::sqlite::SqliteConnection;
+use diesel_async::pooled_connection::bb8::Pool;
+use diesel_async::pooled_connection::AsyncDieselConnectionManager;
+use diesel_async::sync_connection_wrapper::SyncConnectionWrapper;
+use diesel_async::RunQueryDsl;
+use diesel_migrations::{embed_migrations, EmbeddedMigrations, MigrationHarness};
 use feedsignal_core::{Article, ReadingEvent, ReadingOutcome, TopicAffinities};
-use sqlx::sqlite::{SqlitePool, SqlitePoolOptions};
 use uuid::Uuid;
 
+const MIGRATIONS: EmbeddedMigrations = embed_migrations!("./migrations");
+
+/// SQLite has no native async driver, so `diesel-async` wraps a blocking
+/// `SqliteConnection` and runs it on a blocking thread under the hood.
+type AsyncSqliteConnection = SyncConnectionWrapper<SqliteConnection>;
+
 #[derive(Clone)]
 pub struct Db {
-    pool: SqlitePool,
+    pool: Pool<AsyncSqliteConnection>,
 }
 
 impl Db {
-    /// `path` e.g. "sqlite://feedsignal.db?mode=rwc" — everything lives in
+    /// `database_url` is a plain SQLite file path (e.g. "feedsignal.db"),
-    /// one file, no separate database server to run.
+    /// created if it doesn't exist. Everything lives in one file, no
-    pub async fn connect(url: &str) -> Result<Self> {
+    /// separate database server to run.
-        let pool = SqlitePoolOptions::new().max_connections(5).connect(url).await?;
+    pub async fn connect(database_url: &str) -> Result<Self> {
-        sqlx::migrate!("./migrations").run(&pool).await?;
+        // `MigrationHarness` only works on a synchronous `Connection`, and
+        // this only runs once at boot, so open a throwaway sync connection
+        // just for it rather than pulling migrations through the async pool.
+        // Each pending migration runs in its own transaction (SQLite
+        // supports transactional DDL), and `down.sql` gives each one a
+        // matching revert.
+        let url = database_url.to_string();
+        tokio::task::spawn_blocking(move || -> Result<()> {
+            let mut conn = SqliteConnection::establish(&url)?;
+            conn.run_pending_migrations(MIGRATIONS).map_err(|e| anyhow::anyhow!(e))?;
+            Ok(())
+        })
+        .await??;
+
+        let manager = AsyncDieselConnectionManager::<AsyncSqliteConnection>::new(database_url);
+        let pool = Pool::builder().max_size(5).build(manager).await?;
         Ok(Self { pool })
     }
 
     pub async fn upsert_feed(&self, url: &str, title: &str) -> Result<Uuid> {
-        let id = Uuid::new_v4();
+        use schema::feeds::dsl;
-        sqlx::query(
+        let mut conn = self.pool.get().await?;
-            "INSERT INTO feeds (id, url, title) VALUES (?, ?, ?)
+        let new_id = Uuid::new_v4().to_string();
-             ON CONFLICT(url) DO UPDATE SET title = excluded.title",
+        diesel::insert_into(dsl::feeds)
-        )
+            .values((dsl::id.eq(&new_id), dsl::url.eq(url), dsl::title.eq(title)))
-        .bind(id.to_string())
+            .on_conflict(dsl::url)
-        .bind(url)
+            .do_update()
-        .bind(title)
+            .set(dsl::title.eq(title))
-        .execute(&self.pool)
+            .execute(&mut conn)
             .await?;
-        let row: (String,) = sqlx::query_as("SELECT id FROM feeds WHERE url = ?")
+        let id: String = dsl::feeds.filter(dsl::url.eq(url)).select(dsl::id).first(&mut conn).await?;
-            .bind(url)
+        Ok(Uuid::parse_str(&id)?)
-            .fetch_one(&self.pool)
-            .await?;
-        Ok(Uuid::parse_str(&row.0)?)
     }
 
     pub async fn insert_article(&self, article: &Article) -> Result<()> {
+        use schema::articles::dsl;
+        let mut conn = self.pool.get().await?;
         let topics = serde_json::to_string(&article.topics)?;
-        sqlx::query(
+        diesel::insert_into(dsl::articles)
-            "INSERT OR IGNORE INTO articles
+            .values((
-                (id, feed_id, url, title, summary, fetched_at, topics, embedding_score, llm_score, final_score, estimated_read_seconds)
+                dsl::id.eq(article.id.to_string()),
-             VALUES (?, ?, ?, ?, ?, datetime('now'), ?, ?, ?, ?, ?)",
+                dsl::feed_id.eq(article.feed_id.to_string()),
-        )
+                dsl::url.eq(&article.url),
-        .bind(article.id.to_string())
+                dsl::title.eq(&article.title),
-        .bind(article.feed_id.to_string())
+                dsl::summary.eq(&article.summary),
-        .bind(&article.url)
+                dsl::fetched_at.eq(Utc::now().to_rfc3339()),
-        .bind(&article.title)
+                dsl::topics.eq(topics),
-        .bind(&article.summary)
+                dsl::embedding_score.eq(article.embedding_score),
-        .bind(topics)
+                dsl::llm_score.eq(article.llm_score),
-        .bind(article.embedding_score)
+                dsl::final_score.eq(article.final_score),
-        .bind(article.llm_score)
+                dsl::estimated_read_seconds.eq(article.estimated_read_seconds.map(|v| v as i32)),
-        .bind(article.final_score)
+            ))
-        .bind(article.estimated_read_seconds)
+            .on_conflict_do_nothing()
-        .execute(&self.pool)
+            .execute(&mut conn)
             .await?;
         Ok(())
     }
@@ -60,63 +89,118 @@ impl Db {
     /// 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(
+        use schema::articles::dsl;
-            "SELECT id FROM articles
+        let mut conn = self.pool.get().await?;
-             WHERE embedding_score >= ? AND llm_score IS NULL
+        let ids: Vec<String> = dsl::articles
-             ORDER BY embedding_score DESC
+            .filter(dsl::embedding_score.ge(threshold))
-             LIMIT ?",
+            .filter(dsl::llm_score.is_null())
-        )
+            .order(dsl::embedding_score.desc())
-        .bind(threshold)
+            .limit(limit)
-        .bind(limit)
+            .select(dsl::id)
-        .fetch_all(&self.pool)
+            .load(&mut conn)
             .await?;
-        rows.into_iter().map(|(s,)| Ok(Uuid::parse_str(&s)?)).collect()
+        ids.into_iter().map(|s| Ok(Uuid::parse_str(&s)?)).collect()
+    }
+
+    /// Title/summary/topics/embedding_score for one article, used to build
+    /// the LLM-judging prompt for it.
+    pub async fn article_scoring_fields(&self, article_id: Uuid) -> Result<Option<(String, String, Vec<String>, f32)>> {
+        use schema::articles::dsl;
+        let mut conn = self.pool.get().await?;
+        let row: Option<(String, String, String, Option<f32>)> = dsl::articles
+            .filter(dsl::id.eq(article_id.to_string()))
+            .select((dsl::title, dsl::summary, dsl::topics, dsl::embedding_score))
+            .first(&mut conn)
+            .await
+            .optional()?;
+        Ok(match row {
+            Some((title, summary, topics_json, score)) => {
+                let topics: Vec<String> = serde_json::from_str(&topics_json).unwrap_or_default();
+                Some((title, summary, topics, score.unwrap_or(0.0)))
+            }
+            None => None,
+        })
+    }
+
+    pub async fn store_llm_result(&self, article_id: Uuid, llm_score: f32, rationale: &str, final_score: f32) -> Result<()> {
+        use schema::articles::dsl;
+        let mut conn = self.pool.get().await?;
+        diesel::update(dsl::articles.filter(dsl::id.eq(article_id.to_string())))
+            .set((dsl::llm_score.eq(llm_score), dsl::llm_rationale.eq(rationale), dsl::final_score.eq(final_score)))
+            .execute(&mut conn)
+            .await?;
+        Ok(())
+    }
+
+    /// Highest-ranked articles for display, most relevant first.
+    pub async fn list_ranked_articles(&self, limit: i64) -> Result<Vec<(String, String, String, String, Vec<String>, Option<f32>)>> {
+        use schema::articles::dsl;
+        let mut conn = self.pool.get().await?;
+        // SQLite sorts NULL before any value, so `DESC` already puts NULL
+        // `final_score`s last — no separate NULLS LAST clause needed here.
+        let rows: Vec<(String, String, String, String, String, Option<f32>)> = dsl::articles
+            .order(dsl::final_score.desc())
+            .limit(limit)
+            .select((dsl::id, dsl::title, dsl::url, dsl::summary, dsl::topics, dsl::final_score))
+            .load(&mut conn)
+            .await?;
+        Ok(rows
+            .into_iter()
+            .map(|(id, title, url, summary, topics_json, final_score)| {
+                (id, title, url, summary, serde_json::from_str(&topics_json).unwrap_or_default(), final_score)
+            })
+            .collect())
     }
 
     pub async fn record_event(&self, event: &ReadingEvent) -> Result<()> {
+        use schema::reading_events::dsl;
+        let mut conn = self.pool.get().await?;
         let (outcome, dwell) = match &event.outcome {
             ReadingOutcome::Impression => ("impression", None),
-            ReadingOutcome::Opened { dwell_seconds } => ("opened", *dwell_seconds),
+            ReadingOutcome::Opened { dwell_seconds } => ("opened", dwell_seconds.map(|v| v as i32)),
             ReadingOutcome::Starred => ("starred", None),
             ReadingOutcome::Dismissed => ("dismissed", None),
         };
-        sqlx::query(
+        diesel::insert_into(dsl::reading_events)
-            "INSERT INTO reading_events (id, article_id, occurred_at, outcome, dwell_seconds)
+            .values((
-             VALUES (?, ?, ?, ?, ?)",
+                dsl::id.eq(event.id.to_string()),
-        )
+                dsl::article_id.eq(event.article_id.to_string()),
-        .bind(event.id.to_string())
+                dsl::occurred_at.eq(event.occurred_at.to_rfc3339()),
-        .bind(event.article_id.to_string())
+                dsl::outcome.eq(outcome),
-        .bind(event.occurred_at.to_rfc3339())
+                dsl::dwell_seconds.eq(dwell),
-        .bind(outcome)
+            ))
-        .bind(dwell)
+            .execute(&mut conn)
-        .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'")
+        use schema::topic_affinities::dsl;
-            .fetch_optional(&self.pool)
+        let mut conn = self.pool.get().await?;
-            .await?;
+        let row: Option<String> = dsl::topic_affinities
+            .filter(dsl::user_id.eq("default"))
+            .select(dsl::affinities)
+            .first(&mut conn)
+            .await
+            .optional()?;
         Ok(match row {
-            Some((json,)) => serde_json::from_str(&json)?,
+            Some(json) => serde_json::from_str(&json)?,
             None => TopicAffinities::default(),
         })
     }
 
     pub async fn save_affinities(&self, affinities: &TopicAffinities) -> Result<()> {
+        use schema::topic_affinities::dsl;
+        let mut conn = self.pool.get().await?;
         let json = serde_json::to_string(affinities)?;
-        sqlx::query(
+        let updated_at = Utc::now().to_rfc3339();
-            "INSERT INTO topic_affinities (user_id, affinities, updated_at) VALUES ('default', ?, datetime('now'))
+        diesel::insert_into(dsl::topic_affinities)
-             ON CONFLICT(user_id) DO UPDATE SET affinities = excluded.affinities, updated_at = excluded.updated_at",
+            .values((dsl::user_id.eq("default"), dsl::affinities.eq(&json), dsl::updated_at.eq(&updated_at)))
-        )
+            .on_conflict(dsl::user_id)
-        .bind(json)
+            .do_update()
-        .execute(&self.pool)
+            .set((dsl::affinities.eq(&json), dsl::updated_at.eq(&updated_at)))
+            .execute(&mut conn)
             .await?;
         Ok(())
     }
-
-    pub fn pool(&self) -> &SqlitePool {
-        &self.pool
-    }
 }

crates/db/src/schema.rs

@@ -0,0 +1,60 @@
+// Hand-maintained to match `migrations/2026-08-21-000000_init/up.sql`. Diesel
+// checks every query in `lib.rs` against these table! definitions at compile
+// time, so a query referencing a dropped column or wrong type fails to build
+// rather than failing at runtime.
+//
+// id/date columns are plain `Text` (not a native Uuid/DateTime SQL type):
+// Diesel's SQLite backend has no built-in UUID or timestamp type (that's
+// Postgres-only), so ids are stored as UUID strings and dates as RFC3339
+// strings, parsed at the Rust boundary in `lib.rs`.
+
+diesel::table! {
+    feeds (id) {
+        id -> Text,
+        url -> Text,
+        title -> Text,
+        last_fetched_at -> Nullable<Text>,
+    }
+}
+
+diesel::table! {
+    articles (id) {
+        id -> Text,
+        feed_id -> Text,
+        url -> Text,
+        title -> Text,
+        summary -> Text,
+        content -> Nullable<Text>,
+        published_at -> Nullable<Text>,
+        fetched_at -> Text,
+        topics -> Text,
+        embedding -> Nullable<Text>,
+        embedding_score -> Nullable<Float>,
+        llm_score -> Nullable<Float>,
+        llm_rationale -> Nullable<Text>,
+        final_score -> Nullable<Float>,
+        estimated_read_seconds -> Nullable<Integer>,
+    }
+}
+
+diesel::table! {
+    reading_events (id) {
+        id -> Text,
+        article_id -> Text,
+        occurred_at -> Text,
+        outcome -> Text,
+        dwell_seconds -> Nullable<Integer>,
+    }
+}
+
+diesel::table! {
+    topic_affinities (user_id) {
+        user_id -> Text,
+        affinities -> Text,
+        updated_at -> Text,
+    }
+}
+
+diesel::joinable!(articles -> feeds (feed_id));
+diesel::joinable!(reading_events -> articles (article_id));
+diesel::allow_tables_to_appear_in_same_query!(feeds, articles, reading_events, topic_affinities,);

crates/web/Cargo.toml

@@ -19,7 +19,6 @@ 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 }
@@ -37,7 +36,6 @@ server = [
     "dep:tracing",
     "dep:tracing-subscriber",
     "dep:anyhow",
-    "dep:sqlx",
     "dep:uuid",
     "dep:chrono",
     "dep:serde_json",

crates/web/src/server.rs

@@ -35,7 +35,7 @@ async fn ensure_background_jobs_started(db: Arc<Db>) {
 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")
+    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;
@@ -43,25 +43,11 @@ pub async fn db() -> Result<Db, dioxus::prelude::ServerFnError> {
 }
 
 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(
+    Ok(db
-        "SELECT id, title, url, summary, topics, final_score
+        .list_ranked_articles(100)
-         FROM articles
+        .await?
-         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 {
+        .map(|(id, title, url, summary, topics, final_score)| ArticleView { id, title, url, summary, topics, final_score })
-            id,
-            title,
-            url,
-            summary,
-            topics: serde_json::from_str(&topics_json).unwrap_or_default(),
-            final_score,
-        })
         .collect())
 }
 

crates/web/src/server/pipeline.rs

@@ -72,10 +72,9 @@ async fn run_scoring_pipeline(db: &Db, llm: &Llm) -> Result<()> {
     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 {
+        let Some((title, summary, topics, embedding_score)) = db.article_scoring_fields(article_id).await? else {
             continue;
         };
-        let topics: Vec<String> = serde_json::from_str(&topics_json).unwrap_or_default();
 
         let judgment = llm
             .judge_relevance(
@@ -93,33 +92,12 @@ async fn run_scoring_pipeline(db: &Db, llm: &Llm) -> Result<()> {
             affinities: &affinities,
         });
 
-        store_llm_result(db, article_id, judgment.score, &judgment.rationale, final_score).await?;
+        db.store_llm_result(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();