raw/net estimated revenue fields

apps/labrinth/migrations/20260720114640_creator_payouts.sql

@@ -0,0 +1,16 @@
+CREATE TABLE payouts_runs(
+	id BIGINT PRIMARY KEY,
+	-- timestamp on the 1st of a month at midnight,
+	-- representing what month this run is for.
+	-- if a row exists for a month, then a payout run
+	-- is running/has completed for this month (see
+	-- `completed_at`).
+	period_start TIMESTAMPTZ NOT NULL,
+	started_at TIMESTAMPTZ NOT NULL,
+	started_by BIGINT REFERENCES users(id)
+		ON DELETE SET NULL,
+	completed_at TIMESTAMPTZ,
+	completed_result JSONB,
+	adjustments JSONB NOT NULL
+);
+CREATE INDEX payouts_runs_period_start ON payouts_runs(period_start);

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/mod.rs

@@ -15,6 +15,7 @@ pub use v3::oauth_clients;
 pub use v3::organizations;
 pub use v3::pack;
 pub use v3::pats;
+pub use v3::payout_runs;
 pub use v3::payouts;
 pub use v3::projects;
 pub use v3::reports;

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

@@ -12,6 +12,7 @@ pub mod oauth_clients;
 pub mod organizations;
 pub mod pack;
 pub mod pats;
+pub mod payout_runs;
 pub mod payouts;
 pub mod projects;
 pub mod reports;

apps/labrinth/src/models/v3/payout_runs.rs

