diff --git a/apps/labrinth/src/models/v3/payout_runs.rs b/apps/labrinth/src/models/v3/payout_runs.rs index a0ddb7006c..907616a104 100644 --- a/apps/labrinth/src/models/v3/payout_runs.rs +++ b/apps/labrinth/src/models/v3/payout_runs.rs @@ -1,5 +1,6 @@ use ariadne::ids::UserId; -use chrono::{DateTime, Utc}; +use chrono::{DateTime, NaiveDate, Utc}; +use rust_decimal::Decimal; use serde::{Deserialize, Serialize}; use crate::util::time::YearMonth; @@ -13,6 +14,8 @@ pub struct PayoutRun { 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. @@ -36,15 +39,55 @@ pub struct PayoutRun { )] #[serde(rename_all = "snake_case")] pub enum PayoutRunStatus { - /// We are still waiting on the ad provider to issue payouts to us. + /// 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. - InReview, - /// Payouts run is currently being performed. - Running, + Review, /// Payouts run is complete and payouts have been distributed to users. - Done, + Paid, +} + +#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)] +pub struct PayoutRunReport { + pub revenue: PayoutRunRevenue, + #[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)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum PayoutRunRevenue { + Estimated { + days: Vec, + }, + Actual { + #[serde(with = "rust_decimal::serde::float")] + amount_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)] diff --git a/apps/labrinth/src/queue/payouts/mod.rs b/apps/labrinth/src/queue/payouts/mod.rs index 944946e26b..8377e04796 100644 --- a/apps/labrinth/src/queue/payouts/mod.rs +++ b/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, @@ -15,7 +16,7 @@ use crate::util::webhook::{ }; use arc_swap::ArcSwapOption; use base64::Engine; -use chrono::{DateTime, Duration, NaiveTime, Utc}; +use chrono::{DateTime, Duration, Months, NaiveTime, Utc}; use dashmap::DashMap; use eyre::Result; use futures::TryStreamExt; @@ -887,6 +888,106 @@ pub async fn make_aditude_request( Ok(json) } +const ADITUDE_MONTH_ESTIMATE_CACHE_NAMESPACE: &str = "aditude_month_estimates"; +const ADITUDE_MONTH_ESTIMATE_CACHE_EXPIRY: i64 = 60 * 60 * 24; + +pub async fn get_cached_aditude_month_estimates( + periods: &[YearMonth], + redis: &RedisPool, +) -> Result>, ApiError> { + redis + .get_cached_keys_raw_with_expiry( + ADITUDE_MONTH_ESTIMATE_CACHE_NAMESPACE, + periods, + ADITUDE_MONTH_ESTIMATE_CACHE_EXPIRY, + fetch_aditude_month_estimates, + ) + .await +} + +async fn fetch_aditude_month_estimates( + periods: Vec, +) -> Result>, 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 = format!( + "{}/{}", + first_period.date().format("%Y-%m-%d"), + range_end.format("%Y-%m-%d") + ); + 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(DayRevenue { + date, + amount_usd: Decimal::ZERO, + }) + }) + .collect::, ApiError>>()?; + Ok((period, days)) + }) + .collect::, ApiError>>()?; + + let response = + make_aditude_request(&["METRIC_REVENUE"], &range, "1d").await?; + for point in response.into_iter().flat_map(|points| points.points_list) { + let Some(revenue) = point.metric.revenue else { + continue; + }; + let timestamp = i64::try_from(point.time.seconds) + .ok() + .and_then(|seconds| DateTime::from_timestamp(seconds, 0)) + .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].date = date; + days[day_index].amount_usd += revenue; + } + + Ok(estimates) +} + pub async fn process_payout( pool: &PgPool, client: &clickhouse::Client, diff --git a/apps/labrinth/src/routes/internal/payouts.rs b/apps/labrinth/src/routes/internal/payouts.rs index abd77fa8dc..d28c834f88 100644 --- a/apps/labrinth/src/routes/internal/payouts.rs +++ b/apps/labrinth/src/routes/internal/payouts.rs @@ -8,7 +8,11 @@ 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, PayoutRun, PayoutRunStatus}; +use crate::models::payout_runs::{ + Adjustment, PayoutRun, PayoutRunCompletion, PayoutRunReport, + PayoutRunRevenue, PayoutRunStatus, +}; +use crate::queue::payouts::get_cached_aditude_month_estimates; use crate::queue::session::AuthQueue; use crate::routes::ApiError; use crate::util::error::Context; @@ -49,6 +53,7 @@ pub async fn get( started_at, started_by, completed_at, + completed_result AS "completed_result?: sqlx::types::Json", adjustments AS "adjustments!: sqlx::types::Json>" FROM payouts_runs ORDER BY period_start DESC @@ -71,19 +76,48 @@ pub async fn get( .wrap_internal_err("failed to fetch newest payout value")?; let mut stored_periods = HashSet::with_capacity(stored_runs.len()); + let mut estimate_periods = HashSet::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 = if run.completed_at.is_some() { - PayoutRunStatus::Done + let (status, report) = 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, + PayoutRunReport { + revenue: PayoutRunRevenue::Actual { amount_usd }, + fees_deducted_usd: todo!(), + variance_adjustment_usd: todo!(), + net_estimated_revenue_usd: todo!(), + creator_net_estimated_revenue_usd: todo!(), + modrinth_net_estimated_revenue_usd: todo!(), + }, + ) } else { - PayoutRunStatus::Running + estimate_periods.insert(period_start); + ( + PayoutRunStatus::Review, + PayoutRunReport { + revenue: PayoutRunRevenue::Estimated { days: Vec::new() }, + fees_deducted_usd: todo!(), + variance_adjustment_usd: todo!(), + net_estimated_revenue_usd: todo!(), + creator_net_estimated_revenue_usd: todo!(), + modrinth_net_estimated_revenue_usd: todo!(), + }, + ) }; stored_periods.insert(period_start); runs.push(PayoutRun { period_start, status, + report, started_at: is_admin.then_some(run.started_at), started_by: is_admin .then_some(run.started_by.map(|id| DBUserId(id).into())) @@ -100,19 +134,32 @@ pub async fn get( while period <= newest_period { if !stored_periods.contains(&period) { - let status = - if net_60_payout_available_at(period).wrap_internal_err( - "failed to calculate payout review date", - )? <= newest_created - { - PayoutRunStatus::InReview - } else { - PayoutRunStatus::Pending - }; + 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 + }; + + estimate_periods.insert(period); runs.push(PayoutRun { period_start: period, status, + report: PayoutRunReport { + revenue: PayoutRunRevenue::Estimated { + days: Vec::new(), + }, + fees_deducted_usd: todo!(), + variance_adjustment_usd: todo!(), + net_estimated_revenue_usd: todo!(), + creator_net_estimated_revenue_usd: todo!(), + modrinth_net_estimated_revenue_usd: todo!(), + }, started_at: None, started_by: None, completed_at: None, @@ -134,6 +181,18 @@ pub async fn get( } } + let estimate_periods = estimate_periods.into_iter().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) { + run.report.net_estimated_revenue_usd = + days.iter().map(|day| day.amount_usd).sum(); + run.report.revenue = + PayoutRunRevenue::Estimated { days: days.clone() }; + } + } + runs.sort_by_key(|run| Reverse(run.period_start)); Ok(web::Json(runs)) diff --git a/packages/xredis/src/cache.rs b/packages/xredis/src/cache.rs index 5e2aba678b..25654e93ee 100644 --- a/packages/xredis/src/cache.rs +++ b/packages/xredis/src/cache.rs @@ -254,6 +254,48 @@ impl CacheManager { None, false, keys, + None, + |ids| async move { + Ok(closure(ids) + .await? + .into_iter() + .map(|(key, value)| (key, (None::, value))) + .collect()) + }, + ) + .await + } + + pub async fn get_cached_keys_raw_with_expiry( + &self, + provider: &P, + namespace: &str, + keys: &[K], + expiry: i64, + closure: F, + ) -> Result, E> + where + P: ConnectionProvider, + F: FnOnce(Vec) -> Fut, + Fut: Future, E>>, + E: From, + 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 { Ok(closure(ids) .await? @@ -298,6 +340,7 @@ impl CacheManager { Some(slug_namespace), case_sensitive, keys, + None, closure, ) .await? @@ -313,6 +356,7 @@ impl CacheManager { slug_namespace: Option<&str>, case_sensitive: bool, keys: &[I], + expiry_override: Option<(i64, i64)>, closure: F, ) -> Result, E> where @@ -430,7 +474,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(); diff --git a/packages/xredis/src/lib.rs b/packages/xredis/src/lib.rs index e84fce15eb..34c8619b3b 100644 --- a/packages/xredis/src/lib.rs +++ b/packages/xredis/src/lib.rs @@ -168,6 +168,34 @@ impl RedisPool { .await } + pub async fn get_cached_keys_raw_with_expiry( + &self, + namespace: &str, + keys: &[K], + expiry: i64, + closure: F, + ) -> Result, E> + where + F: FnOnce(Vec) -> Fut, + Fut: Future, E>>, + E: From, + 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( &self, namespace: &str, @@ -233,6 +261,7 @@ impl RedisPool { slug_namespace, case_sensitive, keys, + None, closure, ) .await