You've already forked AstralRinth
forked from didirus/AstralRinth
Xander
6fba33d443
* fix FK violation when inserting rows into `notifications_deliveries` * add test for FK violation when inserting into notifications_deliveries * sqlx prepare * add migration to prevent stale notifications from being dequeued all at once upon fix * Revert "add migration to prevent stale notifications from being dequeued all at once upon fix" This reverts commit 446f398752bbddb632196a549501f9ce0b2da67f.
1572 lines
49 KiB
Rust
1572 lines
49 KiB
Rust
use crate::database::models::notification_item::NotificationBuilder;
|
|
use crate::database::models::payouts_values_notifications;
|
|
use crate::database::redis::RedisPool;
|
|
use crate::database::{PgPool, PgTransaction};
|
|
use crate::env::ENV;
|
|
use crate::models::payouts::{
|
|
PayoutDecimal, PayoutInterval, PayoutMethod, PayoutMethodType,
|
|
TremendousForexResponse,
|
|
};
|
|
use crate::models::projects::MonetizationStatus;
|
|
use crate::routes::ApiError;
|
|
use crate::util::error::Context;
|
|
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 dashmap::DashMap;
|
|
use eyre::Result;
|
|
use futures::TryStreamExt;
|
|
use modrinth_util::decimal::Decimal2dp;
|
|
use muralpay::FiatAndRailCode;
|
|
use reqwest::Method;
|
|
use rust_decimal::Decimal;
|
|
use rust_decimal::prelude::ToPrimitive;
|
|
use serde::de::DeserializeOwned;
|
|
use serde::{Deserialize, Serialize};
|
|
use serde_json::Value;
|
|
use sqlx::postgres::PgQueryResult;
|
|
use std::collections::HashMap;
|
|
use tokio::sync::RwLock;
|
|
use tracing::{error, info, warn};
|
|
|
|
mod affiliate;
|
|
pub mod flow;
|
|
pub mod mural;
|
|
pub use affiliate::{
|
|
process_affiliate_payouts, remove_payouts_for_refunded_charges,
|
|
};
|
|
|
|
pub struct PayoutsQueue {
|
|
credential: RwLock<Option<PayPalCredentials>>,
|
|
payout_options: RwLock<Option<PayoutMethods>>,
|
|
pub muralpay: ArcSwapOption<MuralPayConfig>,
|
|
}
|
|
|
|
pub struct MuralPayConfig {
|
|
pub client: muralpay::Client,
|
|
pub source_account_id: muralpay::AccountId,
|
|
}
|
|
|
|
#[derive(Clone, Debug)]
|
|
struct PayPalCredentials {
|
|
access_token: String,
|
|
token_type: String,
|
|
expires: DateTime<Utc>,
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
struct PayoutMethods {
|
|
options: Vec<PayoutMethod>,
|
|
expires: DateTime<Utc>,
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
pub struct AccountBalance {
|
|
pub available: Decimal,
|
|
pub pending: Decimal,
|
|
}
|
|
|
|
impl Default for PayoutsQueue {
|
|
fn default() -> Self {
|
|
Self::new()
|
|
}
|
|
}
|
|
|
|
pub fn create_muralpay_client() -> Result<muralpay::Client> {
|
|
Ok(muralpay::Client::new(
|
|
&ENV.MURALPAY_API_URL,
|
|
ENV.MURALPAY_API_KEY.as_str(),
|
|
ENV.MURALPAY_TRANSFER_API_KEY.as_str(),
|
|
))
|
|
}
|
|
|
|
pub fn create_muralpay() -> Result<MuralPayConfig> {
|
|
let client = create_muralpay_client()?;
|
|
Ok(MuralPayConfig {
|
|
client,
|
|
source_account_id: ENV.MURALPAY_SOURCE_ACCOUNT_ID,
|
|
})
|
|
}
|
|
|
|
fn create_muralpay_methods() -> Vec<PayoutMethod> {
|
|
let all_countries = rust_iso3166::ALL
|
|
.iter()
|
|
.map(|x| x.alpha2)
|
|
.collect::<Vec<_>>();
|
|
|
|
let currencies = vec![
|
|
(
|
|
"blockchain_usdc_polygon",
|
|
"USDC on Polygon",
|
|
all_countries,
|
|
None,
|
|
),
|
|
("fiat_mxn", "MXN", vec!["MX"], Some(FiatAndRailCode::Mxn)),
|
|
("fiat_brl", "BRL", vec!["BR"], Some(FiatAndRailCode::Brl)),
|
|
("fiat_clp", "CLP", vec!["CL"], Some(FiatAndRailCode::Clp)),
|
|
("fiat_crc", "CRC", vec!["CR"], Some(FiatAndRailCode::Crc)),
|
|
("fiat_pen", "PEN", vec!["PE"], Some(FiatAndRailCode::Pen)),
|
|
// ("fiat_dop", "DOP"), // unsupported in API
|
|
// ("fiat_uyu", "UYU"), // unsupported in API
|
|
("fiat_ars", "ARS", vec!["AR"], Some(FiatAndRailCode::Ars)),
|
|
("fiat_cop", "COP", vec!["CO"], Some(FiatAndRailCode::Cop)),
|
|
("fiat_usd", "USD", vec!["US"], Some(FiatAndRailCode::Usd)),
|
|
(
|
|
"fiat_usd-peru",
|
|
"USD Peru",
|
|
vec!["PE"],
|
|
Some(FiatAndRailCode::UsdPeru),
|
|
),
|
|
// ("fiat_usd-panama", "USD Panama"), // by request
|
|
(
|
|
"fiat_eur",
|
|
"EUR",
|
|
vec![
|
|
"DE", "FR", "IT", "ES", "NL", "BE", "AT", "PT", "FI", "IE",
|
|
"GR", "LU", "CY", "MT", "SK", "SI", "EE", "LV", "LT",
|
|
],
|
|
Some(FiatAndRailCode::Eur),
|
|
),
|
|
];
|
|
|
|
currencies
|
|
.into_iter()
|
|
.map(
|
|
|(id, currency, countries, fiat_and_rail_code)| PayoutMethod {
|
|
id: id.to_string(),
|
|
type_: PayoutMethodType::MuralPay,
|
|
name: format!("Mural Pay - {currency}"),
|
|
category: None,
|
|
supported_countries: countries
|
|
.iter()
|
|
.map(|s| s.to_string())
|
|
.collect(),
|
|
image_url: None,
|
|
image_logo_url: None,
|
|
interval: PayoutInterval::Standard {
|
|
min: if let Some(fiat_and_rail_code) = fiat_and_rail_code {
|
|
flow::mural::min_usd_fiat(fiat_and_rail_code)
|
|
} else {
|
|
flow::mural::MIN_USD_BLOCKCHAIN
|
|
}
|
|
.get(),
|
|
max: flow::mural::MAX_USD.get(),
|
|
},
|
|
currency_code: None,
|
|
exchange_rate: None,
|
|
},
|
|
)
|
|
.collect()
|
|
}
|
|
|
|
// Batches payouts and handles token refresh
|
|
impl PayoutsQueue {
|
|
pub fn new() -> Self {
|
|
let muralpay = create_muralpay()
|
|
.inspect_err(|err| {
|
|
warn!("Failed to create Mural Pay client: {err:#?}")
|
|
})
|
|
.ok();
|
|
|
|
PayoutsQueue {
|
|
credential: RwLock::new(None),
|
|
payout_options: RwLock::new(None),
|
|
muralpay: ArcSwapOption::from_pointee(muralpay),
|
|
}
|
|
}
|
|
|
|
async fn refresh_token(&self) -> Result<PayPalCredentials, ApiError> {
|
|
let mut creds = self.credential.write().await;
|
|
let client = reqwest::Client::new();
|
|
|
|
let combined_key =
|
|
format!("{}:{}", ENV.PAYPAL_CLIENT_ID, ENV.PAYPAL_CLIENT_SECRET);
|
|
let formatted_key = format!(
|
|
"Basic {}",
|
|
base64::engine::general_purpose::STANDARD.encode(combined_key)
|
|
);
|
|
|
|
let mut form = HashMap::new();
|
|
form.insert("grant_type", "client_credentials");
|
|
|
|
#[derive(Deserialize)]
|
|
struct PaypalCredential {
|
|
access_token: String,
|
|
token_type: String,
|
|
expires_in: i64,
|
|
}
|
|
|
|
let credential: PaypalCredential = client
|
|
.post(format!("{}oauth2/token", ENV.PAYPAL_API_URL))
|
|
.header("Accept", "application/json")
|
|
.header("Accept-Language", "en_US")
|
|
.header("Authorization", formatted_key)
|
|
.form(&form)
|
|
.send()
|
|
.await
|
|
.map_err(|_| {
|
|
ApiError::Payments(
|
|
"Error while authenticating with PayPal".to_string(),
|
|
)
|
|
})?
|
|
.json()
|
|
.await
|
|
.map_err(|_| {
|
|
ApiError::Payments(
|
|
"Error while authenticating with PayPal (deser error)"
|
|
.to_string(),
|
|
)
|
|
})?;
|
|
|
|
let new_creds = PayPalCredentials {
|
|
access_token: credential.access_token,
|
|
token_type: credential.token_type,
|
|
expires: Utc::now() + Duration::seconds(credential.expires_in),
|
|
};
|
|
|
|
*creds = Some(new_creds.clone());
|
|
|
|
Ok(new_creds)
|
|
}
|
|
|
|
pub async fn make_paypal_request<T: Serialize, X: DeserializeOwned>(
|
|
&self,
|
|
method: Method,
|
|
path: &str,
|
|
body: Option<T>,
|
|
raw_text: Option<String>,
|
|
no_api_prefix: Option<bool>,
|
|
) -> Result<X, ApiError> {
|
|
let read = self.credential.read().await;
|
|
let credentials = if let Some(credentials) = read.as_ref() {
|
|
if credentials.expires < Utc::now() {
|
|
drop(read);
|
|
self.refresh_token().await.map_err(|_| {
|
|
ApiError::Payments(
|
|
"Error while authenticating with PayPal".to_string(),
|
|
)
|
|
})?
|
|
} else {
|
|
credentials.clone()
|
|
}
|
|
} else {
|
|
drop(read);
|
|
self.refresh_token().await.map_err(|_| {
|
|
ApiError::Payments(
|
|
"Error while authenticating with PayPal".to_string(),
|
|
)
|
|
})?
|
|
};
|
|
|
|
let client = reqwest::Client::new();
|
|
let mut request = client
|
|
.request(
|
|
method,
|
|
if no_api_prefix.unwrap_or(false) {
|
|
path.to_string()
|
|
} else {
|
|
format!("{}{path}", ENV.PAYPAL_API_URL)
|
|
},
|
|
)
|
|
.header(
|
|
"Authorization",
|
|
format!(
|
|
"{} {}",
|
|
credentials.token_type, credentials.access_token
|
|
),
|
|
);
|
|
|
|
if let Some(body) = body {
|
|
request = request.json(&body);
|
|
} else if let Some(body) = raw_text {
|
|
request = request
|
|
.header(reqwest::header::CONTENT_TYPE, "application/json")
|
|
.body(body);
|
|
}
|
|
|
|
let resp = request.send().await.map_err(|_| {
|
|
ApiError::Payments("could not communicate with PayPal".to_string())
|
|
})?;
|
|
|
|
let status = resp.status();
|
|
|
|
let value = resp.json::<Value>().await.map_err(|_| {
|
|
ApiError::Payments(
|
|
"could not retrieve PayPal response body".to_string(),
|
|
)
|
|
})?;
|
|
|
|
if !status.is_success() {
|
|
#[derive(Deserialize)]
|
|
struct PayPalError {
|
|
pub name: String,
|
|
pub message: String,
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
struct PayPalIdentityError {
|
|
pub error: String,
|
|
pub error_description: String,
|
|
}
|
|
|
|
if let Ok(mut error) =
|
|
serde_json::from_value::<PayPalError>(value.clone())
|
|
{
|
|
if error.name == "INSUFFICIENT_FUNDS" {
|
|
error.message = "We're currently transferring funds to our PayPal account. Please try again in a couple days.".to_string();
|
|
}
|
|
return Err(ApiError::Payments(format!(
|
|
"error name: {}, message: {}",
|
|
error.name, error.message
|
|
)));
|
|
}
|
|
|
|
if let Ok(error) =
|
|
serde_json::from_value::<PayPalIdentityError>(value)
|
|
{
|
|
return Err(ApiError::Payments(format!(
|
|
"error name: {}, message: {}",
|
|
error.error, error.error_description
|
|
)));
|
|
}
|
|
|
|
return Err(ApiError::Payments(
|
|
"could not retrieve PayPal error body".to_string(),
|
|
));
|
|
}
|
|
|
|
Ok(serde_json::from_value(value)?)
|
|
}
|
|
|
|
pub async fn make_tremendous_request<T: Serialize, X: DeserializeOwned>(
|
|
&self,
|
|
method: Method,
|
|
path: &str,
|
|
body: Option<T>,
|
|
) -> Result<X, ApiError> {
|
|
let client = reqwest::Client::new();
|
|
let mut request = client
|
|
.request(method, format!("{}{path}", ENV.TREMENDOUS_API_URL))
|
|
.header(
|
|
"Authorization",
|
|
format!("Bearer {}", ENV.TREMENDOUS_API_KEY),
|
|
);
|
|
|
|
if let Some(body) = body {
|
|
request = request.json(&body);
|
|
}
|
|
|
|
let resp = request.send().await.map_err(|_| {
|
|
ApiError::Payments(
|
|
"could not communicate with Tremendous".to_string(),
|
|
)
|
|
})?;
|
|
|
|
let status = resp.status();
|
|
|
|
let value = resp.json::<Value>().await.map_err(|_| {
|
|
ApiError::Payments(
|
|
"could not retrieve Tremendous response body".to_string(),
|
|
)
|
|
})?;
|
|
|
|
if !status.is_success()
|
|
&& let Some(obj) = value.as_object()
|
|
{
|
|
if let Some(array) = obj.get("errors") {
|
|
#[derive(Deserialize)]
|
|
struct TremendousError {
|
|
message: String,
|
|
payload: Option<serde_json::Value>,
|
|
}
|
|
|
|
let err =
|
|
serde_json::from_value::<TremendousError>(array.clone())
|
|
.map_err(|_| {
|
|
ApiError::Payments(
|
|
"could not retrieve Tremendous error json body"
|
|
.to_string(),
|
|
)
|
|
})?;
|
|
|
|
return Err(ApiError::Payments(format!(
|
|
"Tremendous error: {} ({:?})",
|
|
err.message, err.payload
|
|
)));
|
|
}
|
|
|
|
return Err(ApiError::Payments(
|
|
"could not retrieve Tremendous error body".to_string(),
|
|
));
|
|
}
|
|
|
|
Ok(serde_json::from_value(value)?)
|
|
}
|
|
|
|
pub async fn get_payout_methods(
|
|
&self,
|
|
) -> Result<Vec<PayoutMethod>, ApiError> {
|
|
async fn refresh_payout_methods(
|
|
queue: &PayoutsQueue,
|
|
) -> Result<PayoutMethods, ApiError> {
|
|
let mut options = queue.payout_options.write().await;
|
|
|
|
let mut methods = Vec::new();
|
|
|
|
match get_tremendous_payout_methods(queue).await {
|
|
Ok(mut tremendous_methods) => {
|
|
methods.append(&mut tremendous_methods);
|
|
}
|
|
Err(err) => {
|
|
warn!(
|
|
"Failed to fetch Tremendous payout methods: {err:#?}"
|
|
);
|
|
}
|
|
}
|
|
|
|
{
|
|
let paypal_us = PayoutMethod {
|
|
id: "paypal_us".to_string(),
|
|
type_: PayoutMethodType::PayPal,
|
|
name: "PayPal".to_string(),
|
|
category: None,
|
|
supported_countries: vec!["US".to_string()],
|
|
image_url: None,
|
|
image_logo_url: None,
|
|
interval: PayoutInterval::Standard {
|
|
min: flow::paypal::MIN_USD.get(),
|
|
max: flow::paypal::MAX_USD.get(),
|
|
},
|
|
currency_code: None,
|
|
exchange_rate: None,
|
|
};
|
|
|
|
let mut venmo = paypal_us.clone();
|
|
venmo.id = "venmo".to_string();
|
|
venmo.name = "Venmo".to_string();
|
|
venmo.type_ = PayoutMethodType::Venmo;
|
|
|
|
methods.insert(0, paypal_us);
|
|
methods.insert(1, venmo)
|
|
}
|
|
|
|
methods.extend(create_muralpay_methods());
|
|
|
|
let new_options = PayoutMethods {
|
|
options: methods,
|
|
expires: Utc::now() + Duration::hours(6),
|
|
};
|
|
|
|
*options = Some(new_options.clone());
|
|
|
|
Ok(new_options)
|
|
}
|
|
|
|
let read = self.payout_options.read().await;
|
|
let options = if let Some(options) = read.as_ref() {
|
|
if options.expires < Utc::now() {
|
|
drop(read);
|
|
refresh_payout_methods(self).await?
|
|
} else {
|
|
options.clone()
|
|
}
|
|
} else {
|
|
drop(read);
|
|
refresh_payout_methods(self).await?
|
|
};
|
|
|
|
Ok(options.options)
|
|
}
|
|
|
|
pub async fn get_brex_balance() -> Result<Option<AccountBalance>, ApiError>
|
|
{
|
|
#[derive(Deserialize)]
|
|
struct BrexBalance {
|
|
pub amount: i64,
|
|
// pub currency: String,
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
struct BrexAccount {
|
|
pub current_balance: BrexBalance,
|
|
pub available_balance: BrexBalance,
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
struct BrexResponse {
|
|
pub items: Vec<BrexAccount>,
|
|
}
|
|
|
|
let client = reqwest::Client::new();
|
|
let res = client
|
|
.get(format!("{}accounts/cash", ENV.BREX_API_URL))
|
|
.bearer_auth(&ENV.BREX_API_KEY)
|
|
.send()
|
|
.await?
|
|
.json::<BrexResponse>()
|
|
.await?;
|
|
|
|
Ok(Some(AccountBalance {
|
|
available: Decimal::from(
|
|
res.items
|
|
.iter()
|
|
.map(|x| x.available_balance.amount)
|
|
.sum::<i64>(),
|
|
) / Decimal::from(100),
|
|
pending: Decimal::from(
|
|
res.items
|
|
.iter()
|
|
.map(|x| {
|
|
x.current_balance.amount - x.available_balance.amount
|
|
})
|
|
.sum::<i64>(),
|
|
) / Decimal::from(100),
|
|
}))
|
|
}
|
|
|
|
pub async fn get_paypal_balance() -> Result<Option<AccountBalance>, ApiError>
|
|
{
|
|
let api_username = &ENV.PAYPAL_NVP_USERNAME;
|
|
let api_password = &ENV.PAYPAL_NVP_PASSWORD;
|
|
let api_signature = &ENV.PAYPAL_NVP_SIGNATURE;
|
|
|
|
let mut params = HashMap::new();
|
|
params.insert("METHOD", "GetBalance");
|
|
params.insert("VERSION", "204");
|
|
params.insert("USER", api_username);
|
|
params.insert("PWD", api_password);
|
|
params.insert("SIGNATURE", api_signature);
|
|
params.insert("RETURNALLCURRENCIES", "1");
|
|
|
|
let endpoint = "https://api-3t.paypal.com/nvp";
|
|
|
|
let client = reqwest::Client::new();
|
|
let response = client.post(endpoint).form(¶ms).send().await?;
|
|
|
|
let text = response.text().await?;
|
|
let body = urlencoding::decode(&text).unwrap_or_default();
|
|
|
|
let mut key_value_map = HashMap::new();
|
|
|
|
for pair in body.split('&') {
|
|
let mut iter = pair.splitn(2, '=');
|
|
if let (Some(key), Some(value)) = (iter.next(), iter.next()) {
|
|
key_value_map.insert(key.to_string(), value.to_string());
|
|
}
|
|
}
|
|
|
|
if let Some(amount) = key_value_map
|
|
.get("L_AMT0")
|
|
.and_then(|x| Decimal::from_str_exact(x).ok())
|
|
{
|
|
Ok(Some(AccountBalance {
|
|
available: amount,
|
|
pending: Decimal::ZERO,
|
|
}))
|
|
} else {
|
|
Ok(None)
|
|
}
|
|
}
|
|
|
|
pub async fn get_tremendous_balance(
|
|
&self,
|
|
) -> Result<Option<AccountBalance>, ApiError> {
|
|
#[derive(Deserialize)]
|
|
struct FundingSourceMeta {
|
|
available_cents: Option<u64>,
|
|
pending_cents: Option<u64>,
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
struct FundingSource {
|
|
method: String,
|
|
meta: FundingSourceMeta,
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
struct FundingSourceRequest {
|
|
pub funding_sources: Vec<FundingSource>,
|
|
}
|
|
|
|
let val = self
|
|
.make_tremendous_request::<(), FundingSourceRequest>(
|
|
Method::GET,
|
|
"funding_sources",
|
|
None,
|
|
)
|
|
.await?;
|
|
|
|
Ok(val
|
|
.funding_sources
|
|
.into_iter()
|
|
.find(|x| x.method == "balance")
|
|
.map(|x| AccountBalance {
|
|
available: Decimal::from(x.meta.available_cents.unwrap_or(0))
|
|
/ Decimal::from(100),
|
|
pending: Decimal::from(x.meta.pending_cents.unwrap_or(0))
|
|
/ Decimal::from(100),
|
|
}))
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, Clone, Copy)]
|
|
pub struct PayoutFees {
|
|
/// Fee which is taken by the underlying method we're using.
|
|
///
|
|
/// For example, if a user withdraws $10.00 and the method takes a
|
|
/// 10% cut, then we submit a payout request of $10.00 to the method,
|
|
/// and only $9.00 will be sent to the recipient.
|
|
pub method_fee: Decimal2dp,
|
|
/// Fee which we keep and don't pass to the underlying method.
|
|
///
|
|
/// For example, if a user withdraws $10.00 and the method takes a
|
|
/// 10% cut, then we submit a payout request of $9.00, and the $1.00 stays
|
|
/// in our account.
|
|
pub platform_fee: Decimal2dp,
|
|
/// How much is 1 USD worth in the target currency?
|
|
pub exchange_rate: Option<Decimal>,
|
|
}
|
|
|
|
impl PayoutFees {
|
|
pub fn total_fee(&self) -> Decimal2dp {
|
|
self.method_fee + self.platform_fee
|
|
}
|
|
}
|
|
|
|
async fn get_tremendous_payout_methods(
|
|
queue: &PayoutsQueue,
|
|
) -> Result<Vec<PayoutMethod>> {
|
|
#[derive(Debug, Deserialize)]
|
|
struct Sku {
|
|
min: Decimal,
|
|
max: Decimal,
|
|
}
|
|
|
|
#[derive(Deserialize, Eq, PartialEq)]
|
|
#[serde(rename_all = "snake_case")]
|
|
enum ProductImageType {
|
|
Card,
|
|
Logo,
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
struct ProductImage {
|
|
src: String,
|
|
#[serde(rename = "type")]
|
|
type_: ProductImageType,
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
struct ProductCountry {
|
|
abbr: String,
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
struct Product {
|
|
id: String,
|
|
category: String,
|
|
name: String,
|
|
// description: String,
|
|
// disclosure: String,
|
|
skus: Vec<Sku>,
|
|
currency_codes: Vec<String>,
|
|
countries: Vec<ProductCountry>,
|
|
images: Vec<ProductImage>,
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
struct TremendousResponse {
|
|
products: Vec<Product>,
|
|
}
|
|
|
|
let forex: TremendousForexResponse = queue
|
|
.make_tremendous_request(Method::GET, "forex", None::<()>)
|
|
.await
|
|
.wrap_err("failed to fetch Tremendous forex data")?;
|
|
|
|
let response = queue
|
|
.make_tremendous_request::<(), TremendousResponse>(
|
|
Method::GET,
|
|
"products",
|
|
None,
|
|
)
|
|
.await
|
|
.wrap_err("failed to fetch Tremendous products data")?;
|
|
|
|
let mut methods = Vec::new();
|
|
|
|
for product in response.products {
|
|
const BLACKLISTED_IDS: &[&str] = &[
|
|
// physical visa
|
|
"A2J05SWPI2QG",
|
|
// crypto
|
|
"1UOOSHUUYTAM",
|
|
"5EVJN47HPDFT",
|
|
"NI9M4EVAVGFJ",
|
|
"VLY29QHTMNGT",
|
|
"7XU98H109Y3A",
|
|
"0CGEDFP2UIKV",
|
|
"PDYLQU0K073Y",
|
|
"HCS5Z7O2NV5G",
|
|
"IY1VMST1MOXS",
|
|
"VRPZLJ7HCA8X",
|
|
// bitcard (crypto)
|
|
"GWQQS5RM8IZS",
|
|
"896MYD4SGOGZ",
|
|
"PWLEN1VZGMZA",
|
|
"A2VRM96J5K5W",
|
|
"HV9ICIM3JT7P",
|
|
"K2KLSPVWC2Q4",
|
|
"HRBRQLLTDF95",
|
|
"UUBYLZVK7QAB",
|
|
"BH8W3XEDEOJN",
|
|
"7WGE043X1RYQ",
|
|
"2B13MHUZZVTF",
|
|
"JN6R44P86EYX",
|
|
"DA8H43GU84SO",
|
|
"QK2XAQHSDEH4",
|
|
"J7K1IQFS76DK",
|
|
"NL4JQ2G7UPRZ",
|
|
"OEFTMSBA5ELH",
|
|
"A3CQK6UHNV27",
|
|
];
|
|
const SUPPORTED_METHODS: &[&str] = &[
|
|
"merchant_cards",
|
|
"merchant_card",
|
|
"bank",
|
|
"charity",
|
|
"visa",
|
|
"visa_card",
|
|
"paypal",
|
|
"venmo",
|
|
];
|
|
|
|
if !SUPPORTED_METHODS.contains(&&*product.category)
|
|
|| BLACKLISTED_IDS.contains(&&*product.id)
|
|
{
|
|
continue;
|
|
};
|
|
|
|
let Some(currency) = product.currency_codes.first() else {
|
|
// cards with multiple currencies are not supported
|
|
continue;
|
|
};
|
|
let Some(&usd_to_currency) = forex.forex.get(currency) else {
|
|
warn!("No Tremendous forex data for {currency}");
|
|
continue;
|
|
};
|
|
|
|
let method = PayoutMethod {
|
|
id: product.id,
|
|
type_: PayoutMethodType::Tremendous,
|
|
name: product.name.clone(),
|
|
category: Some(product.category.clone()),
|
|
supported_countries: product
|
|
.countries
|
|
.into_iter()
|
|
.map(|x| x.abbr)
|
|
.collect(),
|
|
image_logo_url: product
|
|
.images
|
|
.iter()
|
|
.find(|x| x.type_ == ProductImageType::Logo)
|
|
.map(|x| x.src.clone()),
|
|
image_url: product
|
|
.images
|
|
.into_iter()
|
|
.find(|x| x.type_ == ProductImageType::Card)
|
|
.map(|x| x.src),
|
|
interval: if product.skus.len() > 1 {
|
|
let mut values = product
|
|
.skus
|
|
.into_iter()
|
|
.map(|x| PayoutDecimal(x.min))
|
|
.collect::<Vec<_>>();
|
|
values.sort_by(|a, b| a.0.cmp(&b.0));
|
|
|
|
PayoutInterval::Fixed { values }
|
|
} else if let Some(first) = product.skus.first() {
|
|
PayoutInterval::Standard {
|
|
min: first.min,
|
|
max: first.max,
|
|
}
|
|
} else {
|
|
PayoutInterval::Standard {
|
|
min: Decimal::ZERO,
|
|
max: Decimal::from(5_000),
|
|
}
|
|
},
|
|
currency_code: Some(currency.clone()),
|
|
exchange_rate: Some(usd_to_currency),
|
|
};
|
|
|
|
methods.push(method);
|
|
}
|
|
|
|
const UPRANK_IDS: &[&str] =
|
|
&["ET0ZVETV5ILN", "Q24BD9EZ332JT", "UIL1ZYJU5MKN"];
|
|
const DOWNRANK_IDS: &[&str] = &["EIPF8Q00EMM1", "OU2MWXYWPNWQ"];
|
|
|
|
methods.sort_by(|a, b| {
|
|
let a_top = UPRANK_IDS.contains(&&*a.id);
|
|
let a_bottom = DOWNRANK_IDS.contains(&&*a.id);
|
|
let b_top = UPRANK_IDS.contains(&&*b.id);
|
|
let b_bottom = DOWNRANK_IDS.contains(&&*b.id);
|
|
|
|
match (a_top, a_bottom, b_top, b_bottom) {
|
|
(true, _, true, _) => a.name.cmp(&b.name), // Both in top_priority: sort alphabetically
|
|
(_, true, _, true) => a.name.cmp(&b.name), // Both in bottom_priority: sort alphabetically
|
|
(true, _, _, _) => std::cmp::Ordering::Less, // a in top_priority: a comes first
|
|
(_, _, true, _) => std::cmp::Ordering::Greater, // b in top_priority: b comes first
|
|
(_, true, _, _) => std::cmp::Ordering::Greater, // a in bottom_priority: b comes first
|
|
(_, _, _, true) => std::cmp::Ordering::Less, // b in bottom_priority: a comes first
|
|
(_, _, _, _) => a.name.cmp(&b.name), // Neither in priority: sort alphabetically
|
|
}
|
|
});
|
|
|
|
Ok(methods)
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
pub struct AditudePoints {
|
|
#[serde(rename = "pointsList")]
|
|
pub points_list: Vec<AditudePoint>,
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
pub struct AditudePoint {
|
|
pub metric: AditudeMetric,
|
|
pub time: AditudeTime,
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
pub struct AditudeMetric {
|
|
pub revenue: Option<Decimal>,
|
|
pub impressions: Option<u128>,
|
|
pub cpm: Option<Decimal>,
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
pub struct AditudeTime {
|
|
pub seconds: u64,
|
|
}
|
|
|
|
pub async fn make_aditude_request(
|
|
metrics: &[&str],
|
|
range: &str,
|
|
interval: &str,
|
|
) -> Result<Vec<AditudePoints>, ApiError> {
|
|
let request = reqwest::Client::new()
|
|
.post("https://cloud.aditude.io/api/public/insights/metrics")
|
|
.bearer_auth(&ENV.ADITUDE_API_KEY)
|
|
.json(&serde_json::json!({
|
|
"metrics": metrics,
|
|
"range": range,
|
|
"interval": interval
|
|
}))
|
|
.send()
|
|
.await?
|
|
.error_for_status()?;
|
|
|
|
let text = request.text().await?;
|
|
|
|
let json: Vec<AditudePoints> = serde_json::from_str(&text)?;
|
|
|
|
Ok(json)
|
|
}
|
|
|
|
pub async fn process_payout(
|
|
pool: &PgPool,
|
|
client: &clickhouse::Client,
|
|
) -> Result<(), ApiError> {
|
|
sqlx::query!(
|
|
"
|
|
UPDATE payouts
|
|
SET status = $1
|
|
WHERE status = $2 AND created < NOW() - INTERVAL '30 days'
|
|
",
|
|
crate::models::payouts::PayoutStatus::Failed.as_str(),
|
|
crate::models::payouts::PayoutStatus::InTransit.as_str(),
|
|
)
|
|
.execute(pool)
|
|
.await?;
|
|
|
|
let start: DateTime<Utc> = DateTime::from_naive_utc_and_offset(
|
|
(Utc::now() - Duration::days(1))
|
|
.date_naive()
|
|
.and_hms_nano_opt(0, 0, 0, 0)
|
|
.unwrap_or_default(),
|
|
Utc,
|
|
);
|
|
|
|
let results = sqlx::query!(
|
|
"SELECT EXISTS(SELECT 1 FROM payouts_values WHERE created = $1)",
|
|
start,
|
|
)
|
|
.fetch_one(pool)
|
|
.await?;
|
|
|
|
if results.exists.unwrap_or(false) {
|
|
return Ok(());
|
|
}
|
|
|
|
let end = start + Duration::days(1);
|
|
#[derive(Deserialize, clickhouse::Row)]
|
|
struct ProjectMultiplier {
|
|
pub page_views: u64,
|
|
pub project_id: u64,
|
|
}
|
|
|
|
let (views_values, views_sum, downloads_values, downloads_sum) = futures::future::try_join4(
|
|
client
|
|
.query(
|
|
r#"
|
|
SELECT COUNT(1) page_views, project_id
|
|
FROM views
|
|
WHERE (recorded BETWEEN ? AND ?) AND (project_id != 0) AND (monetized = TRUE)
|
|
GROUP BY project_id
|
|
ORDER BY page_views DESC
|
|
"#,
|
|
)
|
|
.bind(start.timestamp())
|
|
.bind(end.timestamp())
|
|
.fetch_all::<ProjectMultiplier>(),
|
|
client
|
|
.query("SELECT COUNT(1) FROM views WHERE (recorded BETWEEN ? AND ?) AND (project_id != 0) AND (monetized = TRUE)")
|
|
.bind(start.timestamp())
|
|
.bind(end.timestamp())
|
|
.fetch_one::<u64>(),
|
|
client
|
|
.query(
|
|
r#"
|
|
SELECT COUNT(1) page_views, project_id
|
|
FROM downloads
|
|
WHERE (recorded BETWEEN ? AND ?) AND (user_id != 0)
|
|
GROUP BY project_id
|
|
ORDER BY page_views DESC
|
|
"#,
|
|
)
|
|
.bind(start.timestamp())
|
|
.bind(end.timestamp())
|
|
.fetch_all::<ProjectMultiplier>(),
|
|
client
|
|
.query("SELECT COUNT(1) FROM downloads WHERE (recorded BETWEEN ? AND ?) AND (user_id != 0)")
|
|
.bind(start.timestamp())
|
|
.bind(end.timestamp())
|
|
.fetch_one::<u64>(),
|
|
)
|
|
.await?;
|
|
|
|
let mut transaction = pool.begin().await?;
|
|
|
|
struct PayoutMultipliers {
|
|
sum: u64,
|
|
values: HashMap<u64, u64>,
|
|
}
|
|
|
|
let mut views_values = views_values
|
|
.into_iter()
|
|
.map(|x| (x.project_id, x.page_views))
|
|
.collect::<HashMap<u64, u64>>();
|
|
let downloads_values = downloads_values
|
|
.into_iter()
|
|
.map(|x| (x.project_id, x.page_views))
|
|
.collect::<HashMap<u64, u64>>();
|
|
|
|
for (key, value) in &downloads_values {
|
|
let counter = views_values.entry(*key).or_insert(0);
|
|
*counter += *value;
|
|
}
|
|
|
|
let multipliers: PayoutMultipliers = PayoutMultipliers {
|
|
sum: downloads_sum + views_sum,
|
|
values: views_values,
|
|
};
|
|
|
|
struct Project {
|
|
// user_id, payouts_split
|
|
team_members: Vec<(i64, Decimal)>,
|
|
}
|
|
|
|
let mut projects_map: HashMap<i64, Project> = HashMap::new();
|
|
|
|
let project_ids = multipliers
|
|
.values
|
|
.keys()
|
|
.map(|x| *x as i64)
|
|
.collect::<Vec<i64>>();
|
|
|
|
let project_org_members = sqlx::query!(
|
|
"
|
|
SELECT m.id id, tm.user_id user_id, tm.payouts_split payouts_split
|
|
FROM mods m
|
|
INNER JOIN organizations o ON m.organization_id = o.id
|
|
INNER JOIN team_members tm on o.team_id = tm.team_id AND tm.accepted = TRUE
|
|
WHERE m.id = ANY($1) AND m.monetization_status = $2 AND m.status = ANY($3) AND m.organization_id IS NOT NULL
|
|
",
|
|
&project_ids,
|
|
MonetizationStatus::Monetized.as_str(),
|
|
&*crate::models::projects::ProjectStatus::iterator()
|
|
.filter(|x| !x.is_hidden())
|
|
.map(|x| x.to_string())
|
|
.collect::<Vec<String>>(),
|
|
)
|
|
.fetch(&mut transaction)
|
|
.try_fold(DashMap::new(), |acc: DashMap<i64, HashMap<i64, Decimal>>, r| {
|
|
acc.entry(r.id)
|
|
.or_default()
|
|
.insert(r.user_id, r.payouts_split);
|
|
async move { Ok(acc) }
|
|
})
|
|
.await?;
|
|
|
|
let project_team_members = sqlx::query!(
|
|
"
|
|
SELECT m.id id, tm.user_id user_id, tm.payouts_split payouts_split
|
|
FROM mods m
|
|
INNER JOIN team_members tm on m.team_id = tm.team_id AND tm.accepted = TRUE
|
|
WHERE m.id = ANY($1) AND m.monetization_status = $2 AND m.status = ANY($3)
|
|
",
|
|
&project_ids,
|
|
MonetizationStatus::Monetized.as_str(),
|
|
&*crate::models::projects::ProjectStatus::iterator()
|
|
.filter(|x| !x.is_hidden())
|
|
.map(|x| x.to_string())
|
|
.collect::<Vec<String>>(),
|
|
)
|
|
.fetch(&mut transaction)
|
|
.try_fold(
|
|
DashMap::new(),
|
|
|acc: DashMap<i64, HashMap<i64, Decimal>>, r| {
|
|
acc.entry(r.id)
|
|
.or_default()
|
|
.insert(r.user_id, r.payouts_split);
|
|
async move { Ok(acc) }
|
|
},
|
|
)
|
|
.await?;
|
|
|
|
for project_id in project_ids {
|
|
let team_members: HashMap<i64, Decimal> = project_team_members
|
|
.remove(&project_id)
|
|
.unwrap_or((0, HashMap::new()))
|
|
.1;
|
|
let org_team_members: HashMap<i64, Decimal> = project_org_members
|
|
.remove(&project_id)
|
|
.unwrap_or((0, HashMap::new()))
|
|
.1;
|
|
|
|
let mut all_team_members = vec![];
|
|
|
|
for (user_id, payouts_split) in org_team_members {
|
|
if !team_members.contains_key(&user_id) {
|
|
all_team_members.push((user_id, payouts_split));
|
|
}
|
|
}
|
|
for (user_id, payouts_split) in team_members {
|
|
all_team_members.push((user_id, payouts_split));
|
|
}
|
|
|
|
// if all team members are set to zero, we treat as an equal revenue distribution
|
|
if all_team_members.iter().all(|x| x.1 == Decimal::ZERO) {
|
|
all_team_members
|
|
.iter_mut()
|
|
.for_each(|x| x.1 = Decimal::from(1));
|
|
}
|
|
|
|
projects_map.insert(
|
|
project_id,
|
|
Project {
|
|
team_members: all_team_members,
|
|
},
|
|
);
|
|
}
|
|
|
|
let aditude_res = make_aditude_request(
|
|
&["METRIC_IMPRESSIONS", "METRIC_REVENUE"],
|
|
"Yesterday",
|
|
"1d",
|
|
)
|
|
.await?;
|
|
|
|
let aditude_amount: Decimal = aditude_res
|
|
.iter()
|
|
.map(|x| {
|
|
x.points_list
|
|
.iter()
|
|
.filter_map(|x| x.metric.revenue)
|
|
.sum::<Decimal>()
|
|
})
|
|
.sum();
|
|
let aditude_impressions: u128 = aditude_res
|
|
.iter()
|
|
.map(|x| {
|
|
x.points_list
|
|
.iter()
|
|
.filter_map(|x| x.metric.impressions)
|
|
.sum::<u128>()
|
|
})
|
|
.sum();
|
|
|
|
// 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);
|
|
// 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));
|
|
|
|
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 (
|
|
mut insert_user_ids,
|
|
mut insert_project_ids,
|
|
mut insert_payouts,
|
|
mut insert_starts,
|
|
mut insert_availables,
|
|
) = (Vec::new(), Vec::new(), Vec::new(), Vec::new(), Vec::new());
|
|
for (id, project) in projects_map {
|
|
if let Some(value) = &multipliers.values.get(&(id as u64)) {
|
|
let project_multiplier: Decimal =
|
|
Decimal::from(**value) / Decimal::from(multipliers.sum);
|
|
|
|
let sum_splits: Decimal =
|
|
project.team_members.iter().map(|x| x.1).sum();
|
|
|
|
if sum_splits > Decimal::ZERO {
|
|
for (user_id, split) in project.team_members {
|
|
let payout: Decimal =
|
|
payout * project_multiplier * (split / sum_splits);
|
|
|
|
if payout > Decimal::ZERO {
|
|
insert_user_ids.push(user_id);
|
|
insert_project_ids.push(id);
|
|
insert_payouts.push(payout);
|
|
insert_starts.push(start);
|
|
insert_availables.push(available);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
sqlx::query!(
|
|
"
|
|
INSERT INTO payouts_values (user_id, mod_id, amount, created, date_available)
|
|
SELECT * FROM UNNEST ($1::bigint[], $2::bigint[], $3::numeric[], $4::timestamptz[], $5::timestamptz[])
|
|
",
|
|
&insert_user_ids[..],
|
|
&insert_project_ids[..],
|
|
&insert_payouts[..],
|
|
&insert_starts[..],
|
|
&insert_availables[..]
|
|
)
|
|
.execute(&mut transaction)
|
|
.await?;
|
|
|
|
transaction.commit().await?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
// Used for testing, should be the same as the above function
|
|
pub async fn insert_payouts(
|
|
insert_user_ids: Vec<i64>,
|
|
insert_project_ids: Vec<i64>,
|
|
insert_payouts: Vec<Decimal>,
|
|
insert_starts: Vec<DateTime<Utc>>,
|
|
insert_availables: Vec<DateTime<Utc>>,
|
|
transaction: &mut PgTransaction<'_>,
|
|
) -> sqlx::Result<PgQueryResult> {
|
|
sqlx::query!(
|
|
"
|
|
INSERT INTO payouts_values (user_id, mod_id, amount, created, date_available)
|
|
SELECT * FROM UNNEST ($1::bigint[], $2::bigint[], $3::numeric[], $4::timestamptz[], $5::timestamptz[])
|
|
",
|
|
&insert_user_ids[..],
|
|
&insert_project_ids[..],
|
|
&insert_payouts[..],
|
|
&insert_starts[..],
|
|
&insert_availables[..],
|
|
)
|
|
.execute(&mut *transaction)
|
|
.await
|
|
}
|
|
|
|
pub async fn index_payouts_notifications(
|
|
pool: &PgPool,
|
|
redis: &RedisPool,
|
|
) -> Result<(), ApiError> {
|
|
info!("Updating payout notifications");
|
|
|
|
let mut transaction = pool.begin().await?;
|
|
|
|
payouts_values_notifications::synchronize_future_payout_values(
|
|
&mut transaction,
|
|
200,
|
|
)
|
|
.await?;
|
|
let items = payouts_values_notifications::PayoutsValuesNotification::unnotified_users_with_available_payouts_with_limit(&mut transaction, 200).await?;
|
|
|
|
let payout_ref_ids = items.iter().map(|x| x.id).collect::<Vec<_>>();
|
|
let dates_available =
|
|
items.iter().map(|x| x.date_available).collect::<Vec<_>>();
|
|
let user_ids = items.iter().map(|x| x.user_id).collect::<Vec<_>>();
|
|
|
|
NotificationBuilder::insert_many_payout_notifications(
|
|
user_ids,
|
|
dates_available,
|
|
&mut transaction,
|
|
redis,
|
|
)
|
|
.await?;
|
|
payouts_values_notifications::PayoutsValuesNotification::set_notified_many(
|
|
&payout_ref_ids,
|
|
&mut transaction,
|
|
)
|
|
.await?;
|
|
|
|
transaction.commit().await?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn insert_bank_balances_and_webhook(
|
|
payouts: &PayoutsQueue,
|
|
pool: &PgPool,
|
|
) -> Result<(), ApiError> {
|
|
let mut transaction = pool.begin().await?;
|
|
|
|
let paypal_result = PayoutsQueue::get_paypal_balance().await;
|
|
let brex_result = PayoutsQueue::get_brex_balance().await;
|
|
let tremendous_result = payouts.get_tremendous_balance().await;
|
|
let mural_result = payouts.get_mural_balance().await;
|
|
|
|
let mut insert_account_types = Vec::new();
|
|
let mut insert_amounts = Vec::new();
|
|
let mut insert_pending = Vec::new();
|
|
let mut insert_recorded = Vec::new();
|
|
|
|
let now = Utc::now();
|
|
let today = now.date_naive().and_time(NaiveTime::MIN).and_utc();
|
|
|
|
let mut add_balance = |account_type: &str, balance: &AccountBalance| {
|
|
insert_account_types.push(account_type.to_string());
|
|
insert_amounts.push(balance.available);
|
|
insert_pending.push(false);
|
|
insert_recorded.push(today);
|
|
|
|
insert_account_types.push(account_type.to_string());
|
|
insert_amounts.push(balance.pending);
|
|
insert_pending.push(true);
|
|
insert_recorded.push(today);
|
|
};
|
|
|
|
if let Ok(Some(ref paypal)) = paypal_result {
|
|
add_balance("paypal", paypal);
|
|
}
|
|
if let Ok(Some(ref brex)) = brex_result {
|
|
add_balance("brex", brex);
|
|
}
|
|
if let Ok(Some(ref tremendous)) = tremendous_result {
|
|
add_balance("tremendous", tremendous);
|
|
}
|
|
if let Ok(Some(ref mural)) = mural_result {
|
|
add_balance("mural", mural);
|
|
}
|
|
|
|
let inserted = sqlx::query_scalar!(
|
|
r#"
|
|
INSERT INTO payout_sources_balance (account_type, amount, pending, recorded)
|
|
SELECT * FROM UNNEST ($1::text[], $2::numeric[], $3::boolean[], $4::timestamptz[])
|
|
ON CONFLICT (recorded, account_type, pending)
|
|
DO UPDATE SET amount = EXCLUDED.amount
|
|
RETURNING xmax = 0 "xmax!"
|
|
"#,
|
|
&insert_account_types[..],
|
|
&insert_amounts[..],
|
|
&insert_pending[..],
|
|
&insert_recorded[..],
|
|
)
|
|
.fetch_one(&mut transaction)
|
|
.await?;
|
|
|
|
if inserted {
|
|
check_balance_with_webhook(
|
|
"paypal",
|
|
ENV.PAYPAL_BALANCE_ALERT_THRESHOLD,
|
|
paypal_result,
|
|
)
|
|
.await?;
|
|
check_balance_with_webhook(
|
|
"brex",
|
|
ENV.BREX_BALANCE_ALERT_THRESHOLD,
|
|
brex_result,
|
|
)
|
|
.await?;
|
|
check_balance_with_webhook(
|
|
"tremendous",
|
|
ENV.TREMENDOUS_BALANCE_ALERT_THRESHOLD,
|
|
tremendous_result,
|
|
)
|
|
.await?;
|
|
check_balance_with_webhook(
|
|
"mural",
|
|
ENV.MURAL_BALANCE_ALERT_THRESHOLD,
|
|
mural_result,
|
|
)
|
|
.await?;
|
|
}
|
|
|
|
transaction.commit().await?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
async fn check_balance_with_webhook(
|
|
source: &str,
|
|
threshold: u64,
|
|
result: Result<Option<AccountBalance>, ApiError>,
|
|
) -> Result<Option<AccountBalance>, ApiError> {
|
|
let maybe_threshold = if threshold > 0 { Some(threshold) } else { None };
|
|
let payout_alert_webhook = &ENV.PAYOUT_ALERT_SLACK_WEBHOOK;
|
|
|
|
match &result {
|
|
Ok(Some(account_balance)) => {
|
|
if let Some(threshold) = maybe_threshold
|
|
&& let Some(available) =
|
|
account_balance.available.trunc().to_u64()
|
|
&& available <= threshold
|
|
{
|
|
send_slack_payout_source_alert_webhook(
|
|
PayoutSourceAlertType::UnderThreshold {
|
|
source: source.to_owned(),
|
|
threshold,
|
|
current_balance: available,
|
|
},
|
|
payout_alert_webhook,
|
|
)
|
|
.await?;
|
|
}
|
|
}
|
|
|
|
Err(error) => {
|
|
error!(%error, "Failure getting balance for payout source '{source}'");
|
|
|
|
if maybe_threshold.is_some() {
|
|
send_slack_payout_source_alert_webhook(
|
|
PayoutSourceAlertType::CheckFailure {
|
|
source: source.to_owned(),
|
|
display_error: error.to_string(),
|
|
},
|
|
payout_alert_webhook,
|
|
)
|
|
.await?;
|
|
}
|
|
}
|
|
|
|
_ => {}
|
|
}
|
|
|
|
Ok(result.ok().flatten())
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::test::{
|
|
api_v3::ApiV3,
|
|
database::USER_USER_ID_PARSED,
|
|
environment::{TestEnvironment, with_test_environment},
|
|
};
|
|
use rust_decimal::dec;
|
|
|
|
async fn setup_payouts_values(
|
|
db: &PgPool,
|
|
entries: Vec<(i64, Decimal, DateTime<Utc>)>, // (user_id, amount, date_available)
|
|
) {
|
|
for (user_id, amount, date_available) in &entries {
|
|
sqlx::query!(
|
|
"INSERT INTO payouts_values (user_id, mod_id, amount, created, date_available)
|
|
VALUES ($1, NULL, $2, NOW(), $3)",
|
|
user_id,
|
|
amount,
|
|
date_available,
|
|
)
|
|
.execute(db)
|
|
.await
|
|
.unwrap();
|
|
}
|
|
|
|
for (user_id, _amount, date_available) in &entries {
|
|
sqlx::query!(
|
|
"INSERT INTO payouts_values_notifications (date_available, user_id, notified)
|
|
VALUES ($1, $2, FALSE)
|
|
ON CONFLICT (date_available, user_id) DO NOTHING",
|
|
date_available,
|
|
user_id,
|
|
)
|
|
.execute(db)
|
|
.await
|
|
.unwrap();
|
|
}
|
|
}
|
|
|
|
/// When a user's payout amount is below
|
|
/// the $1.00 (100 cents) threshold, the `WHERE sum >= 100` filter in
|
|
/// insert_many_payout_notifications skips the INSERT into notifications
|
|
/// for that user, but the old code still passed ALL pre-generated
|
|
/// notification_ids to insert_many_deliveries. Since
|
|
/// notifications_deliveries.notification_id has a FK constraint on
|
|
/// notifications(id), this caused a violation that failed
|
|
/// the entire transaction. The notified flag was never set to `true`,
|
|
/// so the rows accumulated and the same failure repeated every run.
|
|
#[actix_rt::test]
|
|
async fn test_payout_notification_below_threshold_does_not_fail() {
|
|
with_test_environment(None, |env: TestEnvironment<ApiV3>| async move {
|
|
let db = &env.db.pool;
|
|
let redis = &env.db.redis_pool;
|
|
|
|
let user_id = USER_USER_ID_PARSED;
|
|
|
|
// date_available must be in the past so the notification query
|
|
// picks it up (date_available <= NOW()).
|
|
let date_available = Utc::now() - Duration::hours(1);
|
|
|
|
// Amount of $0.50 -- below the $1.00 threshold (sum < 100 cents).
|
|
setup_payouts_values(
|
|
db,
|
|
vec![(user_id, dec!(0.50), date_available)],
|
|
)
|
|
.await;
|
|
|
|
// This should succeed, NOT return an error.
|
|
// Before the fix, this would fail with a FK constraint violation.
|
|
let result = index_payouts_notifications(db, redis).await;
|
|
assert!(
|
|
result.is_ok(),
|
|
"index_payouts_notifications should succeed for below-threshold payouts, got: {:?}",
|
|
result.err()
|
|
);
|
|
|
|
// Verify the notification row was marked as notified (not stuck).
|
|
let remaining = sqlx::query_scalar!(
|
|
"SELECT COUNT(*) FROM payouts_values_notifications WHERE notified = FALSE AND user_id = $1",
|
|
user_id,
|
|
)
|
|
.fetch_one(db)
|
|
.await
|
|
.unwrap()
|
|
.unwrap_or(0);
|
|
|
|
assert_eq!(
|
|
remaining, 0,
|
|
"All payouts_values_notifications rows should be marked notified"
|
|
);
|
|
})
|
|
.await;
|
|
}
|
|
|
|
/// When there is a mix of users, some above and some below the
|
|
/// threshold, the above-threshold users should get notifications
|
|
/// while the below-threshold ones are silently skipped. The entire
|
|
/// transaction must succeed (not fail due to the below-threshold user).
|
|
#[actix_rt::test]
|
|
async fn test_payout_notification_mixed_threshold_users() {
|
|
with_test_environment(None, |env: TestEnvironment<ApiV3>| async move {
|
|
let db = &env.db.pool;
|
|
let redis = &env.db.redis_pool;
|
|
|
|
let above_user_id = USER_USER_ID_PARSED; // user 3
|
|
let below_user_id = 4i64; // FRIEND_USER_ID
|
|
let date_available = Utc::now() - Duration::hours(1);
|
|
|
|
setup_payouts_values(
|
|
db,
|
|
vec![
|
|
// Above threshold
|
|
(above_user_id, dec!(5.00), date_available),
|
|
// Below threshold
|
|
(below_user_id, dec!(0.25), date_available),
|
|
],
|
|
)
|
|
.await;
|
|
|
|
let result = index_payouts_notifications(db, redis).await;
|
|
assert!(
|
|
result.is_ok(),
|
|
"index_payouts_notifications should succeed with mixed users, got: {:?}",
|
|
result.err()
|
|
);
|
|
|
|
// Above-threshold user should have a notification.
|
|
let above_count = sqlx::query_scalar!(
|
|
"SELECT COUNT(*) FROM notifications WHERE user_id = $1 AND body->>'type' = 'payout_available'",
|
|
above_user_id,
|
|
)
|
|
.fetch_one(db)
|
|
.await
|
|
.unwrap()
|
|
.unwrap_or(0);
|
|
|
|
assert!(
|
|
above_count > 0,
|
|
"Above-threshold user should have a payout notification"
|
|
);
|
|
|
|
// Below-threshold user should NOT have a notification.
|
|
let below_count = sqlx::query_scalar!(
|
|
"SELECT COUNT(*) FROM notifications WHERE user_id = $1 AND body->>'type' = 'payout_available'",
|
|
below_user_id,
|
|
)
|
|
.fetch_one(db)
|
|
.await
|
|
.unwrap()
|
|
.unwrap_or(0);
|
|
|
|
assert_eq!(
|
|
below_count, 0,
|
|
"Below-threshold user should NOT have a payout notification"
|
|
);
|
|
|
|
// Both should be marked notified.
|
|
let remaining = sqlx::query_scalar!(
|
|
"SELECT COUNT(*) FROM payouts_values_notifications WHERE notified = FALSE",
|
|
)
|
|
.fetch_one(db)
|
|
.await
|
|
.unwrap()
|
|
.unwrap_or(0);
|
|
|
|
assert_eq!(
|
|
remaining, 0,
|
|
"All payouts_values_notifications rows should be marked notified"
|
|
);
|
|
})
|
|
.await;
|
|
}
|
|
}
|