fetch from db first instead of aditude

apps/labrinth/src/queue/payout_run/estimate.rs

@@ -1,8 +1,8 @@
 //! Logic for fetching and caching revenue estimations from our ad provider.
 
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 
-use chrono::{Months, NaiveDate};
+use chrono::{DateTime, Days, NaiveDate, Utc};
 use dashmap::DashMap;
 use eyre::Result;
 use rust_decimal::Decimal;
@@ -29,15 +29,25 @@ pub struct DayEstimate {
     pub impressions: u128,
 }
 
-/// Get per-month and per-day estimated ad provider info.
+/// Get per-month and per-day estimated ad provider info for an inclusive date
+/// range.
+///
+/// Dates are interpreted as Phoenix calendar dates. They are converted to UTC
+/// instants only when constructing the Aditude API request.
 pub async fn estimate(
     aditude: &aditude::Client,
     redis: &RedisPool,
     periods: &[YearMonth],
+    start_date: NaiveDate,
+    end_date: NaiveDate,
 ) -> Result<Vec<PeriodEstimate>> {
+    if start_date > end_date {
+        return Err(eyre::eyre!("estimate start date is after end date"));
+    }
+
     let mut periods = redis
         .get_cached_keys(REDIS_KEY, periods, |periods| async move {
-            fetch_estimates(aditude, &periods)
+            fetch_estimates(aditude, &periods, start_date, end_date)
                 .await
                 .map_err(ApiError::Internal)
         })
@@ -49,16 +59,13 @@ pub async fn estimate(
 async fn fetch_estimates(
     aditude: &aditude::Client,
     periods: &[YearMonth],
+    start_date: NaiveDate,
+    end_date: NaiveDate,
 ) -> Result<DashMap<YearMonth, PeriodEstimate>> {
-    let mut periods_iter = periods.iter();
+    let range_start = aditude::phoenix_midnight(start_date);
-    let first_period = periods_iter.next().wrap_err("no first period")?;
+    let range_end_date = end_date
-    let last_period = periods_iter.last().unwrap_or(first_period);
+        .checked_add_days(Days::new(1))
-
+        .wrap_err("calculating day after estimate range end")?;
-    let range_start = aditude::phoenix_midnight(first_period.date());
-    let range_end_date = last_period
-        .date()
-        .checked_add_months(Months::new(1))
-        .wrap_err("calculating month after payout period end")?;
     let range_end = aditude::phoenix_midnight(range_end_date);
 
     let metrics = aditude
@@ -76,33 +83,59 @@ async fn fetch_estimates(
         .await
         .wrap_err("fetching metrics from Aditude")?;
 
-    let mut map = HashMap::<YearMonth, PeriodEstimate>::new();
+    Ok(period_estimates(metrics, periods))
+}
+
+#[derive(Default)]
+struct PartialDayEstimate {
+    raw_estimated_revenue_usd: Option<Decimal>,
+    impressions: Option<u128>,
+}
+
+fn period_estimates(
+    metrics: aditude::v2::Metrics,
+    periods: &[YearMonth],
+) -> DashMap<YearMonth, PeriodEstimate> {
+    let requested_periods = periods.iter().copied().collect::<HashSet<_>>();
+    let mut partial_days = HashMap::<DateTime<Utc>, PartialDayEstimate>::new();
     for response in metrics.responses {
         for row in response.rows {
-            let Some(raw_estimated_revenue_usd) = row.revenue else {
+            let day = partial_days.entry(row.time).or_default();
-                continue;
+            if let Some(raw_estimated_revenue_usd) = row.revenue {
-            };
+                day.raw_estimated_revenue_usd = Some(raw_estimated_revenue_usd);
-            let Some(impressions) = row.impressions else {
+            }
-                continue;
+            if let Some(impressions) = row.impressions {
-            };
+                day.impressions = Some(impressions);
-
+            }
-            let date = aditude::phoenix_date(row.time);
-            let period = YearMonth::from_day1(date);
-
-            let period_estimate = map.entry(period).or_insert(PeriodEstimate {
-                period,
-                days: Vec::new(),
-            });
-            let days = &mut period_estimate.days;
-
-            days.push(DayEstimate {
-                date,
-                raw_estimated_revenue_usd,
-                impressions,
-            });
         }
     }
 
+    let mut map = HashMap::<YearMonth, PeriodEstimate>::new();
+    for (time, day) in partial_days {
+        let Some(raw_estimated_revenue_usd) = day.raw_estimated_revenue_usd
+        else {
+            continue;
+        };
+        let Some(impressions) = day.impressions else {
+            continue;
+        };
+
+        let date = aditude::phoenix_date(time);
+        let period = YearMonth::from_day1(date);
+        if !requested_periods.contains(&period) {
+            continue;
+        }
+        let period_estimate = map.entry(period).or_insert(PeriodEstimate {
+            period,
+            days: Vec::new(),
+        });
+        period_estimate.days.push(DayEstimate {
+            date,
+            raw_estimated_revenue_usd,
+            impressions,
+        });
+    }
+
     // we have no clue if the Aditude row return order is stable,
     // so for safety sort the days here
     for period in map.values_mut() {
@@ -118,5 +151,46 @@ async fn fetch_estimates(
         });
     }
 
-    Ok(map.into_iter().collect::<DashMap<_, _>>())
+    map.into_iter().collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use chrono::TimeZone;
+    use rust_decimal::dec;
+
+    use super::*;
+
+    #[test]
+    fn combines_metrics_from_separate_responses() {
+        let time = Utc.with_ymd_and_hms(2026, 8, 1, 7, 0, 0).unwrap();
+        let metrics = aditude::v2::Metrics {
+            responses: vec![
+                aditude::v2::Response {
+                    rows: vec![aditude::v2::Row {
+                        impressions: None,
+                        revenue: Some(dec!(12.34)),
+                        time,
+                    }],
+                },
+                aditude::v2::Response {
+                    rows: vec![aditude::v2::Row {
+                        impressions: Some(5678),
+                        revenue: None,
+                        time,
+                    }],
+                },
+            ],
+        };
+        let period =
+            YearMonth::from_day1(NaiveDate::from_ymd_opt(2026, 8, 1).unwrap());
+
+        let estimates = period_estimates(metrics, &[period]);
+        let estimate = estimates.get(&period).unwrap();
+
+        assert_eq!(estimate.days.len(), 1);
+        assert_eq!(estimate.days[0].date, period.date());
+        assert_eq!(estimate.days[0].raw_estimated_revenue_usd, dec!(12.34));
+        assert_eq!(estimate.days[0].impressions, 5678);
+    }
 }

apps/labrinth/src/routes/internal/payout_runs/admin.rs

@@ -1,5 +1,5 @@
 use actix_web::{HttpRequest, post, web};
-use chrono::{DateTime, Utc};
+use chrono::{DateTime, Months, Utc};
 use rust_decimal::Decimal;
 use serde::{Deserialize, Serialize};
 use xredis::RedisPool;
@@ -133,10 +133,21 @@ pub async fn start_run(
         )));
     }
 
