From 97c00f2e33f75c5b3e97039edafaaeae72dadf7d Mon Sep 17 00:00:00 2001 From: Jon Seager Date: Sun, 8 Feb 2026 15:20:01 +0000 Subject: [PATCH] feat(stats): add pre-computed stats cache with background recomputation - Add CachedStats domain types (roast summary, consumption, brewing) - Extend StatsRepository with summary queries and cache get/store - Add StatsInvalidator + background task with 2s debounce - Add invalidate() calls to all entity create/update/delete handlers - Add POST /api/v1/stats/recompute endpoint for manual refresh - Add stats_cache migration and include in database reset - Add GeoStats::from_counts() and Serialize/Deserialize derives - Add 6 integration tests for stats API - Document Stats Cache pattern in CLAUDE.md --- CLAUDE.md | 20 ++ migrations/0006_stats_cache.sql | 5 + src/application/routes/api/bags.rs | 2 + src/application/routes/api/brews.rs | 1 + src/application/routes/api/cafes.rs | 2 + src/application/routes/api/cups.rs | 1 + src/application/routes/api/gear.rs | 2 + src/application/routes/api/macros.rs | 1 + src/application/routes/api/mod.rs | 2 + src/application/routes/api/roasters.rs | 2 + src/application/routes/api/roasts.rs | 2 + src/application/routes/api/stats.rs | 27 +++ src/application/server.rs | 17 ++ src/application/services/mod.rs | 2 + src/application/services/stats.rs | 94 +++++++++ src/application/state.rs | 4 + src/domain/country_stats.rs | 38 +++- src/domain/mod.rs | 1 + src/domain/repositories.rs | 16 ++ src/domain/stats.rs | 50 +++++ src/infrastructure/backup.rs | 1 + src/infrastructure/repositories/stats.rs | 246 ++++++++++++++++++++++- tests/cli/helpers.rs | 4 + tests/server/helpers.rs | 2 + tests/server/main.rs | 1 + tests/server/stats_api.rs | 195 ++++++++++++++++++ 26 files changed, 735 insertions(+), 3 deletions(-) create mode 100644 migrations/0006_stats_cache.sql create mode 100644 src/application/routes/api/stats.rs create mode 100644 src/application/services/stats.rs create mode 100644 src/domain/stats.rs create mode 100644 tests/server/stats_api.rs diff --git a/CLAUDE.md b/CLAUDE.md index 3602f74..3b518a9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -226,6 +226,26 @@ Use `QueryBuilder` for dynamic queries. For UPDATE, use `push_update_field!` (se **`format_weight`** — displays grams up to 999g, switches to kg for 1000g+. Whole-gram values omit the decimal ("250g"), fractional values show one decimal ("15.5g"). Kilogram values always show one decimal ("1.5kg"). All weight values in the database are stored in grams. +### Stats Cache + +Statistics are pre-computed and stored as a single JSON row in `stats_cache`. A background `tokio::spawn` task (`stats_recomputation_task` in `application/services/stats.rs`) recomputes all stats when signalled, with 2-second debouncing to collapse rapid mutations (e.g., bootstrap script). + +**`StatsInvalidator`** — lives on `AppState`, provides `invalidate()` which sends a non-blocking signal to the background task. Every entity create/update/delete handler must call `state.stats_invalidator.invalidate()` after a successful mutation. The `define_delete_handler!` macro does this automatically. + +**Stats page** (`application/routes/app/stats.rs`) — reads from cache via `stats_repo.get_cached()`, falling back to live computation on cache miss (first startup before background task completes). + +**Force recompute** — `POST /api/v1/stats/recompute` (authenticated) bypasses the debounce and recomputes synchronously. Available on the admin page. + +**Database reset** clears `stats_cache` along with all other coffee data (see `infrastructure/backup.rs`). + +**Adding new stats:** +1. Add the field to the relevant domain struct (`RoastSummaryStats`, `ConsumptionStats`, `BrewingSummaryStats`, or `GeoStats`) +2. Add the query in `SqlStatsRepository` +3. The `CachedStats` struct inherits the change via serde +4. The stats page template can reference the new field — it will be populated from the cache + +**Gotcha:** If a new entity type is added that affects stats, its create/update/delete handlers must call `state.stats_invalidator.invalidate()`. + ### Error Handling & Logging **Error types**: `RepositoryError` (domain), `AppError` (HTTP with status code mapping), `anyhow::Result` (CLI). diff --git a/migrations/0006_stats_cache.sql b/migrations/0006_stats_cache.sql new file mode 100644 index 0000000..588c6e7 --- /dev/null +++ b/migrations/0006_stats_cache.sql @@ -0,0 +1,5 @@ +CREATE TABLE stats_cache ( + id INTEGER PRIMARY KEY CHECK (id = 1), + data TEXT NOT NULL, + computed_at TEXT NOT NULL DEFAULT (datetime('now')) +); diff --git a/src/application/routes/api/bags.rs b/src/application/routes/api/bags.rs index ba4cebf..25b25ab 100644 --- a/src/application/routes/api/bags.rs +++ b/src/application/routes/api/bags.rs @@ -70,6 +70,7 @@ pub(crate) async fn create_bag( .map_err(AppError::from)?; info!(bag_id = %bag.id, "bag created"); + state.stats_invalidator.invalidate(); if is_datastar_request(&headers) { render_bag_list_fragment(state, request, search, true) @@ -153,6 +154,7 @@ pub(crate) async fn update_bag( }; info!(%id, closed = ?update.closed, "bag updated"); + state.stats_invalidator.invalidate(); if is_datastar_request(&headers) { render_bag_list_fragment(state, request, search, true) diff --git a/src/application/routes/api/brews.rs b/src/application/routes/api/brews.rs index 53534b4..808a5c0 100644 --- a/src/application/routes/api/brews.rs +++ b/src/application/routes/api/brews.rs @@ -252,6 +252,7 @@ pub(crate) async fn create_brew( .map_err(AppError::from)?; info!(brew_id = %enriched.brew.id, "brew created"); + state.stats_invalidator.invalidate(); if is_datastar_request(&headers) { // If the request came from a page that has #brew-list, return the updated fragment. diff --git a/src/application/routes/api/cafes.rs b/src/application/routes/api/cafes.rs index 9825299..c32ed1d 100644 --- a/src/application/routes/api/cafes.rs +++ b/src/application/routes/api/cafes.rs @@ -72,6 +72,7 @@ pub(crate) async fn create_cafe( .map_err(AppError::from)?; info!(cafe_id = %cafe.id, name = %cafe.name, "cafe created"); + state.stats_invalidator.invalidate(); if is_datastar_request(&headers) { render_cafe_list_fragment(state, request, search, true) @@ -113,6 +114,7 @@ pub(crate) async fn update_cafe( .await .map_err(AppError::from)?; info!(%id, "cafe updated"); + state.stats_invalidator.invalidate(); Ok(Json(cafe)) } diff --git a/src/application/routes/api/cups.rs b/src/application/routes/api/cups.rs index d00d2a0..222c030 100644 --- a/src/application/routes/api/cups.rs +++ b/src/application/routes/api/cups.rs @@ -62,6 +62,7 @@ pub(crate) async fn create_cup( .map_err(AppError::from)?; info!(cup_id = %cup.id, "cup created"); + state.stats_invalidator.invalidate(); if is_datastar_request(&headers) { render_cup_list_fragment(state, request, search, true) diff --git a/src/application/routes/api/gear.rs b/src/application/routes/api/gear.rs index e8c917f..97d6069 100644 --- a/src/application/routes/api/gear.rs +++ b/src/application/routes/api/gear.rs @@ -65,6 +65,7 @@ pub(crate) async fn create_gear( .map_err(AppError::from)?; info!(gear_id = %gear.id, make = %gear.make, model = %gear.model, "gear created"); + state.stats_invalidator.invalidate(); if is_datastar_request(&headers) { render_gear_list_fragment(state, request, search, true) @@ -121,6 +122,7 @@ pub(crate) async fn update_gear( .map_err(AppError::from)?; info!(%id, "gear updated"); + state.stats_invalidator.invalidate(); if is_datastar_request(&headers) { render_gear_list_fragment(state, request, search, true) diff --git a/src/application/routes/api/macros.rs b/src/application/routes/api/macros.rs index 9d28e8a..d0dc262 100644 --- a/src/application/routes/api/macros.rs +++ b/src/application/routes/api/macros.rs @@ -96,6 +96,7 @@ macro_rules! define_delete_handler { .map_err(crate::application::errors::AppError::from)?; tracing::info!(%id, "entity deleted"); + state.stats_invalidator.invalidate(); if crate::application::routes::support::is_datastar_request(&headers) { $render_fragment(state, request, search, true) diff --git a/src/application/routes/api/mod.rs b/src/application/routes/api/mod.rs index 59ba854..b2f49c9 100644 --- a/src/application/routes/api/mod.rs +++ b/src/application/routes/api/mod.rs @@ -10,6 +10,7 @@ mod macros; pub(crate) mod roasters; pub(crate) mod roasts; pub(crate) mod scan; +pub(crate) mod stats; pub(crate) mod tokens; pub(crate) mod webauthn; @@ -91,6 +92,7 @@ pub(super) fn router() -> axum::Router { post(backup::restore_backup).layer(DefaultBodyLimit::max(50 * 1024 * 1024)), ) .route("/backup/reset", post(backup::reset_database)) + .route("/stats/recompute", post(stats::recompute_stats)) } pub(super) fn webauthn_router() -> axum::Router { diff --git a/src/application/routes/api/roasters.rs b/src/application/routes/api/roasters.rs index 357e0b9..0e402c0 100644 --- a/src/application/routes/api/roasters.rs +++ b/src/application/routes/api/roasters.rs @@ -73,6 +73,7 @@ pub(crate) async fn create_roaster( .map_err(AppError::from)?; info!(roaster_id = %roaster.id, name = %roaster.name, "roaster created"); + state.stats_invalidator.invalidate(); if is_datastar_request(&headers) { render_roaster_list_fragment(state, request, search, true) @@ -113,6 +114,7 @@ pub(crate) async fn update_roaster( .await .map_err(AppError::from)?; info!(%id, "roaster updated"); + state.stats_invalidator.invalidate(); Ok(Json(roaster)) } diff --git a/src/application/routes/api/roasts.rs b/src/application/routes/api/roasts.rs index 3dac3e3..32f4b62 100644 --- a/src/application/routes/api/roasts.rs +++ b/src/application/routes/api/roasts.rs @@ -72,6 +72,7 @@ pub(crate) async fn create_roast( .map_err(AppError::from)?; info!(roast_id = %roast.id, name = %roast.name, "roast created"); + state.stats_invalidator.invalidate(); if is_datastar_request(&headers) { render_roast_list_fragment(state, request, search, true) @@ -175,6 +176,7 @@ pub(crate) async fn update_roast( .map_err(AppError::from)?; info!(%id, "roast updated"); + state.stats_invalidator.invalidate(); let enriched = state .roast_repo diff --git a/src/application/routes/api/stats.rs b/src/application/routes/api/stats.rs new file mode 100644 index 0000000..ee2cfc5 --- /dev/null +++ b/src/application/routes/api/stats.rs @@ -0,0 +1,27 @@ +use axum::Json; +use axum::extract::State; + +use crate::application::auth::AuthenticatedUser; +use crate::application::errors::{ApiError, AppError}; +use crate::application::services::stats::compute_all_stats; +use crate::application::state::AppState; +use crate::domain::stats::CachedStats; + +/// Force an immediate stats recomputation, bypassing the debounce timer. +#[tracing::instrument(skip(state, _auth_user))] +pub(crate) async fn recompute_stats( + State(state): State, + _auth_user: AuthenticatedUser, +) -> Result, ApiError> { + let cached = compute_all_stats(&*state.stats_repo) + .await + .map_err(AppError::from)?; + + state + .stats_repo + .store_cached(&cached) + .await + .map_err(AppError::from)?; + + Ok(Json(cached)) +} diff --git a/src/application/server.rs b/src/application/server.rs index fee29b0..305127f 100644 --- a/src/application/server.rs +++ b/src/application/server.rs @@ -9,6 +9,8 @@ use tracing::info; use webauthn_rs::prelude::*; use crate::application::routes::app_router; +use crate::application::services::StatsInvalidator; +use crate::application::services::stats::stats_recomputation_task; use crate::application::state::{AppState, AppStateConfig}; use crate::domain::registration_tokens::NewRegistrationToken; use crate::domain::repositories::{RegistrationTokenRepository, UserRepository}; @@ -39,6 +41,9 @@ pub async fn serve(config: ServerConfig) -> anyhow::Result<()> { .context("failed to build WebAuthn instance")?, ); + let (stats_tx, stats_rx) = tokio::sync::mpsc::channel::<()>(32); + let stats_invalidator = StatsInvalidator::new(stats_tx); + let state = AppState::from_database( &database, AppStateConfig { @@ -48,9 +53,21 @@ pub async fn serve(config: ServerConfig) -> anyhow::Result<()> { openrouter_url: crate::infrastructure::ai::OPENROUTER_URL.to_string(), openrouter_api_key: config.openrouter_api_key, openrouter_model: config.openrouter_model, + stats_invalidator: stats_invalidator.clone(), }, ); + // Spawn background stats recomputation task + let stats_repo = Arc::clone(&state.stats_repo); + tokio::spawn(stats_recomputation_task( + stats_rx, + stats_repo, + std::time::Duration::from_secs(2), + )); + + // Seed the stats cache on startup + stats_invalidator.invalidate(); + // Clean up expired sessions on startup if let Err(err) = state.session_repo.delete_expired().await { tracing::warn!(error = %err, "failed to clean up expired sessions on startup"); diff --git a/src/application/services/mod.rs b/src/application/services/mod.rs index e3bb2cb..7fe9bee 100644 --- a/src/application/services/mod.rs +++ b/src/application/services/mod.rs @@ -2,11 +2,13 @@ mod bags; mod brews; mod cups; mod roasts; +pub mod stats; pub use bags::BagService; pub use brews::BrewService; pub use cups::CupService; pub use roasts::RoastService; +pub use stats::StatsInvalidator; use std::sync::Arc; diff --git a/src/application/services/stats.rs b/src/application/services/stats.rs new file mode 100644 index 0000000..48dd4d7 --- /dev/null +++ b/src/application/services/stats.rs @@ -0,0 +1,94 @@ +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use tokio::sync::mpsc; +use tracing::{error, info}; + +use crate::domain::country_stats::GeoStats; +use crate::domain::repositories::StatsRepository; +use crate::domain::stats::CachedStats; + +/// Sends invalidation signals to the background stats recomputer. +/// Non-blocking and fire-and-forget — safe to call from any handler. +#[derive(Clone)] +pub struct StatsInvalidator { + tx: mpsc::Sender<()>, +} + +impl StatsInvalidator { + pub fn new(tx: mpsc::Sender<()>) -> Self { + Self { tx } + } + + /// Signal that stats need recomputation. + pub fn invalidate(&self) { + let _ = self.tx.try_send(()); + } +} + +/// Listens for invalidation signals, debounces, and recomputes all stats. +/// Runs as a long-lived background task — spawn with `tokio::spawn`. +pub async fn stats_recomputation_task( + mut rx: mpsc::Receiver<()>, + stats_repo: Arc, + debounce: Duration, +) { + loop { + if rx.recv().await.is_none() { + break; + } + + // Debounce: wait then drain any accumulated signals + tokio::time::sleep(debounce).await; + while rx.try_recv().is_ok() {} + + match compute_all_stats(&*stats_repo).await { + Ok(cached) => { + if let Err(err) = stats_repo.store_cached(&cached).await { + error!(error = %err, "failed to store stats cache"); + } + } + Err(err) => error!(error = %err, "stats recomputation failed"), + } + } +} + +/// Runs all stats queries and assembles a complete `CachedStats` snapshot. +/// Logs the total computation time on success. +pub async fn compute_all_stats( + repo: &dyn StatsRepository, +) -> Result { + let start = Instant::now(); + + let ( + roast_summary, + consumption, + brewing_summary, + roaster_counts, + roast_counts, + cup_counts, + cafe_counts, + ) = tokio::join!( + repo.roast_summary(), + repo.consumption_summary(), + repo.brewing_summary(), + repo.roaster_country_counts(), + repo.roast_origin_counts(), + repo.cup_country_counts(), + repo.cafe_country_counts(), + ); + + let cached = CachedStats { + roast_summary: roast_summary?, + consumption: consumption?, + brewing_summary: brewing_summary?, + geo_roasters: GeoStats::from_counts(roaster_counts?), + geo_roasts: GeoStats::from_counts(roast_counts?), + geo_cups: GeoStats::from_counts(cup_counts?), + geo_cafes: GeoStats::from_counts(cafe_counts?), + computed_at: chrono::Utc::now().to_rfc3339(), + }; + + info!(duration_ms = start.elapsed().as_millis(), "stats computed"); + Ok(cached) +} diff --git a/src/application/state.rs b/src/application/state.rs index 9f83f95..0f68843 100644 --- a/src/application/state.rs +++ b/src/application/state.rs @@ -4,6 +4,7 @@ use webauthn_rs::prelude::*; use crate::application::services::{ BagService, BrewService, CafeService, CupService, GearService, RoastService, RoasterService, + StatsInvalidator, }; use crate::domain::repositories::{ AiUsageRepository, BagRepository, BrewRepository, CafeRepository, CupRepository, @@ -40,6 +41,7 @@ pub struct AppStateConfig { pub openrouter_url: String, pub openrouter_api_key: String, pub openrouter_model: String, + pub stats_invalidator: StatsInvalidator, } #[derive(Clone)] @@ -75,6 +77,7 @@ pub struct AppState { pub gear_service: GearService, pub cafe_service: CafeService, pub cup_service: CupService, + pub stats_invalidator: StatsInvalidator, } impl AppState { @@ -161,6 +164,7 @@ impl AppState { gear_service, cafe_service, cup_service, + stats_invalidator: config.stats_invalidator, } } } diff --git a/src/domain/country_stats.rs b/src/domain/country_stats.rs index 457e4f8..a672557 100644 --- a/src/domain/country_stats.rs +++ b/src/domain/country_stats.rs @@ -1,8 +1,10 @@ use std::collections::HashMap; use std::sync::LazyLock; +use serde::{Deserialize, Serialize}; + /// A single country's count for geographic statistics. -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct CountryStat { pub country_name: String, pub iso_code: String, @@ -11,13 +13,45 @@ pub struct CountryStat { } /// Aggregated geographic stats for one entity type. -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct GeoStats { pub entries: Vec, pub total_countries: usize, pub max_count: u64, } +impl GeoStats { + /// Build from raw (`country_name`, count) pairs, resolving ISO codes and flag emoji. + pub fn from_counts(raw: Vec<(String, u64)>) -> Self { + let entries: Vec = raw + .into_iter() + .map(|(name, count)| { + let iso = country_to_iso(&name).unwrap_or("").to_string(); + let flag = if iso.is_empty() { + String::new() + } else { + iso_to_flag_emoji(&iso) + }; + CountryStat { + country_name: name, + iso_code: iso, + flag_emoji: flag, + count, + } + }) + .collect(); + + let total_countries = entries.len(); + let max_count = entries.iter().map(|e| e.count).max().unwrap_or(0); + + Self { + entries, + total_countries, + max_count, + } + } +} + static COUNTRY_MAP: LazyLock> = LazyLock::new(|| { HashMap::from([ // Coffee-producing countries diff --git a/src/domain/mod.rs b/src/domain/mod.rs index 7890bab..bd16395 100644 --- a/src/domain/mod.rs +++ b/src/domain/mod.rs @@ -15,6 +15,7 @@ pub mod repositories; pub mod roasters; pub mod roasts; pub mod sessions; +pub mod stats; pub mod timeline; pub mod tokens; pub mod users; diff --git a/src/domain/repositories.rs b/src/domain/repositories.rs index 7ae92ce..90413ce 100644 --- a/src/domain/repositories.rs +++ b/src/domain/repositories.rs @@ -275,4 +275,20 @@ pub trait StatsRepository: Send + Sync { async fn roast_origin_counts(&self) -> Result, RepositoryError>; async fn cup_country_counts(&self) -> Result, RepositoryError>; async fn cafe_country_counts(&self) -> Result, RepositoryError>; + async fn roast_summary( + &self, + ) -> Result; + async fn consumption_summary( + &self, + ) -> Result; + async fn brewing_summary( + &self, + ) -> Result; + async fn get_cached( + &self, + ) -> Result, RepositoryError>; + async fn store_cached( + &self, + stats: &crate::domain::stats::CachedStats, + ) -> Result<(), RepositoryError>; } diff --git a/src/domain/stats.rs b/src/domain/stats.rs new file mode 100644 index 0000000..d6aab35 --- /dev/null +++ b/src/domain/stats.rs @@ -0,0 +1,50 @@ +use serde::{Deserialize, Serialize}; + +use super::country_stats::GeoStats; + +/// Summary statistics for roasts: origins, flavours, and roasters. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RoastSummaryStats { + pub unique_origins: u64, + pub top_origin: Option, + pub top_roaster: Option, + pub origin_counts: Vec<(String, u64)>, + pub max_origin_count: u64, + pub flavour_counts: Vec<(String, u64)>, + pub max_flavour_count: u64, +} + +/// Coffee consumption totals. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ConsumptionStats { + pub last_30_days_grams: f64, + pub all_time_grams: f64, + pub brews_last_30_days: u64, + pub brews_all_time: u64, +} + +/// Brewing equipment statistics. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct BrewingSummaryStats { + pub brewer_counts: Vec<(String, u64)>, + pub grinder_counts: Vec<(String, u64)>, + #[serde(alias = "grinder_kg_counts")] + pub grinder_weight_counts: Vec<(String, f64)>, + #[serde(alias = "max_grinder_kg")] + pub max_grinder_weight: f64, + pub brew_time_distribution: Vec<(String, u64)>, + pub max_brew_time_count: u64, +} + +/// Pre-computed snapshot of all statistics, stored as JSON in the cache table. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CachedStats { + pub roast_summary: RoastSummaryStats, + pub consumption: ConsumptionStats, + pub brewing_summary: BrewingSummaryStats, + pub geo_roasters: GeoStats, + pub geo_roasts: GeoStats, + pub geo_cups: GeoStats, + pub geo_cafes: GeoStats, + pub computed_at: String, +} diff --git a/src/infrastructure/backup.rs b/src/infrastructure/backup.rs index bc8a6ba..1bffa75 100644 --- a/src/infrastructure/backup.rs +++ b/src/infrastructure/backup.rs @@ -138,6 +138,7 @@ impl BackupService { "gear", "cafes", "roasters", + "stats_cache", ]; for table in tables { diff --git a/src/infrastructure/repositories/stats.rs b/src/infrastructure/repositories/stats.rs index 18ee52b..e72bd91 100644 --- a/src/infrastructure/repositories/stats.rs +++ b/src/infrastructure/repositories/stats.rs @@ -1,8 +1,10 @@ use async_trait::async_trait; -use sqlx::query_as; +use sqlx::{Row, query_as, query_scalar}; +use tracing::info; use crate::domain::RepositoryError; use crate::domain::repositories::StatsRepository; +use crate::domain::stats::{BrewingSummaryStats, CachedStats, ConsumptionStats, RoastSummaryStats}; use crate::infrastructure::database::DatabasePool; #[derive(Clone)] @@ -28,6 +30,20 @@ impl CountryCount { } } +#[derive(sqlx::FromRow)] +#[allow(dead_code)] +struct NameCount { + name: String, + count: i64, +} + +#[derive(sqlx::FromRow)] +#[allow(dead_code)] +struct NameWeight { + name: String, + total_grams: f64, +} + #[async_trait] impl StatsRepository for SqlStatsRepository { async fn roaster_country_counts(&self) -> Result, RepositoryError> { @@ -87,4 +103,232 @@ impl StatsRepository for SqlStatsRepository { Ok(rows.into_iter().map(CountryCount::into_tuple).collect()) } + + async fn roast_summary(&self) -> Result { + let unique_origins: i64 = query_scalar( + r"SELECT COUNT(DISTINCT origin) FROM roasts + WHERE origin IS NOT NULL AND origin != ''", + ) + .fetch_one(&self.pool) + .await + .map_err(|err| RepositoryError::unexpected(err.to_string()))?; + + let top_origin = query_as::<_, NameCount>( + r"SELECT origin as name, COUNT(*) as count FROM roasts + WHERE origin IS NOT NULL AND origin != '' + GROUP BY origin ORDER BY count DESC LIMIT 1", + ) + .fetch_optional(&self.pool) + .await + .map_err(|err| RepositoryError::unexpected(err.to_string()))? + .map(|r| r.name); + + let top_roaster = query_as::<_, NameCount>( + r"SELECT ro.name as name, COUNT(*) as count + FROM roasts r JOIN roasters ro ON r.roaster_id = ro.id + GROUP BY ro.id ORDER BY count DESC LIMIT 1", + ) + .fetch_optional(&self.pool) + .await + .map_err(|err| RepositoryError::unexpected(err.to_string()))? + .map(|r| r.name); + + let all_origin_counts = self.roast_origin_counts().await?; + let origin_counts: Vec<(String, u64)> = all_origin_counts.into_iter().take(5).collect(); + let max_origin_count = origin_counts.iter().map(|(_, c)| *c).max().unwrap_or(0); + + let all_flavour_counts: Vec<(String, u64)> = query_as::<_, NameCount>( + r"WITH RECURSIVE raw(val) AS ( + SELECT TRIM(j.value) + FROM roasts, json_each(roasts.tasting_notes) j + WHERE roasts.tasting_notes IS NOT NULL AND roasts.tasting_notes != '[]' + ), + split(note, rest) AS ( + SELECT TRIM(SUBSTR(val, 1, INSTR(val || ',', ',') - 1)), + TRIM(SUBSTR(val, INSTR(val || ',', ',') + 1)) + FROM raw + UNION ALL + SELECT TRIM(SUBSTR(rest, 1, INSTR(rest || ',', ',') - 1)), + TRIM(SUBSTR(rest, INSTR(rest || ',', ',') + 1)) + FROM split WHERE rest != '' + ) + SELECT note as name, COUNT(*) as count + FROM split WHERE note != '' + GROUP BY LOWER(note) ORDER BY count DESC", + ) + .fetch_all(&self.pool) + .await + .map_err(|err| RepositoryError::unexpected(err.to_string()))? + .into_iter() + .map(|r| (r.name, r.count as u64)) + .collect(); + + let flavour_counts: Vec<(String, u64)> = all_flavour_counts.into_iter().take(5).collect(); + let max_flavour_count = flavour_counts.iter().map(|(_, c)| *c).max().unwrap_or(0); + + Ok(RoastSummaryStats { + unique_origins: unique_origins as u64, + top_origin, + top_roaster, + origin_counts, + max_origin_count, + flavour_counts, + max_flavour_count, + }) + } + + async fn consumption_summary(&self) -> Result { + let last_30_days_grams: f64 = query_scalar( + r"SELECT COALESCE(SUM(coffee_weight), 0.0) FROM brews + WHERE created_at >= datetime('now', '-30 days')", + ) + .fetch_one(&self.pool) + .await + .map_err(|err| RepositoryError::unexpected(err.to_string()))?; + + let all_time_grams: f64 = + query_scalar(r"SELECT COALESCE(SUM(coffee_weight), 0.0) FROM brews") + .fetch_one(&self.pool) + .await + .map_err(|err| RepositoryError::unexpected(err.to_string()))?; + + let brews_last_30_days: i64 = query_scalar( + r"SELECT COUNT(*) FROM brews + WHERE created_at >= datetime('now', '-30 days')", + ) + .fetch_one(&self.pool) + .await + .map_err(|err| RepositoryError::unexpected(err.to_string()))?; + + let brews_all_time: i64 = query_scalar(r"SELECT COUNT(*) FROM brews") + .fetch_one(&self.pool) + .await + .map_err(|err| RepositoryError::unexpected(err.to_string()))?; + + Ok(ConsumptionStats { + last_30_days_grams, + all_time_grams, + brews_last_30_days: brews_last_30_days as u64, + brews_all_time: brews_all_time as u64, + }) + } + + async fn brewing_summary(&self) -> Result { + let brewer_counts: Vec<(String, u64)> = query_as::<_, NameCount>( + r"SELECT g.make || ' ' || g.model as name, COUNT(*) as count + FROM brews b JOIN gear g ON b.brewer_id = g.id + GROUP BY g.id ORDER BY count DESC", + ) + .fetch_all(&self.pool) + .await + .map_err(|err| RepositoryError::unexpected(err.to_string()))? + .into_iter() + .map(|r| (r.name, r.count as u64)) + .collect(); + + let grinder_counts: Vec<(String, u64)> = query_as::<_, NameCount>( + r"SELECT g.make || ' ' || g.model as name, COUNT(*) as count + FROM brews b JOIN gear g ON b.grinder_id = g.id + GROUP BY g.id ORDER BY count DESC", + ) + .fetch_all(&self.pool) + .await + .map_err(|err| RepositoryError::unexpected(err.to_string()))? + .into_iter() + .map(|r| (r.name, r.count as u64)) + .collect(); + + let grinder_weight_counts: Vec<(String, f64)> = query_as::<_, NameWeight>( + r"SELECT g.make || ' ' || g.model as name, + ROUND(COALESCE(SUM(b.coffee_weight), 0), 1) as total_grams + FROM brews b JOIN gear g ON b.grinder_id = g.id + GROUP BY g.id ORDER BY total_grams DESC + LIMIT 5", + ) + .fetch_all(&self.pool) + .await + .map_err(|err| RepositoryError::unexpected(err.to_string()))? + .into_iter() + .map(|r| (r.name, r.total_grams)) + .collect(); + + let max_grinder_weight = grinder_weight_counts + .iter() + .map(|(_, g)| *g) + .fold(0.0_f64, f64::max); + + let brew_time_distribution: Vec<(String, u64)> = query_as::<_, NameCount>( + r"SELECT + CASE + WHEN brew_time < 60 THEN '< 1:00' + WHEN brew_time < 90 THEN '1:00–1:30' + WHEN brew_time < 120 THEN '1:30–2:00' + WHEN brew_time < 150 THEN '2:00–2:30' + WHEN brew_time < 180 THEN '2:30–3:00' + ELSE '3:00+' + END as name, + COUNT(*) as count + FROM brews + WHERE brew_time IS NOT NULL + GROUP BY name + ORDER BY MIN(brew_time)", + ) + .fetch_all(&self.pool) + .await + .map_err(|err| RepositoryError::unexpected(err.to_string()))? + .into_iter() + .map(|r| (r.name, r.count as u64)) + .collect(); + + let max_brew_time_count = brew_time_distribution + .iter() + .map(|(_, c)| *c) + .max() + .unwrap_or(0); + + Ok(BrewingSummaryStats { + brewer_counts, + grinder_counts, + grinder_weight_counts, + max_grinder_weight, + brew_time_distribution, + max_brew_time_count, + }) + } + + async fn get_cached(&self) -> Result, RepositoryError> { + let row = sqlx::query(r"SELECT data FROM stats_cache WHERE id = 1") + .fetch_optional(&self.pool) + .await + .map_err(|err| RepositoryError::unexpected(err.to_string()))?; + + match row { + Some(row) => { + let json: String = row + .try_get("data") + .map_err(|err| RepositoryError::unexpected(err.to_string()))?; + let stats: CachedStats = serde_json::from_str(&json) + .map_err(|err| RepositoryError::unexpected(err.to_string()))?; + Ok(Some(stats)) + } + None => Ok(None), + } + } + + async fn store_cached(&self, stats: &CachedStats) -> Result<(), RepositoryError> { + let json = serde_json::to_string(stats) + .map_err(|err| RepositoryError::unexpected(err.to_string()))?; + + sqlx::query( + r"INSERT OR REPLACE INTO stats_cache (id, data, computed_at) + VALUES (1, ?, datetime('now'))", + ) + .bind(&json) + .execute(&self.pool) + .await + .map_err(|err| RepositoryError::unexpected(err.to_string()))?; + + info!("stats cache updated"); + Ok(()) + } } diff --git a/tests/cli/helpers.rs b/tests/cli/helpers.rs index 6a419f0..05aef08 100644 --- a/tests/cli/helpers.rs +++ b/tests/cli/helpers.rs @@ -85,6 +85,7 @@ fn ensure_server_started() -> Result<(String, String), String> { .await .expect("Failed to connect to test database"); + let (stats_tx, _stats_rx) = tokio::sync::mpsc::channel(1); let state = AppState::from_database( &database, AppStateConfig { @@ -95,6 +96,9 @@ fn ensure_server_started() -> Result<(String, String), String> { openrouter_url: brewlog::infrastructure::ai::OPENROUTER_URL.to_string(), openrouter_api_key: String::new(), openrouter_model: "openrouter/free".to_string(), + stats_invalidator: brewlog::application::services::StatsInvalidator::new( + stats_tx, + ), }, ); diff --git a/tests/server/helpers.rs b/tests/server/helpers.rs index 42fa310..efc8385 100644 --- a/tests/server/helpers.rs +++ b/tests/server/helpers.rs @@ -73,6 +73,7 @@ pub async fn spawn_app() -> TestApp { } fn test_state_config() -> AppStateConfig { + let (tx, _rx) = tokio::sync::mpsc::channel(1); AppStateConfig { webauthn: test_webauthn(), foursquare_url: brewlog::infrastructure::foursquare::FOURSQUARE_SEARCH_URL.to_string(), @@ -80,6 +81,7 @@ fn test_state_config() -> AppStateConfig { openrouter_url: brewlog::infrastructure::ai::OPENROUTER_URL.to_string(), openrouter_api_key: String::new(), openrouter_model: "openrouter/free".to_string(), + stats_invalidator: brewlog::application::services::StatsInvalidator::new(tx), } } diff --git a/tests/server/main.rs b/tests/server/main.rs index d864078..cb5254b 100644 --- a/tests/server/main.rs +++ b/tests/server/main.rs @@ -14,5 +14,6 @@ pub mod pages; pub mod roasters_api; pub mod roasts_api; pub mod scan_api; +pub mod stats_api; pub mod test_macros; pub mod timeline; diff --git a/tests/server/stats_api.rs b/tests/server/stats_api.rs new file mode 100644 index 0000000..bc8a4c7 --- /dev/null +++ b/tests/server/stats_api.rs @@ -0,0 +1,195 @@ +use reqwest::Client; + +use crate::helpers::{ + assert_datastar_headers_with_mode, assert_full_page, assert_html_fragment, create_default_cafe, + create_default_roast, create_default_roaster, spawn_app, spawn_app_with_auth, +}; + +#[tokio::test] +async fn stats_page_returns_200_with_empty_database() { + let app = spawn_app().await; + let client = Client::new(); + + let response = client + .get(app.page_url("/stats")) + .send() + .await + .expect("Failed to execute request"); + + assert_eq!(response.status(), 200); + + let body = response.text().await.expect("Failed to read body"); + assert_full_page(&body); + assert!(body.contains("Stats"), "Page should contain title"); +} + +#[tokio::test] +async fn stats_page_returns_200_with_data() { + let app = spawn_app_with_auth().await; + + let roaster = create_default_roaster(&app).await; + let _roast = create_default_roast(&app, roaster.id).await; + + let client = Client::new(); + let response = client + .get(app.page_url("/stats")) + .send() + .await + .expect("Failed to execute request"); + + assert_eq!(response.status(), 200); + + let body = response.text().await.expect("Failed to read body"); + assert_full_page(&body); + assert!( + body.contains("Ethiopia"), + "Stats page should contain roast origin" + ); +} + +#[tokio::test] +async fn stats_page_datastar_tab_switch_returns_fragment() { + let app = spawn_app().await; + let client = Client::new(); + + let response = client + .get(app.page_url("/stats?type=roasts")) + .header("datastar-request", "true") + .send() + .await + .expect("Failed to execute request"); + + assert_eq!(response.status(), 200); + assert_datastar_headers_with_mode(&response, "#stats-content", "inner"); + + let body = response.text().await.expect("Failed to read body"); + assert_html_fragment(&body); +} + +#[tokio::test] +async fn recompute_stats_requires_authentication() { + let app = spawn_app().await; + let client = Client::new(); + + let response = client + .post(app.api_url("/stats/recompute")) + .send() + .await + .expect("Failed to execute request"); + + assert_eq!(response.status(), 401); +} + +#[tokio::test] +async fn recompute_stats_returns_json() { + let app = spawn_app_with_auth().await; + let client = Client::new(); + + let response = client + .post(app.api_url("/stats/recompute")) + .bearer_auth(app.auth_token.as_ref().unwrap()) + .send() + .await + .expect("Failed to execute request"); + + assert_eq!(response.status(), 200); + + let body: serde_json::Value = response.json().await.expect("Failed to parse JSON"); + assert!( + body.get("computed_at").is_some(), + "Response should contain computed_at" + ); + assert!( + body.get("roast_summary").is_some(), + "Response should contain roast_summary" + ); + assert!( + body.get("consumption").is_some(), + "Response should contain consumption" + ); + assert!( + body.get("brewing_summary").is_some(), + "Response should contain brewing_summary" + ); +} + +#[tokio::test] +async fn recompute_stats_reflects_created_data() { + let app = spawn_app_with_auth().await; + + let roaster = create_default_roaster(&app).await; + let _roast = create_default_roast(&app, roaster.id).await; + let _cafe = create_default_cafe(&app).await; + + let client = Client::new(); + let response = client + .post(app.api_url("/stats/recompute")) + .bearer_auth(app.auth_token.as_ref().unwrap()) + .send() + .await + .expect("Failed to execute request"); + + assert_eq!(response.status(), 200); + + let body: serde_json::Value = response.json().await.expect("Failed to parse JSON"); + + // Roaster is in UK + let geo_roasters = &body["geo_roasters"]["entries"]; + assert!( + geo_roasters + .as_array() + .unwrap() + .iter() + .any(|e| e["country_name"] == "UK"), + "geo_roasters should contain UK: {geo_roasters}" + ); + + // Roast origin is Ethiopia + let geo_roasts = &body["geo_roasts"]["entries"]; + assert!( + geo_roasts + .as_array() + .unwrap() + .iter() + .any(|e| e["country_name"] == "Ethiopia"), + "geo_roasts should contain Ethiopia: {geo_roasts}" + ); + + // Cafe is in US + let geo_cafes = &body["geo_cafes"]["entries"]; + assert!( + geo_cafes + .as_array() + .unwrap() + .iter() + .any(|e| e["country_name"] == "US"), + "geo_cafes should contain US: {geo_cafes}" + ); +} + +#[tokio::test] +async fn stats_page_loads_after_recompute() { + let app = spawn_app_with_auth().await; + let client = Client::new(); + + // Populate the cache + let recompute_response = client + .post(app.api_url("/stats/recompute")) + .bearer_auth(app.auth_token.as_ref().unwrap()) + .send() + .await + .expect("Failed to recompute"); + assert_eq!(recompute_response.status(), 200); + + // Load the page — should read from cache + let response = client + .get(app.page_url("/stats")) + .send() + .await + .expect("Failed to execute request"); + + assert_eq!(response.status(), 200); + + let body = response.text().await.expect("Failed to read body"); + assert_full_page(&body); +}