raw/net estimated revenue fields

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);

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<DayRevenue>,
-	#[serde(with = "rust_decimal::serde::float")]
-	pub fees_deducted_usd: Decimal,
+    pub days: Vec<DayRevenue>,
+    #[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")]

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

@@ -882,20 +882,28 @@ pub struct AditudeTime {
 
 #[derive(Deserialize)]
 struct AditudeMetricsV2Response {
-	responses: Vec<AditudeMetricsV2Table>,
+    responses: Vec<AditudeMetricsV2Table>,
 }
 
 #[derive(Deserialize)]
 struct AditudeMetricsV2Table {
-	rows: Vec<AditudeRevenueRow>,
+    rows: Vec<AditudeMetricRow>,
 }
 
 #[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<Decimal>,
+    #[serde(rename = "IMPRESSIONS")]
+    impressions: Option<u128>,
+}
+
+#[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<AditudeMetricsV2Response, ApiError> {
-	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<HashMap<YearMonth, Vec<DayRevenue>>, ApiError> {
+) -> Result<HashMap<YearMonth, Vec<AditudeDayEstimate>>, 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<YearMonth>,
-) -> Result<DashMap<YearMonth, Vec<DayRevenue>>, ApiError> {
+) -> Result<DashMap<YearMonth, Vec<AditudeDayEstimate>>, 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::<Result<Vec<_>, ApiError>>()?;
@@ -1035,15 +1046,15 @@ async fn fetch_aditude_month_estimates(
         })
         .collect::<Result<DashMap<_, _>, 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);
 

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<Utc>,
+    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::<Vec<_>>();
+
     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::<Vec<_>>();
-	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::<Vec<_>>();
+    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<PayoutRunReport, ApiError> {
+    let estimated_days = estimates
+        .iter()
+        .map(|estimate| estimate.revenue.clone())
+        .collect::<Vec<_>>();
+    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::<Decimal>();
+    let fees_deducted_usd = estimates
+        .iter()
+        .map(|estimate| clean_io_fee_usd(estimate.impressions))
+        .sum::<Decimal>();
+    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::<Result<Vec<Decimal>, ApiError>>()?
+        .into_iter()
+        .sum::<Decimal>();
+    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<Vec<DayRevenue>, 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::<Decimal>();
-	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::<Decimal>();
+    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())
 }