Batch dashboard DB queries to eliminate N+1

build_watchlist_dashboard previously issued about five DB queries per
ticker (list_runs_for_ticker, latest score for latest run, latest
score for previous run, latest source timestamp for latest run,
latest source timestamp for previous run). A 50-ticker watchlist
fired ~250 round-trips to render once.

Add three batched Database methods that collapse those into three
queries total regardless of watchlist size:
- recent_runs_for_tickers: window-function query returning up to N
  most-recent runs per ticker
- latest_scores_for_runs: MAX(iteration_number)-per-run join
- latest_source_timestamps_for_runs: grouped MAX timestamp

Rewire build_watchlist_dashboard to pre-fetch everything up front
and look it up from hash maps.

Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com>

Author: morganlinton

src/db.rs

@@ -1727,6 +1727,56 @@ impl Database {
         }))
     }
 
+    /// Batch variant of [`Database::get_latest_iteration_evaluation_score`].
+    /// Returns a map from run_id to score for the latest iteration with an
+    /// `evaluation_json` payload. Missing runs / runs with no evaluated
+    /// iteration are simply absent from the map. Used by the dashboard and
+    /// alerts path to avoid N+1 score lookups.
+    pub async fn latest_scores_for_runs(
+        &self,
+        run_ids: &[String],
+    ) -> Result<std::collections::HashMap<String, f64>> {
+        use std::collections::HashMap;
+        if run_ids.is_empty() {
+            return Ok(HashMap::new());
+        }
+        let conn = self.open_connection()?;
+        let placeholders: String = (0..run_ids.len())
+            .map(|i| format!("?{}", i + 1))
+            .collect::<Vec<_>>()
+            .join(",");
+        // Pick the iteration with the highest iteration_number per run_id.
+        let sql = format!(
+            "SELECT i.run_id, i.evaluation_json
+             FROM iterations i
+             INNER JOIN (
+                SELECT run_id, MAX(iteration_number) AS max_n
+                FROM iterations
+                WHERE run_id IN ({placeholders}) AND evaluation_json IS NOT NULL
+                GROUP BY run_id
+             ) latest ON latest.run_id = i.run_id AND latest.max_n = i.iteration_number"
+        );
+        let params: Vec<&dyn rusqlite::ToSql> =
+            run_ids.iter().map(|r| r as &dyn rusqlite::ToSql).collect();
+        let mut statement = conn.prepare(&sql)?;
+        let rows = statement.query_map(params.as_slice(), |row| {
+            Ok((row.get::<_, String>(0)?, row.get::<_, Option<String>>(1)?))
+        })?;
+        let mut map = HashMap::new();
+        for row in rows {
+            let (run_id, raw) = row?;
+            if let Some(raw) = raw {
+                if let Some(score) = serde_json::from_str::<serde_json::Value>(&raw)
+                    .ok()
+                    .and_then(|value| value.get("score").and_then(|s| s.as_f64()))
+                {
+                    map.insert(run_id, score);
+                }
+            }
+        }
+        Ok(map)
+    }
+
     pub async fn get_latest_source_timestamp_for_run(
         &self,
         run_id: &str,
@@ -1750,6 +1800,87 @@ impl Database {
         }
     }
 
