diff --git a/backend/.env.example b/backend/.env.example index e2007dd26..2d17eeca2 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -17,3 +17,9 @@ INACTIVITY_WATCHDOG_BATCH_SIZE=500 # webhooks (X-KYC-Signature header). Required: without it /api/kyc/webhook # rejects every request with 503. KYC_WEBHOOK_SECRET= + +# Stellar Anchor API for fiat payouts +# Base URL of the Stellar Anchor's fiat off-ramp API +ANCHOR_API_URL= +# API key for authenticating with the anchor endpoint +ANCHOR_API_KEY= diff --git a/backend/migrations/20260726000000_add_anchor_payout_id.down.sql b/backend/migrations/20260726000000_add_anchor_payout_id.down.sql new file mode 100644 index 000000000..b53d445a3 --- /dev/null +++ b/backend/migrations/20260726000000_add_anchor_payout_id.down.sql @@ -0,0 +1,4 @@ +DROP INDEX IF EXISTS payouts_anchor_payout_id_idx; + +ALTER TABLE payouts +DROP COLUMN IF EXISTS anchor_payout_id; diff --git a/backend/migrations/20260726000000_add_anchor_payout_id.up.sql b/backend/migrations/20260726000000_add_anchor_payout_id.up.sql new file mode 100644 index 000000000..e0831ee2e --- /dev/null +++ b/backend/migrations/20260726000000_add_anchor_payout_id.up.sql @@ -0,0 +1,7 @@ +-- Add anchor_payout_id column to payouts table for tracking Stellar Anchor +-- transaction references. This allows the system to correlate internal payout +-- records with the anchor's external transaction ID for status polling. +ALTER TABLE payouts +ADD COLUMN anchor_payout_id TEXT; + +CREATE INDEX payouts_anchor_payout_id_idx ON payouts (anchor_payout_id); diff --git a/backend/src/api.rs b/backend/src/api.rs index e0e877929..9a04a1dad 100644 --- a/backend/src/api.rs +++ b/backend/src/api.rs @@ -19,7 +19,7 @@ use rust_decimal::Decimal; use serde::{Deserialize, Serialize}; use std::sync::Arc; use tower_http::cors::CorsLayer; -use tracing::{error, warn}; +use tracing::{error, info, warn}; use uuid::Uuid; use crate::auth::{jwt_auth_middleware, signature_auth_middleware, Claims}; @@ -122,6 +122,7 @@ pub struct PayoutRow { pub amount: String, pub payout_type: String, pub status: String, + pub anchor_payout_id: Option, pub created_at: DateTime, } @@ -1251,8 +1252,8 @@ async fn ping_plan( .into_response() } // Handler: Trigger Payout -// Contributors: Implement calculating final payout with yield, parsing fiat payout details, -// submitting fiat payouts to AnchorRegistry, and marking the plan inactive +// Initiates payouts for all beneficiaries — calls AnchorRegistry for fiat payouts +// and marks the plan as PAID_OUT. async fn trigger_payout( State(state): State>, Json(payload): Json, @@ -1404,11 +1405,48 @@ async fn trigger_payout( } } + let anchor_payout_id: Option = if is_fiat { + let (beneficiary_name, fiat_currency, bank_name, account_number) = + parse_fiat_anchor_info(&b.fiat_anchor_info, &b.wallet_address); + let token_amount_f64 = share.to_string().parse::().unwrap_or(0.0); + let req = crate::stellar_anchor::AnchorPayoutRequest { + beneficiary_address: b.wallet_address.clone(), + beneficiary_name, + token: plan.token_address.clone(), + token_amount: token_amount_f64, + fiat_currency, + bank_name, + account_number, + }; + match state.anchor.create_payout(req).await { + Ok(anchor_payout) => { + info!( + anchor_payout_id = %anchor_payout.id, + beneficiary = %b.wallet_address, + plan_id = %plan.id, + "Anchor payout initiated" + ); + Some(anchor_payout.id) + } + Err(e) => { + error!( + error = %e, + beneficiary = %b.wallet_address, + plan_id = %plan.id, + "Failed to initiate anchor payout" + ); + None + } + } + } else { + None + }; + let payout_row = match sqlx::query_as::<_, PayoutRow>( r#" - INSERT INTO payouts (plan_id, beneficiary_address, amount, payout_type, status) - VALUES ($1, $2, $3, $4::payout_type, $5::payout_status) - RETURNING id, plan_id, beneficiary_address, amount::text, payout_type::text, status::text, created_at + INSERT INTO payouts (plan_id, beneficiary_address, amount, payout_type, status, anchor_payout_id) + VALUES ($1, $2, $3, $4::payout_type, $5::payout_status, $6) + RETURNING id, plan_id, beneficiary_address, amount::text, payout_type::text, status::text, anchor_payout_id, created_at "#, ) .bind(plan.id) @@ -1416,6 +1454,7 @@ async fn trigger_payout( .bind(share) .bind(payout_type_str) .bind(payout_status_str) + .bind(&anchor_payout_id) .fetch_one(&mut *tx) .await { Ok(row) => row, @@ -1428,22 +1467,7 @@ async fn trigger_payout( } }; - // Initiate payout distribution if is_fiat { - let (beneficiary_name, fiat_currency, bank_name, account_number) = - parse_fiat_anchor_info(&b.fiat_anchor_info, &b.wallet_address); - let token_amount_f64 = share.to_string().parse::().unwrap_or(0.0); - let req = crate::stellar_anchor::AnchorPayoutRequest { - beneficiary_address: b.wallet_address.clone(), - beneficiary_name, - token: plan.token_address.clone(), - token_amount: token_amount_f64, - fiat_currency, - bank_name, - account_number, - }; - state.anchor.create_payout(req); - if b.fiat_daily_limit > Decimal::ZERO { let today = chrono::Utc::now().naive_utc().date(); if let Err(e) = sqlx::query( @@ -1659,6 +1683,7 @@ async fn get_anchor_payouts( amount::text AS amount, payout_type::text AS payout_type, status::text AS status, + anchor_payout_id, created_at FROM payouts WHERE ($1::text IS NULL OR beneficiary_address = $1) diff --git a/backend/src/config.rs b/backend/src/config.rs index d8d79585a..546fab425 100644 --- a/backend/src/config.rs +++ b/backend/src/config.rs @@ -11,6 +11,8 @@ pub struct Config { pub kyc_webhook_secret: Option, pub stellar_horizon_url: String, pub fiat_daily_limit_default: rust_decimal::Decimal, + pub anchor_api_url: Option, + pub anchor_api_key: Option, } impl Config { @@ -44,6 +46,16 @@ impl Config { .filter(|value| !value.is_empty()) .unwrap_or_else(|| "https://horizon-testnet.stellar.org".to_string()); + let anchor_api_url = std::env::var("ANCHOR_API_URL") + .ok() + .map(|value| value.trim().to_string()) + .filter(|value| !value.is_empty()); + + let anchor_api_key = std::env::var("ANCHOR_API_KEY") + .ok() + .map(|value| value.trim().to_string()) + .filter(|value| !value.is_empty()); + Ok(Config { port, database_url, @@ -52,6 +64,8 @@ impl Config { kyc_webhook_secret, stellar_horizon_url, fiat_daily_limit_default, + anchor_api_url, + anchor_api_key, }) } } diff --git a/backend/src/main.rs b/backend/src/main.rs index 3d416ac24..479c7821d 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -60,7 +60,10 @@ async fn main() -> Result<(), Box> { let (kyc_tx, _) = tokio::sync::broadcast::channel(100); // Initialize state let state = Arc::new(AppState { - anchor: Arc::new(inheritx_backend::stellar_anchor::AnchorRegistry::new()), + anchor: Arc::new(inheritx_backend::stellar_anchor::AnchorRegistry::new( + config.anchor_api_url.clone(), + config.anchor_api_key.clone(), + )), db_pool: db_pool.clone(), kyc_webhook_secret: config.kyc_webhook_secret.clone(), apy_config: inheritx_backend::yield_calculator::ApyConfig::from_env(), diff --git a/backend/src/stellar_anchor.rs b/backend/src/stellar_anchor.rs index cae88c840..815764a5f 100644 --- a/backend/src/stellar_anchor.rs +++ b/backend/src/stellar_anchor.rs @@ -1,5 +1,6 @@ use serde::{Deserialize, Serialize}; use std::sync::Arc; +use tracing::{error, info, warn}; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct AnchorPayoutRequest { @@ -13,6 +14,7 @@ pub struct AnchorPayoutRequest { } #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq)] +#[serde(rename_all = "snake_case")] pub enum AnchorPayoutStatus { Pending, Processing, @@ -20,6 +22,17 @@ pub enum AnchorPayoutStatus { Failed, } +impl AnchorPayoutStatus { + pub fn as_str(&self) -> &'static str { + match self { + AnchorPayoutStatus::Pending => "pending", + AnchorPayoutStatus::Processing => "processing", + AnchorPayoutStatus::Completed => "completed", + AnchorPayoutStatus::Failed => "failed", + } + } +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct AnchorPayout { pub id: String, @@ -32,39 +45,404 @@ pub struct AnchorPayout { pub updated_at: String, } -#[derive(Default)] -pub struct AnchorRegistry; +#[derive(Debug, Serialize, Deserialize)] +struct AnchorTransactionRequest { + amount: String, + asset_code: String, + destination_asset: String, + sender_id: String, + fields: TransactionFields, +} + +#[derive(Debug, Serialize, Deserialize)] +struct TransactionFields { + #[serde(rename = "transaction")] + transaction: TransactionDetail, +} + +#[derive(Debug, Serialize, Deserialize)] +struct TransactionDetail { + beneficiary_name: String, + bank_name: String, + account_number: String, +} + +#[derive(Debug, Deserialize)] +struct AnchorTransactionResponse { + id: String, + status: String, + #[serde(default)] + amount_in: Option, + #[serde(default)] + amount_out: Option, + #[serde(default)] + amount_fee: Option, + #[serde(default)] + #[allow(dead_code)] + stellar_transaction_id: Option, +} + +#[derive(Debug, Deserialize)] +struct AnchorTransactionStatusResponse { + transaction: AnchorTransactionStatus, +} + +#[derive(Debug, Deserialize)] +struct AnchorTransactionStatus { + id: String, + status: String, + #[serde(default)] + amount_in: Option, + #[serde(default)] + amount_out: Option, + #[serde(default)] + amount_fee: Option, + #[serde(default)] + stellar_account: Option, + #[serde(default)] + updated_at: Option, +} + +#[derive(Debug, Deserialize)] +struct AnchorTransactionListResponse { + transactions: Vec, +} + +#[derive(Debug, Deserialize)] +struct AnchorTransactionListItem { + id: String, + status: String, + #[serde(default)] + amount_in: Option, + #[serde(default)] + amount_out: Option, + #[serde(default)] + created_at: Option, +} + +#[derive(Debug, thiserror::Error)] +pub enum AnchorError { + #[error("HTTP request failed: {0}")] + Http(#[from] reqwest::Error), + #[error("Anchor returned error status {status}: {body}")] + Api { status: u16, body: String }, + #[error("Anchor API not configured")] + NotConfigured, + #[error("Payout not found")] + NotFound, + #[error("Invalid response: {0}")] + InvalidResponse(String), +} + +pub struct AnchorRegistry { + client: reqwest::Client, + api_url: Option, + api_key: Option, +} impl AnchorRegistry { - pub fn new() -> Self { - Self + pub fn new(api_url: Option, api_key: Option) -> Self { + Self { + client: reqwest::Client::new(), + api_url, + api_key, + } + } + + pub async fn create_payout( + self: &Arc, + req: AnchorPayoutRequest, + ) -> Result { + let api_url = match &self.api_url { + Some(url) => url.trim_end_matches('/').to_string(), + None => { + warn!( + "Anchor API not configured — returning simulated payout for {}", + req.beneficiary_address + ); + return Ok(simulated_payout(req)); + } + }; + + let payload = AnchorTransactionRequest { + amount: format!("{:.7}", req.token_amount), + asset_code: req.token.clone(), + destination_asset: req.fiat_currency.clone(), + sender_id: req.beneficiary_address.clone(), + fields: TransactionFields { + transaction: TransactionDetail { + beneficiary_name: req.beneficiary_name.clone(), + bank_name: req.bank_name.clone(), + account_number: req.account_number.clone(), + }, + }, + }; + + let mut request_builder = self + .client + .post(format!("{api_url}/transactions")) + .json(&payload); + + if let Some(key) = &self.api_key { + request_builder = request_builder.header("Authorization", format!("Bearer {key}")); + } + + let response = match request_builder.send().await { + Ok(resp) => resp, + Err(e) => { + error!(error = %e, beneficiary = %req.beneficiary_address, "Failed to send create payout request to anchor"); + return Err(AnchorError::Http(e)); + } + }; + + let status = response.status(); + if !status.is_success() { + let body = response.text().await.unwrap_or_default(); + error!( + status = %status, + body = %body, + beneficiary = %req.beneficiary_address, + "Anchor API returned error" + ); + return Err(AnchorError::Api { + status: status.as_u16(), + body, + }); + } + + match response.json::().await { + Ok(anchor_resp) => { + info!( + anchor_tx_id = %anchor_resp.id, + status = %anchor_resp.status, + beneficiary = %req.beneficiary_address, + "Anchor payout created successfully" + ); + + let exchange_rate = if let (Some(_in_val), Some(out_val)) = + (&anchor_resp.amount_in, &anchor_resp.amount_out) + { + if req.token_amount > 0.0 { + out_val.parse::().unwrap_or(1.0) / req.token_amount + } else { + 1.0 + } + } else { + 1.0 + }; + + let fiat_amount = anchor_resp + .amount_out + .as_deref() + .and_then(|v| v.parse::().ok()) + .unwrap_or(req.token_amount * exchange_rate); + + let anchor_fee = anchor_resp + .amount_fee + .as_deref() + .and_then(|v| v.parse::().ok()) + .unwrap_or(0.0); + + let mapped_status = map_anchor_status(&anchor_resp.status); + let now = chrono::Utc::now().to_rfc3339(); + + Ok(AnchorPayout { + id: anchor_resp.id, + request: req, + exchange_rate, + fiat_amount, + anchor_fee_usd: anchor_fee, + status: mapped_status, + created_at: now.clone(), + updated_at: now, + }) + } + Err(e) => { + error!(error = %e, "Failed to parse anchor create payout response"); + Err(AnchorError::InvalidResponse(e.to_string())) + } + } + } + + pub async fn get_payout(&self, id: &str) -> Result { + let api_url = match &self.api_url { + Some(url) => url.trim_end_matches('/').to_string(), + None => return Err(AnchorError::NotConfigured), + }; + + let mut request_builder = self.client.get(format!("{api_url}/transactions/{id}")); + + if let Some(key) = &self.api_key { + request_builder = request_builder.header("Authorization", format!("Bearer {key}")); + } + + let response = match request_builder.send().await { + Ok(resp) => resp, + Err(e) => { + error!(error = %e, anchor_id = %id, "Failed to get payout from anchor"); + return Err(AnchorError::Http(e)); + } + }; + + let status = response.status(); + if status == reqwest::StatusCode::NOT_FOUND { + return Err(AnchorError::NotFound); + } + if !status.is_success() { + let body = response.text().await.unwrap_or_default(); + return Err(AnchorError::Api { + status: status.as_u16(), + body, + }); + } + + match response.json::().await { + Ok(status_resp) => { + let tx = status_resp.transaction; + let now = chrono::Utc::now().to_rfc3339(); + Ok(AnchorPayout { + id: tx.id, + request: AnchorPayoutRequest { + beneficiary_address: tx.stellar_account.clone().unwrap_or_default(), + beneficiary_name: String::new(), + token: String::new(), + token_amount: tx + .amount_in + .as_deref() + .and_then(|v| v.parse().ok()) + .unwrap_or(0.0), + fiat_currency: String::new(), + bank_name: String::new(), + account_number: String::new(), + }, + exchange_rate: 1.0, + fiat_amount: tx + .amount_out + .as_deref() + .and_then(|v| v.parse().ok()) + .unwrap_or(0.0), + anchor_fee_usd: tx + .amount_fee + .as_deref() + .and_then(|v| v.parse().ok()) + .unwrap_or(0.0), + status: map_anchor_status(&tx.status), + created_at: now.clone(), + updated_at: tx.updated_at.unwrap_or(now), + }) + } + Err(e) => { + error!(error = %e, anchor_id = %id, "Failed to parse anchor payout status response"); + Err(AnchorError::InvalidResponse(e.to_string())) + } + } } - /// Simulate creating an anchor payout request. - /// Contributors: Implement the registry storage, rate matching, fees, and async status update thread. - pub fn create_payout(self: &Arc, req: AnchorPayoutRequest) -> AnchorPayout { - // TODO: Implement anchor payout off-ramp creation and state machine transition - AnchorPayout { - id: "".to_string(), - request: req, - exchange_rate: 1.0, - fiat_amount: 0.0, - anchor_fee_usd: 0.0, - status: AnchorPayoutStatus::Pending, - created_at: "".to_string(), - updated_at: "".to_string(), + pub async fn list_payouts( + &self, + address: Option, + ) -> Result, AnchorError> { + let api_url = match &self.api_url { + Some(url) => url.trim_end_matches('/').to_string(), + None => return Err(AnchorError::NotConfigured), + }; + + let mut request_builder = self.client.get(format!("{api_url}/transactions")); + + if let Some(ref addr) = address { + request_builder = request_builder.query(&[("sender_id", addr.as_str())]); + } + + if let Some(key) = &self.api_key { + request_builder = request_builder.header("Authorization", format!("Bearer {key}")); + } + + let response = match request_builder.send().await { + Ok(resp) => resp, + Err(e) => { + error!(error = %e, "Failed to list payouts from anchor"); + return Err(AnchorError::Http(e)); + } + }; + + let status = response.status(); + if !status.is_success() { + let body = response.text().await.unwrap_or_default(); + return Err(AnchorError::Api { + status: status.as_u16(), + body, + }); + } + + match response.json::().await { + Ok(list_resp) => { + let now = chrono::Utc::now().to_rfc3339(); + let payouts = list_resp + .transactions + .into_iter() + .map(|tx| AnchorPayout { + id: tx.id, + request: AnchorPayoutRequest { + beneficiary_address: String::new(), + beneficiary_name: String::new(), + token: String::new(), + token_amount: tx + .amount_in + .as_deref() + .and_then(|v| v.parse().ok()) + .unwrap_or(0.0), + fiat_currency: String::new(), + bank_name: String::new(), + account_number: String::new(), + }, + exchange_rate: 1.0, + fiat_amount: tx + .amount_out + .as_deref() + .and_then(|v| v.parse().ok()) + .unwrap_or(0.0), + anchor_fee_usd: 0.0, + status: map_anchor_status(&tx.status), + created_at: tx.created_at.clone().unwrap_or_else(|| now.clone()), + updated_at: tx.created_at.unwrap_or(now.clone()), + }) + .collect(); + + Ok(payouts) + } + Err(e) => { + error!(error = %e, "Failed to parse anchor list payouts response"); + Err(AnchorError::InvalidResponse(e.to_string())) + } } } +} - /// Retrieve anchor payout by transaction ID. - pub fn get_payout(&self, _id: &str) -> Option { - // TODO: Implement get payout logic - None +fn map_anchor_status(status: &str) -> AnchorPayoutStatus { + match status.to_lowercase().as_str() { + "pending" => AnchorPayoutStatus::Pending, + "processing" | "in_progress" => AnchorPayoutStatus::Processing, + "completed" | "success" => AnchorPayoutStatus::Completed, + "failed" | "error" | "rejected" => AnchorPayoutStatus::Failed, + _ => { + warn!(status = %status, "Unknown anchor payout status, defaulting to Pending"); + AnchorPayoutStatus::Pending + } } +} - /// List all anchor payouts. - pub fn list_payouts(&self, _address: Option) -> Vec { - // TODO: Implement listing payouts - Vec::new() +/// Creates a simulated payout response when the anchor API is not configured. +/// This allows the system to function in development/test mode. +fn simulated_payout(req: AnchorPayoutRequest) -> AnchorPayout { + let now = chrono::Utc::now().to_rfc3339(); + AnchorPayout { + id: uuid::Uuid::new_v4().to_string(), + request: req, + exchange_rate: 1.0, + fiat_amount: 0.0, + anchor_fee_usd: 0.0, + status: AnchorPayoutStatus::Pending, + created_at: now.clone(), + updated_at: now, } } diff --git a/backend/tests/api_tests.rs b/backend/tests/api_tests.rs index 52ba0da78..0f42c2dbd 100644 --- a/backend/tests/api_tests.rs +++ b/backend/tests/api_tests.rs @@ -42,7 +42,9 @@ fn setup_app_with_cache(plan_cache: PlanCache) -> axum::Router { .connect_lazy(&database_url) .unwrap(); let state = Arc::new(AppState { - anchor: Arc::new(inheritx_backend::stellar_anchor::AnchorRegistry::new()), + anchor: Arc::new(inheritx_backend::stellar_anchor::AnchorRegistry::new( + None, None, + )), db_pool, kyc_tx: tokio::sync::broadcast::channel(16).0, kyc_webhook_secret: None, @@ -509,7 +511,7 @@ async fn test_health_endpoint_without_db_yields_service_unavailable() { .connect_lazy("postgres://localhost:1/nonexistent") .unwrap(); let state = Arc::new(AppState { - anchor: Arc::new(inheritx_backend::stellar_anchor::AnchorRegistry::new()), + anchor: Arc::new(inheritx_backend::stellar_anchor::AnchorRegistry::new(None, None)), db_pool, kyc_tx: tokio::sync::broadcast::channel(16).0, kyc_webhook_secret: None, @@ -560,7 +562,7 @@ async fn test_get_current_rate_cached() { .connect_lazy("postgres://postgres:password@localhost:5432/test") .unwrap(); let state = Arc::new(AppState { - anchor: Arc::new(inheritx_backend::stellar_anchor::AnchorRegistry::new()), + anchor: Arc::new(inheritx_backend::stellar_anchor::AnchorRegistry::new(None, None)), db_pool, kyc_tx: tokio::sync::broadcast::channel(16).0, kyc_webhook_secret: None, diff --git a/backend/tests/kyc_webhook_test.rs b/backend/tests/kyc_webhook_test.rs index a061afd3f..9facea22a 100644 --- a/backend/tests/kyc_webhook_test.rs +++ b/backend/tests/kyc_webhook_test.rs @@ -27,7 +27,7 @@ fn test_state(secret: Option<&str>) -> std::sync::Arc