From efe20653c77a3e6c7cb6b1adc796d0269657004f Mon Sep 17 00:00:00 2001 From: aecsocket <43144841+aecsocket@users.noreply.github.com> Date: Wed, 19 Aug 2026 16:42:39 +0900 Subject: [PATCH] compute actual rev for period --- ...70c36b476340e3c9bca62517b074ff8038e89.json | 40 +++++ ...f2b1876f262e5c9e165f9f1256aaa58850a0c.json | 46 ++++++ apps/labrinth/src/database/models/mod.rs | 1 + .../src/database/models/payout_period_item.rs | 111 +++++++++++++ apps/labrinth/src/queue/payout_run/mod.rs | 71 ++++++-- .../src/routes/internal/payout_runs.rs | 155 ++++++++++++++---- 6 files changed, 382 insertions(+), 42 deletions(-) create mode 100644 apps/labrinth/.sqlx/query-53e45fef7122a91a4509b232ede70c36b476340e3c9bca62517b074ff8038e89.json create mode 100644 apps/labrinth/.sqlx/query-90237a2f470766b030df7bb9642f2b1876f262e5c9e165f9f1256aaa58850a0c.json create mode 100644 apps/labrinth/src/database/models/payout_period_item.rs diff --git a/apps/labrinth/.sqlx/query-53e45fef7122a91a4509b232ede70c36b476340e3c9bca62517b074ff8038e89.json b/apps/labrinth/.sqlx/query-53e45fef7122a91a4509b232ede70c36b476340e3c9bca62517b074ff8038e89.json new file mode 100644 index 0000000000..05e84d612c --- /dev/null +++ b/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" +} diff --git a/apps/labrinth/.sqlx/query-90237a2f470766b030df7bb9642f2b1876f262e5c9e165f9f1256aaa58850a0c.json b/apps/labrinth/.sqlx/query-90237a2f470766b030df7bb9642f2b1876f262e5c9e165f9f1256aaa58850a0c.json new file mode 100644 index 0000000000..13c2635f07 --- /dev/null +++ b/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" +} diff --git a/apps/labrinth/src/database/models/mod.rs b/apps/labrinth/src/database/models/mod.rs index 484742c621..3f8d9100cf 100644 --- a/apps/labrinth/src/database/models/mod.rs +++ b/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; diff --git a/apps/labrinth/src/database/models/payout_period_item.rs b/apps/labrinth/src/database/models/payout_period_item.rs new file mode 100644 index 0000000000..4e58261dc8 --- /dev/null +++ b/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, + 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, 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::>(); + + let stored_period_dates = periods.keys().copied().collect::>(); + 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::>(); + periods.sort_unstable_by_key(|period| period.period); + Ok(periods) + } +} diff --git a/apps/labrinth/src/queue/payout_run/mod.rs b/apps/labrinth/src/queue/payout_run/mod.rs index 099b87dbc8..1c15da375c 100644 --- a/apps/labrinth/src/queue/payout_run/mod.rs +++ b/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, + ) + } +} diff --git a/apps/labrinth/src/routes/internal/payout_runs.rs b/apps/labrinth/src/routes/internal/payout_runs.rs index 96a2240eeb..1cbf8e9883 100644 --- a/apps/labrinth/src/routes/internal/payout_runs.rs +++ b/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::>(); + 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::>(); + 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, + >(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::(); + 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::, _>>()?; + 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::>(); + .collect::, _>>()?; Ok(web::Json(PayoutRuns { periods })) }