+    /// Batch variant of [`Database::get_latest_source_timestamp_for_run`].
+    /// Returns a map of run_id to the most recent source timestamp. Missing
+    /// runs / runs without sources are absent from the map.
+    pub async fn latest_source_timestamps_for_runs(
+        &self,
+        run_ids: &[String],
+    ) -> Result<std::collections::HashMap<String, DateTime<Utc>>> {
+        use std::collections::HashMap;
+        if run_ids.is_empty() {
+            return Ok(HashMap::new());
+        }
+        let conn = self.open_connection()?;
+        let placeholders: String = (0..run_ids.len())
+            .map(|i| format!("?{}", i + 1))
+            .collect::<Vec<_>>()
+            .join(",");
+        let sql = format!(
+            "SELECT run_id, MAX(COALESCE(published_at, created_at)) AS latest
+             FROM sources
+             WHERE run_id IN ({placeholders})
+             GROUP BY run_id"
+        );
+        let params: Vec<&dyn rusqlite::ToSql> =
+            run_ids.iter().map(|r| r as &dyn rusqlite::ToSql).collect();
+        let mut statement = conn.prepare(&sql)?;
+        let rows = statement.query_map(params.as_slice(), |row| {
+            Ok((row.get::<_, String>(0)?, row.get::<_, Option<String>>(1)?))
+        })?;
+        let mut map = HashMap::new();
+        for row in rows {
+            let (run_id, raw) = row?;
+            if let Some(raw) = raw {
+                if let Some(ts) = parse_time_opt(&raw) {
+                    map.insert(run_id, ts);
+                }
+            }
+        }
+        Ok(map)
+    }
+
+    /// Fetch up to `limit` most-recent runs for each ticker in one query.
+    /// Used by the dashboard which needs the latest two runs per ticker.
+    pub async fn recent_runs_for_tickers(
+        &self,
+        tickers: &[String],
+        limit: i64,
+    ) -> Result<std::collections::HashMap<String, Vec<Run>>> {
+        use std::collections::HashMap;
+        if tickers.is_empty() || limit <= 0 {
+            return Ok(HashMap::new());
+        }
+        let conn = self.open_connection()?;
+        let placeholders: String = (0..tickers.len())
+            .map(|i| format!("?{}", i + 1))
+            .collect::<Vec<_>>()
+            .join(",");
+        let limit_placeholder = tickers.len() + 1;
+        let sql = format!(
+            "SELECT id, ticker, question, status, created_at, updated_at,
+                    final_iteration_number, final_memo_markdown, final_memo_html, summary
+             FROM (
+                 SELECT r.*,
+                        ROW_NUMBER() OVER (PARTITION BY ticker ORDER BY created_at DESC) AS rn
+                 FROM runs r
+                 WHERE ticker IN ({placeholders})
+             ) WHERE rn <= ?{limit_placeholder}
+             ORDER BY ticker, created_at DESC"
+        );
+        let mut params: Vec<&dyn rusqlite::ToSql> =
+            tickers.iter().map(|t| t as &dyn rusqlite::ToSql).collect();
+        params.push(&limit as &dyn rusqlite::ToSql);
+        let mut statement = conn.prepare(&sql)?;
+        let rows = statement.query_map(params.as_slice(), map_run)?;
+        let mut map: HashMap<String, Vec<Run>> = HashMap::new();
+        for row in rows {
+            let run = row?;
+            map.entry(run.ticker.clone()).or_default().push(run);
+        }
+        Ok(map)
+    }
+
     // Scanner database methods
 
     pub async fn upsert_ticker_universe(

src/services/dashboard.rs

@@ -19,69 +19,56 @@ pub async fn build_watchlist_dashboard(
     let watchlist_tickers = state.db.list_watchlist_tickers(watchlist_id).await?;
     let alert_rules = state.db.list_or_create_alert_rules(watchlist_id).await?;
 
+    // --- Single batched pre-fetch replaces the per-ticker N+1. ---
+    let ticker_ids: Vec<String> = watchlist_tickers
+        .iter()
+        .map(|wt| wt.ticker.clone())
+        .collect();
+    // Up to the two most-recent runs per ticker (latest + previous).
+    let runs_by_ticker = state.db.recent_runs_for_tickers(&ticker_ids, 2).await?;
+    // Collect run_ids so we can batch-fetch their scores / source freshness.
+    let run_ids: Vec<String> = runs_by_ticker
+        .values()
+        .flat_map(|runs| runs.iter().map(|r| r.id.clone()))
+        .collect();
+    let scores_by_run = state.db.latest_scores_for_runs(&run_ids).await?;
+    let source_ts_by_run = state.db.latest_source_timestamps_for_runs(&run_ids).await?;
+
     let mut rows = Vec::new();
     for watchlist_ticker in watchlist_tickers {
-        let recent_runs = state
-            .db
-            .list_runs_for_ticker(&watchlist_ticker.ticker, 2)
-            .await?;
-        let latest_run = recent_runs.first().cloned();
-        let previous_run = recent_runs.get(1).cloned();
+        let recent_runs: &[crate::models::Run] = runs_by_ticker
+            .get(&watchlist_ticker.ticker)
+            .map(Vec::as_slice)
+            .unwrap_or(&[]);
+        let latest_run = recent_runs.first();
+        let previous_run = recent_runs.get(1);
 
-        let latest_score = if let Some(run) = latest_run.as_ref() {
-            state
-                .db
-                .get_latest_iteration_evaluation_score(&run.id)
-                .await?
-        } else {
-            None
-        };
-        let previous_score = if let Some(run) = previous_run.as_ref() {
-            state
-                .db
-                .get_latest_iteration_evaluation_score(&run.id)
-                .await?
-        } else {
-            None
-        };
+        let latest_score = latest_run.and_then(|run| scores_by_run.get(&run.id).copied());
+        let previous_score = previous_run.and_then(|run| scores_by_run.get(&run.id).copied());
         let score_delta = match (latest_score, previous_score) {
             (Some(latest), Some(previous)) => Some(latest - previous),
             _ => None,
         };
         let trend = classify_trend(score_delta).to_string();
 
-        let latest_source_timestamp = if let Some(run) = latest_run.as_ref() {
-            state
-                .db
-                .get_latest_source_timestamp_for_run(&run.id)
-                .await?
-        } else {
-            None
-        };
+        let latest_source_timestamp =
+            latest_run.and_then(|run| source_ts_by_run.get(&run.id).copied());
         let evidence_freshness = classify_freshness(latest_source_timestamp);
-        let previous_evidence_freshness = if let Some(run) = previous_run.as_ref() {
-            classify_freshness(
-                state
-                    .db
-                    .get_latest_source_timestamp_for_run(&run.id)
-                    .await?,
-            )
-        } else {
-            "no_evidence".to_string()
-        };
+        let previous_evidence_freshness = previous_run
+            .map(|run| classify_freshness(source_ts_by_run.get(&run.id).copied()))
+            .unwrap_or_else(|| "no_evidence".to_string());
 
         let latest_status = latest_run
-            .as_ref()
             .map(|run| run.status.clone())
             .unwrap_or_else(|| "no_data".to_string());
         let decision_state = classify_decision(&latest_status, latest_score, &evidence_freshness);
-        let previous_decision_state = previous_run.as_ref().map(|run| {
+        let previous_decision_state = previous_run.map(|run| {
             classify_decision(&run.status, previous_score, &previous_evidence_freshness)
         });
 
         let snapshot = TickerAlertSnapshot {
             ticker: watchlist_ticker.ticker.clone(),
-            latest_run_id: latest_run.as_ref().map(|run| run.id.clone()),
+            latest_run_id: latest_run.map(|run| run.id.clone()),
             latest_status: latest_status.clone(),
             latest_score,
             previous_score,
@@ -93,17 +80,17 @@ pub async fn build_watchlist_dashboard(
 
         rows.push(DashboardTickerRow {
             ticker: watchlist_ticker.ticker,
-            latest_run_id: latest_run.as_ref().map(|run| run.id.clone()),
+            latest_run_id: latest_run.map(|run| run.id.clone()),
             latest_status,
             latest_score,
             previous_score,
             score_delta,
             trend,
-            summary: latest_run.as_ref().and_then(|run| run.summary.clone()),
+            summary: latest_run.and_then(|run| run.summary.clone()),
             evidence_freshness,
             decision_state,
             active_alert_count: 0,
-            last_run_updated_at: latest_run.as_ref().map(|run| run.updated_at),
+            last_run_updated_at: latest_run.map(|run| run.updated_at),
         });
     }