diff --git a/src/db/usage_store.rs b/src/db/usage_store.rs index e62762f..fc17de7 100644 --- a/src/db/usage_store.rs +++ b/src/db/usage_store.rs @@ -122,11 +122,13 @@ impl Store { Ok(true) } - pub(crate) fn list_usage_events( + pub(crate) fn fold_usage_events( &self, sources: Option<&[String]>, time_range: TimeRange, - ) -> Result> { + mut state: T, + mut fold: impl FnMut(&mut T, UsageEventRecord), + ) -> Result { let mut sql = String::from( "SELECT session_id, source, source_id, event_key, timestamp, model, provider, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, @@ -182,7 +184,10 @@ impl Store { }) })?; - rows.collect::, _>>().map_err(Into::into) + for row in rows { + fold(&mut state, row?); + } + Ok(state) } pub(crate) fn list_usage_events_for_session( diff --git a/src/tui/app.rs b/src/tui/app.rs index cfe5503..34d13ec 100644 --- a/src/tui/app.rs +++ b/src/tui/app.rs @@ -11,7 +11,7 @@ use crate::db::store::{ProjectDirectory, Store}; use crate::handoff; use crate::project_scope::ProjectScope; use crate::session_action; -use crate::skill_audit::{self, SkillAuditFilters, SkillAuditReport}; +use crate::skill_audit::SkillAuditReport; use crate::transcript; use crate::tui::layout::{ MessagePane, SearchLayout, ViewingLayout, search_layout, vertical_scrollbar_position, @@ -26,6 +26,7 @@ use crate::tui::share_state::{ }; use crate::tui::text_layout::wrap_visual_rows; use crate::tui::usage_state::UsageTab; +use crate::tui::usage_worker::{UsageRequest, UsageResponse}; use crate::tui::viewing_state::{ SanitizedLine, ViewingFrame, ViewingLineage, ViewingParent, ViewingSessionSummary, build_viewing_caches, @@ -33,7 +34,7 @@ use crate::tui::viewing_state::{ use crate::types::{ BackgroundJobStatus, MatchSource, Message, SearchResult, SemanticProgress, Session, }; -use crate::usage::{self, UsageFilters, UsageReport}; +use crate::usage::UsageReport; const USAGE_LOADING_MIN_MS: u128 = 75; const SEARCH_DEBOUNCE_MS: u64 = 250; @@ -198,6 +199,9 @@ pub(crate) struct App { pub(crate) usage_error: Option, pub(crate) usage_time_filter: TimeRange, pub(crate) usage_refresh_requested_at: Option, + pub(crate) usage_in_flight: bool, + pub(crate) usage_request_id: u64, + pub(crate) active_usage_request_id: u64, pub(crate) usage_breakdown_scroll: u16, pub(crate) usage_tab: UsageTab, pub(crate) skill_audit_report: Option, @@ -290,6 +294,9 @@ impl App { usage_error: None, usage_time_filter: TimeRange::All, usage_refresh_requested_at: None, + usage_in_flight: false, + usage_request_id: 0, + active_usage_request_id: 0, usage_breakdown_scroll: 0, usage_tab: UsageTab::Tokens, skill_audit_report: None, @@ -1199,7 +1206,7 @@ impl App { self.reset_usage_dashboard(); self.request_usage_refresh(); } - KeyCode::Enter if self.usage_tab == UsageTab::Skills => { + KeyCode::Enter if self.usage_tab == UsageTab::Skills && !self.usage_is_loading() => { self.open_skill_sessions(store); } KeyCode::Up | KeyCode::Char('k') => self.handle_scroll_up(store), @@ -1951,60 +1958,33 @@ impl App { } } - pub(crate) fn refresh_usage(&mut self, store: &Store) { - self.usage_refresh_requested_at = None; - self.usage_error = None; - self.skill_audit_error = None; - self.usage_breakdown_scroll = 0; - self.skill_audit_selected = 0; - let sources = self.source_filter_ids(); - let current_filters = - UsageFilters { sources: sources.clone(), time_range: self.usage_time_filter }; - match usage::build_usage_report(store, ¤t_filters) { - Ok(report) => { - self.usage_report = Some(report); - } - Err(err) => { - self.usage_report = None; - self.usage_error = Some(format!("Usage unavailable: {err}")); - } - } - - let year_filters = UsageFilters { sources, time_range: TimeRange::All }; - match usage::build_usage_report(store, &year_filters) { - Ok(report) => { - self.usage_year_report = Some(report); - } - Err(err) => { - self.usage_year_report = None; - if self.usage_error.is_none() { - self.usage_error = Some(format!("Usage unavailable: {err}")); - } - } + pub(crate) fn take_usage_request(&mut self, sync: bool) -> Option { + if !self.usage_refresh_is_due() { + return None; } - let skill_filters = SkillAuditFilters { + self.usage_refresh_requested_at = None; + self.usage_in_flight = true; + Some(UsageRequest { + id: self.active_usage_request_id, sources: self.source_filter_ids(), time_range: self.usage_time_filter, - }; - match skill_audit::build_skill_audit_report(store, &skill_filters) { - Ok(report) => { - self.skill_audit_report = Some(report); - } - Err(err) => { - self.skill_audit_report = None; - self.skill_audit_error = Some(format!("Skill audit unavailable: {err}")); - } - } + sync, + }) } pub(crate) fn request_usage_refresh(&mut self) { self.usage_error = None; + self.skill_audit_error = None; + self.usage_request_id = self.usage_request_id.saturating_add(1); + self.active_usage_request_id = self.usage_request_id; self.usage_refresh_requested_at = Some(Instant::now()); + self.usage_breakdown_scroll = 0; + self.skill_audit_selected = 0; } pub(crate) fn usage_is_loading(&self) -> bool { - self.usage_refresh_requested_at.is_some() + self.usage_refresh_requested_at.is_some() || self.usage_in_flight } pub(crate) fn usage_refresh_is_due(&self) -> bool { @@ -2015,6 +1995,7 @@ impl App { pub(crate) fn fail_usage_refresh(&mut self, error: impl std::fmt::Display) { self.usage_refresh_requested_at = None; + self.usage_in_flight = false; self.usage_report = None; self.usage_year_report = None; self.skill_audit_report = None; @@ -2022,6 +2003,42 @@ impl App { self.skill_audit_error = Some(format!("Skill audit unavailable: {error}")); } + pub(crate) fn apply_usage_response(&mut self, response: UsageResponse) { + if response.id != self.active_usage_request_id + || response.sources != self.source_filter_ids() + || response.time_range != self.usage_time_filter + { + return; + } + + self.usage_in_flight = false; + match response.current_report { + Ok(report) => self.usage_report = Some(report), + Err(error) => { + self.usage_report = None; + self.usage_error = Some(format!("Usage unavailable: {error}")); + } + } + + match response.all_time_report { + Ok(report) => self.usage_year_report = Some(report), + Err(error) => { + self.usage_year_report = None; + if self.usage_error.is_none() { + self.usage_error = Some(format!("Usage unavailable: {error}")); + } + } + } + + match response.skill_audit_report { + Ok(report) => self.skill_audit_report = Some(report), + Err(error) => { + self.skill_audit_report = None; + self.skill_audit_error = Some(format!("Skill audit unavailable: {error}")); + } + } + } + pub(crate) fn try_search(&mut self, store: &Store, worker: &SearchWorker) { self.refresh_semantic_progress(store); if !self.search_pending { @@ -2829,6 +2846,9 @@ mod tests { usage_error: None, usage_time_filter: TimeRange::All, usage_refresh_requested_at: None, + usage_in_flight: false, + usage_request_id: 0, + active_usage_request_id: 0, usage_breakdown_scroll: 0, usage_tab: UsageTab::Tokens, skill_audit_report: None, @@ -3861,4 +3881,75 @@ mod tests { assert!(app.filters_dirty); assert!(!app.search_pending); } + + #[test] + fn usage_refresh_yields_background_request() { + let mut app = app_with_sources(); + app.request_usage_refresh(); + app.usage_refresh_requested_at = Some(Instant::now() - Duration::from_millis(100)); + + let request = app.take_usage_request(false).expect("usage request"); + + assert_eq!(request.time_range, TimeRange::All); + assert!(!request.sync); + assert!(app.usage_is_loading()); + } + + #[test] + fn stale_usage_response_does_not_replace_latest_filter_state() { + use crate::tui::usage_worker::UsageResponse; + + let mut app = app_with_sources(); + app.source_filter_selection = vec!["codex".to_string()]; + app.usage_time_filter = TimeRange::Month; + app.usage_request_id = 1; + app.active_usage_request_id = 1; + app.request_usage_refresh(); + + app.apply_usage_response(UsageResponse { + id: 1, + sources: Some(vec!["codex".to_string()]), + time_range: TimeRange::Month, + current_report: Err("stale current report".to_string()), + all_time_report: Err("stale all-time report".to_string()), + skill_audit_report: Err("stale skill audit".to_string()), + }); + + assert_eq!(app.active_usage_request_id, 2); + assert!(app.usage_error.is_none()); + assert!(app.skill_audit_error.is_none()); + assert_eq!(app.usage_time_filter, TimeRange::Month); + } + + #[test] + fn skill_drill_down_is_disabled_while_usage_is_loading() { + use crate::skill_audit::{SkillAuditSummary, SkillTier, SkillUsageEntry}; + + crate::db::schema::register_sqlite_vec(); + let store = Store::open_in_memory().unwrap(); + let mut app = app_with_sources(); + app.mode = AppMode::Usage; + app.usage_tab = UsageTab::Skills; + app.usage_in_flight = true; + app.skill_audit_report = Some(SkillAuditReport { + summary: SkillAuditSummary { installed: 1, core: 0, occasional: 1, dormant: 0 }, + core: Vec::new(), + occasional: vec![SkillUsageEntry { + id: "stale-skill".to_string(), + tier: SkillTier::Occasional, + invocations: 1, + last_used: None, + signals: Vec::new(), + install_path: None, + session_ids: vec!["stale-session".to_string()], + }], + dormant: Vec::new(), + coverage_note: None, + }); + + app.handle_usage_key(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE), &store); + + assert!(matches!(app.mode, AppMode::Usage)); + assert!(app.status_message.is_none()); + } } diff --git a/src/tui/mod.rs b/src/tui/mod.rs index 0a01e17..f4429eb 100644 --- a/src/tui/mod.rs +++ b/src/tui/mod.rs @@ -9,4 +9,5 @@ pub(crate) mod text_layout; pub(crate) mod theme; pub(crate) mod ui; pub(crate) mod usage_state; +pub(crate) mod usage_worker; pub(crate) mod viewing_state; diff --git a/src/tui/runner.rs b/src/tui/runner.rs index 3d47bf6..3c3cdbc 100644 --- a/src/tui/runner.rs +++ b/src/tui/runner.rs @@ -5,8 +5,8 @@ use crate::config::AppConfig; use crate::db::search::TimeRange; use crate::db::store::Store; use crate::semantic; -use crate::sync::run_dashboard_sync_job; use crate::tui::search_worker::SearchWorker; +use crate::tui::usage_worker::UsageWorker; pub(crate) fn run(usage_start: Option<(Option>, Option)>) -> Result<()> { use std::io; @@ -61,6 +61,7 @@ pub(crate) fn run(usage_start: Option<(Option>, Option)>) Some("Debug builds do not start semantic indexing; run cargo run -- sync first".into()); } let search_worker = SearchWorker::spawn(); + let usage_worker = UsageWorker::spawn(); if let Some((source_filter, time_filter)) = usage_start { app.source_filter_selection = source_filter.unwrap_or_default(); if let Some(time_filter) = time_filter { @@ -77,6 +78,9 @@ pub(crate) fn run(usage_start: Option<(Option>, Option)>) while let Some(response) = search_worker.try_recv() { app.apply_search_response(&store, response); } + while let Some(response) = usage_worker.try_recv() { + app.apply_usage_response(response); + } terminal.draw(|f| ui::render(f, &app))?; let size = terminal.size()?; @@ -97,15 +101,10 @@ pub(crate) fn run(usage_start: Option<(Option>, Option)>) break; } - if app.usage_refresh_is_due() { - if usage_sync_pending { - usage_sync_pending = false; - match run_dashboard_sync_job() { - Ok(()) => app.refresh_usage(&store), - Err(err) => app.fail_usage_refresh(err), - } - } else { - app.refresh_usage(&store); + if let Some(request) = app.take_usage_request(usage_sync_pending) { + usage_sync_pending = false; + if !usage_worker.refresh(request) { + app.fail_usage_refresh("Usage worker unavailable"); } } @@ -113,6 +112,9 @@ pub(crate) fn run(usage_start: Option<(Option>, Option)>) while let Some(response) = search_worker.try_recv() { app.apply_search_response(&store, response); } + while let Some(response) = usage_worker.try_recv() { + app.apply_usage_response(response); + } if app.should_quit { break; diff --git a/src/tui/usage_worker.rs b/src/tui/usage_worker.rs new file mode 100644 index 0000000..ce6f3b9 --- /dev/null +++ b/src/tui/usage_worker.rs @@ -0,0 +1,234 @@ +use std::sync::mpsc::{self, Receiver, Sender}; +use std::thread; + +use crate::db::search::TimeRange; +use crate::db::store::Store; +use crate::skill_audit::{self, SkillAuditFilters, SkillAuditReport}; +use crate::sync::run_dashboard_sync_job; +use crate::usage::{self, UsageFilters, UsageReport}; + +#[derive(Debug)] +pub(crate) struct UsageRequest { + pub(crate) id: u64, + pub(crate) sources: Option>, + pub(crate) time_range: TimeRange, + pub(crate) sync: bool, +} + +pub(crate) struct UsageResponse { + pub(crate) id: u64, + pub(crate) sources: Option>, + pub(crate) time_range: TimeRange, + pub(crate) current_report: Result, + pub(crate) all_time_report: Result, + pub(crate) skill_audit_report: Result, +} + +pub(crate) struct UsageWorker { + request_tx: Sender, + response_rx: Receiver, +} + +impl UsageWorker { + pub(crate) fn spawn() -> Self { + Self::spawn_with(run_worker) + } + + fn spawn_with( + run: impl FnOnce(Receiver, Sender) + Send + 'static, + ) -> Self { + let (request_tx, request_rx) = mpsc::channel(); + let (response_tx, response_rx) = mpsc::channel(); + thread::spawn(move || run(request_rx, response_tx)); + Self { request_tx, response_rx } + } + + pub(crate) fn refresh(&self, request: UsageRequest) -> bool { + self.request_tx.send(request).is_ok() + } + + pub(crate) fn try_recv(&self) -> Option { + self.response_rx.try_recv().ok() + } +} + +fn run_worker(request_rx: Receiver, response_tx: Sender) { + crate::db::schema::register_sqlite_vec(); + + let store = match Store::open() { + Ok(store) => store, + Err(error) => { + while let Ok(request) = request_rx.recv() { + let _ = response_tx + .send(failed_response(request, format!("Database unavailable: {error}"))); + } + return; + } + }; + + let mut sync_pending = false; + while let Ok(mut request) = request_rx.recv() { + while let Ok(next) = request_rx.try_recv() { + let sync = request.sync || next.sync; + request = UsageRequest { sync, ..next }; + } + + if let Err(error) = run_sync(&mut sync_pending, request.sync, run_dashboard_sync_job) { + if response_tx.send(failed_response(request, error)).is_err() { + return; + } + continue; + } + + let response = run_request(&store, request); + if response_tx.send(response).is_err() { + return; + } + } +} + +fn run_sync( + pending: &mut bool, + requested: bool, + sync: impl FnOnce() -> anyhow::Result<()>, +) -> Result<(), String> { + *pending |= requested; + if *pending { + sync().map_err(|error| format!("Sync failed: {error}"))?; + *pending = false; + } + Ok(()) +} + +fn run_request(store: &Store, request: UsageRequest) -> UsageResponse { + let (current_report, all_time_report) = + build_usage_reports(&request, |filters| usage::build_usage_report(store, filters)); + let skill_filters = + SkillAuditFilters { sources: request.sources.clone(), time_range: request.time_range }; + let skill_audit_report = skill_audit::build_skill_audit_report(store, &skill_filters) + .map_err(|error| error.to_string()); + + UsageResponse { + id: request.id, + sources: request.sources, + time_range: request.time_range, + current_report, + all_time_report, + skill_audit_report, + } +} + +fn build_usage_reports( + request: &UsageRequest, + mut build: impl FnMut(&UsageFilters) -> anyhow::Result, +) -> (Result, Result) { + let current_filters = + UsageFilters { sources: request.sources.clone(), time_range: request.time_range }; + let current_report = build(¤t_filters).map_err(|error| error.to_string()); + + let all_time_report = if request.time_range == TimeRange::All { + current_report.clone() + } else { + let all_time_filters = + UsageFilters { sources: request.sources.clone(), time_range: TimeRange::All }; + build(&all_time_filters).map_err(|error| error.to_string()) + }; + + (current_report, all_time_report) +} + +fn failed_response(request: UsageRequest, error: String) -> UsageResponse { + UsageResponse { + id: request.id, + sources: request.sources, + time_range: request.time_range, + current_report: Err(error.clone()), + all_time_report: Err(error.clone()), + skill_audit_report: Err(error), + } +} + +#[cfg(test)] +mod tests { + use std::sync::mpsc; + use std::time::Duration; + + use super::*; + + fn request(time_range: TimeRange) -> UsageRequest { + UsageRequest { id: 1, sources: None, time_range, sync: false } + } + + #[test] + fn identical_filters_build_one_usage_report() { + let mut builds = 0; + let (current, all_time) = build_usage_reports(&request(TimeRange::All), |_| { + builds += 1; + Ok(usage::aggregate_usage_events(&[])) + }); + + assert_eq!(builds, 1); + assert!(current.is_ok()); + assert!(all_time.is_ok()); + } + + #[test] + fn different_filters_keep_current_and_all_time_ranges() { + let mut ranges = Vec::new(); + let (current, all_time) = build_usage_reports(&request(TimeRange::Month), |filters| { + ranges.push(filters.time_range); + Ok(usage::aggregate_usage_events(&[])) + }); + + assert_eq!(ranges, vec![TimeRange::Month, TimeRange::All]); + assert!(current.is_ok()); + assert!(all_time.is_ok()); + } + + #[test] + fn request_returns_while_worker_is_busy() { + let (started_tx, started_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + let worker = UsageWorker::spawn_with(move |request_rx, _| { + request_rx.recv().unwrap(); + started_tx.send(()).unwrap(); + release_rx.recv().unwrap(); + }); + + assert!(worker.refresh(request(TimeRange::All))); + started_rx.recv_timeout(Duration::from_secs(1)).unwrap(); + assert!(worker.try_recv().is_none()); + release_tx.send(()).unwrap(); + } + + #[test] + fn failed_initial_sync_is_retried_by_the_next_request() { + let mut pending = false; + + assert!(run_sync(&mut pending, true, || anyhow::bail!("offline")).is_err()); + assert!(pending); + assert!(run_sync(&mut pending, false, || Ok(())).is_ok()); + assert!(!pending); + } + + #[test] + fn request_returns_usage_and_skill_reports() { + crate::db::schema::register_sqlite_vec(); + let store = Store::open_in_memory().unwrap(); + + let response = run_request(&store, request(TimeRange::Month)); + + assert!(response.current_report.is_ok()); + assert!(response.all_time_report.is_ok()); + assert!(response.skill_audit_report.is_ok()); + } + + #[test] + fn failure_reaches_every_report() { + let response = failed_response(request(TimeRange::All), "sync failed".to_string()); + + assert_eq!(response.current_report.unwrap_err(), "sync failed"); + assert_eq!(response.all_time_report.unwrap_err(), "sync failed"); + assert_eq!(response.skill_audit_report.unwrap_err(), "sync failed"); + } +} diff --git a/src/usage.rs b/src/usage.rs index c56d97c..8700abc 100644 --- a/src/usage.rs +++ b/src/usage.rs @@ -130,87 +130,21 @@ impl Accumulator { } } -pub(crate) fn build_usage_report(store: &Store, filters: &UsageFilters) -> Result { - let events = store.list_usage_events(filters.sources.as_deref(), filters.time_range)?; - Ok(aggregate_usage_events(&events)) -} - -pub(crate) fn aggregate_usage_events(events: &[UsageEventRecord]) -> UsageReport { - let events = dedupe_report_events(events); - let mut total = Accumulator::default(); - let mut by_source: BTreeMap = BTreeMap::new(); - let mut by_model: BTreeMap<(String, String, String), Accumulator> = BTreeMap::new(); - let mut daily: BTreeMap = BTreeMap::new(); - let mut weekly: BTreeMap = BTreeMap::new(); - let mut monthly: BTreeMap = BTreeMap::new(); - let mut token_source_events = BTreeMap::new(); - - for event in events { - total.add(event); - *token_source_events.entry(event.token_source.clone()).or_insert(0) += 1; - - by_source.entry(event.source.clone()).or_default().add(event); - by_model - .entry((event.source.clone(), event.provider.clone(), event.model.clone())) - .or_default() - .add(event); - - let (day, week, month) = period_keys(event.timestamp); - daily.entry(day).or_default().add(event); - weekly.entry(week).or_default().add(event); - monthly.entry(month).or_default().add(event); - } - - let source_count = by_source.len(); - let model_count = by_model.len(); - - let mut by_source = by_source - .into_iter() - .map(|(source, acc)| SourceUsage { - source, - sessions: acc.sessions.len(), - events: acc.events, - tokens: acc.tokens, - }) - .collect::>(); - by_source.sort_by_key(|source| Reverse(source.tokens.total_tokens)); - - let mut by_model = by_model - .into_iter() - .map(|((source, provider, model), acc)| ModelUsage { - source, - provider, - model, - sessions: acc.sessions.len(), - events: acc.events, - tokens: acc.tokens, - }) - .collect::>(); - by_model.sort_by_key(|model| Reverse(model.tokens.total_tokens)); - - UsageReport { - summary: UsageSummary { - events: total.events, - sessions: total.sessions.len(), - sources: source_count, - models: model_count, - token_source_events, - tokens: total.tokens, - }, - by_source, - by_model, - daily: period_vec(daily), - weekly: period_vec(weekly), - monthly: period_vec(monthly), - } +#[derive(Default)] +struct UsageReportAccumulator { + total: Accumulator, + by_source: BTreeMap, + by_model: BTreeMap<(String, String, String), Accumulator>, + daily: BTreeMap, + weekly: BTreeMap, + monthly: BTreeMap, + token_source_events: BTreeMap, + codex_seen: HashSet, + claude_seen: HashSet, } -fn dedupe_report_events(events: &[UsageEventRecord]) -> Vec<&UsageEventRecord> { - let mut codex_seen: HashSet = HashSet::new(); - let mut claude_seen: HashSet = HashSet::new(); - let mut deduped = Vec::with_capacity(events.len()); - - for event in events { +impl UsageReportAccumulator { + fn add(&mut self, event: &UsageEventRecord) { if event.source == "codex" { let key = format!( "codex:token_count:{}:{}:{}:{}:{}:{}:{}:{}", @@ -223,21 +157,96 @@ fn dedupe_report_events(events: &[UsageEventRecord]) -> Vec<&UsageEventRecord> { event.cache_write_tokens, event.reasoning_tokens ); - if !codex_seen.insert(key) { - continue; + if !self.codex_seen.insert(key) { + return; } } else if event.source == "claude-code" && event.event_key.starts_with("assistant:") && !event.event_key.contains(":line:") - && !claude_seen.insert(event.event_key.clone()) + && !self.claude_seen.insert(event.event_key.clone()) { - continue; + return; } - deduped.push(event); + self.total.add(event); + *self.token_source_events.entry(event.token_source.clone()).or_insert(0) += 1; + self.by_source.entry(event.source.clone()).or_default().add(event); + self.by_model + .entry((event.source.clone(), event.provider.clone(), event.model.clone())) + .or_default() + .add(event); + + let (day, week, month) = period_keys(event.timestamp); + self.daily.entry(day).or_default().add(event); + self.weekly.entry(week).or_default().add(event); + self.monthly.entry(month).or_default().add(event); + } + + fn finish(self) -> UsageReport { + let source_count = self.by_source.len(); + let model_count = self.by_model.len(); + + let mut by_source = self + .by_source + .into_iter() + .map(|(source, acc)| SourceUsage { + source, + sessions: acc.sessions.len(), + events: acc.events, + tokens: acc.tokens, + }) + .collect::>(); + by_source.sort_by_key(|source| Reverse(source.tokens.total_tokens)); + + let mut by_model = self + .by_model + .into_iter() + .map(|((source, provider, model), acc)| ModelUsage { + source, + provider, + model, + sessions: acc.sessions.len(), + events: acc.events, + tokens: acc.tokens, + }) + .collect::>(); + by_model.sort_by_key(|model| Reverse(model.tokens.total_tokens)); + + UsageReport { + summary: UsageSummary { + events: self.total.events, + sessions: self.total.sessions.len(), + sources: source_count, + models: model_count, + token_source_events: self.token_source_events, + tokens: self.total.tokens, + }, + by_source, + by_model, + daily: period_vec(self.daily), + weekly: period_vec(self.weekly), + monthly: period_vec(self.monthly), + } } +} - deduped +pub(crate) fn build_usage_report(store: &Store, filters: &UsageFilters) -> Result { + let report = store.fold_usage_events( + filters.sources.as_deref(), + filters.time_range, + UsageReportAccumulator::default(), + |report, event| report.add(&event), + )?; + Ok(report.finish()) +} + +#[cfg(any(test, feature = "bench"))] +pub(crate) fn aggregate_usage_events(events: &[UsageEventRecord]) -> UsageReport { + let mut report = UsageReportAccumulator::default(); + for event in events { + report.add(event); + } + report.finish() } fn period_vec(map: BTreeMap) -> Vec {