@@ -0,0 +1,84 @@
+use ariadne::ids::UserId;
+use chrono::{DateTime, NaiveDate, Utc};
+use rust_decimal::Decimal;
+use serde::{Deserialize, Serialize};
+
+use crate::util::time::YearMonth;
+
+#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
+pub struct PayoutRun {
+    /// What period this payout run is for.
+    ///
+    /// Payout runs are always for the period of a specific year and month -
+    /// they are not associated with any specific day.
+    pub period_start: YearMonth,
+    /// What state this run is in.
+    pub status: PayoutRunStatus,
+    #[serde(flatten)]
+    pub report: PayoutRunReport,
+    /// When this run started running.
+    ///
+    /// Only accessible to admins.
+    pub started_at: Option<DateTime<Utc>>,
+    /// What user started this run.
+    ///
+    /// Only accessible to admins.
+    pub started_by: Option<UserId>,
+    /// When this run completed.
+    ///
+    /// Only accessible to admins.
+    pub completed_at: Option<DateTime<Utc>>,
+    /// What payout adjustments were specified in this run.
+    ///
+    /// Only accessible to admins.
+    pub adjustments: Option<Vec<Adjustment>>,
+}
+
+#[derive(
+    Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, utoipa::ToSchema,
+)]
+#[serde(rename_all = "snake_case")]
+pub enum PayoutRunStatus {
+    /// The payout period is still receiving revenue estimates.
+    Open,
+    /// The payout period is closed, but is still within Net-60 terms.
+    Pending,
+    /// The ad provider should have issued payouts to us by now, and we will
+    /// soon run the payouts.
+    Review,
+    /// Payouts run is complete and payouts have been distributed to users.
+    Paid,
+}
+
+#[derive(Debug, Clone, Default, Serialize, Deserialize, utoipa::ToSchema)]
+pub struct PayoutRunReport {
+    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")]
+    pub net_estimated_revenue_usd: Decimal,
+    #[serde(with = "rust_decimal::serde::float")]
+    pub creator_net_estimated_revenue_usd: Decimal,
+    #[serde(with = "rust_decimal::serde::float")]
+    pub modrinth_net_estimated_revenue_usd: Decimal,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
+pub struct PayoutRunCompletion {
+    #[serde(with = "rust_decimal::serde::float")]
+    pub revenue_usd: Decimal,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
+pub struct DayRevenue {
+    pub date: NaiveDate,
+    #[serde(with = "rust_decimal::serde::float")]
+    pub amount_usd: Decimal,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
+pub struct Adjustment {}

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

@@ -1,11 +1,11 @@
 use crate::database::PgPool;
 use crate::env::ENV;
-use chrono::{Datelike, Duration, TimeZone, Utc};
 use eyre::{Context, Result, eyre};
 use rust_decimal::{Decimal, dec};
 use tracing::warn;
 
 use crate::database::models::{DBAffiliateCodeId, DBUserId};
+use crate::util::time::{YearMonth, net_60_payout_available_at};
 
 pub async fn process_affiliate_payouts(postgres: &PgPool) -> Result<()> {
     // process:
@@ -91,21 +91,10 @@ pub async fn process_affiliate_payouts(postgres: &PgPool) -> Result<()> {
             continue;
         };
 
-        // affiliate payouts are Net 60 from the end of the month
-        // this is net 60 relative to the time of the charge's last attempt, not from now
-        let available = {
-            let year = last_attempt.year();
-            let month = last_attempt.month();
-
-            // get the first day of the next month
-            let last_day_of_month = if month == 12 {
-                Utc.with_ymd_and_hms(year + 1, 1, 1, 0, 0, 0).unwrap()
-            } else {
-                Utc.with_ymd_and_hms(year, month + 1, 1, 0, 0, 0).unwrap()
-            };
-
-            last_day_of_month + Duration::days(59)
-        };
+        let available = net_60_payout_available_at(YearMonth::from_day1(
+            last_attempt.date_naive(),
+        ))
+        .ok_or_else(|| eyre!("failed to calculate affiliate payout date"))?;
 
         let revenue_split = row
             .revenue_split

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

@@ -2,6 +2,7 @@ use crate::database::models::notification_item::NotificationBuilder;
 use crate::database::models::payouts_values_notifications;
 use crate::database::{PgPool, PgTransaction};
 use crate::env::ENV;
+use crate::models::payout_runs::DayRevenue;
 use crate::models::payouts::{
     PayoutDecimal, PayoutInterval, PayoutMethod, PayoutMethodType,
     TremendousForexResponse,
@@ -10,12 +11,13 @@ use crate::models::projects::MonetizationStatus;
 use crate::routes::ApiError;
 use crate::util::error::ApiContext as _;
 use crate::util::error::Context;
+use crate::util::time::{YearMonth, net_60_payout_available_at};
 use crate::util::webhook::{
     PayoutSourceAlertType, send_slack_payout_source_alert_webhook,
 };
 use arc_swap::ArcSwapOption;
 use base64::Engine;
-use chrono::{DateTime, Datelike, Duration, NaiveTime, TimeZone, Utc};
+use chrono::{DateTime, Duration, Months, NaiveTime, Utc};
 use dashmap::DashMap;
 use eyre::Result;
 use futures::TryStreamExt;
@@ -878,6 +880,32 @@ pub struct AditudeTime {
     pub seconds: u64,
 }
 
+#[derive(Deserialize)]
+struct AditudeMetricsV2Response {
+    responses: Vec<AditudeMetricsV2Table>,
+}
+
+#[derive(Deserialize)]
+struct AditudeMetricsV2Table {
+    rows: Vec<AditudeMetricRow>,
+}
+
+#[derive(Deserialize)]
+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(
     metrics: &[&str],
     range: &str,
@@ -908,6 +936,152 @@ pub async fn make_aditude_request(
     Ok(json)
 }
 
+async fn make_aditude_revenue_request(
+    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", "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_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<AditudeDayEstimate>>, ApiError> {
+    redis
+        .get_cached_keys_raw_with_expiry(
+            ADITUDE_MONTH_ESTIMATE_CACHE_NAMESPACE,
+            periods,
+            ADITUDE_MONTH_ESTIMATE_CACHE_EXPIRY,
+            fetch_aditude_month_estimates,
+        )
+        .await
+        .wrap_internal_err("failed to fetch cached Aditude month estimates")
+}
+
+async fn fetch_aditude_month_estimates(
+    periods: Vec<YearMonth>,
+) -> Result<DashMap<YearMonth, Vec<AditudeDayEstimate>>, ApiError> {
+    let first_period = periods
+        .iter()
+        .min()
+        .copied()
+        .wrap_internal_err("missing first Aditude estimate period")?;
+    let last_period = periods
+        .iter()
+        .max()
+        .copied()
+        .wrap_internal_err("missing last Aditude estimate period")?;
+    let range_end = last_period
+        .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 estimates = periods
+        .into_iter()
+        .map(|period| {
+            let period_end = period
+                .date()
+                .checked_add_months(Months::new(1))
+                .wrap_internal_err(
+                "failed to calculate payout period end",
+            )?;
+            let day_count = usize::try_from(
+                period_end.signed_duration_since(period.date()).num_days(),
+            )
+            .wrap_internal_err("failed to calculate payout period day count")?;
+            let days = (0..day_count)
+                .map(|day| {
+                    let day = i64::try_from(day).wrap_internal_err(
+                        "failed to calculate payout period day",
+                    )?;
+                    let date = period
+                        .date()
+                        .checked_add_signed(Duration::days(day))
+                        .wrap_internal_err(
+                            "failed to calculate payout period day",
+                        )?;
+                    Ok(AditudeDayEstimate {
+                        revenue: DayRevenue {
+                            date,
+                            amount_usd: Decimal::ZERO,
+                        },
+                        impressions: 0,
+                    })
+                })
+                .collect::<Result<Vec<_>, ApiError>>()?;
+            Ok((period, days))
+        })
+        .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 date = timestamp.date_naive();
+        let period = YearMonth::from_day1(date);
+        let Some(mut days) = estimates.get_mut(&period) else {
+            continue;
+        };
+
+        let day_index = usize::try_from(
+            date.signed_duration_since(period.date()).num_days(),
+        )
+        .wrap_internal_err("invalid Aditude estimate day")?;
+        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,
@@ -1150,32 +1324,20 @@ 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);
 
-    // Ad payouts are Net 60 from the end of the month
-    let available = {
-        let now = Utc::now().date_naive();
-
-        let year = now.year();
-        let month = now.month();
-
-        // Get the first day of the next month
-        let last_day_of_month = if month == 12 {
-            Utc.with_ymd_and_hms(year + 1, 1, 1, 0, 0, 0).unwrap()
-        } else {
-            Utc.with_ymd_and_hms(year, month + 1, 1, 0, 0, 0).unwrap()
-        };
-
-        last_day_of_month + Duration::days(59)
-    };
+    let available = net_60_payout_available_at(YearMonth::from_day1(
+        Utc::now().date_naive(),
+    ))
+    .wrap_internal_err("failed to calculate creator payout date")?;
 
     let (
         mut insert_user_ids,

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

@@ -15,6 +15,7 @@ pub mod medal;
 pub mod moderation;
 pub mod mural;
 pub mod pats;
+pub mod payouts;
 pub mod search;
 pub mod server_ping;
 pub mod session;
@@ -35,6 +36,7 @@ pub fn config(cfg: &mut web::ServiceConfig) {
             .configure(session::config)
             .configure(flows::config)
             .configure(pats::config)
+            .configure(payouts::config)
             .configure(oauth_clients::config)
             .service(
                 web::scope("/analytics-event")
@@ -103,6 +105,7 @@ pub fn config(cfg: &mut web::ServiceConfig) {
 		pats::create_pat,
 		pats::edit_pat,
 		pats::delete_pat,
+		payouts::get,
 		moderation::get_projects,
 		moderation::get_project_ids,
 		moderation::get_project_meta,

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

@@ -0,0 +1,318 @@
+use std::cmp::Reverse;
+use std::collections::{HashMap, HashSet};
+
+use actix_web::{HttpRequest, get, web};
+use chrono::{DateTime, Months, Utc};
+use rust_decimal::Decimal;
+
+use crate::auth::get_user_from_headers;
+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,
+};
+use crate::queue::payouts::{
+    AditudeDayEstimate, clean_io_fee_usd, get_cached_aditude_month_estimates,
+};
+use crate::queue::session::AuthQueue;
+use crate::routes::ApiError;
+use crate::util::error::Context;
+use crate::util::time::{YearMonth, net_60_payout_available_at};
+use xredis::RedisPool;
+
+#[derive(Debug, Clone, Copy)]
+enum DayRevenueEstimate {
+    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) {
+    cfg.service(get);
+}
+
+/// List creator payout runs.
+#[utoipa::path(
+    tag = "payouts",
+    responses((status = OK, body = inline(Vec<PayoutRun>)))
+)]
+#[get("/payout-runs")]
+pub async fn get(
+    req: HttpRequest,
+    pool: web::Data<PgPool>,
+    ro_pool: web::Data<ReadOnlyPgPool>,
+    redis: web::Data<RedisPool>,
+    session_queue: web::Data<AuthQueue>,
+) -> Result<web::Json<Vec<PayoutRun>>, ApiError> {
+    let is_admin = get_user_from_headers(
+        &req,
+        &**pool,
+        &redis,
+        &session_queue,
+        Scopes::empty(),
+    )
+    .await
+    .is_ok_and(|(_, user)| user.role.is_admin());
+
+    let stored_runs = sqlx::query!(
+        r#"
+        SELECT
+            period_start,
+            started_at,
+            started_by,
+            completed_at,
+            completed_result AS "completed_result?: sqlx::types::Json<PayoutRunCompletion>",
+            adjustments AS "adjustments!: sqlx::types::Json<Vec<Adjustment>>"
+        FROM payouts_runs
+        ORDER BY period_start DESC
+        "#,
+    )
+    .fetch_all(&***ro_pool)
+    .await
+    .wrap_internal_err("failed to fetch payout runs")?;
+
+    let newest_created = sqlx::query_scalar!(
+        r#"
+        SELECT created
+        FROM payouts_values
+        ORDER BY created DESC
+        LIMIT 1
+        "#,
+    )
+    .fetch_optional(&***ro_pool)
+    .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
+                .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)
+        };
+
+        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()))
+                .flatten(),
+            completed_at: is_admin.then_some(run.completed_at).flatten(),
+            adjustments: is_admin.then_some(run.adjustments.0),
+        });
+    }
+
+    if let Some(newest_created) = newest_created {
+        let now = Utc::now();
+        let newest_period = YearMonth::from_day1(newest_created.date_naive());
+        let mut period = YearMonth::from_day1(now.date_naive());
+
+        while period <= newest_period {
+            if !stored_periods.contains(&period) {
+                let status = if period == newest_period {
+                    PayoutRunStatus::Open
+                } else if net_60_payout_available_at(period).wrap_internal_err(
+                    "failed to calculate payout review date",
+                )? <= newest_created
+                {
+                    PayoutRunStatus::Review
+                } else {
+                    PayoutRunStatus::Pending
+                };
+
+                revenue_estimates.insert(period, DayRevenueEstimate::Raw);
+
+                runs.push(PayoutRun {
+                    period_start: period,
+                    status,
+                    report: empty_payout_report(),
+                    started_at: None,
+                    started_by: None,
+                    completed_at: None,
+                    adjustments: None,
+                });
+            }
+
+            if period == newest_period {
+                break;
+            }
+
+            let next_month = period
+                .date()
+                .checked_add_months(Months::new(1))
+                .wrap_internal_err(
+                "failed to calculate next payout month",
+            )?;
+            period = YearMonth::from_day1(next_month);
+        }
+    }
+
+    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))
+}
+
+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,
+) -> Result<Vec<DayRevenue>, ApiError> {
+    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;
+
+    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())
+}

