use async_trait::async_trait; use sqlx::{PgPool, Row}; use super::{SettlementWriteRepository, StoredUsageSettlement, UsageSettlementInput}; use crate::error::SqlxResultExt; use crate::postgres::PostgresTransactionRunner; use crate::DataLayerError; const FIND_USAGE_FOR_SETTLEMENT_SQL: &str = r#" SELECT usage_record.request_id, COALESCE(usage_settlement_snapshots.wallet_id, usage_record.wallet_id) AS wallet_id, COALESCE(usage_settlement_snapshots.billing_status, usage_record.billing_status) AS billing_status, COALESCE( CAST(usage_settlement_snapshots.wallet_balance_before AS DOUBLE PRECISION), CAST(usage_record.wallet_balance_before AS DOUBLE PRECISION) ) AS wallet_balance_before, COALESCE( CAST(usage_settlement_snapshots.wallet_balance_after AS DOUBLE PRECISION), CAST(usage_record.wallet_balance_after AS DOUBLE PRECISION) ) AS wallet_balance_after, COALESCE( CAST(usage_settlement_snapshots.wallet_recharge_balance_before AS DOUBLE PRECISION), CAST(usage_record.wallet_recharge_balance_before AS DOUBLE PRECISION) ) AS wallet_recharge_balance_before, COALESCE( CAST(usage_settlement_snapshots.wallet_recharge_balance_after AS DOUBLE PRECISION), CAST(usage_record.wallet_recharge_balance_after AS DOUBLE PRECISION) ) AS wallet_recharge_balance_after, COALESCE( CAST(usage_settlement_snapshots.wallet_gift_balance_before AS DOUBLE PRECISION), CAST(usage_record.wallet_gift_balance_before AS DOUBLE PRECISION) ) AS wallet_gift_balance_before, COALESCE( CAST(usage_settlement_snapshots.wallet_gift_balance_after AS DOUBLE PRECISION), CAST(usage_record.wallet_gift_balance_after AS DOUBLE PRECISION) ) AS wallet_gift_balance_after, CAST(usage_settlement_snapshots.provider_monthly_used_usd AS DOUBLE PRECISION) AS provider_monthly_used_usd, usage_record.provider_id, CAST( EXTRACT( EPOCH FROM COALESCE(usage_settlement_snapshots.finalized_at, usage_record.finalized_at) ) AS BIGINT ) AS finalized_at_unix_secs FROM "usage" AS usage_record LEFT JOIN usage_settlement_snapshots ON usage_settlement_snapshots.request_id = usage_record.request_id WHERE usage_record.request_id = $1 FOR UPDATE OF usage_record "#; const FINALIZE_USAGE_BILLING_SQL: &str = r#" UPDATE "usage" SET billing_status = $2, finalized_at = COALESCE(finalized_at, to_timestamp($3)) WHERE request_id = $1 "#; const UPSERT_USAGE_SETTLEMENT_SNAPSHOT_SQL: &str = r#" INSERT INTO usage_settlement_snapshots ( request_id, billing_status, wallet_id, wallet_balance_before, wallet_balance_after, wallet_recharge_balance_before, wallet_recharge_balance_after, wallet_gift_balance_before, wallet_gift_balance_after, provider_monthly_used_usd, finalized_at ) VALUES ( $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, CASE WHEN $11 IS NULL THEN NULL ELSE TO_TIMESTAMP($11::double precision) END ) ON CONFLICT (request_id) DO UPDATE SET billing_status = EXCLUDED.billing_status, wallet_id = COALESCE(EXCLUDED.wallet_id, usage_settlement_snapshots.wallet_id), wallet_balance_before = COALESCE( EXCLUDED.wallet_balance_before, usage_settlement_snapshots.wallet_balance_before ), wallet_balance_after = COALESCE( EXCLUDED.wallet_balance_after, usage_settlement_snapshots.wallet_balance_after ), wallet_recharge_balance_before = COALESCE( EXCLUDED.wallet_recharge_balance_before, usage_settlement_snapshots.wallet_recharge_balance_before ), wallet_recharge_balance_after = COALESCE( EXCLUDED.wallet_recharge_balance_after, usage_settlement_snapshots.wallet_recharge_balance_after ), wallet_gift_balance_before = COALESCE( EXCLUDED.wallet_gift_balance_before, usage_settlement_snapshots.wallet_gift_balance_before ), wallet_gift_balance_after = COALESCE( EXCLUDED.wallet_gift_balance_after, usage_settlement_snapshots.wallet_gift_balance_after ), provider_monthly_used_usd = COALESCE( EXCLUDED.provider_monthly_used_usd, usage_settlement_snapshots.provider_monthly_used_usd ), finalized_at = COALESCE(EXCLUDED.finalized_at, usage_settlement_snapshots.finalized_at), updated_at = NOW() "#; #[derive(Debug, Clone)] pub struct SqlxSettlementRepository { tx_runner: PostgresTransactionRunner, } impl SqlxSettlementRepository { pub fn new(pool: PgPool) -> Self { let tx_runner = PostgresTransactionRunner::new(pool); Self { tx_runner } } } fn settlement_from_row( row: &sqlx::postgres::PgRow, ) -> Result { Ok(StoredUsageSettlement { request_id: row.try_get("request_id").map_postgres_err()?, wallet_id: row.try_get("wallet_id").map_postgres_err()?, billing_status: row.try_get("billing_status").map_postgres_err()?, wallet_balance_before: row.try_get("wallet_balance_before").map_postgres_err()?, wallet_balance_after: row.try_get("wallet_balance_after").map_postgres_err()?, wallet_recharge_balance_before: row .try_get("wallet_recharge_balance_before") .map_postgres_err()?, wallet_recharge_balance_after: row .try_get("wallet_recharge_balance_after") .map_postgres_err()?, wallet_gift_balance_before: row .try_get("wallet_gift_balance_before") .map_postgres_err()?, wallet_gift_balance_after: row .try_get("wallet_gift_balance_after") .map_postgres_err()?, provider_monthly_used_usd: row .try_get("provider_monthly_used_usd") .map_postgres_err()?, finalized_at_unix_secs: row .try_get::, _>("finalized_at_unix_secs") .map_postgres_err()? .map(|value| value as u64), }) } async fn sync_usage_settlement_snapshot<'e, E>( executor: E, settlement: &StoredUsageSettlement, ) -> Result<(), DataLayerError> where E: sqlx::Executor<'e, Database = sqlx::Postgres>, { sqlx::query(UPSERT_USAGE_SETTLEMENT_SNAPSHOT_SQL) .bind(&settlement.request_id) .bind(&settlement.billing_status) .bind(settlement.wallet_id.as_deref()) .bind(settlement.wallet_balance_before) .bind(settlement.wallet_balance_after) .bind(settlement.wallet_recharge_balance_before) .bind(settlement.wallet_recharge_balance_after) .bind(settlement.wallet_gift_balance_before) .bind(settlement.wallet_gift_balance_after) .bind(settlement.provider_monthly_used_usd) .bind(settlement.finalized_at_unix_secs.map(|value| value as f64)) .execute(executor) .await .map_postgres_err()?; Ok(()) } #[async_trait] impl SettlementWriteRepository for SqlxSettlementRepository { async fn settle_usage( &self, input: UsageSettlementInput, ) -> Result, DataLayerError> { input.validate()?; self.tx_runner .run_read_write(|tx| { Box::pin(async move { let row = sqlx::query(FIND_USAGE_FOR_SETTLEMENT_SQL) .bind(&input.request_id) .fetch_optional(&mut **tx) .await .map_postgres_err()?; let Some(usage_row) = row else { return Ok(None); }; let current_billing_status: String = usage_row.try_get("billing_status").map_postgres_err()?; if current_billing_status == "settled" || current_billing_status == "void" { return settlement_from_row(&usage_row).map(Some); } let final_billing_status = if input.status == "completed" { "settled" } else { "void" }; let finalized_at = i64::try_from(input.finalized_at_unix_secs.unwrap_or_else(|| { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() .as_secs() })) .map_err(|_| { DataLayerError::InvalidInput("finalized_at overflow".to_string()) })?; let mut settlement = StoredUsageSettlement { request_id: input.request_id.clone(), wallet_id: None, billing_status: final_billing_status.to_string(), wallet_balance_before: None, wallet_balance_after: None, wallet_recharge_balance_before: None, wallet_recharge_balance_after: None, wallet_gift_balance_before: None, wallet_gift_balance_after: None, provider_monthly_used_usd: None, finalized_at_unix_secs: Some(finalized_at as u64), }; if final_billing_status == "settled" { let wallet_row = if let Some(api_key_id) = input .api_key_id .as_deref() .filter(|value| !value.is_empty()) { sqlx::query( r#" SELECT id, CAST(balance AS DOUBLE PRECISION) AS balance, CAST(gift_balance AS DOUBLE PRECISION) AS gift_balance, limit_mode FROM wallets WHERE api_key_id = $1 FOR UPDATE LIMIT 1 "#, ) .bind(api_key_id) .fetch_optional(&mut **tx) .await .map_postgres_err()? } else { None }; let wallet_row = if wallet_row.is_some() { wallet_row } else if let Some(user_id) = input.user_id.as_deref().filter(|value| !value.is_empty()) { sqlx::query( r#" SELECT id, CAST(balance AS DOUBLE PRECISION) AS balance, CAST(gift_balance AS DOUBLE PRECISION) AS gift_balance, limit_mode FROM wallets WHERE user_id = $1 FOR UPDATE LIMIT 1 "#, ) .bind(user_id) .fetch_optional(&mut **tx) .await .map_postgres_err()? } else { None }; if let Some(wallet_row) = wallet_row { let wallet_id: String = wallet_row.try_get("id").map_postgres_err()?; let before_recharge: f64 = wallet_row.try_get("balance").map_postgres_err()?; let before_gift: f64 = wallet_row.try_get("gift_balance").map_postgres_err()?; let limit_mode: String = wallet_row.try_get("limit_mode").map_postgres_err()?; let before_total = before_recharge + before_gift; let mut after_recharge = before_recharge; let mut after_gift = before_gift; if !limit_mode.eq_ignore_ascii_case("unlimited") { let gift_deduction = before_gift.max(0.0).min(input.total_cost_usd); let recharge_deduction = input.total_cost_usd - gift_deduction; after_gift = before_gift - gift_deduction; after_recharge = before_recharge - recharge_deduction; } sqlx::query( r#" UPDATE wallets SET balance = $2, gift_balance = $3, total_consumed = CAST(total_consumed AS DOUBLE PRECISION) + $4, updated_at = NOW() WHERE id = $1 "#, ) .bind(&wallet_id) .bind(after_recharge) .bind(after_gift) .bind(input.total_cost_usd) .execute(&mut **tx) .await .map_postgres_err()?; settlement.wallet_id = Some(wallet_id.clone()); settlement.wallet_balance_before = Some(before_total); settlement.wallet_balance_after = Some(after_recharge + after_gift); settlement.wallet_recharge_balance_before = Some(before_recharge); settlement.wallet_recharge_balance_after = Some(after_recharge); settlement.wallet_gift_balance_before = Some(before_gift); settlement.wallet_gift_balance_after = Some(after_gift); } if let Some(provider_id) = input .provider_id .as_deref() .filter(|value| !value.is_empty()) { let quota_row = sqlx::query( r#" UPDATE providers SET monthly_used_usd = COALESCE(monthly_used_usd, 0) + $2, updated_at = NOW() WHERE id = $1 RETURNING CAST(monthly_used_usd AS DOUBLE PRECISION) AS monthly_used_usd "#, ) .bind(provider_id) .bind(input.actual_total_cost_usd) .fetch_optional(&mut **tx) .await .map_postgres_err()?; settlement.provider_monthly_used_usd = quota_row.and_then(|row| row.try_get("monthly_used_usd").ok()); } } sync_usage_settlement_snapshot(&mut **tx, &settlement).await?; sqlx::query(FINALIZE_USAGE_BILLING_SQL) .bind(&input.request_id) .bind(final_billing_status) .bind(finalized_at) .execute(&mut **tx) .await .map_postgres_err()?; Ok(Some(settlement)) }) }) .await } } #[cfg(test)] mod tests { #[test] fn finalize_usage_billing_sql_does_not_require_usage_updated_at_column() { assert!(!super::FINALIZE_USAGE_BILLING_SQL.contains("updated_at")); } #[test] fn settlement_sql_reads_settlement_snapshots_before_legacy_usage_columns() { assert!( super::FIND_USAGE_FOR_SETTLEMENT_SQL.contains("LEFT JOIN usage_settlement_snapshots") ); assert!(super::FIND_USAGE_FOR_SETTLEMENT_SQL.contains( "COALESCE(usage_settlement_snapshots.billing_status, usage_record.billing_status)" )); assert!(super::FIND_USAGE_FOR_SETTLEMENT_SQL.contains("FOR UPDATE OF usage_record")); } #[test] fn settlement_sql_dual_writes_usage_settlement_snapshots() { assert!(super::UPSERT_USAGE_SETTLEMENT_SNAPSHOT_SQL .contains("INSERT INTO usage_settlement_snapshots")); assert!(super::UPSERT_USAGE_SETTLEMENT_SNAPSHOT_SQL.contains("provider_monthly_used_usd")); assert!(super::UPSERT_USAGE_SETTLEMENT_SNAPSHOT_SQL .contains("TO_TIMESTAMP($11::double precision)")); } #[test] fn settlement_sql_no_longer_dual_writes_wallet_snapshots_to_usage_rows() { let source = include_str!("sql.rs"); assert!(!source.contains("UPDATE \"usage\"\nSET\n wallet_id = $2")); } }