src/application/reconciliation.rs

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<usize, sqlx::Error> {
    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::<Vec<()>>().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::<Vec<()>>().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::<Vec<()>>().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)
}