compute actual rev for period

apps/labrinth/.sqlx/query-53e45fef7122a91a4509b232ede70c36b476340e3c9bca62517b074ff8038e89.json

@@ -0,0 +1,40 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n\t\t\tSELECT\n\t\t\t\tperiod,\n\t\t\t\tdate,\n\t\t\t\traw_estimated_aditude_revenue_usd,\n\t\t\t\taditude_impressions\n\t\t\tFROM payout_period_days\n\t\t\tWHERE period = ANY($1)\n\t\t\tORDER BY period, date\n\t\t\t",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "period",
+        "type_info": "Date"
+      },
+      {
+        "ordinal": 1,
+        "name": "date",
+        "type_info": "Date"
+      },
+      {
+        "ordinal": 2,
+        "name": "raw_estimated_aditude_revenue_usd",
+        "type_info": "Numeric"
+      },
+      {
+        "ordinal": 3,
+        "name": "aditude_impressions",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "DateArray"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "53e45fef7122a91a4509b232ede70c36b476340e3c9bca62517b074ff8038e89"
+}

apps/labrinth/.sqlx/query-90237a2f470766b030df7bb9642f2b1876f262e5c9e165f9f1256aaa58850a0c.json

@@ -0,0 +1,46 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n\t\t\tSELECT\n\t\t\t\tpayout_periods.period,\n\t\t\t\tpayout_periods.raw_actual_aditude_revenue_usd,\n\t\t\t\tpayout_periods.adjustments,\n\t\t\t\tEXISTS (\n\t\t\t\t\tSELECT 1\n\t\t\t\t\tFROM payout_runs\n\t\t\t\t\tWHERE payout_runs.period = payout_periods.period\n\t\t\t\t\t\tAND payout_runs.status IN ('scheduled', 'running')\n\t\t\t\t) AS \"has_active_run!\",\n\t\t\t\tEXISTS (\n\t\t\t\t\tSELECT 1\n\t\t\t\t\tFROM payout_runs\n\t\t\t\t\tWHERE payout_runs.period = payout_periods.period\n\t\t\t\t\t\tAND payout_runs.status = 'succeeded'\n\t\t\t\t) AS \"has_succeeded_run!\"\n\t\t\tFROM payout_periods\n\t\t\tWHERE payout_periods.period = ANY($1)\n\t\t\t",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "period",
+        "type_info": "Date"
+      },
+      {
+        "ordinal": 1,
+        "name": "raw_actual_aditude_revenue_usd",
+        "type_info": "Numeric"
+      },
+      {
+        "ordinal": 2,
+        "name": "adjustments",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 3,
+        "name": "has_active_run!",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 4,
+        "name": "has_succeeded_run!",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "DateArray"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      null,
+      null
+    ]
+  },
+  "hash": "90237a2f470766b030df7bb9642f2b1876f262e5c9e165f9f1256aaa58850a0c"
+}

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

@@ -27,6 +27,7 @@ pub mod organization_item;
 pub mod passkey_item;
 pub mod pat_item;
 pub mod payout_item;
+pub mod payout_period_item;
 pub mod payout_run_item;
 pub mod payout_variance_item;
 pub mod payouts_values_notifications;

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