-    let mut estimates =
+    let estimate_end_date = body
-        estimate(aditude.get_ref(), redis.get_ref(), &[body.period])
+        .period
-            .await
+        .date()
-            .wrap_internal_err("fetching payout estimate")?;
+        .checked_add_months(Months::new(1))
+        .and_then(|date| date.pred_opt())
+        .wrap_internal_err("calculating payout estimate end date")?;
+    let mut estimates = estimate(
+        aditude.get_ref(),
+        redis.get_ref(),
+        &[body.period],
+        body.period.date(),
+        estimate_end_date,
+    )
+    .await
+    .wrap_internal_err("fetching payout estimate")?;
     let estimate = estimates
         .pop()
         .wrap_internal_err("missing requested payout estimate")?;

apps/labrinth/src/routes/internal/payout_runs/fetch.rs

@@ -2,6 +2,7 @@ use std::collections::HashMap;
 
 use actix_web::{HttpRequest, get, web};
 use chrono::{Months, NaiveDate, Utc};
+use rust_decimal::Decimal;
 use serde::{Deserialize, Serialize};
 use xredis::RedisPool;
 
@@ -40,7 +41,14 @@ pub struct PayoutRunPeriod {
     pub period: YearMonth,
     pub status: PayoutPeriodStatus,
     pub days: Vec<PayoutRunDay>,
-    pub adjustments: Vec<Adjustment>,
+    /// Sum of all adjustments applied on top of actual revenue.
+    #[serde(with = "rust_decimal::serde::float")]
+    pub total_adjustments: Decimal,
+    /// Individual adjustments, including their admin-provided descriptions.
+    ///
+    /// Only visible to admins.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub adjustments: Option<Vec<Adjustment>>,
 }
 
 /// Has revenue been distributed for a specific payout period month yet?
