feat(backend): fix #812 implement trigger payout endpoint with grace …

backend/src/api.rs

@@ -852,12 +852,279 @@ async fn ping_plan(
 // Contributors: Implement calculating final payout with yield, parsing fiat payout details,
 // submitting fiat payouts to AnchorRegistry, and marking the plan inactive
 async fn trigger_payout(
-    State(_state): State<Arc<AppState>>,
-    Json(_payload): Json<PayoutRequest>,
+    State(state): State<Arc<AppState>>,
+    Json(payload): Json<PayoutRequest>,
 ) -> impl IntoResponse {
+    // 1. Begin database transaction
+    let mut tx = match state.db_pool.begin().await {
+        Ok(tx) => tx,
+        Err(e) => {
+            error!(error = %e, "Failed to begin database transaction");
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(serde_json::json!({ "error": format!("Failed to begin database transaction: {}", e) })),
+            ).into_response();
+        }
+    };
+
+    // 2. Fetch the active plan for the owner
+    let plan = match sqlx::query_as::<_, PlanRow>(
+        "SELECT id, owner_address, token_address, amount, grace_period, grace_period_seconds, earn_yield, last_ping, is_active, status, yield_rate_bps, accrued_yield, created_at FROM plans WHERE owner_address = $1 AND is_active = true FOR UPDATE",
+    )
+    .bind(&payload.owner)
+    .fetch_optional(&mut *tx)
+    .await
+    {
+        Ok(Some(p)) => p,
+        Ok(None) => {
+            return (
+                StatusCode::NOT_FOUND,
+                Json(serde_json::json!({ "error": "No active plan found for this owner" })),
+            ).into_response();
+        }
+        Err(e) => {
+            error!(owner = %payload.owner, error = %e, "Database error fetching plan");
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(serde_json::json!({ "error": format!("Database error: {}", e) })),
+            ).into_response();
+        }
+    };
+
+    // 3. Verify if the grace period has elapsed
+    let now = chrono::Utc::now().timestamp();
+    let deadline = plan.last_ping + plan.grace_period_seconds;
+    if now < deadline {
+        return (
+            StatusCode::BAD_REQUEST,
+            Json(serde_json::json!({ "error": "Grace period has not elapsed" })),
+        )
+            .into_response();
+    }
+
+    // 4. Compute final locked amount + yield
+    let accrued_yield_f64 = compute_projected_accrued_yield(&plan);
+    let accrued_yield_dec = match Decimal::from_f64_retain(accrued_yield_f64) {
+        Some(d) => d.normalize(),
+        None => Decimal::ZERO,
+    };
+    let total_payout_dec = plan.amount + accrued_yield_dec;
+
+    // 5. Load beneficiaries for the plan
+    let beneficiaries_rows = match sqlx::query_as::<_, BeneficiaryRow>(
+        r#"
+        SELECT id, plan_id, wallet_address, allocation_bps, fiat_anchor_info
+        FROM beneficiaries
+        WHERE plan_id = $1
+        "#,
+    )
+    .bind(plan.id)
+    .fetch_all(&mut *tx)
+    .await
+    {
+        Ok(rows) => rows,
+        Err(e) => {
+            error!(plan_id = %plan.id, error = %e, "Failed to load beneficiaries");
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(
+                    serde_json::json!({ "error": format!("Failed to load beneficiaries: {}", e) }),
+                ),
+            )
+                .into_response();
+        }
+    };
+
+    let n = beneficiaries_rows.len();
+    if n == 0 {
+        return (
+            StatusCode::BAD_REQUEST,
+            Json(serde_json::json!({ "error": "Plan has no beneficiaries" })),
+        )
+            .into_response();
+    }
+
+    // 6. Iterate over beneficiaries and insert payout records
+    let mut remaining = total_payout_dec;
+    let mut payout_rows = Vec::with_capacity(n);
+
+    for (i, b) in beneficiaries_rows.iter().enumerate() {
+        let share = if i == n - 1 {
+            remaining
+        } else {
+            let amount =
+                (total_payout_dec * Decimal::from(b.allocation_bps)) / Decimal::from(10000);
+            let amount = amount.floor();
+            remaining -= amount;
+            amount
+        };
+
+        if share <= Decimal::ZERO {
+            continue;
+        }
+
+        let is_fiat = !b.fiat_anchor_info.trim().is_empty();
+        let payout_type_str = if is_fiat { "fiat" } else { "crypto" };
+        let payout_status_str = "processing";
+
+        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
+            "#,
+        )
+        .bind(plan.id)
+        .bind(&b.wallet_address)
+        .bind(share)
+        .bind(payout_type_str)
+        .bind(payout_status_str)
+        .fetch_one(&mut *tx)
+        .await {
+            Ok(row) => row,
+            Err(e) => {
+                error!(plan_id = %plan.id, beneficiary = %b.wallet_address, error = %e, "Failed to insert payout record");
+                return (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    Json(serde_json::json!({ "error": format!("Failed to insert payout record: {}", e) })),
+                ).into_response();
+            }
+        };
+
+        // 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::<f64>().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);
+        } else {
+            tracing::info!(
+                plan_id = %plan.id,
+                beneficiary = %b.wallet_address,
+                amount = %share,
+                "Initiated on-chain crypto distribution"
+            );
+        }
+
+        payout_rows.push(payout_row);
+    }
+
+    // 7. Mark the plan as inactive
+    if let Err(e) = sqlx::query(
+        "UPDATE plans SET is_active = false, status = 'PAID_OUT', accrued_yield = $1, last_ping = $2 WHERE id = $3"
+    )
+    .bind(accrued_yield_dec)
+    .bind(now)
+    .bind(plan.id)
+    .execute(&mut *tx)
+    .await {
+        error!(plan_id = %plan.id, error = %e, "Failed to mark plan as inactive");
+        return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(serde_json::json!({ "error": format!("Failed to mark plan as inactive: {}", e) })),
+        ).into_response();
+    }
+
+    // 8. Commit transaction
+    if let Err(e) = tx.commit().await {
+        error!(error = %e, "Failed to commit database transaction");
+        return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(serde_json::json!({ "error": format!("Failed to commit database transaction: {}", e) })),
+        ).into_response();
+    }
+
+    // 9. Invalidate cache
+    let beneficiary_addresses: Vec<String> = beneficiaries_rows
+        .iter()
+        .map(|b| b.wallet_address.clone())
+        .collect();
+    invalidate_plan_cache(
+        &state.plan_cache,
+        &plan.owner_address,
+        &beneficiary_addresses,
+    )
+    .await;
+
+    (StatusCode::OK, Json(payout_rows)).into_response()
+}
+
+fn parse_fiat_anchor_info(info: &str, wallet_address: &str) -> (String, String, String, String) {
+    #[derive(Deserialize)]
+    struct LocalAnchorInfo {
+        name: Option<String>,
+        currency: Option<String>,
+        bank: Option<String>,
+        account: Option<String>,
+    }
+
+    if let Ok(parsed) = serde_json::from_str::<LocalAnchorInfo>(info) {
+        return (
+            parsed.name.unwrap_or_else(|| "Beneficiary".to_string()),
+            parsed.currency.unwrap_or_else(|| "USD".to_string()),
+            parsed
+                .bank
+                .unwrap_or_else(|| "Stellar Anchor Bank".to_string()),
+            parsed.account.unwrap_or_else(|| {
+                format!("ACC-{}", &wallet_address[..8.min(wallet_address.len())])
+            }),
+        );
+    }
+
+    let parts: Vec<&str> = if info.contains(';') {
+        info.split(';').map(|s| s.trim()).collect()
+    } else if info.contains(',') {
+        info.split(',').map(|s| s.trim()).collect()
+    } else {
+        vec![info]
+    };
+
+    if parts.len() >= 3 {
+        return (
+            "Beneficiary".to_string(),
+            parts[2].to_string(),
+            parts[0].to_string(),
+            parts[1].to_string(),
+        );
+    }
+
+    let info_upper = info.to_uppercase();
+    let fiat_currency = if info_upper.contains("NGN") {
+        "NGN"
+    } else if info_upper.contains("KES") {
+        "KES"
+    } else if info_upper.contains("BRL") {
+        "BRL"
+    } else if info_upper.contains("PHP") {
+        "PHP"
+    } else if info_upper.contains("EUR") {
+        "EUR"
+    } else {
+        "USD"
+    };
+
+    let bank_name = if info.trim().is_empty() {
+        "Stellar Anchor Bank".to_string()
+    } else {
+        info.to_string()
+    };
+
+    let account_number = format!("ACC-{}", &wallet_address[..8.min(wallet_address.len())]);
+
     (
-        StatusCode::NOT_IMPLEMENTED,
-        "Payout trigger logic not implemented",
+        "Beneficiary".to_string(),
+        fiat_currency.to_string(),
+        bank_name,
+        account_number,
     )
 }
 //