@@ -0,0 +1,111 @@
+use std::collections::HashMap;
+
+use chrono::NaiveDate;
+use rust_decimal::Decimal;
+
+use super::DatabaseError;
+
+#[derive(Debug, Clone)]
+pub struct DBPayoutPeriod {
+    pub period: NaiveDate,
+    pub raw_actual_aditude_revenue_usd: Decimal,
+    pub adjustments: serde_json::Value,
+    pub days: Vec<DBPayoutPeriodDay>,
+    pub has_active_run: bool,
+    pub has_succeeded_run: bool,
+}
+
+#[derive(Debug, Clone)]
+pub struct DBPayoutPeriodDay {
+    pub period: NaiveDate,
+    pub date: NaiveDate,
+    pub raw_estimated_aditude_revenue_usd: Decimal,
+    pub aditude_impressions: i64,
+}
+
+impl DBPayoutPeriod {
+    pub async fn get_many<'a, E>(
+        periods: &[NaiveDate],
+        exec: E,
+    ) -> Result<Vec<Self>, DatabaseError>
+    where
+        E: crate::database::Executor<'a, Database = sqlx::Postgres> + Copy,
+    {
+        let period_rows = sqlx::query!(
+            r#"
+			SELECT
+				payout_periods.period,
+				payout_periods.raw_actual_aditude_revenue_usd,
+				payout_periods.adjustments,
+				EXISTS (
+					SELECT 1
+					FROM payout_runs
+					WHERE payout_runs.period = payout_periods.period
+						AND payout_runs.status IN ('scheduled', 'running')
+				) AS "has_active_run!",
+				EXISTS (
+					SELECT 1
+					FROM payout_runs
+					WHERE payout_runs.period = payout_periods.period
+						AND payout_runs.status = 'succeeded'
+				) AS "has_succeeded_run!"
+			FROM payout_periods
+			WHERE payout_periods.period = ANY($1)
+			"#,
+            periods,
+        )
+        .fetch_all(exec)
+        .await?;
+
+        let mut periods = period_rows
+            .into_iter()
+            .map(|row| {
+                (
+                    row.period,
+                    Self {
+                        period: row.period,
+                        raw_actual_aditude_revenue_usd: row
+                            .raw_actual_aditude_revenue_usd,
+                        adjustments: row.adjustments,
+                        days: Vec::new(),
+                        has_active_run: row.has_active_run,
+                        has_succeeded_run: row.has_succeeded_run,
+                    },
+                )
+            })
+            .collect::<HashMap<_, _>>();
+
+        let stored_period_dates = periods.keys().copied().collect::<Vec<_>>();
+        let days = sqlx::query!(
+            r#"
+			SELECT
+				period,
+				date,
+				raw_estimated_aditude_revenue_usd,
+				aditude_impressions
+			FROM payout_period_days
+			WHERE period = ANY($1)
+			ORDER BY period, date
+			"#,
+            &stored_period_dates,
+        )
+        .fetch_all(exec)
+        .await?;
+
+        for day in days {
+            if let Some(period) = periods.get_mut(&day.period) {
+                period.days.push(DBPayoutPeriodDay {
+                    period: day.period,
+                    date: day.date,
+                    raw_estimated_aditude_revenue_usd: day
+                        .raw_estimated_aditude_revenue_usd,
+                    aditude_impressions: day.aditude_impressions,
+                });
+            }
+        }
+
+        let mut periods = periods.into_values().collect::<Vec<_>>();
+        periods.sort_unstable_by_key(|period| period.period);
+        Ok(periods)
+    }
+}

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

@@ -50,10 +50,11 @@
 //!       ```
 //!   - (fees stay the same, since they're based on impressions, not revenue)
 //!   - (variance is ignored, since that's purely an estimation value)
-//!   - `adjustments_usd`: sum of all manual adjustments input by the admin
-//!   - `actual.net_revenue_usd`: raw actual revenue - fees + adjustments
+//!   - `actual.net_revenue_usd`: raw actual revenue - fees
 //!   - `actual.(platform|creator)_net_revenue_usd`: same logic as estimated,
 //!     but using the net actual revenue
+//! - Manual adjustments are stored separately on the payout period and applied
+//!   on top of its actual distribution.
 //!
 //! ## Variance
 //!
