diff --git a/apps/labrinth/src/queue/payout_run/estimate.rs b/apps/labrinth/src/queue/payout_run/estimate.rs index a17e56b591..250ae86edd 100644 --- a/apps/labrinth/src/queue/payout_run/estimate.rs +++ b/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> { + 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> { - let mut periods_iter = periods.iter(); - let first_period = periods_iter.next().wrap_err("no first period")?; - let last_period = periods_iter.last().unwrap_or(first_period); - - 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_start = aditude::phoenix_midnight(start_date); + let range_end_date = end_date + .checked_add_days(Days::new(1)) + .wrap_err("calculating day after estimate range 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::::new(); + Ok(period_estimates(metrics, periods)) +} + +#[derive(Default)] +struct PartialDayEstimate { + raw_estimated_revenue_usd: Option, + impressions: Option, +} + +fn period_estimates( + metrics: aditude::v2::Metrics, + periods: &[YearMonth], +) -> DashMap { + let requested_periods = periods.iter().copied().collect::>(); + let mut partial_days = HashMap::, PartialDayEstimate>::new(); for response in metrics.responses { for row in response.rows { - let Some(raw_estimated_revenue_usd) = row.revenue else { - continue; - }; - let Some(impressions) = row.impressions else { - continue; - }; - - 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 day = partial_days.entry(row.time).or_default(); + if let Some(raw_estimated_revenue_usd) = row.revenue { + day.raw_estimated_revenue_usd = Some(raw_estimated_revenue_usd); + } + if let Some(impressions) = row.impressions { + day.impressions = Some(impressions); + } } } + let mut map = HashMap::::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::>()) + 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); + } } diff --git a/apps/labrinth/src/routes/internal/payout_runs/admin.rs b/apps/labrinth/src/routes/internal/payout_runs/admin.rs index 9a4f68485b..123bd42801 100644 --- a/apps/labrinth/src/routes/internal/payout_runs/admin.rs +++ b/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 = - estimate(aditude.get_ref(), redis.get_ref(), &[body.period]) - .await - .wrap_internal_err("fetching payout estimate")?; + let estimate_end_date = body + .period + .date() + .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")?; diff --git a/apps/labrinth/src/routes/internal/payout_runs/fetch.rs b/apps/labrinth/src/routes/internal/payout_runs/fetch.rs index a3b15d27f3..82cc7b8392 100644 --- a/apps/labrinth/src/routes/internal/payout_runs/fetch.rs +++ b/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, - pub adjustments: Vec, + /// 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>, } /// Has revenue been distributed for a specific payout period month yet? @@ -83,7 +91,7 @@ pub async fn get_runs( aditude: web::Data, session_queue: web::Data, ) -> Result, 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) - FROM payouts_values + SELECT LEAST( + 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")? - .unwrap_or(now) - .min(now); + .wrap_internal_err("fetching first payout period")?; - let current_period = YearMonth::from_day1(now.date_naive()); - let mut period = YearMonth::from_day1(latest_payout_value.date_naive()); + let current_period = YearMonth::from_day1(current_date); + 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::>(); + 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 = - estimate(aditude.get_ref(), redis.get_ref(), &requested_periods) + let live_periods = requested_periods + .iter() + .copied() + .filter(|period| { + stored_periods + .get(&period.date()) + .is_none_or(|stored| stored.days.is_empty()) + }) + .collect::>(); + 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::>() + } else { + HashMap::new() + }; - let periods = estimates + let periods = requested_periods .into_iter() - .map(|estimate| -> Result<_, ApiError> { - if let Some(period) = stored_periods.get(&estimate.period.date()) { - let mut adjustments = + .map(|requested_period| -> Result<_, ApiError> { + if let Some(period) = stored_periods + .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 { - for adjustment in &mut adjustments { - adjustment.description = None; - } - } + let total_adjustments = period_adjustments + .iter() + .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), }) } })