From 9a717962fb676511f266fe26b30d296c0c450eb2 Mon Sep 17 00:00:00 2001 From: aecsocket <43144841+aecsocket@users.noreply.github.com> Date: Fri, 14 Aug 2026 16:54:41 +0000 Subject: [PATCH] raw/net estimated revenue fields --- .../20260814130000_payouts_variance.sql | 8 + apps/labrinth/src/models/v3/payout_runs.rs | 10 +- apps/labrinth/src/queue/payouts/mod.rs | 141 ++++++---- apps/labrinth/src/routes/internal/payouts.rs | 259 ++++++++++++------ 4 files changed, 265 insertions(+), 153 deletions(-) create mode 100644 apps/labrinth/migrations/20260814130000_payouts_variance.sql diff --git a/apps/labrinth/migrations/20260814130000_payouts_variance.sql b/apps/labrinth/migrations/20260814130000_payouts_variance.sql new file mode 100644 index 0000000000..fe1dea183a --- /dev/null +++ b/apps/labrinth/migrations/20260814130000_payouts_variance.sql @@ -0,0 +1,8 @@ +CREATE TABLE payouts_variance ( + applied_at TIMESTAMPTZ PRIMARY KEY, + variance NUMERIC(40, 20) NOT NULL + CHECK (variance BETWEEN 0 AND 1) +); + +INSERT INTO payouts_variance (applied_at, variance) +VALUES ('1970-01-01 00:00:00+00', 0.1); diff --git a/apps/labrinth/src/models/v3/payout_runs.rs b/apps/labrinth/src/models/v3/payout_runs.rs index e195f4c251..6ed5c47562 100644 --- a/apps/labrinth/src/models/v3/payout_runs.rs +++ b/apps/labrinth/src/models/v3/payout_runs.rs @@ -50,11 +50,13 @@ pub enum PayoutRunStatus { Paid, } -#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)] +#[derive(Debug, Clone, Default, Serialize, Deserialize, utoipa::ToSchema)] pub struct PayoutRunReport { - pub days: Vec, - #[serde(with = "rust_decimal::serde::float")] - pub fees_deducted_usd: Decimal, + pub days: Vec, + #[serde(with = "rust_decimal::serde::float")] + pub raw_estimated_revenue_usd: Decimal, + #[serde(with = "rust_decimal::serde::float")] + pub fees_deducted_usd: Decimal, #[serde(with = "rust_decimal::serde::float")] pub variance_adjustment_usd: Decimal, #[serde(with = "rust_decimal::serde::float")] diff --git a/apps/labrinth/src/queue/payouts/mod.rs b/apps/labrinth/src/queue/payouts/mod.rs index e1e3753f43..0205dc9e08 100644 --- a/apps/labrinth/src/queue/payouts/mod.rs +++ b/apps/labrinth/src/queue/payouts/mod.rs @@ -882,20 +882,28 @@ pub struct AditudeTime { #[derive(Deserialize)] struct AditudeMetricsV2Response { - responses: Vec, + responses: Vec, } #[derive(Deserialize)] struct AditudeMetricsV2Table { - rows: Vec, + rows: Vec, } #[derive(Deserialize)] -struct AditudeRevenueRow { - #[serde(rename = "_TIME")] - time_millis: i64, - #[serde(rename = "REVENUE")] - revenue: Decimal, +struct AditudeMetricRow { + #[serde(rename = "_TIME")] + time_millis: i64, + #[serde(rename = "REVENUE")] + revenue: Option, + #[serde(rename = "IMPRESSIONS")] + impressions: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AditudeDayEstimate { + pub revenue: DayRevenue, + pub impressions: u128, } pub async fn make_aditude_request( @@ -929,37 +937,37 @@ pub async fn make_aditude_request( } async fn make_aditude_revenue_request( - start_time: i64, - end_time: i64, + start_time: i64, + end_time: i64, ) -> Result { - reqwest::Client::new() - .post("https://cloud.aditude.io/api/public/insights/metrics/v2") - .bearer_auth(&ENV.ADITUDE_API_KEY) - .json(&serde_json::json!({ - "metrics": ["REVENUE"], - "range": "custom", - "startTime": start_time, - "endTime": end_time, - "interval": "1d" - })) - .send() - .await - .wrap_internal_err("failed to request Aditude revenue estimates")? - .error_for_status() - .wrap_internal_err("Aditude revenue estimate request failed")? - .json() - .await - .wrap_internal_err("failed to deserialize Aditude revenue estimates") + reqwest::Client::new() + .post("https://cloud.aditude.io/api/public/insights/metrics/v2") + .bearer_auth(&ENV.ADITUDE_API_KEY) + .json(&serde_json::json!({ + "metrics": ["REVENUE", "IMPRESSIONS"], + "range": "custom", + "startTime": start_time, + "endTime": end_time, + "interval": "1d" + })) + .send() + .await + .wrap_internal_err("failed to request Aditude revenue estimates")? + .error_for_status() + .wrap_internal_err("Aditude revenue estimate request failed")? + .json() + .await + .wrap_internal_err("failed to deserialize Aditude revenue estimates") } const ADITUDE_MONTH_ESTIMATE_CACHE_NAMESPACE: &str = - "aditude_month_estimates_v2"; + "aditude_month_estimates_v1"; const ADITUDE_MONTH_ESTIMATE_CACHE_EXPIRY: i64 = 60 * 60 * 24; pub async fn get_cached_aditude_month_estimates( periods: &[YearMonth], redis: &RedisPool, -) -> Result>, ApiError> { +) -> Result>, ApiError> { redis .get_cached_keys_raw_with_expiry( ADITUDE_MONTH_ESTIMATE_CACHE_NAMESPACE, @@ -973,7 +981,7 @@ pub async fn get_cached_aditude_month_estimates( async fn fetch_aditude_month_estimates( periods: Vec, -) -> Result>, ApiError> { +) -> Result>, ApiError> { let first_period = periods .iter() .min() @@ -988,19 +996,19 @@ async fn fetch_aditude_month_estimates( .date() .checked_add_months(Months::new(1)) .wrap_internal_err("failed to calculate payout period end")?; - let range_start_time = first_period - .date() - .and_hms_opt(0, 0, 0) - .wrap_internal_err("failed to calculate payout period start")? - .and_utc() - .timestamp_millis(); - let range_end_time = range_end - .and_hms_opt(0, 0, 0) - .wrap_internal_err("failed to calculate payout period end")? - .and_utc() - .timestamp_millis() - .checked_sub(1) - .wrap_internal_err("failed to calculate inclusive payout period end")?; + let range_start_time = first_period + .date() + .and_hms_opt(0, 0, 0) + .wrap_internal_err("failed to calculate payout period start")? + .and_utc() + .timestamp_millis(); + let range_end_time = range_end + .and_hms_opt(0, 0, 0) + .wrap_internal_err("failed to calculate payout period end")? + .and_utc() + .timestamp_millis() + .checked_sub(1) + .wrap_internal_err("failed to calculate inclusive payout period end")?; let estimates = periods .into_iter() .map(|period| { @@ -1025,9 +1033,12 @@ async fn fetch_aditude_month_estimates( .wrap_internal_err( "failed to calculate payout period day", )?; - Ok(DayRevenue { - date, - amount_usd: Decimal::ZERO, + Ok(AditudeDayEstimate { + revenue: DayRevenue { + date, + amount_usd: Decimal::ZERO, + }, + impressions: 0, }) }) .collect::, ApiError>>()?; @@ -1035,15 +1046,15 @@ async fn fetch_aditude_month_estimates( }) .collect::, ApiError>>()?; - let response = - make_aditude_revenue_request(range_start_time, range_end_time).await?; - for row in response - .responses - .into_iter() - .flat_map(|response| response.rows) - { - let timestamp = DateTime::from_timestamp_millis(row.time_millis) - .wrap_internal_err("invalid Aditude estimate timestamp")?; + let response = + make_aditude_revenue_request(range_start_time, range_end_time).await?; + for row in response + .responses + .into_iter() + .flat_map(|response| response.rows) + { + let timestamp = DateTime::from_timestamp_millis(row.time_millis) + .wrap_internal_err("invalid Aditude estimate timestamp")?; let date = timestamp.date_naive(); let period = YearMonth::from_day1(date); let Some(mut days) = estimates.get_mut(&period) else { @@ -1054,13 +1065,23 @@ async fn fetch_aditude_month_estimates( date.signed_duration_since(period.date()).num_days(), ) .wrap_internal_err("invalid Aditude estimate day")?; - days[day_index].date = date; - days[day_index].amount_usd += row.revenue; + days[day_index].revenue.date = date; + if let Some(revenue) = row.revenue { + days[day_index].revenue.amount_usd += revenue; + } + if let Some(impressions) = row.impressions { + days[day_index].impressions += impressions; + } } Ok(estimates) } +pub fn clean_io_fee_usd(impressions: u128) -> Decimal { + let clean_io_cpm = Decimal::from(8) / Decimal::from(1000); + clean_io_cpm * Decimal::from(impressions) / Decimal::from(1000) +} + pub async fn process_payout( pool: &PgPool, client: &clickhouse::Client, @@ -1303,13 +1324,13 @@ pub async fn process_payout( // Modrinth's share of ad revenue let modrinth_cut = Decimal::from(1) / Decimal::from(4); // Clean.io fee (ad antimalware). Per 1000 impressions. 0.008 CPM - let clean_io_fee = Decimal::from(8) / Decimal::from(1000); + let clean_io_fee = clean_io_fee_usd(aditude_impressions); // Google Ad Manager fee. Per 1000 impressions. 0.015400 CPM let gam_fee = Decimal::from(154) / Decimal::from(10000); let net_revenue = aditude_amount - - ((clean_io_fee + gam_fee) * Decimal::from(aditude_impressions) - / Decimal::from(1000)); + - clean_io_fee + - (gam_fee * Decimal::from(aditude_impressions) / Decimal::from(1000)); let payout = net_revenue * (Decimal::from(1) - modrinth_cut); diff --git a/apps/labrinth/src/routes/internal/payouts.rs b/apps/labrinth/src/routes/internal/payouts.rs index 82fb899be1..07ddc16106 100644 --- a/apps/labrinth/src/routes/internal/payouts.rs +++ b/apps/labrinth/src/routes/internal/payouts.rs @@ -2,7 +2,7 @@ use std::cmp::Reverse; use std::collections::{HashMap, HashSet}; use actix_web::{HttpRequest, get, web}; -use chrono::{Months, Utc}; +use chrono::{DateTime, Months, Utc}; use rust_decimal::Decimal; use crate::auth::get_user_from_headers; @@ -10,10 +10,12 @@ use crate::database::models::DBUserId; use crate::database::{PgPool, ReadOnlyPgPool}; use crate::models::pats::Scopes; use crate::models::payout_runs::{ - Adjustment, DayRevenue, PayoutRun, PayoutRunCompletion, PayoutRunReport, - PayoutRunStatus, + Adjustment, DayRevenue, PayoutRun, PayoutRunCompletion, PayoutRunReport, + PayoutRunStatus, +}; +use crate::queue::payouts::{ + AditudeDayEstimate, clean_io_fee_usd, get_cached_aditude_month_estimates, }; -use crate::queue::payouts::get_cached_aditude_month_estimates; use crate::queue::session::AuthQueue; use crate::routes::ApiError; use crate::util::error::Context; @@ -22,8 +24,14 @@ use xredis::RedisPool; #[derive(Debug, Clone, Copy)] enum DayRevenueEstimate { - Raw, - AdjustedToActual { actual_revenue_usd: Decimal }, + Raw, + AdjustedToActual { actual_revenue_usd: Decimal }, +} + +#[derive(Debug, Clone, Copy)] +struct PayoutVariance { + applied_at: DateTime, + variance: Decimal, } pub fn config(cfg: &mut web::ServiceConfig) { @@ -82,41 +90,51 @@ pub async fn get( .await .wrap_internal_err("failed to fetch newest payout value")?; + let payout_variances = sqlx::query!( + r#" + SELECT applied_at, variance + FROM payouts_variance + ORDER BY applied_at ASC + "#, + ) + .fetch_all(&***ro_pool) + .await + .wrap_internal_err("failed to fetch payout variances")? + .into_iter() + .map(|row| PayoutVariance { + applied_at: row.applied_at, + variance: row.variance, + }) + .collect::>(); + let mut stored_periods = HashSet::with_capacity(stored_runs.len()); - let mut revenue_estimates = HashMap::new(); - let mut runs = Vec::with_capacity(stored_runs.len()); - for run in stored_runs { - let period_start = YearMonth::from_day1(run.period_start.date_naive()); - let (status, revenue_estimate) = if run.completed_at.is_some() { - let amount_usd = run - .completed_result + let mut revenue_estimates = HashMap::new(); + let mut runs = Vec::with_capacity(stored_runs.len()); + for run in stored_runs { + let period_start = YearMonth::from_day1(run.period_start.date_naive()); + let (status, revenue_estimate) = if run.completed_at.is_some() { + let amount_usd = run + .completed_result .map(|completion| completion.revenue_usd) .wrap_internal_err( "paid payout run is missing its completion result", )?; - ( - PayoutRunStatus::Paid, - DayRevenueEstimate::AdjustedToActual { - actual_revenue_usd: amount_usd, - }, - ) - } else { - (PayoutRunStatus::Review, DayRevenueEstimate::Raw) - }; + ( + PayoutRunStatus::Paid, + DayRevenueEstimate::AdjustedToActual { + actual_revenue_usd: amount_usd, + }, + ) + } else { + (PayoutRunStatus::Review, DayRevenueEstimate::Raw) + }; - stored_periods.insert(period_start); - revenue_estimates.insert(period_start, revenue_estimate); - runs.push(PayoutRun { - period_start, - status, - report: PayoutRunReport { - days: Vec::new(), - fees_deducted_usd: Decimal::ZERO, // TODO: calculate deducted fees - variance_adjustment_usd: Decimal::ZERO, // TODO: calculate variance adjustment - net_estimated_revenue_usd: Decimal::ZERO, // TODO: calculate net revenue - creator_net_estimated_revenue_usd: Decimal::ZERO, // TODO: calculate creator share - modrinth_net_estimated_revenue_usd: Decimal::ZERO, // TODO: calculate Modrinth share - }, + stored_periods.insert(period_start); + revenue_estimates.insert(period_start, revenue_estimate); + runs.push(PayoutRun { + period_start, + status, + report: empty_payout_report(), started_at: is_admin.then_some(run.started_at), started_by: is_admin .then_some(run.started_by.map(|id| DBUserId(id).into())) @@ -144,19 +162,12 @@ pub async fn get( PayoutRunStatus::Pending }; - revenue_estimates.insert(period, DayRevenueEstimate::Raw); + revenue_estimates.insert(period, DayRevenueEstimate::Raw); runs.push(PayoutRun { period_start: period, status, - report: PayoutRunReport { - days: Vec::new(), - fees_deducted_usd: Decimal::ZERO, // TODO: calculate deducted fees - variance_adjustment_usd: Decimal::ZERO, // TODO: calculate variance adjustment - net_estimated_revenue_usd: Decimal::ZERO, // TODO: calculate net revenue - creator_net_estimated_revenue_usd: Decimal::ZERO, // TODO: calculate creator share - modrinth_net_estimated_revenue_usd: Decimal::ZERO, // TODO: calculate Modrinth share - }, + report: empty_payout_report(), started_at: None, started_by: None, completed_at: None, @@ -178,60 +189,130 @@ pub async fn get( } } - let estimate_periods = revenue_estimates.keys().copied().collect::>(); - let estimates = - get_cached_aditude_month_estimates(&estimate_periods, &redis).await?; - for run in &mut runs { - if let Some(days) = estimates.get(&run.period_start) { - let days = match revenue_estimates.get(&run.period_start) { - Some(DayRevenueEstimate::Raw) | None => days.clone(), - Some(DayRevenueEstimate::AdjustedToActual { - actual_revenue_usd, - }) => adjust_estimates_to_actual(days, *actual_revenue_usd)?, - }; - run.report.net_estimated_revenue_usd = - days.iter().map(|day| day.amount_usd).sum(); - run.report.days = days; - } + let estimate_periods = + revenue_estimates.keys().copied().collect::>(); + let estimates = + get_cached_aditude_month_estimates(&estimate_periods, &redis).await?; + for run in &mut runs { + let estimates = estimates + .get(&run.period_start) + .wrap_internal_err("missing Aditude payout period estimates")?; + let revenue_estimate = revenue_estimates + .get(&run.period_start) + .copied() + .wrap_internal_err("missing payout period revenue estimate type")?; + run.report = calculate_payout_report( + estimates, + revenue_estimate, + &payout_variances, + )?; } runs.sort_by_key(|run| Reverse(run.period_start)); - Ok(web::Json(runs)) + Ok(web::Json(runs)) +} + +fn empty_payout_report() -> PayoutRunReport { + PayoutRunReport { + days: Vec::new(), + raw_estimated_revenue_usd: Decimal::ZERO, + fees_deducted_usd: Decimal::ZERO, + variance_adjustment_usd: Decimal::ZERO, + net_estimated_revenue_usd: Decimal::ZERO, + creator_net_estimated_revenue_usd: Decimal::ZERO, + modrinth_net_estimated_revenue_usd: Decimal::ZERO, + } +} + +fn calculate_payout_report( + estimates: &[AditudeDayEstimate], + revenue_estimate: DayRevenueEstimate, + payout_variances: &[PayoutVariance], +) -> Result { + let estimated_days = estimates + .iter() + .map(|estimate| estimate.revenue.clone()) + .collect::>(); + let days = match revenue_estimate { + DayRevenueEstimate::Raw => estimated_days, + DayRevenueEstimate::AdjustedToActual { actual_revenue_usd } => { + adjust_estimates_to_actual(&estimated_days, actual_revenue_usd)? + } + }; + + let raw_estimated_revenue_usd = + days.iter().map(|day| day.amount_usd).sum::(); + let fees_deducted_usd = estimates + .iter() + .map(|estimate| clean_io_fee_usd(estimate.impressions)) + .sum::(); + let variance_adjustment_usd = days + .iter() + .zip(estimates) + .map(|(day, estimate)| { + let variance = payout_variances + .iter() + .rev() + .find(|variance| variance.applied_at.date_naive() <= day.date) + .map(|variance| variance.variance) + .wrap_internal_err("missing payout variance for revenue day")?; + let fee_usd = clean_io_fee_usd(estimate.impressions); + Ok((day.amount_usd - fee_usd) * variance) + }) + .collect::, ApiError>>()? + .into_iter() + .sum::(); + let net_estimated_revenue_usd = + raw_estimated_revenue_usd - fees_deducted_usd - variance_adjustment_usd; + let creator_net_estimated_revenue_usd = + net_estimated_revenue_usd * Decimal::new(75, 2); + let modrinth_net_estimated_revenue_usd = + net_estimated_revenue_usd - creator_net_estimated_revenue_usd; + + Ok(PayoutRunReport { + days, + raw_estimated_revenue_usd, + fees_deducted_usd, + variance_adjustment_usd, + net_estimated_revenue_usd, + creator_net_estimated_revenue_usd, + modrinth_net_estimated_revenue_usd, + }) } fn adjust_estimates_to_actual( - days: &[DayRevenue], - actual_revenue_usd: Decimal, + days: &[DayRevenue], + actual_revenue_usd: Decimal, ) -> Result, ApiError> { - if days.is_empty() { - return Ok(Vec::new()); - } + if days.is_empty() { + return Ok(Vec::new()); + } - let estimated_revenue_usd = - days.iter().map(|day| day.amount_usd).sum::(); - let day_count = u64::try_from(days.len()) - .wrap_internal_err("failed to calculate payout period day count")?; - let mut allocated_revenue_usd = Decimal::ZERO; - let last_day = days.len() - 1; + let estimated_revenue_usd = + days.iter().map(|day| day.amount_usd).sum::(); + let day_count = u64::try_from(days.len()) + .wrap_internal_err("failed to calculate payout period day count")?; + let mut allocated_revenue_usd = Decimal::ZERO; + let last_day = days.len() - 1; - Ok(days - .iter() - .enumerate() - .map(|(index, day)| { - let amount_usd = if index == last_day { - actual_revenue_usd - allocated_revenue_usd - } else if estimated_revenue_usd.is_zero() { - actual_revenue_usd / Decimal::from(day_count) - } else { - day.amount_usd * actual_revenue_usd / estimated_revenue_usd - }; - allocated_revenue_usd += amount_usd; + Ok(days + .iter() + .enumerate() + .map(|(index, day)| { + let amount_usd = if index == last_day { + actual_revenue_usd - allocated_revenue_usd + } else if estimated_revenue_usd.is_zero() { + actual_revenue_usd / Decimal::from(day_count) + } else { + day.amount_usd * actual_revenue_usd / estimated_revenue_usd + }; + allocated_revenue_usd += amount_usd; - DayRevenue { - date: day.date, - amount_usd, - } - }) - .collect()) + DayRevenue { + date: day.date, + amount_usd, + } + }) + .collect()) }