@@ -86,10 +87,8 @@ pub struct DayDistribution {
     ///
     /// For non-estimates (actual revenue values), this is zero.
     pub variance_usd: Decimal,
-    /// Manually-input adjustments on top of raw revenue.
-    pub sum_adjustments_usd: Decimal,
     /// Total net revenue that we earned;
-    /// `raw_revenue - fees - variance + sum_adjustments`.
+    /// `raw_revenue - fees - variance`.
     pub net_revenue_usd: Decimal,
     /// How much of the net revenue goes to the platform.
     pub platform_net_revenue_usd: Decimal,
@@ -125,7 +124,6 @@ pub fn distribution_for_day(
     date: NaiveDate,
     raw_revenue_usd: Decimal,
     impressions: u128,
-    sum_adjustments_usd: Decimal,
     variances: &PayoutVariances,
 ) -> DayDistribution {
     let fees_usd = {
@@ -135,19 +133,18 @@ pub fn distribution_for_day(
     let variance_frac = variances
         .fracs
         .iter()
-        .find(|v| v.starts_at >= date)
+        .rev()
+        .find(|v| v.starts_at <= date)
         .map(|v| v.frac)
         .unwrap_or(variances.default_frac);
     let variance_usd = raw_revenue_usd * variance_frac;
 
-    let net_estimated_revenue_usd =
-        raw_revenue_usd - fees_usd - variance_usd + sum_adjustments_usd;
+    let net_estimated_revenue_usd = raw_revenue_usd - fees_usd - variance_usd;
 
     DayDistribution {
         raw_revenue_usd,
         fees_usd,
         variance_usd,
-        sum_adjustments_usd,
         net_revenue_usd: net_estimated_revenue_usd,
         platform_net_revenue_usd: net_estimated_revenue_usd
             * PLATFORM_REVENUE_SPLIT,
@@ -155,3 +152,57 @@ pub fn distribution_for_day(
             * (dec!(1) - PLATFORM_REVENUE_SPLIT),
     }
 }
+
+/// Precomputed allocation used to distribute actual period revenue by day.
+#[derive(Debug, Clone, Copy)]
+pub struct ActualDistributionFlow {
+    share: Decimal,
+}
+
+/// Start a flow for computing the actual revenue distribution of a payout
+/// period.
+///
+/// Our ad provider gives us per-day estimates for how much money we earned,
+/// but only provides money in a lump sum per month. Therefore, it is up to us
+/// to figure out how much each day contributed to the lump-sum amount.
+///
+/// We do this using the following formula:
+/// ```text
+/// let share = raw_actual_revenue / raw_estimated_revenue
+/// actual_day_revenue[day] = share * estimated_day_revenue[day]
+/// ```
+///
+/// If the period's estimated revenue is zero, the share is `1`.
+///
+/// We use a type-state-ish pattern here to ensure that the same flow is used
+/// for each day in a period.
+pub fn compute_actual_distribution_flow(
+    raw_estimated_revenue_usd: Decimal,
+    raw_actual_revenue_usd: Decimal,
+) -> ActualDistributionFlow {
+    let share = if raw_estimated_revenue_usd.is_zero() {
+        Decimal::ONE
+    } else {
+        raw_actual_revenue_usd / raw_estimated_revenue_usd
+    };
+
+    ActualDistributionFlow { share }
+}
+
+impl ActualDistributionFlow {
+    /// Compute the actual distribution for a stored day in this period.
+    pub fn distribution_for_day(
+        &self,
+        date: NaiveDate,
+        raw_estimated_revenue_usd: Decimal,
+        impressions: u128,
+    ) -> DayDistribution {
+        distribution_for_day(
+            date,
+            raw_estimated_revenue_usd * self.share,
+            impressions,
+            // actual rev distribution always has no variance
+            &PayoutVariances::ZERO,
+        )
+    }
+}

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

@@ -1,3 +1,5 @@
+use std::collections::HashMap;
+
 use actix_web::{get, web};
 use chrono::{Months, NaiveDate, Utc};
 use rust_decimal::Decimal;
@@ -5,9 +7,16 @@ use serde::{Deserialize, Serialize};
 use xredis::RedisPool;
 
 use crate::{
-    database::PgPool,
+    database::{
+        PgPool,
+        models::{
+            payout_period_item::DBPayoutPeriod,
+            payout_variance_item::DBPayoutVariance,
+        },
+    },
     queue::payout_run::{
-        DayDistribution, PayoutVariances, distribution_for_day, estimate,
+        DayDistribution, PayoutVariance, PayoutVariances,
+        compute_actual_distribution_flow, distribution_for_day, estimate,
     },
     routes::ApiError,
     util::{
@@ -100,6 +109,34 @@ pub async fn get_runs(
         );
     }
 
+    let requested_period_dates = requested_periods
+        .iter()
+        .map(|period| period.date())
+        .collect::<Vec<_>>();
+    let stored_periods =
+        DBPayoutPeriod::get_many(&requested_period_dates, &**pool)
+            .await
+            .wrap_internal_err("fetching stored payout periods")?
+            .into_iter()
+            .map(|period| (period.period, period))
+            .collect::<HashMap<_, _>>();
+    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 estimates =
         estimate(aditude.get_ref(), redis.get_ref(), &requested_periods)
             .await
@@ -107,39 +144,93 @@ pub async fn get_runs(
 
     let periods = estimates
         .into_iter()
-        .map(|estimate| {
-            let status = if let Some(available_at) =
-                net_60_payout_available_at(estimate.period)
-                && now >= available_at
-            {
-                PayoutPeriodStatus::InReview
-            } else {
-                PayoutPeriodStatus::Open
-            };
-            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,
-                        Decimal::ZERO,
-                        &PayoutVariances::ZERO,
-                    ),
-                    actual: None,
-                })
-                .collect();
+        .map(|estimate| -> Result<_, ApiError> {
+            if let Some(period) = stored_periods.get(&estimate.period.date()) {
+                let adjustments = serde_json::from_value::<
+                    Vec<PayoutRunAdjustment>,
+                >(period.adjustments.clone())
+                .wrap_internal_err("deserializing payout adjustments")?;
+                let total_estimated_revenue_usd = period
+                    .days
+                    .iter()
+                    .map(|day| day.raw_estimated_aditude_revenue_usd)
+                    .sum::<Decimal>();
+                let actual_flow = compute_actual_distribution_flow(
+                    total_estimated_revenue_usd,
+                    period.raw_actual_aditude_revenue_usd,
+                );
+                let days = period
+                    .days
+                    .iter()
+                    .map(|day| -> Result<_, ApiError> {
+                        let impressions =
+                            u128::try_from(day.aditude_impressions)
+                                .wrap_internal_err(
+                                    "converting stored Aditude impressions",
+                                )?;
+                        Ok(PayoutRunDay {
+                            date: day.date,
+                            estimated: distribution_for_day(
+                                day.date,
+                                day.raw_estimated_aditude_revenue_usd,
+                                impressions,
+                                &variances,
+                            ),
+                            actual: Some(actual_flow.distribution_for_day(
+                                day.date,
+                                day.raw_estimated_aditude_revenue_usd,
+                                impressions,
+                            )),
+                        })
+                    })
+                    .collect::<Result<Vec<_>, _>>()?;
+                let status = if period.has_succeeded_run {
+                    PayoutPeriodStatus::Paid
+                } else if period.has_active_run {
+                    PayoutPeriodStatus::Running
+                } else {
+                    PayoutPeriodStatus::InReview
+                };
 
-            PayoutRunPeriod {
-                period: estimate.period,
-                status,
-                days,
-                adjustments: Vec::new(),
+                Ok(PayoutRunPeriod {
+                    period: estimate.period,
+                    status,
+                    days,
+                    adjustments,
+                })
+            } else {
+                let status = if let Some(available_at) =
+                    net_60_payout_available_at(estimate.period)
+                    && now >= available_at
+                {
+                    PayoutPeriodStatus::InReview
+                } else {
+                    PayoutPeriodStatus::Open
+                };
+                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: None,
+                    })
+                    .collect();
+
+                Ok(PayoutRunPeriod {
+                    period: estimate.period,
+                    status,
+                    days,
+                    adjustments: Vec::new(),
+                })
             }
         })
-        .collect::<Vec<_>>();
+        .collect::<Result<Vec<_>, _>>()?;
 
     Ok(web::Json(PayoutRuns { periods }))
 }