diff --git a/apps/labrinth/.sqlx/query-01ac9da2281dd65c1b4a3eb5a2b941ba117decea2e953f400bbcabb1ad873a3f.json b/apps/labrinth/.sqlx/query-01ac9da2281dd65c1b4a3eb5a2b941ba117decea2e953f400bbcabb1ad873a3f.json deleted file mode 100644 index 5c42f5cf07..0000000000 --- a/apps/labrinth/.sqlx/query-01ac9da2281dd65c1b4a3eb5a2b941ba117decea2e953f400bbcabb1ad873a3f.json +++ /dev/null @@ -1,34 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n SELECT\n date_available AS \"date_available!\",\n SUM(amount) FILTER (WHERE NOT provisional) finalized_sum,\n SUM(amount) FILTER (WHERE provisional) estimate_sum\n FROM (\n SELECT date_available, amount, FALSE provisional\n FROM payouts_values\n WHERE user_id = $1\n\n UNION ALL\n\n SELECT date_available, amount, TRUE provisional\n FROM payout_estimates\n WHERE user_id = $1\n ) payout_amounts\n GROUP BY date_available\n ORDER BY date_available DESC\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "date_available!", - "type_info": "Timestamptz" - }, - { - "ordinal": 1, - "name": "finalized_sum", - "type_info": "Numeric" - }, - { - "ordinal": 2, - "name": "estimate_sum", - "type_info": "Numeric" - } - ], - "parameters": { - "Left": [ - "Int8" - ] - }, - "nullable": [ - null, - null, - null - ] - }, - "hash": "01ac9da2281dd65c1b4a3eb5a2b941ba117decea2e953f400bbcabb1ad873a3f" -} diff --git a/apps/labrinth/.sqlx/query-1d68f3585ac26a38955abf04703e26538972770ea7454fab132da87dad747669.json b/apps/labrinth/.sqlx/query-1d68f3585ac26a38955abf04703e26538972770ea7454fab132da87dad747669.json deleted file mode 100644 index e38f03e144..0000000000 --- a/apps/labrinth/.sqlx/query-1d68f3585ac26a38955abf04703e26538972770ea7454fab132da87dad747669.json +++ /dev/null @@ -1,58 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n\t\t\tSELECT\n\t\t\t\tpayout_periods.period,\n\t\t\t\tpayout_periods.raw_actual_aditude_revenue_usd,\n\t\t\t\tpayout_periods.adjustments AS \"adjustments: Json>\",\n\t\t\t\t(\n\t\t\t\t\tSELECT payout_runs.payload\n\t\t\t\t\tFROM payout_runs\n\t\t\t\t\tWHERE payout_runs.period = payout_periods.period\n\t\t\t\t\t\tAND payout_runs.status IN ('scheduled', 'running')\n\t\t\t\t\tLIMIT 1\n\t\t\t\t) AS \"active_run_payload: Json\",\n\t\t\t\tEXISTS (\n\t\t\t\t\tSELECT 1\n\t\t\t\t\tFROM payout_runs\n\t\t\t\t\tWHERE payout_runs.period = payout_periods.period\n\t\t\t\t\t\tAND payout_runs.status = 'scheduled'\n\t\t\t\t) AS \"has_scheduled_run!\",\n\t\t\t\tEXISTS (\n\t\t\t\t\tSELECT 1\n\t\t\t\t\tFROM payout_runs\n\t\t\t\t\tWHERE payout_runs.period = payout_periods.period\n\t\t\t\t\t\tAND payout_runs.status = 'running'\n\t\t\t\t) AS \"has_running_run!\",\n\t\t\t\tEXISTS (\n\t\t\t\t\tSELECT 1\n\t\t\t\t\tFROM payout_runs\n\t\t\t\t\tWHERE payout_runs.period = payout_periods.period\n\t\t\t\t\t\tAND payout_runs.status = 'succeeded'\n\t\t\t\t) AS \"has_succeeded_run!\"\n\t\t\tFROM payout_periods\n\t\t\tWHERE payout_periods.period = ANY($1)\n\t\t\t", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "period", - "type_info": "Date" - }, - { - "ordinal": 1, - "name": "raw_actual_aditude_revenue_usd", - "type_info": "Numeric" - }, - { - "ordinal": 2, - "name": "adjustments: Json>", - "type_info": "Jsonb" - }, - { - "ordinal": 3, - "name": "active_run_payload: Json", - "type_info": "Jsonb" - }, - { - "ordinal": 4, - "name": "has_scheduled_run!", - "type_info": "Bool" - }, - { - "ordinal": 5, - "name": "has_running_run!", - "type_info": "Bool" - }, - { - "ordinal": 6, - "name": "has_succeeded_run!", - "type_info": "Bool" - } - ], - "parameters": { - "Left": [ - "DateArray" - ] - }, - "nullable": [ - false, - false, - false, - null, - null, - null, - null - ] - }, - "hash": "1d68f3585ac26a38955abf04703e26538972770ea7454fab132da87dad747669" -} diff --git a/apps/labrinth/.sqlx/query-203e4a78f0be92cea79a800018418ab35b74370e7f8d1216e5f7d0a3e1615a4b.json b/apps/labrinth/.sqlx/query-203e4a78f0be92cea79a800018418ab35b74370e7f8d1216e5f7d0a3e1615a4b.json new file mode 100644 index 0000000000..a7c96f551f --- /dev/null +++ b/apps/labrinth/.sqlx/query-203e4a78f0be92cea79a800018418ab35b74370e7f8d1216e5f7d0a3e1615a4b.json @@ -0,0 +1,18 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO payout_estimates (\n period,\n user_id,\n mod_id,\n amount,\n created,\n date_available\n )\n VALUES ($1, $2, $3, 30, $4, $5)\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Date", + "Int8", + "Int8", + "Timestamptz", + "Timestamptz" + ] + }, + "nullable": [] + }, + "hash": "203e4a78f0be92cea79a800018418ab35b74370e7f8d1216e5f7d0a3e1615a4b" +} diff --git a/apps/labrinth/.sqlx/query-291b7a70925bdd5d748048fa08dea1901ce8c2b584ec86921efe6c4d2d279b29.json b/apps/labrinth/.sqlx/query-291b7a70925bdd5d748048fa08dea1901ce8c2b584ec86921efe6c4d2d279b29.json new file mode 100644 index 0000000000..eadd7ae8f1 --- /dev/null +++ b/apps/labrinth/.sqlx/query-291b7a70925bdd5d748048fa08dea1901ce8c2b584ec86921efe6c4d2d279b29.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO payout_periods (\n period,\n raw_actual_aditude_revenue_usd,\n adjustments\n )\n VALUES ($1, 100, '[]'::jsonb)\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Date" + ] + }, + "nullable": [] + }, + "hash": "291b7a70925bdd5d748048fa08dea1901ce8c2b584ec86921efe6c4d2d279b29" +} diff --git a/apps/labrinth/.sqlx/query-3962c046a8877433f2e5afd6064c0dfb11c7093d60093e8265d8cc5937c2c02c.json b/apps/labrinth/.sqlx/query-3962c046a8877433f2e5afd6064c0dfb11c7093d60093e8265d8cc5937c2c02c.json deleted file mode 100644 index 2c8bb1eeaa..0000000000 --- a/apps/labrinth/.sqlx/query-3962c046a8877433f2e5afd6064c0dfb11c7093d60093e8265d8cc5937c2c02c.json +++ /dev/null @@ -1,20 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n SELECT MAX(date)\n FROM payout_period_days\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "max", - "type_info": "Date" - } - ], - "parameters": { - "Left": [] - }, - "nullable": [ - null - ] - }, - "hash": "3962c046a8877433f2e5afd6064c0dfb11c7093d60093e8265d8cc5937c2c02c" -} diff --git a/apps/labrinth/.sqlx/query-56e3d8b2961562ef5a8be80f71814743f0cd95a7c26e4d0dd5b354d2ac3bbaf1.json b/apps/labrinth/.sqlx/query-56e3d8b2961562ef5a8be80f71814743f0cd95a7c26e4d0dd5b354d2ac3bbaf1.json new file mode 100644 index 0000000000..fa1bbf6566 --- /dev/null +++ b/apps/labrinth/.sqlx/query-56e3d8b2961562ef5a8be80f71814743f0cd95a7c26e4d0dd5b354d2ac3bbaf1.json @@ -0,0 +1,52 @@ +{ + "db_name": "PostgreSQL", + "query": "\n\t\t\tSELECT\n\t\t\t\tpayout_periods.period,\n\t\t\t\tpayout_periods.raw_actual_aditude_revenue_usd,\n\t\t\t\tpayout_periods.adjustments AS \"adjustments: Json>\",\n\t\t\t\tactive_run.payload AS \"active_run_payload: Json\",\n\t\t\t\tactive_run.execute_at AS active_run_execute_at,\n\t\t\t\tEXISTS (\n\t\t\t\t\tSELECT 1\n\t\t\t\t\tFROM payout_runs\n\t\t\t\t\tWHERE payout_runs.period = payout_periods.period\n\t\t\t\t\t\tAND payout_runs.status = 'succeeded'\n\t\t\t\t) AS \"has_succeeded_run!\"\n\t\t\tFROM payout_periods\n\t\t\tLEFT JOIN payout_runs active_run\n\t\t\t\tON active_run.period = payout_periods.period\n\t\t\t\tAND active_run.status IN ('scheduled', 'running')\n\t\t\tWHERE payout_periods.period = ANY($1)\n\t\t\t", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "period", + "type_info": "Date" + }, + { + "ordinal": 1, + "name": "raw_actual_aditude_revenue_usd", + "type_info": "Numeric" + }, + { + "ordinal": 2, + "name": "adjustments: Json>", + "type_info": "Jsonb" + }, + { + "ordinal": 3, + "name": "active_run_payload: Json", + "type_info": "Jsonb" + }, + { + "ordinal": 4, + "name": "active_run_execute_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 5, + "name": "has_succeeded_run!", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "DateArray" + ] + }, + "nullable": [ + false, + false, + false, + true, + true, + null + ] + }, + "hash": "56e3d8b2961562ef5a8be80f71814743f0cd95a7c26e4d0dd5b354d2ac3bbaf1" +} diff --git a/apps/labrinth/.sqlx/query-7707d9dba71fde632b52807578af564ffc242b8ec5d081fe1f9272158658e6e1.json b/apps/labrinth/.sqlx/query-7707d9dba71fde632b52807578af564ffc242b8ec5d081fe1f9272158658e6e1.json new file mode 100644 index 0000000000..e59e025afe --- /dev/null +++ b/apps/labrinth/.sqlx/query-7707d9dba71fde632b52807578af564ffc242b8ec5d081fe1f9272158658e6e1.json @@ -0,0 +1,19 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO payouts_values (\n user_id,\n mod_id,\n amount,\n created,\n date_available\n )\n VALUES\n ($1, $2, 70, $3, $4),\n ($1, $2, 30, $5, $6)\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int8", + "Int8", + "Timestamptz", + "Timestamptz", + "Timestamptz", + "Timestamptz" + ] + }, + "nullable": [] + }, + "hash": "7707d9dba71fde632b52807578af564ffc242b8ec5d081fe1f9272158658e6e1" +} diff --git a/apps/labrinth/.sqlx/query-a8928c794adfdaea6673e8dfb9040b0084e18ec880d11c05005c6eb7652e9f20.json b/apps/labrinth/.sqlx/query-a8928c794adfdaea6673e8dfb9040b0084e18ec880d11c05005c6eb7652e9f20.json new file mode 100644 index 0000000000..5f95b43e61 --- /dev/null +++ b/apps/labrinth/.sqlx/query-a8928c794adfdaea6673e8dfb9040b0084e18ec880d11c05005c6eb7652e9f20.json @@ -0,0 +1,17 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO payouts_values (\n user_id,\n mod_id,\n amount,\n created,\n date_available,\n payout_run_id\n )\n VALUES ($1, $2, 80, $3, $4, 20002)\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int8", + "Int8", + "Timestamptz", + "Timestamptz" + ] + }, + "nullable": [] + }, + "hash": "a8928c794adfdaea6673e8dfb9040b0084e18ec880d11c05005c6eb7652e9f20" +} diff --git a/apps/labrinth/.sqlx/query-b0f5c8d0879b5d797a4aeaa6b80d040cc1a57429516061ae3ce33ee19fcb04e3.json b/apps/labrinth/.sqlx/query-b0f5c8d0879b5d797a4aeaa6b80d040cc1a57429516061ae3ce33ee19fcb04e3.json new file mode 100644 index 0000000000..cb69b2752e --- /dev/null +++ b/apps/labrinth/.sqlx/query-b0f5c8d0879b5d797a4aeaa6b80d040cc1a57429516061ae3ce33ee19fcb04e3.json @@ -0,0 +1,17 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO payouts_values (\n user_id,\n mod_id,\n amount,\n created,\n date_available,\n payout_run_id\n )\n VALUES ($1, $2, 40, $3, $4, 20001)\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int8", + "Int8", + "Timestamptz", + "Timestamptz" + ] + }, + "nullable": [] + }, + "hash": "b0f5c8d0879b5d797a4aeaa6b80d040cc1a57429516061ae3ce33ee19fcb04e3" +} diff --git a/apps/labrinth/.sqlx/query-c30e4f1aad62a3a8c278340f051e91476a4fbe3d7774ef9fed712929a5816ecb.json b/apps/labrinth/.sqlx/query-c30e4f1aad62a3a8c278340f051e91476a4fbe3d7774ef9fed712929a5816ecb.json new file mode 100644 index 0000000000..3bcaaf94d6 --- /dev/null +++ b/apps/labrinth/.sqlx/query-c30e4f1aad62a3a8c278340f051e91476a4fbe3d7774ef9fed712929a5816ecb.json @@ -0,0 +1,19 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO payout_runs (\n id,\n period,\n payload,\n status,\n started_at,\n started_by,\n execute_at,\n processing_started_at,\n finished_at\n )\n VALUES (\n $1,\n $2,\n '{\"raw_actual_revenue_usd\":100,\"adjustments\":[]}'::jsonb,\n 'succeeded',\n $3,\n $4,\n $5,\n $5,\n $6\n )\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int8", + "Date", + "Timestamptz", + "Int8", + "Timestamptz", + "Timestamptz" + ] + }, + "nullable": [] + }, + "hash": "c30e4f1aad62a3a8c278340f051e91476a4fbe3d7774ef9fed712929a5816ecb" +} diff --git a/apps/labrinth/.sqlx/query-d3ccc0901d209bc77843ab8d8d3c7dc5159b79d99010a7cd145beed301d24b36.json b/apps/labrinth/.sqlx/query-d3ccc0901d209bc77843ab8d8d3c7dc5159b79d99010a7cd145beed301d24b36.json new file mode 100644 index 0000000000..23ab73239f --- /dev/null +++ b/apps/labrinth/.sqlx/query-d3ccc0901d209bc77843ab8d8d3c7dc5159b79d99010a7cd145beed301d24b36.json @@ -0,0 +1,46 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT\n date_available AS \"date_available!\",\n SUM(amount) FILTER (WHERE NOT provisional) finalized_sum,\n SUM(amount) FILTER (WHERE provisional) estimate_sum,\n SUM(amount) FILTER (WHERE authoritative) authoritative_sum,\n SUM(amount) FILTER (\n WHERE provisional AND authoritative\n ) actionable_estimate_sum\n FROM (\n SELECT\n payouts_values.date_available,\n payouts_values.amount,\n FALSE provisional,\n payouts_values.payout_run_id IS NULL\n OR payout_runs.status = 'succeeded' authoritative\n FROM payouts_values\n LEFT JOIN payout_runs\n ON payout_runs.id = payouts_values.payout_run_id\n WHERE payouts_values.user_id = $1\n\n UNION ALL\n\n SELECT\n payout_estimates.date_available,\n payout_estimates.amount,\n TRUE provisional,\n succeeded_periods.period IS NULL authoritative\n FROM payout_estimates\n LEFT JOIN (\n SELECT period\n FROM payout_runs\n WHERE status = 'succeeded'\n ) succeeded_periods\n ON succeeded_periods.period = payout_estimates.period\n WHERE payout_estimates.user_id = $1\n ) payout_amounts\n GROUP BY date_available\n ORDER BY date_available DESC\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "date_available!", + "type_info": "Timestamptz" + }, + { + "ordinal": 1, + "name": "finalized_sum", + "type_info": "Numeric" + }, + { + "ordinal": 2, + "name": "estimate_sum", + "type_info": "Numeric" + }, + { + "ordinal": 3, + "name": "authoritative_sum", + "type_info": "Numeric" + }, + { + "ordinal": 4, + "name": "actionable_estimate_sum", + "type_info": "Numeric" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + null, + null, + null, + null, + null + ] + }, + "hash": "d3ccc0901d209bc77843ab8d8d3c7dc5159b79d99010a7cd145beed301d24b36" +} diff --git a/apps/labrinth/.sqlx/query-f469644456f40b35ad4b56aec37bc8d02a346b77ccb033988aa2d28435d69732.json b/apps/labrinth/.sqlx/query-f469644456f40b35ad4b56aec37bc8d02a346b77ccb033988aa2d28435d69732.json new file mode 100644 index 0000000000..844b402665 --- /dev/null +++ b/apps/labrinth/.sqlx/query-f469644456f40b35ad4b56aec37bc8d02a346b77ccb033988aa2d28435d69732.json @@ -0,0 +1,18 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO payout_estimates (\n period,\n user_id,\n mod_id,\n amount,\n created,\n date_available\n )\n VALUES ($1, $2, $3, 100, $4, $5)\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Date", + "Int8", + "Int8", + "Timestamptz", + "Timestamptz" + ] + }, + "nullable": [] + }, + "hash": "f469644456f40b35ad4b56aec37bc8d02a346b77ccb033988aa2d28435d69732" +} diff --git a/apps/labrinth/src/database/models/payout_period_item.rs b/apps/labrinth/src/database/models/payout_period_item.rs index 9537d495ef..2dca554d90 100644 --- a/apps/labrinth/src/database/models/payout_period_item.rs +++ b/apps/labrinth/src/database/models/payout_period_item.rs @@ -1,6 +1,6 @@ use std::collections::HashMap; -use chrono::NaiveDate; +use chrono::{DateTime, NaiveDate, Utc}; use rust_decimal::Decimal; use sqlx::types::Json; @@ -13,9 +13,8 @@ pub struct DBPayoutPeriod { pub raw_actual_aditude_revenue_usd: Decimal, pub adjustments: Vec, pub active_run_payload: Option, + pub active_run_execute_at: Option>, pub days: Vec, - pub has_scheduled_run: bool, - pub has_running_run: bool, pub has_succeeded_run: bool, } @@ -41,25 +40,8 @@ impl DBPayoutPeriod { payout_periods.period, payout_periods.raw_actual_aditude_revenue_usd, payout_periods.adjustments AS "adjustments: Json>", - ( - SELECT payout_runs.payload - FROM payout_runs - WHERE payout_runs.period = payout_periods.period - AND payout_runs.status IN ('scheduled', 'running') - LIMIT 1 - ) AS "active_run_payload: Json", - EXISTS ( - SELECT 1 - FROM payout_runs - WHERE payout_runs.period = payout_periods.period - AND payout_runs.status = 'scheduled' - ) AS "has_scheduled_run!", - EXISTS ( - SELECT 1 - FROM payout_runs - WHERE payout_runs.period = payout_periods.period - AND payout_runs.status = 'running' - ) AS "has_running_run!", + active_run.payload AS "active_run_payload: Json", + active_run.execute_at AS active_run_execute_at, EXISTS ( SELECT 1 FROM payout_runs @@ -67,6 +49,9 @@ impl DBPayoutPeriod { AND payout_runs.status = 'succeeded' ) AS "has_succeeded_run!" FROM payout_periods + LEFT JOIN payout_runs active_run + ON active_run.period = payout_periods.period + AND active_run.status IN ('scheduled', 'running') WHERE payout_periods.period = ANY($1) "#, periods, @@ -87,9 +72,8 @@ impl DBPayoutPeriod { active_run_payload: row .active_run_payload .map(|payload| payload.0), + active_run_execute_at: row.active_run_execute_at, days: Vec::new(), - has_scheduled_run: row.has_scheduled_run, - has_running_run: row.has_running_run, has_succeeded_run: row.has_succeeded_run, }, ) diff --git a/apps/labrinth/src/queue/payout_run/estimate.rs b/apps/labrinth/src/queue/payout_run/estimate.rs index 48c5cfe72c..e3f8a742f3 100644 --- a/apps/labrinth/src/queue/payout_run/estimate.rs +++ b/apps/labrinth/src/queue/payout_run/estimate.rs @@ -89,6 +89,26 @@ pub async fn estimate( Ok(periods) } +/// Fetch fresh per-day estimated ad provider info for one payout period, +/// bypassing Redis. +pub async fn refresh_estimate( + aditude: &aditude::Client, + period: YearMonth, + start_date: NaiveDate, + end_date: NaiveDate, +) -> Result { + if start_date > end_date { + return Err(eyre::eyre!("estimate start date is after end date")); + } + + let estimates = + fetch_estimates(aditude, &[period], start_date, end_date).await?; + estimates + .remove(&period) + .map(|(_, estimate)| estimate) + .wrap_err("missing refreshed payout estimate") +} + async fn fetch_estimates( aditude: &aditude::Client, periods: &[YearMonth], diff --git a/apps/labrinth/src/routes/internal/mod.rs b/apps/labrinth/src/routes/internal/mod.rs index e8c207d67c..266ef1f98b 100644 --- a/apps/labrinth/src/routes/internal/mod.rs +++ b/apps/labrinth/src/routes/internal/mod.rs @@ -186,6 +186,7 @@ pub fn config(cfg: &mut web::ServiceConfig) { analytics_event::analytics_event_edit, analytics_event::analytics_event_delete, payout_runs::get_runs, + payout_runs::calculate_run, payout_runs::start_run, payout_runs::cancel_runs, ), diff --git a/apps/labrinth/src/routes/internal/payout_runs/admin.rs b/apps/labrinth/src/routes/internal/payout_runs/admin.rs index 6bdeb65add..8aae8cdc75 100644 --- a/apps/labrinth/src/routes/internal/payout_runs/admin.rs +++ b/apps/labrinth/src/routes/internal/payout_runs/admin.rs @@ -4,7 +4,7 @@ use rust_decimal::Decimal; use serde::{Deserialize, Serialize}; use xredis::RedisPool; -use super::{Adjustment, PayoutRunPayload}; +use super::{Adjustment, PayoutRunDay, PayoutRunPayload}; use crate::{ auth::{ AuthenticationError, get_user_from_headers, two_factor::verify_2fa_code, @@ -14,11 +14,16 @@ use crate::{ models::{ DBUserId, DatabaseError, generate_payout_run_id, payout_run_item::{DBPayoutRun, PayoutRunStatus}, + payout_variance_item::DBPayoutVariance, }, }, models::{ids::PayoutRunId, pats::Scopes}, queue::{ - payout_run::{estimate, validate_complete_period_estimate}, + payout_run::{ + PayoutVariance, PayoutVariances, compute_actual_distribution_flow, + distribution_for_day, refresh_estimate, + validate_complete_period_estimate, + }, session::AuthQueue, }, routes::ApiError, @@ -49,6 +54,136 @@ pub struct StartPayoutRunResponse { pub execute_at: DateTime, } +#[derive(Debug, Serialize, utoipa::ToSchema)] +pub struct CalculatePayoutRunResponse { + pub period: YearMonth, + pub days: Vec, + /// Sum of all adjustments that would be applied on top of actual revenue. + #[serde(with = "rust_decimal::serde::float")] + pub total_adjustments: Decimal, + pub adjustments: Vec, +} + +/// Calculate a payout run without scheduling it. +/// +/// Admin-only. This fetches fresh Aditude data and does not require TOTP. +#[utoipa::path( + tag = "payout runs", + request_body = StartPayoutRun, + responses( + (status = OK, body = CalculatePayoutRunResponse), + (status = BAD_REQUEST, description = "Invalid payout run input"), + (status = UNAUTHORIZED, description = "Invalid authentication"), + (status = FAILED_DEPENDENCY, description = "Aditude returned an incomplete period"), + ), + security(("bearer_auth" = ["SESSION_ACCESS"])), +)] +#[post("/calculate")] +pub async fn calculate_run( + req: HttpRequest, + pool: web::Data, + redis: web::Data, + aditude: web::Data, + session_queue: web::Data, + web::Json(body): web::Json, +) -> Result, ApiError> { + let user = get_user_from_headers( + &req, + &**pool, + &redis, + &session_queue, + Scopes::SESSION_ACCESS, + ) + .await + .wrap_auth_err("authenticating API request")? + .1; + if !user.role.is_admin() { + return Err(ApiError::Auth(eyre::eyre!( + AuthenticationError::InvalidCredentials, + ))); + } + + if body.raw_actual_revenue_usd.is_sign_negative() { + return Err(ApiError::Request(eyre::eyre!( + "`raw_actual_revenue_usd` cannot be negative", + ))); + } + + let estimate_end_date = body + .period + .date() + .checked_add_months(Months::new(1)) + .and_then(|date| date.pred_opt()) + .wrap_internal_err("calculating payout estimate end date")?; + let estimate = refresh_estimate( + aditude.get_ref(), + body.period, + body.period.date(), + estimate_end_date, + ) + .await + .wrap_internal_err("fetching payout estimate")?; + validate_complete_period_estimate(&estimate) + .wrap_failed_dependency_err("validating payout estimate")?; + + let stored_variances = DBPayoutVariance::get_all(&**pool) + .await + .wrap_internal_err("fetching payout variances")?; + let default_variance = stored_variances + .first() + .wrap_internal_err("no payout variance configured")? + .variance; + let variances = PayoutVariances { + default_frac: default_variance, + fracs: stored_variances + .into_iter() + .map(|variance| PayoutVariance { + starts_at: variance.applied_on, + frac: variance.variance, + }) + .collect(), + }; + let total_estimated_revenue_usd = estimate + .days + .iter() + .map(|day| day.raw_estimated_revenue_usd) + .sum(); + let actual_flow = compute_actual_distribution_flow( + total_estimated_revenue_usd, + body.raw_actual_revenue_usd, + ); + let days = estimate + .days + .into_iter() + .map(|day| PayoutRunDay { + date: day.date, + estimated: distribution_for_day( + day.date, + day.raw_estimated_revenue_usd, + day.impressions, + &variances, + ), + actual: Some(actual_flow.distribution_for_day( + day.date, + day.raw_estimated_revenue_usd, + day.impressions, + )), + }) + .collect(); + let total_adjustments = body + .adjustments + .iter() + .map(|adjustment| adjustment.amount_usd) + .sum(); + + Ok(web::Json(CalculatePayoutRunResponse { + period: body.period, + days, + total_adjustments, + adjustments: body.adjustments, + })) +} + /// Start a payout run. /// /// Admin-only. @@ -143,18 +278,14 @@ pub async fn start_run( .checked_add_months(Months::new(1)) .and_then(|date| date.pred_opt()) .wrap_internal_err("calculating payout estimate end date")?; - let mut estimates = estimate( + let estimate = refresh_estimate( aditude.get_ref(), - redis.get_ref(), - &[body.period], + body.period, body.period.date(), estimate_end_date, ) .await .wrap_internal_err("fetching payout estimate")?; - let estimate = estimates - .pop() - .wrap_internal_err("missing requested payout estimate")?; validate_complete_period_estimate(&estimate) .wrap_failed_dependency_err("validating payout estimate")?; let payload = PayoutRunPayload { diff --git a/apps/labrinth/src/routes/internal/payout_runs/fetch.rs b/apps/labrinth/src/routes/internal/payout_runs/fetch.rs index 82cc7b8392..d8b8110c5b 100644 --- a/apps/labrinth/src/routes/internal/payout_runs/fetch.rs +++ b/apps/labrinth/src/routes/internal/payout_runs/fetch.rs @@ -1,7 +1,7 @@ use std::collections::HashMap; use actix_web::{HttpRequest, get, web}; -use chrono::{Months, NaiveDate, Utc}; +use chrono::{DateTime, Months, NaiveDate, Utc}; use rust_decimal::Decimal; use serde::{Deserialize, Serialize}; use xredis::RedisPool; @@ -40,6 +40,8 @@ pub struct PayoutRuns { pub struct PayoutRunPeriod { pub period: YearMonth, pub status: PayoutPeriodStatus, + /// When the active payout run is scheduled to begin executing. + pub runs_at: Option>, pub days: Vec, /// Sum of all adjustments applied on top of actual revenue. #[serde(with = "rust_decimal::serde::float")] @@ -65,8 +67,6 @@ pub enum PayoutPeriodStatus { InReview, /// A payout run is waiting for its cancellation window to expire. Scheduled, - /// Payout run is currently executing. - Running, /// Payout run has been paid out to creators. Paid, } @@ -147,15 +147,6 @@ pub async fn get_runs( .into_iter() .map(|period| (period.period, period)) .collect::>(); - let newest_stored_day = sqlx::query_scalar!( - r#" - SELECT MAX(date) - FROM payout_period_days - "#, - ) - .fetch_one(&**pool) - .await - .wrap_internal_err("fetching newest stored payout period day")?; let stored_variances = DBPayoutVariance::get_all(&**pool) .await .wrap_internal_err("fetching payout variances")?; @@ -182,26 +173,25 @@ pub async fn get_runs( .is_none_or(|stored| stored.days.is_empty()) }) .collect::>(); - let mut live_estimates = - if let Some(first_live_period) = live_periods.first() { - let start_date = newest_stored_day - .unwrap_or_else(|| first_live_period.date()) - .min(current_date); - estimate( - aditude.get_ref(), - redis.get_ref(), - &live_periods, - start_date, - current_date, - ) - .await - .wrap_internal_err("fetching payout estimates")? - .into_iter() - .map(|estimate| (estimate.period, estimate)) - .collect::>() - } else { - HashMap::new() - }; + let mut live_estimates = if let Some(first_requested_period) = + requested_periods.first() + && !live_periods.is_empty() + { + estimate( + aditude.get_ref(), + redis.get_ref(), + &live_periods, + first_requested_period.date(), + current_date, + ) + .await + .wrap_internal_err("fetching payout estimates")? + .into_iter() + .map(|estimate| (estimate.period, estimate)) + .collect::>() + } else { + HashMap::new() + }; let periods = requested_periods .into_iter() @@ -257,9 +247,7 @@ pub async fn get_runs( .collect::, _>>()?; let status = if period.has_succeeded_run { PayoutPeriodStatus::Paid - } else if period.has_running_run { - PayoutPeriodStatus::Running - } else if period.has_scheduled_run { + } else if period.active_run_execute_at.is_some() { PayoutPeriodStatus::Scheduled } else { PayoutPeriodStatus::InReview @@ -268,6 +256,7 @@ pub async fn get_runs( Ok(PayoutRunPeriod { period: requested_period, status, + runs_at: period.active_run_execute_at, days, total_adjustments, adjustments, @@ -302,6 +291,7 @@ pub async fn get_runs( Ok(PayoutRunPeriod { period: requested_period, status, + runs_at: None, days, total_adjustments: Decimal::ZERO, adjustments: is_admin.then(Vec::new), diff --git a/apps/labrinth/src/routes/internal/payout_runs/mod.rs b/apps/labrinth/src/routes/internal/payout_runs/mod.rs index a35bf9da73..53ea9a38a1 100644 --- a/apps/labrinth/src/routes/internal/payout_runs/mod.rs +++ b/apps/labrinth/src/routes/internal/payout_runs/mod.rs @@ -7,6 +7,7 @@ pub use fetch::*; pub fn config(cfg: &mut actix_web::web::ServiceConfig) { cfg.service(get_runs) + .service(calculate_run) .service(start_run) .service(cancel_runs); } diff --git a/apps/labrinth/src/routes/v3/payouts.rs b/apps/labrinth/src/routes/v3/payouts.rs index 6153ae2f0e..dedb73048b 100644 --- a/apps/labrinth/src/routes/v3/payouts.rs +++ b/apps/labrinth/src/routes/v3/payouts.rs @@ -1031,16 +1031,25 @@ pub struct UserBalance { pub available: Decimal, pub withdrawn_lifetime: Decimal, pub withdrawn_ytd: Decimal, - /// Finalized revenue not yet available and provisional creator estimates. + /// Finalized revenue not yet available and provisional creator estimates + /// for periods without a succeeded payout run. /// - /// Provisional estimates remain pending regardless of their estimated - /// availability date and may change before finalization. + /// Unfinalized estimates remain pending regardless of their estimated + /// availability date and may change before finalization. Retained + /// historical estimates do not remain actionable after finalization. pub pending: Decimal, /// Revenue grouped by its finalized or estimated availability date. /// - /// These groups may include provisional creator estimates that can change - /// before finalization. + /// Each payout period contributes either its finalized revenue, when it + /// has a succeeded payout run, or its provisional estimate otherwise. pub dates: HashMap, Decimal>, + /// Provisional creator estimates grouped by estimated availability date. + /// + /// Retained historical estimates remain visible after finalization and + /// may differ from the corresponding finalized revenue. + pub estimated_dates: HashMap, Decimal>, + /// Finalized revenue grouped by availability date. + pub actual_dates: HashMap, Decimal>, } #[derive(Serialize, utoipa::ToSchema)] @@ -1127,17 +1136,38 @@ async fn get_user_balance( SELECT date_available AS "date_available!", SUM(amount) FILTER (WHERE NOT provisional) finalized_sum, - SUM(amount) FILTER (WHERE provisional) estimate_sum + SUM(amount) FILTER (WHERE provisional) estimate_sum, + SUM(amount) FILTER (WHERE authoritative) authoritative_sum, + SUM(amount) FILTER ( + WHERE provisional AND authoritative + ) actionable_estimate_sum FROM ( - SELECT date_available, amount, FALSE provisional + SELECT + payouts_values.date_available, + payouts_values.amount, + FALSE provisional, + payouts_values.payout_run_id IS NULL + OR payout_runs.status = 'succeeded' authoritative FROM payouts_values - WHERE user_id = $1 + LEFT JOIN payout_runs + ON payout_runs.id = payouts_values.payout_run_id + WHERE payouts_values.user_id = $1 UNION ALL - SELECT date_available, amount, TRUE provisional + SELECT + payout_estimates.date_available, + payout_estimates.amount, + TRUE provisional, + succeeded_periods.period IS NULL authoritative FROM payout_estimates - WHERE user_id = $1 + LEFT JOIN ( + SELECT period + FROM payout_runs + WHERE status = 'succeeded' + ) succeeded_periods + ON succeeded_periods.period = payout_estimates.period + WHERE payout_estimates.user_id = $1 ) payout_amounts GROUP BY date_available ORDER BY date_available DESC @@ -1161,7 +1191,7 @@ async fn get_user_balance( } else { Decimal::ZERO }; - acc + finalized + x.estimate_sum.unwrap_or(Decimal::ZERO) + acc + finalized + x.actionable_estimate_sum.unwrap_or(Decimal::ZERO) }); let withdrawn = sqlx::query!( @@ -1196,12 +1226,20 @@ async fn get_user_balance( pending, dates: payouts .iter() - .map(|x| { - ( - x.date_available, - x.finalized_sum.unwrap_or(Decimal::ZERO) - + x.estimate_sum.unwrap_or(Decimal::ZERO), - ) + .filter_map(|x| { + x.authoritative_sum.map(|amount| (x.date_available, amount)) + }) + .collect(), + estimated_dates: payouts + .iter() + .filter_map(|x| { + x.estimate_sum.map(|amount| (x.date_available, amount)) + }) + .collect(), + actual_dates: payouts + .iter() + .filter_map(|x| { + x.finalized_sum.map(|amount| (x.date_available, amount)) }) .collect(), }) diff --git a/apps/labrinth/tests/payout_estimates.rs b/apps/labrinth/tests/payout_estimates.rs index c1c0d5f6b1..a033d68782 100644 --- a/apps/labrinth/tests/payout_estimates.rs +++ b/apps/labrinth/tests/payout_estimates.rs @@ -21,6 +21,8 @@ struct TestBalance { withdrawn_ytd: Decimal, pending: Decimal, dates: std::collections::HashMap, Decimal>, + estimated_dates: std::collections::HashMap, Decimal>, + actual_dates: std::collections::HashMap, Decimal>, } async fn get_balance(test_env: &TestEnvironment) -> TestBalance { @@ -33,6 +35,383 @@ async fn get_balance(test_env: &TestEnvironment) -> TestBalance { test::read_body_json(response).await } +fn amount_delta( + current: &std::collections::HashMap, Decimal>, + initial: &std::collections::HashMap, Decimal>, + date: DateTime, +) -> Decimal { + current.get(&date).copied().unwrap_or_default() + - initial.get(&date).copied().unwrap_or_default() +} + +async fn insert_succeeded_period( + pool: &sqlx::PgPool, + period: NaiveDate, + run_id: i64, +) { + sqlx::query!( + r#" + INSERT INTO payout_periods ( + period, + raw_actual_aditude_revenue_usd, + adjustments + ) + VALUES ($1, 100, '[]'::jsonb) + "#, + period, + ) + .execute(pool) + .await + .unwrap(); + + let started_at = Utc.with_ymd_and_hms(2001, 1, 1, 0, 0, 0).unwrap(); + sqlx::query!( + r#" + INSERT INTO payout_runs ( + id, + period, + payload, + status, + started_at, + started_by, + execute_at, + processing_started_at, + finished_at + ) + VALUES ( + $1, + $2, + '{"raw_actual_revenue_usd":100,"adjustments":[]}'::jsonb, + 'succeeded', + $3, + $4, + $5, + $5, + $6 + ) + "#, + run_id, + period, + started_at, + USER_USER_ID_PARSED, + started_at + Duration::minutes(2), + started_at + Duration::minutes(3), + ) + .execute(pool) + .await + .unwrap(); +} + +#[actix_rt::test] +async fn estimate_only_period_uses_estimate_as_pending_revenue() { + with_test_environment( + None, + |test_env: TestEnvironment| async move { + let pool = &test_env.db.pool; + let project_id = + parse_base62(&test_env.dummy.project_alpha.project_id).unwrap() + as i64; + let now = Utc.timestamp_opt(Utc::now().timestamp(), 0).unwrap(); + let date_available = now - Duration::days(20); + let initial_balance = get_balance(&test_env).await; + + sqlx::query!( + r#" + INSERT INTO payout_estimates ( + period, + user_id, + mod_id, + amount, + created, + date_available + ) + VALUES ($1, $2, $3, 30, $4, $5) + "#, + NaiveDate::from_ymd_opt(2001, 1, 1).unwrap(), + USER_USER_ID_PARSED, + project_id, + now - Duration::days(50), + date_available, + ) + .execute(pool) + .await + .unwrap(); + + let balance = get_balance(&test_env).await; + assert_eq!(balance.available, initial_balance.available); + assert_eq!( + balance.pending, + initial_balance.pending + Decimal::from(30), + ); + assert_eq!( + amount_delta( + &balance.dates, + &initial_balance.dates, + date_available, + ), + Decimal::from(30), + ); + assert_eq!( + amount_delta( + &balance.estimated_dates, + &initial_balance.estimated_dates, + date_available, + ), + Decimal::from(30), + ); + assert_eq!( + amount_delta( + &balance.actual_dates, + &initial_balance.actual_dates, + date_available, + ), + Decimal::ZERO, + ); + }, + ) + .await; +} + +#[actix_rt::test] +async fn finalized_only_period_uses_actual_as_pending_revenue() { + with_test_environment( + None, + |test_env: TestEnvironment| async move { + let pool = &test_env.db.pool; + let project_id = + parse_base62(&test_env.dummy.project_alpha.project_id).unwrap() + as i64; + let now = Utc.timestamp_opt(Utc::now().timestamp(), 0).unwrap(); + let date_available = now + Duration::days(20); + let period = NaiveDate::from_ymd_opt(2001, 2, 1).unwrap(); + let initial_balance = get_balance(&test_env).await; + insert_succeeded_period(pool, period, 20_001).await; + + sqlx::query!( + r#" + INSERT INTO payouts_values ( + user_id, + mod_id, + amount, + created, + date_available, + payout_run_id + ) + VALUES ($1, $2, 40, $3, $4, 20001) + "#, + USER_USER_ID_PARSED, + project_id, + now - Duration::days(50), + date_available, + ) + .execute(pool) + .await + .unwrap(); + + let balance = get_balance(&test_env).await; + assert_eq!(balance.available, initial_balance.available); + assert_eq!( + balance.pending, + initial_balance.pending + Decimal::from(40), + ); + assert_eq!( + amount_delta( + &balance.dates, + &initial_balance.dates, + date_available, + ), + Decimal::from(40), + ); + assert_eq!( + amount_delta( + &balance.estimated_dates, + &initial_balance.estimated_dates, + date_available, + ), + Decimal::ZERO, + ); + assert_eq!( + amount_delta( + &balance.actual_dates, + &initial_balance.actual_dates, + date_available, + ), + Decimal::from(40), + ); + }, + ) + .await; +} + +#[actix_rt::test] +async fn finalized_period_retains_estimate_but_uses_actual_balance() { + with_test_environment( + None, + |test_env: TestEnvironment| async move { + let pool = &test_env.db.pool; + let project_id = + parse_base62(&test_env.dummy.project_alpha.project_id).unwrap() + as i64; + let now = Utc.timestamp_opt(Utc::now().timestamp(), 0).unwrap(); + let date_available = now - Duration::days(20); + let period = NaiveDate::from_ymd_opt(2001, 3, 1).unwrap(); + let initial_balance = get_balance(&test_env).await; + insert_succeeded_period(pool, period, 20_002).await; + + sqlx::query!( + r#" + INSERT INTO payout_estimates ( + period, + user_id, + mod_id, + amount, + created, + date_available + ) + VALUES ($1, $2, $3, 100, $4, $5) + "#, + period, + USER_USER_ID_PARSED, + project_id, + now - Duration::days(50), + date_available, + ) + .execute(pool) + .await + .unwrap(); + sqlx::query!( + r#" + INSERT INTO payouts_values ( + user_id, + mod_id, + amount, + created, + date_available, + payout_run_id + ) + VALUES ($1, $2, 80, $3, $4, 20002) + "#, + USER_USER_ID_PARSED, + project_id, + now - Duration::days(50), + date_available, + ) + .execute(pool) + .await + .unwrap(); + + let balance = get_balance(&test_env).await; + assert_eq!( + balance.available, + initial_balance.available + Decimal::from(80), + ); + assert_eq!(balance.pending, initial_balance.pending); + assert_eq!( + amount_delta( + &balance.dates, + &initial_balance.dates, + date_available, + ), + Decimal::from(80), + ); + assert_eq!( + amount_delta( + &balance.estimated_dates, + &initial_balance.estimated_dates, + date_available, + ), + Decimal::from(100), + ); + assert_eq!( + amount_delta( + &balance.actual_dates, + &initial_balance.actual_dates, + date_available, + ), + Decimal::from(80), + ); + }, + ) + .await; +} + +#[actix_rt::test] +async fn legacy_finalized_values_remain_available_and_pending() { + with_test_environment( + None, + |test_env: TestEnvironment| async move { + let pool = &test_env.db.pool; + let project_id = + parse_base62(&test_env.dummy.project_alpha.project_id).unwrap() + as i64; + let now = Utc.timestamp_opt(Utc::now().timestamp(), 0).unwrap(); + let available_at = now - Duration::days(20); + let pending_at = now + Duration::days(20); + let initial_balance = get_balance(&test_env).await; + + sqlx::query!( + r#" + INSERT INTO payouts_values ( + user_id, + mod_id, + amount, + created, + date_available + ) + VALUES + ($1, $2, 70, $3, $4), + ($1, $2, 30, $5, $6) + "#, + USER_USER_ID_PARSED, + project_id, + now - Duration::days(50), + available_at, + now - Duration::days(10), + pending_at, + ) + .execute(pool) + .await + .unwrap(); + + let balance = get_balance(&test_env).await; + assert_eq!( + balance.available, + initial_balance.available + Decimal::from(70), + ); + assert_eq!( + balance.pending, + initial_balance.pending + Decimal::from(30), + ); + for (date, amount) in [ + (available_at, Decimal::from(70)), + (pending_at, Decimal::from(30)), + ] { + assert_eq!( + amount_delta(&balance.dates, &initial_balance.dates, date,), + amount, + ); + assert_eq!( + amount_delta( + &balance.actual_dates, + &initial_balance.actual_dates, + date, + ), + amount, + ); + assert_eq!( + amount_delta( + &balance.estimated_dates, + &initial_balance.estimated_dates, + date, + ), + Decimal::ZERO, + ); + } + }, + ) + .await; +} + #[actix_rt::test] async fn provisional_estimates_only_affect_pending_balance() { with_test_environment(