use std::time::Duration; use futures_util::stream::{self, StreamExt}; use sqlx::Row; use tokio::time::sleep; use crate::domain::constants::{ ORDER_STATUS_DELIVERED_PENDING_APPROVAL, ORDER_STATUS_PAID_PENDING_DELIVERY, ORDER_STATUS_PAYMENT_FAILED, ORDER_STATUS_PENDING_PAYMENT, ORDER_STATUS_SETTLED, }; use crate::interfaces::http::api::{AppState, RealtimeEvent}; pub async fn spawn_reconciliation_worker(state: AppState) { tracing::info!("Reconciliation worker active."); let mut interval = tokio::time::interval(Duration::from_secs(300)); interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { interval.tick().await; metrics::counter!("rtix_reconciliation_cycles_total").increment(1); match reconcile_pending_orders(&state).await { Ok(count) if count > 0 => { tracing::info!("Reconciliation cycle resolved {} orders.", count); metrics::counter!("rtix_reconciliation_resolved_total").increment(count as u64); } Ok(_) => {} Err(e) => { tracing::error!("Reconciliation cycle failed: {:?}", e); metrics::counter!("rtix_reconciliation_errors_total").increment(1); // On DB error, wait a bit before next tick sleep(Duration::from_secs(10)).await; } } } } pub async fn reconcile_pending_orders(state: &AppState) -> Result { let mut reconciled_count = 0; let stale_pending_orders = sqlx::query( "UPDATE orders SET status = $1 WHERE status = $2 AND created_at < NOW() - INTERVAL '60 minutes' RETURNING transaction_id, merchant_id", ) .bind(ORDER_STATUS_PAYMENT_FAILED) .bind(ORDER_STATUS_PENDING_PAYMENT) .fetch_all(&state.pool) .await?; if !stale_pending_orders.is_empty() { let stale_count = stale_pending_orders.len(); let stale_pending_orders_stream = stream::iter(stale_pending_orders) .map(|row| { let state = state.clone(); async move { let txnid: String = row.get("transaction_id"); let merchant_id: String = row.get("merchant_id"); if let Ok(mut conn) = state.pool.acquire().await { crate::domain::audit::log_risk_event( &mut conn, Some(&txnid), &merchant_id, "PAYMENT_TIMEOUT", "MEDIUM", Some("Pending payment expired after 30 minutes without a verified callback."), None, None, None, Some(&state.tx), ) .await; } let _ = state.tx.send(RealtimeEvent::OrderStatusChanged { transaction_id: txnid, merchant_id, new_status: ORDER_STATUS_PAYMENT_FAILED.to_string(), }); } }) .buffer_unordered(10); stale_pending_orders_stream.collect::>().await; reconciled_count += stale_count; } let aged_shipped_orders = sqlx::query( "UPDATE orders SET status = $1 WHERE status = $2 AND shipped_at IS NOT NULL AND shipped_at < NOW() - INTERVAL '7 days' RETURNING transaction_id, merchant_id", ) .bind(ORDER_STATUS_DELIVERED_PENDING_APPROVAL) .bind(ORDER_STATUS_PAID_PENDING_DELIVERY) .fetch_all(&state.pool) .await?; if !aged_shipped_orders.is_empty() { let aged_count = aged_shipped_orders.len(); let aged_shipped_orders_stream = stream::iter(aged_shipped_orders) .map(|row| { let state = state.clone(); async move { let txnid: String = row.get("transaction_id"); let merchant_id: String = row.get("merchant_id"); if let Ok(mut conn) = state.pool.acquire().await { crate::domain::audit::log_risk_event( &mut conn, Some(&txnid), &merchant_id, "AUTO_DELIVERY_CONFIRMATION", "LOW", Some("Order auto-transitioned after a 7-day shipped window."), None, None, None, Some(&state.tx), ) .await; } let _ = state.tx.send(RealtimeEvent::OrderStatusChanged { transaction_id: txnid, merchant_id, new_status: ORDER_STATUS_DELIVERED_PENDING_APPROVAL.to_string(), }); } }) .buffer_unordered(10); aged_shipped_orders_stream.collect::>().await; reconciled_count += aged_count; } // 3. Autonomous Settlement for Aged Delivered Orders (48-hour window) let aged_delivered_orders = sqlx::query( "UPDATE orders SET status = $1, settled_at = CURRENT_TIMESTAMP WHERE status = $2 AND delivered_at IS NOT NULL AND delivered_at < NOW() - INTERVAL '48 hours' RETURNING transaction_id, merchant_id, price_inr", ) .bind(crate::domain::constants::ORDER_STATUS_SETTLED) .bind(ORDER_STATUS_DELIVERED_PENDING_APPROVAL) .fetch_all(&state.pool) .await?; if !aged_delivered_orders.is_empty() { let aged_delivered_count = aged_delivered_orders.len(); let aged_delivered_orders_stream = stream::iter(aged_delivered_orders) .map(|row| { let state = state.clone(); async move { let txnid: String = row.get("transaction_id"); let merchant_id: String = row.get("merchant_id"); let price: f64 = row.get("price_inr"); if let Ok(mut conn) = state.pool.acquire().await { crate::domain::audit::log_risk_event( &mut conn, Some(&txnid), &merchant_id, "AUTONOMOUS_SETTLEMENT", "LOW", Some( "Order auto-settled after 48-hour verification window closed without disputes.", ), None, None, None, Some(&state.tx), ) .await; // Dynamic Trust Scoring: Reward successful autonomous settlement // Reuse the acquired connection `conn` here instead of checkout from pool let _ = sqlx::query("UPDATE merchants SET trust_score = LEAST(100.0, trust_score + 0.1) WHERE merchant_id = $1") .bind(&merchant_id) .execute(&mut *conn) .await; } let _ = state.tx.send(RealtimeEvent::OrderStatusChanged { transaction_id: txnid, merchant_id: merchant_id.clone(), new_status: ORDER_STATUS_SETTLED.to_string(), }); metrics::counter!("rtix_settlement_volume_inr_total").increment(price as u64); } }) .buffer_unordered(10); aged_delivered_orders_stream.collect::>().await; reconciled_count += aged_delivered_count; } // 4. Handle Stale Unpaid Orders (Expiration) let expired_count = sqlx::query( "UPDATE orders SET status = $1 WHERE status = $2 AND created_at < NOW() - INTERVAL '2 hours'" ) .bind(crate::domain::constants::ORDER_STATUS_PAYMENT_FAILED) .bind(crate::domain::constants::ORDER_STATUS_PENDING_PAYMENT) .execute(&state.pool) .await? .rows_affected(); if expired_count > 0 { tracing::info!( "Reconciliation: Expired {} stale unpaid orders.", expired_count ); } Ok(reconciled_count) }