@@ -83,7 +91,7 @@ pub async fn get_runs(
     aditude: web::Data<aditude::Client>,
     session_queue: web::Data<AuthQueue>,
 ) -> Result<web::Json<PayoutRuns>, ApiError> {
-    let show_adjustment_descriptions = get_user_from_headers(
+    let is_admin = get_user_from_headers(
         &req,
         &**pool,
         &redis,
@@ -93,20 +101,29 @@ pub async fn get_runs(
     .await
     .is_ok_and(|(_, user)| user.role.is_admin());
     let now = Utc::now();
-    let latest_payout_value = sqlx::query_scalar!(
+    let current_date = aditude::phoenix_date(now);
+    let first_period_date = sqlx::query_scalar!(
         r#"
-        SELECT MAX(created)
+        SELECT LEAST(
-        FROM payouts_values
+            COALESCE(
+                (SELECT MAX(created)::date FROM payouts_values),
+                $1
+            ),
+            COALESCE(
+                (SELECT MIN(period) FROM payout_periods),
+                $1
+            ),
+            $1
+        ) AS "first_period!"
         "#,
+        current_date,
     )
     .fetch_one(&**pool)
     .await
-    .wrap_internal_err("fetching latest payout value")?
+    .wrap_internal_err("fetching first payout period")?;
-    .unwrap_or(now)
-    .min(now);
 
-    let current_period = YearMonth::from_day1(now.date_naive());
+    let current_period = YearMonth::from_day1(current_date);
-    let mut period = YearMonth::from_day1(latest_payout_value.date_naive());
+    let mut period = YearMonth::from_day1(first_period_date);
     let mut requested_periods = Vec::new();
 
     while period <= current_period {
@@ -130,6 +147,15 @@ pub async fn get_runs(
             .into_iter()
             .map(|period| (period.period, period))
             .collect::<HashMap<_, _>>();
+    let newest_stored_day = sqlx::query_scalar!(
+        r#"
+        SELECT MAX(date)
+        FROM payout_period_days
+        "#,
+    )
+    .fetch_one(&**pool)
+    .await
+    .wrap_internal_err("fetching newest stored payout period day")?;
     let stored_variances = DBPayoutVariance::get_all(&**pool)
         .await
         .wrap_internal_err("fetching payout variances")?;
@@ -147,26 +173,54 @@ pub async fn get_runs(
             })
             .collect(),
     };
-    let estimates =
+    let live_periods = requested_periods
-        estimate(aditude.get_ref(), redis.get_ref(), &requested_periods)
+        .iter()
+        .copied()
+        .filter(|period| {
+            stored_periods
+                .get(&period.date())
+                .is_none_or(|stored| stored.days.is_empty())
+        })
+        .collect::<Vec<_>>();
+    let mut live_estimates =
+        if let Some(first_live_period) = live_periods.first() {
+            let start_date = newest_stored_day
+                .unwrap_or_else(|| first_live_period.date())
+                .min(current_date);
+            estimate(
+                aditude.get_ref(),
+                redis.get_ref(),
+                &live_periods,
+                start_date,
+                current_date,
+            )
             .await
-            .wrap_internal_err("fetching payout estimates")?;
+            .wrap_internal_err("fetching payout estimates")?
+            .into_iter()
+            .map(|estimate| (estimate.period, estimate))
+            .collect::<HashMap<_, _>>()
+        } else {
+            HashMap::new()
+        };
 
-    let periods = estimates
+    let periods = requested_periods
         .into_iter()
-        .map(|estimate| -> Result<_, ApiError> {
+        .map(|requested_period| -> Result<_, ApiError> {
-            if let Some(period) = stored_periods.get(&estimate.period.date()) {
+            if let Some(period) = stored_periods
-                let mut adjustments =
+                .get(&requested_period.date())
+                .filter(|period| !period.days.is_empty())
+            {
+                let period_adjustments =
                     if let Some(payload) = &period.active_run_payload {
-                        payload.adjustments.clone()
+                        &payload.adjustments
                     } else {
-                        period.adjustments.clone()
+                        &period.adjustments
                     };
-                if !show_adjustment_descriptions {
+                let total_adjustments = period_adjustments
-                    for adjustment in &mut adjustments {
+                    .iter()
-                        adjustment.description = None;
+                    .map(|adjustment| adjustment.amount_usd)
-                    }
+                    .sum();
-                }
+                let adjustments = is_admin.then(|| period_adjustments.clone());
                 let total_estimated_revenue_usd = period
                     .days
                     .iter()
@@ -212,14 +266,18 @@ pub async fn get_runs(
                 };
 
                 Ok(PayoutRunPeriod {
-                    period: estimate.period,
+                    period: requested_period,
                     status,
                     days,
+                    total_adjustments,
                     adjustments,
                 })
             } else {
+                let estimate = live_estimates
+                    .remove(&requested_period)
+                    .wrap_internal_err("missing live payout estimate")?;
                 let status = if let Some(available_at) =
-                    net_60_payout_available_at(estimate.period)
+                    net_60_payout_available_at(requested_period)
                     && now >= available_at
                 {
                     PayoutPeriodStatus::InReview
@@ -242,10 +300,11 @@ pub async fn get_runs(
                     .collect();
 
                 Ok(PayoutRunPeriod {
-                    period: estimate.period,
+                    period: requested_period,
                     status,
                     days,
-                    adjustments: Vec::new(),
+                    total_adjustments: Decimal::ZERO,
+                    adjustments: is_admin.then(Vec::new),
                 })
             }
         })