backend/tests/api_tests.rs

@@ -318,3 +318,83 @@ async fn test_ping_plan_invalid_signature() {
 
     assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
 }
+
+#[tokio::test]
+async fn test_trigger_payout_invalid_signature() {
+    let app = setup_app();
+
+    let body = json!({
+        "owner": "GDIW7P2XUXC4XZB452Y5Z774N4V27PUDHWTKWTQZ3KHYUGB743WEXG7T"
+    })
+    .to_string();
+
+    // Generate a valid signature for a different body
+    let (public_key, _correct_sig) = generate_valid_signature(
+        &body,
+        "0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef",
+    );
+    let (_different_pub_key, invalid_signature) = generate_valid_signature(
+        "different body",
+        "0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef",
+    );
+
+    let response = app
+        .oneshot(
+            Request::builder()
+                .method(http::Method::POST)
+                .uri("/api/plans/payout")
+                .header(http::header::CONTENT_TYPE, "application/json")
+                .header("X-Public-Key", public_key)
+                .header("X-Signature", invalid_signature)
+                .body(Body::from(body))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+    let status = response.status();
+    if status != StatusCode::UNAUTHORIZED {
+        let body_bytes = axum::body::to_bytes(response.into_body(), usize::MAX)
+            .await
+            .unwrap();
+        let body_str = String::from_utf8(body_bytes.to_vec()).unwrap();
+        panic!(
+            "Expected 401 Unauthorized, got {}. Response body: {}",
+            status, body_str
+        );
+    }
+}
+
+#[tokio::test]
+async fn test_trigger_payout_valid_signature_not_found() {
+    let app = setup_app();
+
+    let body = json!({
+        "owner": "owner_address"
+    })
+    .to_string();
+
+    let (public_key, signature) = generate_valid_signature(
+        &body,
+        "0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef",
+    );
+
+    let response = app
+        .oneshot(
+            Request::builder()
+                .method(http::Method::POST)
+                .uri("/api/plans/payout")
+                .header(http::header::CONTENT_TYPE, "application/json")
+                .header("X-Public-Key", public_key)
+                .header("X-Signature", signature)
+                .body(Body::from(body))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+    // Since the database is not actually running, this should return a DB connection error (500)
+    // rather than an unauthorized error (401), proving that the request successfully passed auth
+    // and reached the handler.
+    assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
+}