feedsignal/crates/db/src/lib.rs

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
    }
}