apps/labrinth/src/util/mod.rs

@@ -20,5 +20,6 @@ pub mod routes;
 pub mod sentry;
 pub mod tags;
 pub mod tiltify;
+pub mod time;
 pub mod validate;
 pub mod webhook;

apps/labrinth/src/util/time.rs

@@ -0,0 +1,211 @@
+use std::{fmt, str::FromStr};
+
+use chrono::{DateTime, Datelike, Days, Months, NaiveDate, Utc};
+use serde::{Deserialize, Deserializer, Serialize, Serializer, de::Error};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct YearMonth(NaiveDate);
+
+impl YearMonth {
+    pub fn new(date: NaiveDate) -> Result<Self, InvalidYearMonth> {
+        if date.day() == 1 {
+            Ok(Self(date))
+        } else {
+            return Err(InvalidYearMonth);
+        }
+    }
+
+    pub fn from_year_month(year: i32, month: u32) -> Option<Self> {
+        NaiveDate::from_ymd_opt(year, month, 1).map(Self)
+    }
+
+    pub fn from_day1(date: NaiveDate) -> Self {
+        Self(date.with_day(1).expect("every month has a first day"))
+    }
+
+    pub fn date(self) -> NaiveDate {
+        self.0
+    }
+}
+
+/// Calculate when a payout period becomes available under Net-60 terms.
+pub fn net_60_payout_available_at(period: YearMonth) -> Option<DateTime<Utc>> {
+    period
+        .date()
+        .checked_add_months(Months::new(1))?
+        .and_hms_opt(0, 0, 0)?
+        .and_utc()
+        .checked_add_days(Days::new(59))
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
+#[error("a year-month must use the first day of its month")]
+pub struct InvalidYearMonth;
+
+impl TryFrom<NaiveDate> for YearMonth {
+    type Error = InvalidYearMonth;
+
+    fn try_from(date: NaiveDate) -> Result<Self, Self::Error> {
+        Self::new(date)
+    }
+}
+
+impl From<YearMonth> for NaiveDate {
+    fn from(year_month: YearMonth) -> Self {
+        year_month.0
+    }
+}
+
+impl fmt::Display for YearMonth {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(formatter, "{}", self.0.format("%Y-%m"))
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
+#[error("expected a valid year and month in `YYYY-MM` format")]
+pub struct ParseYearMonthError;
+
+impl FromStr for YearMonth {
+    type Err = ParseYearMonthError;
+
+    fn from_str(value: &str) -> Result<Self, Self::Err> {
+        let mut segments = value.split('-');
+        let (Some(year), Some(month), None) =
+            (segments.next(), segments.next(), segments.next())
+        else {
+            return Err(ParseYearMonthError);
+        };
+
+        let year = year.parse().map_err(|_| ParseYearMonthError)?;
+        let month = month.parse().map_err(|_| ParseYearMonthError)?;
+        Self::from_year_month(year, month).ok_or(ParseYearMonthError)
+    }
+}
+
+impl Serialize for YearMonth {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        serializer.collect_str(self)
+    }
+}
+
+impl<'de> Deserialize<'de> for YearMonth {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let value = String::deserialize(deserializer)?;
+        value.parse().map_err(D::Error::custom)
+    }
+}
+
+impl utoipa::PartialSchema for YearMonth {
+    fn schema() -> utoipa::openapi::RefOr<utoipa::openapi::schema::Schema> {
+        utoipa::openapi::ObjectBuilder::new()
+            .schema_type(utoipa::openapi::schema::Type::String)
+            .pattern(Some(r"^\d{4}-(0[1-9]|1[0-2])$"))
+            .into()
+    }
+}
+
+impl utoipa::ToSchema for YearMonth {}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn rejects_dates_after_the_first() {
+        let date = NaiveDate::from_ymd_opt(2026, 7, 2).unwrap();
+
+        assert_eq!(YearMonth::new(date), Err(InvalidYearMonth));
+    }
+
+    #[test]
+    fn constructs_from_year_and_month() {
+        let year_month = YearMonth::from_year_month(2026, 7).unwrap();
+
+        assert_eq!(
+            year_month.date(),
+            NaiveDate::from_ymd_opt(2026, 7, 1).unwrap()
+        );
+        assert!(YearMonth::from_year_month(2026, 13).is_none());
+    }
+
+    #[test]
+    fn constructs_from_date_using_first_day() {
+        let date = NaiveDate::from_ymd_opt(2026, 7, 20).unwrap();
+
+        assert_eq!(
+            YearMonth::from_day1(date).date(),
+            NaiveDate::from_ymd_opt(2026, 7, 1).unwrap()
+        );
+    }
+
+    #[test]
+    fn serializes_as_year_and_month() {
+        let date = NaiveDate::from_ymd_opt(2026, 7, 1).unwrap();
+        let year_month = YearMonth::new(date).unwrap();
+
+        assert_eq!(serde_json::to_string(&year_month).unwrap(), r#""2026-07""#);
+    }
+
+    #[test]
+    fn deserializes_year_and_month_to_the_first() {
+        let year_month: YearMonth =
+            serde_json::from_str(r#""2026-07""#).unwrap();
+
+        assert_eq!(
+            year_month.date(),
+            NaiveDate::from_ymd_opt(2026, 7, 1).unwrap()
+        );
+    }
+
+    #[test]
+    fn parses_year_and_month_to_the_first() {
+        let year_month = YearMonth::from_str("2026-7").unwrap();
+
+        assert_eq!(
+            year_month.date(),
+            NaiveDate::from_ymd_opt(2026, 7, 1).unwrap()
+        );
+        assert_eq!(year_month.to_string(), "2026-07");
+    }
+
+    #[test]
+    fn rejects_other_serialized_formats() {
+        for value in [r#""2026""#, r#""2026-07-01""#, r#""2026-13""#] {
+            assert!(serde_json::from_str::<YearMonth>(value).is_err());
+        }
+    }
+
+    #[test]
+    fn calculates_net_60_payout_availability() {
+        let august = YearMonth::from_year_month(2026, 8).unwrap();
+        let december = YearMonth::from_year_month(2026, 12).unwrap();
+
+        assert_eq!(
+            net_60_payout_available_at(august),
+            Some(
+                NaiveDate::from_ymd_opt(2026, 10, 30)
+                    .unwrap()
+                    .and_hms_opt(0, 0, 0)
+                    .unwrap()
+                    .and_utc()
+            )
+        );
+        assert_eq!(
+            net_60_payout_available_at(december),
+            Some(
+                NaiveDate::from_ymd_opt(2027, 3, 1)
+                    .unwrap()
+                    .and_hms_opt(0, 0, 0)
+                    .unwrap()
+                    .and_utc()
+            )
+        );
+    }
+}

packages/xredis/src/cache.rs

@@ -258,6 +258,50 @@ impl CacheManager {
             None,
             false,
             keys,
+            None,
+            |ids| async move {
+                Ok::<_, E>(
+                    closure(ids)
+                        .await?
+                        .into_iter()
+                        .map(|(key, value)| (key, (None::<String>, value)))
+                        .collect(),
+                )
+            },
+        )
+        .await
+    }
+
+    pub async fn get_cached_keys_raw_with_expiry<P, F, Fut, T, K, E>(
+        &self,
+        provider: &P,
+        namespace: &str,
+        keys: &[K],
+        expiry: i64,
+        closure: F,
+    ) -> Result<HashMap<K, T>>
+    where
+        P: ConnectionProvider,
+        F: FnOnce(Vec<K>) -> Fut,
+        Fut: Future<Output = Result<DashMap<K, T>, E>>,
+        E: std::error::Error + Send + Sync + 'static,
+        T: Serialize + DeserializeOwned,
+        K: Display
+            + Hash
+            + Eq
+            + PartialEq
+            + Clone
+            + DeserializeOwned
+            + Serialize
+            + Debug,
+    {
+        self.get_cached_keys_raw_with_slug(
+            provider,
+            namespace,
+            None,
+            false,
+            keys,
+            Some((expiry, expiry)),
             |ids| async move {
                 let values = match closure(ids).await {
                     Ok(values) => values,
@@ -307,6 +351,7 @@ impl CacheManager {
                 Some(slug_namespace),
                 case_sensitive,
                 keys,
+                None,
                 closure,
             )
             .await
@@ -323,6 +368,7 @@ impl CacheManager {
         slug_namespace: Option<&str>,
         case_sensitive: bool,
         keys: &[I],
+        expiry_override: Option<(i64, i64)>,
         closure: F,
     ) -> Result<HashMap<K, T>>
     where
@@ -445,7 +491,8 @@ impl CacheManager {
                 .instrument(info_span!("get_cached_values_closure"))
             };
 
-        let (default_expiry, actual_expiry) = self.settings.expiries(namespace);
+        let (default_expiry, actual_expiry) = expiry_override
+            .unwrap_or_else(|| self.settings.expiries(namespace));
         let current_time = Utc::now();
         let mut expired_values = HashMap::new();
         let mut expired_identities = HashMap::new();

packages/xredis/src/lib.rs

@@ -163,6 +163,34 @@ impl RedisPool {
             .await
     }
 
+    pub async fn get_cached_keys_raw_with_expiry<F, Fut, T, K, E>(
+        &self,
+        namespace: &str,
+        keys: &[K],
+        expiry: i64,
+        closure: F,
+    ) -> Result<std::collections::HashMap<K, T>>
+    where
+        F: FnOnce(Vec<K>) -> Fut,
+        Fut: Future<Output = Result<DashMap<K, T>, E>>,
+        E: std::error::Error + Send + Sync + 'static,
+        T: Serialize + DeserializeOwned,
+        K: Display
+            + Hash
+            + Eq
+            + PartialEq
+            + Clone
+            + DeserializeOwned
+            + Serialize
+            + Debug,
+    {
+        self.cache
+            .get_cached_keys_raw_with_expiry(
+                self, namespace, keys, expiry, closure,
+            )
+            .await
+    }
+
     pub async fn get_cached_keys_with_slug<F, Fut, T, I, K, S, E>(
         &self,
         namespace: &str,
@@ -228,6 +256,7 @@ impl RedisPool {
                 slug_namespace,
                 case_sensitive,
                 keys,
+                None,
                 closure,
             )
             .await