fix how estimate range is fetched, /calculate route

apps/labrinth/src/database/models/payout_period_item.rs

@@ -1,6 +1,6 @@
 use std::collections::HashMap;
 
-use chrono::NaiveDate;
+use chrono::{DateTime, NaiveDate, Utc};
 use rust_decimal::Decimal;
 use sqlx::types::Json;
 
@@ -13,9 +13,8 @@ pub struct DBPayoutPeriod {
     pub raw_actual_aditude_revenue_usd: Decimal,
     pub adjustments: Vec<Adjustment>,
     pub active_run_payload: Option<PayoutRunPayload>,
+    pub active_run_execute_at: Option<DateTime<Utc>>,
     pub days: Vec<DBPayoutPeriodDay>,
-    pub has_scheduled_run: bool,
-    pub has_running_run: bool,
     pub has_succeeded_run: bool,
 }
 
@@ -41,25 +40,8 @@ impl DBPayoutPeriod {
 				payout_periods.period,
 				payout_periods.raw_actual_aditude_revenue_usd,
 				payout_periods.adjustments AS "adjustments: Json<Vec<Adjustment>>",
-				(
-					SELECT payout_runs.payload
-					FROM payout_runs
-					WHERE payout_runs.period = payout_periods.period
-						AND payout_runs.status IN ('scheduled', 'running')
-					LIMIT 1
-				) AS "active_run_payload: Json<PayoutRunPayload>",
-				EXISTS (
-					SELECT 1
-					FROM payout_runs
-					WHERE payout_runs.period = payout_periods.period
-						AND payout_runs.status = 'scheduled'
-				) AS "has_scheduled_run!",
-				EXISTS (
-					SELECT 1
-					FROM payout_runs
-					WHERE payout_runs.period = payout_periods.period
-						AND payout_runs.status = 'running'
-				) AS "has_running_run!",
+				active_run.payload AS "active_run_payload: Json<PayoutRunPayload>",
+				active_run.execute_at AS active_run_execute_at,
 				EXISTS (
 					SELECT 1
 					FROM payout_runs
@@ -67,6 +49,9 @@ impl DBPayoutPeriod {
 						AND payout_runs.status = 'succeeded'
 				) AS "has_succeeded_run!"
 			FROM payout_periods
+			LEFT JOIN payout_runs active_run
+				ON active_run.period = payout_periods.period
+				AND active_run.status IN ('scheduled', 'running')
 			WHERE payout_periods.period = ANY($1)
 			"#,
             periods,
@@ -87,9 +72,8 @@ impl DBPayoutPeriod {
                         active_run_payload: row
                             .active_run_payload
                             .map(|payload| payload.0),
+                        active_run_execute_at: row.active_run_execute_at,
                         days: Vec::new(),
-                        has_scheduled_run: row.has_scheduled_run,
-                        has_running_run: row.has_running_run,
                         has_succeeded_run: row.has_succeeded_run,
                     },
                 )

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

@@ -89,6 +89,26 @@ pub async fn estimate(
     Ok(periods)
 }
 
+/// Fetch fresh per-day estimated ad provider info for one payout period,
+/// bypassing Redis.
+pub async fn refresh_estimate(
+    aditude: &aditude::Client,
+    period: YearMonth,
+    start_date: NaiveDate,
+    end_date: NaiveDate,
+) -> Result<PeriodEstimate> {
+    if start_date > end_date {
+        return Err(eyre::eyre!("estimate start date is after end date"));
+    }
+
+    let estimates =
+        fetch_estimates(aditude, &[period], start_date, end_date).await?;
+    estimates
+        .remove(&period)
+        .map(|(_, estimate)| estimate)
+        .wrap_err("missing refreshed payout estimate")
+}
+
 async fn fetch_estimates(
     aditude: &aditude::Client,
     periods: &[YearMonth],

apps/labrinth/src/routes/internal/mod.rs

@@ -186,6 +186,7 @@ pub fn config(cfg: &mut web::ServiceConfig) {
 		analytics_event::analytics_event_edit,
 		analytics_event::analytics_event_delete,
 		payout_runs::get_runs,
+		payout_runs::calculate_run,
 		payout_runs::start_run,
 		payout_runs::cancel_runs,
 	),

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

@@ -4,7 +4,7 @@ use rust_decimal::Decimal;
 use serde::{Deserialize, Serialize};
 use xredis::RedisPool;
 
-use super::{Adjustment, PayoutRunPayload};
+use super::{Adjustment, PayoutRunDay, PayoutRunPayload};
 use crate::{
     auth::{
         AuthenticationError, get_user_from_headers, two_factor::verify_2fa_code,
@@ -14,11 +14,16 @@ use crate::{
         models::{
             DBUserId, DatabaseError, generate_payout_run_id,
             payout_run_item::{DBPayoutRun, PayoutRunStatus},
+            payout_variance_item::DBPayoutVariance,
         },
     },
     models::{ids::PayoutRunId, pats::Scopes},
     queue::{
-        payout_run::{estimate, validate_complete_period_estimate},
+        payout_run::{
+            PayoutVariance, PayoutVariances, compute_actual_distribution_flow,
+            distribution_for_day, refresh_estimate,
+            validate_complete_period_estimate,
+        },
         session::AuthQueue,
     },
     routes::ApiError,
@@ -49,6 +54,136 @@ pub struct StartPayoutRunResponse {
     pub execute_at: DateTime<Utc>,
 }
 
+#[derive(Debug, Serialize, utoipa::ToSchema)]
+pub struct CalculatePayoutRunResponse {
+    pub period: YearMonth,
+    pub days: Vec<PayoutRunDay>,
+    /// Sum of all adjustments that would be applied on top of actual revenue.
+    #[serde(with = "rust_decimal::serde::float")]
+    pub total_adjustments: Decimal,
+    pub adjustments: Vec<Adjustment>,
+}
+
+/// Calculate a payout run without scheduling it.
+///
+/// Admin-only. This fetches fresh Aditude data and does not require TOTP.
+#[utoipa::path(
+    tag = "payout runs",
+    request_body = StartPayoutRun,
+    responses(
+        (status = OK, body = CalculatePayoutRunResponse),
+        (status = BAD_REQUEST, description = "Invalid payout run input"),
+        (status = UNAUTHORIZED, description = "Invalid authentication"),
+        (status = FAILED_DEPENDENCY, description = "Aditude returned an incomplete period"),
+    ),
+    security(("bearer_auth" = ["SESSION_ACCESS"])),
+)]
+#[post("/calculate")]
+pub async fn calculate_run(
+    req: HttpRequest,
+    pool: web::Data<PgPool>,
+    redis: web::Data<RedisPool>,
+    aditude: web::Data<aditude::Client>,
+    session_queue: web::Data<AuthQueue>,
+    web::Json(body): web::Json<StartPayoutRun>,
+) -> Result<web::Json<CalculatePayoutRunResponse>, ApiError> {
+    let user = get_user_from_headers(
+        &req,
+        &**pool,
+        &redis,
+        &session_queue,
+        Scopes::SESSION_ACCESS,
+    )
+    .await
+    .wrap_auth_err("authenticating API request")?
+    .1;
+    if !user.role.is_admin() {
+        return Err(ApiError::Auth(eyre::eyre!(
+            AuthenticationError::InvalidCredentials,
+        )));
+    }
+
+    if body.raw_actual_revenue_usd.is_sign_negative() {
+        return Err(ApiError::Request(eyre::eyre!(
+            "`raw_actual_revenue_usd` cannot be negative",
+        )));
+    }
+
+    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 estimate = refresh_estimate(
+        aditude.get_ref(),
+        body.period,
+        body.period.date(),
+        estimate_end_date,
+    )
+    .await
+    .wrap_internal_err("fetching payout estimate")?;
+    validate_complete_period_estimate(&estimate)
+        .wrap_failed_dependency_err("validating payout estimate")?;
+
+    let stored_variances = DBPayoutVariance::get_all(&**pool)
+        .await
+        .wrap_internal_err("fetching payout variances")?;
+    let default_variance = stored_variances
+        .first()
+        .wrap_internal_err("no payout variance configured")?
+        .variance;
+    let variances = PayoutVariances {
+        default_frac: default_variance,
+        fracs: stored_variances
+            .into_iter()
+            .map(|variance| PayoutVariance {
+                starts_at: variance.applied_on,
+                frac: variance.variance,
+            })
+            .collect(),
+    };
+    let total_estimated_revenue_usd = estimate
+        .days
+        .iter()
+        .map(|day| day.raw_estimated_revenue_usd)
+        .sum();
+    let actual_flow = compute_actual_distribution_flow(
+        total_estimated_revenue_usd,
+        body.raw_actual_revenue_usd,
+    );
+    let days = estimate
+        .days
+        .into_iter()
+        .map(|day| PayoutRunDay {
+            date: day.date,
+            estimated: distribution_for_day(
+                day.date,
+                day.raw_estimated_revenue_usd,
+                day.impressions,
+                &variances,
+            ),
+            actual: Some(actual_flow.distribution_for_day(
+                day.date,
+                day.raw_estimated_revenue_usd,
+                day.impressions,
+            )),
+        })
+        .collect();
+    let total_adjustments = body
+        .adjustments
+        .iter()
+        .map(|adjustment| adjustment.amount_usd)
+        .sum();
+
+    Ok(web::Json(CalculatePayoutRunResponse {
+        period: body.period,
+        days,
+        total_adjustments,
+        adjustments: body.adjustments,
+    }))
+}
+
 /// Start a payout run.
 ///
 /// Admin-only.
@@ -143,18 +278,14 @@ pub async fn start_run(
         .checked_add_months(Months::new(1))
         .and_then(|date| date.pred_opt())
         .wrap_internal_err("calculating payout estimate end date")?;
-    let mut estimates = estimate(
+    let estimate = refresh_estimate(
         aditude.get_ref(),
-        redis.get_ref(),
-        &[body.period],
+        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")?;
     validate_complete_period_estimate(&estimate)
         .wrap_failed_dependency_err("validating payout estimate")?;
     let payload = PayoutRunPayload {

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

@@ -1,7 +1,7 @@
 use std::collections::HashMap;
 
 use actix_web::{HttpRequest, get, web};
-use chrono::{Months, NaiveDate, Utc};
+use chrono::{DateTime, Months, NaiveDate, Utc};
 use rust_decimal::Decimal;
 use serde::{Deserialize, Serialize};
 use xredis::RedisPool;
@@ -40,6 +40,8 @@ pub struct PayoutRuns {
 pub struct PayoutRunPeriod {
     pub period: YearMonth,
     pub status: PayoutPeriodStatus,
+    /// When the active payout run is scheduled to begin executing.
+    pub runs_at: Option<DateTime<Utc>>,
     pub days: Vec<PayoutRunDay>,
     /// Sum of all adjustments applied on top of actual revenue.
     #[serde(with = "rust_decimal::serde::float")]
@@ -65,8 +67,6 @@ pub enum PayoutPeriodStatus {
     InReview,
     /// A payout run is waiting for its cancellation window to expire.
     Scheduled,
-    /// Payout run is currently executing.
-    Running,
     /// Payout run has been paid out to creators.
     Paid,
 }
@@ -147,15 +147,6 @@ 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")?;
@@ -182,26 +173,25 @@ pub async fn get_runs(
                 .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")?
-            .into_iter()
-            .map(|estimate| (estimate.period, estimate))
-            .collect::<HashMap<_, _>>()
-        } else {
-            HashMap::new()
-        };
+    let mut live_estimates = if let Some(first_requested_period) =
+        requested_periods.first()
+        && !live_periods.is_empty()
+    {
+        estimate(
+            aditude.get_ref(),
+            redis.get_ref(),
+            &live_periods,
+            first_requested_period.date(),
+            current_date,
+        )
+        .await
+        .wrap_internal_err("fetching payout estimates")?
+        .into_iter()
+        .map(|estimate| (estimate.period, estimate))
+        .collect::<HashMap<_, _>>()
+    } else {
+        HashMap::new()
+    };
 
     let periods = requested_periods
         .into_iter()
@@ -257,9 +247,7 @@ pub async fn get_runs(
                     .collect::<Result<Vec<_>, _>>()?;
                 let status = if period.has_succeeded_run {
                     PayoutPeriodStatus::Paid
-                } else if period.has_running_run {
-                    PayoutPeriodStatus::Running
-                } else if period.has_scheduled_run {
+                } else if period.active_run_execute_at.is_some() {
                     PayoutPeriodStatus::Scheduled
                 } else {
                     PayoutPeriodStatus::InReview
@@ -268,6 +256,7 @@ pub async fn get_runs(
                 Ok(PayoutRunPeriod {
                     period: requested_period,
                     status,
+                    runs_at: period.active_run_execute_at,
                     days,
                     total_adjustments,
                     adjustments,
@@ -302,6 +291,7 @@ pub async fn get_runs(
                 Ok(PayoutRunPeriod {
                     period: requested_period,
                     status,
+                    runs_at: None,
                     days,
                     total_adjustments: Decimal::ZERO,
                     adjustments: is_admin.then(Vec::new),

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

@@ -7,6 +7,7 @@ pub use fetch::*;
 
 pub fn config(cfg: &mut actix_web::web::ServiceConfig) {
     cfg.service(get_runs)
+        .service(calculate_run)
         .service(start_run)
         .service(cancel_runs);
 }

apps/labrinth/src/routes/v3/payouts.rs

@@ -1031,16 +1031,25 @@ pub struct UserBalance {
     pub available: Decimal,
     pub withdrawn_lifetime: Decimal,
     pub withdrawn_ytd: Decimal,
-    /// Finalized revenue not yet available and provisional creator estimates.
+    /// Finalized revenue not yet available and provisional creator estimates
+    /// for periods without a succeeded payout run.
     ///
-    /// Provisional estimates remain pending regardless of their estimated
-    /// availability date and may change before finalization.
+    /// Unfinalized estimates remain pending regardless of their estimated
+    /// availability date and may change before finalization. Retained
+    /// historical estimates do not remain actionable after finalization.
     pub pending: Decimal,
     /// Revenue grouped by its finalized or estimated availability date.
     ///
-    /// These groups may include provisional creator estimates that can change
-    /// before finalization.
+    /// Each payout period contributes either its finalized revenue, when it
+    /// has a succeeded payout run, or its provisional estimate otherwise.
     pub dates: HashMap<DateTime<Utc>, Decimal>,
+    /// Provisional creator estimates grouped by estimated availability date.
+    ///
+    /// Retained historical estimates remain visible after finalization and
+    /// may differ from the corresponding finalized revenue.
+    pub estimated_dates: HashMap<DateTime<Utc>, Decimal>,
+    /// Finalized revenue grouped by availability date.
+    pub actual_dates: HashMap<DateTime<Utc>, Decimal>,
 }
 
 #[derive(Serialize, utoipa::ToSchema)]
@@ -1127,17 +1136,38 @@ async fn get_user_balance(
         SELECT
             date_available AS "date_available!",
             SUM(amount) FILTER (WHERE NOT provisional) finalized_sum,
-            SUM(amount) FILTER (WHERE provisional) estimate_sum
+            SUM(amount) FILTER (WHERE provisional) estimate_sum,
+            SUM(amount) FILTER (WHERE authoritative) authoritative_sum,
+            SUM(amount) FILTER (
+                WHERE provisional AND authoritative
+            ) actionable_estimate_sum
         FROM (
-            SELECT date_available, amount, FALSE provisional
+            SELECT
+                payouts_values.date_available,
+                payouts_values.amount,
+                FALSE provisional,
+                payouts_values.payout_run_id IS NULL
+                    OR payout_runs.status = 'succeeded' authoritative
             FROM payouts_values
-            WHERE user_id = $1
+            LEFT JOIN payout_runs
+                ON payout_runs.id = payouts_values.payout_run_id
+            WHERE payouts_values.user_id = $1
 
             UNION ALL
 
-            SELECT date_available, amount, TRUE provisional
+            SELECT
+                payout_estimates.date_available,
+                payout_estimates.amount,
+                TRUE provisional,
+                succeeded_periods.period IS NULL authoritative
             FROM payout_estimates
-            WHERE user_id = $1
+            LEFT JOIN (
+                SELECT period
+                FROM payout_runs
+                WHERE status = 'succeeded'
+            ) succeeded_periods
+                ON succeeded_periods.period = payout_estimates.period
+            WHERE payout_estimates.user_id = $1
         ) payout_amounts
         GROUP BY date_available
         ORDER BY date_available DESC
@@ -1161,7 +1191,7 @@ async fn get_user_balance(
         } else {
             Decimal::ZERO
         };
-        acc + finalized + x.estimate_sum.unwrap_or(Decimal::ZERO)
+        acc + finalized + x.actionable_estimate_sum.unwrap_or(Decimal::ZERO)
     });
 
     let withdrawn = sqlx::query!(
@@ -1196,12 +1226,20 @@ async fn get_user_balance(
         pending,
         dates: payouts
             .iter()
-            .map(|x| {
-                (
-                    x.date_available,
-                    x.finalized_sum.unwrap_or(Decimal::ZERO)
-                        + x.estimate_sum.unwrap_or(Decimal::ZERO),
-                )
+            .filter_map(|x| {
+                x.authoritative_sum.map(|amount| (x.date_available, amount))
+            })
+            .collect(),
+        estimated_dates: payouts
+            .iter()
+            .filter_map(|x| {
+                x.estimate_sum.map(|amount| (x.date_available, amount))
+            })
+            .collect(),
+        actual_dates: payouts
+            .iter()
+            .filter_map(|x| {
+                x.finalized_sum.map(|amount| (x.date_available, amount))
             })
             .collect(),
     })