wip: payout run reports

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<DayRevenue>,
+    },
+    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)]

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<HashMap<YearMonth, Vec<DayRevenue>>, 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<YearMonth>,
+) -> Result<DashMap<YearMonth, Vec<DayRevenue>>, 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::<Result<Vec<_>, ApiError>>()?;
+            Ok((period, days))
+        })
+        .collect::<Result<DashMap<_, _>, 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,

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<PayoutRunCompletion>",
             adjustments AS "adjustments!: sqlx::types::Json<Vec<Adjustment>>"
         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::<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) {
+            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))

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::<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>, E>
+    where
+        P: ConnectionProvider,
+        F: FnOnce(Vec<K>) -> Fut,
+        Fut: Future<Output = Result<DashMap<K, T>, E>>,
+        E: From<Error>,
+        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<HashMap<K, T>, 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();

packages/xredis/src/lib.rs

@@ -168,6 +168,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>, E>
+    where
+        F: FnOnce(Vec<K>) -> Fut,
+        Fut: Future<Output = Result<DashMap<K, T>, E>>,
+        E: From<Error>,
+        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,
@@ -233,6 +261,7 @@ impl RedisPool {
                 slug_namespace,
                 case_sensitive,
                 keys,
+                None,
                 closure,
             )
             .await