//! Recovery Module for Broker Gateway Service
//!
//! Implements comprehensive recovery mechanisms for:
//! - FIX session recovery (reconnect, restore sequence numbers)
//! - Order recovery (replay unsent orders from database)
//! - Position recovery (reconcile positions from broker on startup)
//! - Health monitoring with degraded state detection

use sqlx::PgPool;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::RwLock;
use tracing::{error, info};

use crate::proto::broker_gateway::SessionState;

/// Session recovery manager for FIX session lifecycle
pub struct SessionRecovery {
    /// Database connection pool
    db_pool: PgPool,
    /// Current session state
    session_state: Arc<RwLock<SessionState>>,
    /// Session ID (SenderCompID-TargetCompID)
    session_id: String,
}

impl SessionRecovery {
    /// Create a new session recovery manager
    pub fn new(
        db_pool: PgPool,
        session_state: Arc<RwLock<SessionState>>,
        session_id: String,
    ) -> Self {
        Self {
            db_pool,
            session_state,
            session_id,
        }
    }

    /// Recover session from database (restore sequence numbers)
    pub async fn recover_session(&self) -> anyhow::Result<SessionInfo> {
        info!("Recovering FIX session: {}", self.session_id);

        // Query database for persisted session state
        let session: Option<SessionInfoRow> = sqlx::query_as(
            r#"
            SELECT
                sender_seq_num,
                target_seq_num,
                session_state,
                last_heartbeat_sent,
                last_heartbeat_received,
                connected_at
            FROM broker_sessions
            WHERE session_id = $1
            ORDER BY updated_at DESC
            LIMIT 1
            "#,
        )
        .bind(&self.session_id)
        .fetch_optional(&self.db_pool)
        .await?;

        if let Some(session) = session {
            info!(
                "Session {} recovered from database: sender_seq={}, target_seq={}",
                self.session_id, session.sender_seq_num, session.target_seq_num
            );

            Ok(SessionInfo {
                sender_seq_num: session.sender_seq_num,
                target_seq_num: session.target_seq_num,
                session_state: session.session_state,
                last_heartbeat_sent: session.last_heartbeat_sent,
                last_heartbeat_received: session.last_heartbeat_received,
                connected_at: session.connected_at,
            })
        } else {
            info!(
                "No previous session found for {}, starting fresh with seq=1",
                self.session_id
            );

            // Create new session in database
            self.initialize_session().await?;

            Ok(SessionInfo {
                sender_seq_num: 1,
                target_seq_num: 1,
                session_state: "DISCONNECTED".to_string(),
                last_heartbeat_sent: None,
                last_heartbeat_received: None,
                connected_at: None,
            })
        }
    }

    /// Initialize a new session in the database
    async fn initialize_session(&self) -> anyhow::Result<()> {
        info!("Initializing new FIX session: {}", self.session_id);

        // Parse session_id: "FOXHUNT_CLIENT-CQG"
        let parts: Vec<&str> = self.session_id.split('-').collect();
        let (sender_comp_id, target_comp_id) = if parts.len() == 2 {
            (parts.first().copied().unwrap_or("FOXHUNT_CLIENT"), parts.get(1).copied().unwrap_or("CQG"))
        } else {
            ("FOXHUNT_CLIENT", "CQG")
        };

        sqlx::query(
            r#"
            INSERT INTO broker_sessions
            (session_id, sender_comp_id, target_comp_id, sender_seq_num, target_seq_num, session_state, created_at, updated_at)
            VALUES ($1, $2, $3, 1, 1, 'DISCONNECTED', NOW(), NOW())
            ON CONFLICT (session_id) DO NOTHING
            "#,
        )
        .bind(&self.session_id)
        .bind(sender_comp_id)
        .bind(target_comp_id)
        .execute(&self.db_pool)
        .await?;

        Ok(())
    }

    /// Reconnect to FIX session.
    ///
    /// Sets state to LoggingIn, then performs a health check (DB ping) to
    /// verify infrastructure is available before transitioning to Active.
    /// If the health check fails, state is set to Disconnected and an error
    /// is returned instead of faking a successful reconnection.
    pub async fn reconnect(&self) -> anyhow::Result<()> {
        info!("Attempting to reconnect FIX session: {}", self.session_id);

        // Update session state to RECONNECTING
        {
            let mut state = self.session_state.write().await;
            *state = SessionState::LoggingIn;
        }

        sqlx::query(
            r#"
            UPDATE broker_sessions
            SET session_state = 'RECONNECTING', updated_at = NOW()
            WHERE session_id = $1
            "#,
        )
        .bind(&self.session_id)
        .execute(&self.db_pool)
        .await?;

        // Allow a brief delay for session negotiation
        tokio::time::sleep(Duration::from_millis(500)).await;

        // Health check: verify DB is reachable before declaring Active.
        // In production this should also verify the FIX/cTrader TCP session,
        // but until a real broker connection is wired into SessionRecovery,
        // the DB ping is the best available liveness probe.
        let db_healthy = sqlx::query("SELECT 1")
            .fetch_optional(&self.db_pool)
            .await
            .is_ok();

        if !db_healthy {
            error!(
                session_id = %self.session_id,
                "Reconnect health check failed — database unreachable, setting state to Disconnected"
            );

            {
                let mut state = self.session_state.write().await;
                *state = SessionState::Disconnected;
            }

            // Best-effort DB update; may also fail if DB is down.
            let _ = sqlx::query(
                r#"
                UPDATE broker_sessions
                SET session_state = 'DISCONNECTED', updated_at = NOW()
                WHERE session_id = $1
                "#,
            )
            .bind(&self.session_id)
            .execute(&self.db_pool)
            .await;

            anyhow::bail!(
                "Reconnect failed for session {}: health check failed",
                self.session_id
            );
        }

        // Health check passed — transition to Active
        {
            let mut state = self.session_state.write().await;
            *state = SessionState::Active;
        }

        sqlx::query(
            r#"
            UPDATE broker_sessions
            SET session_state = 'ACTIVE', connected_at = NOW(), updated_at = NOW()
            WHERE session_id = $1
            "#,
        )
        .bind(&self.session_id)
        .execute(&self.db_pool)
        .await?;

        info!("FIX session {} reconnected successfully", self.session_id);
        Ok(())
    }

    /// Update sequence numbers in database
    pub async fn update_sequence_numbers(
        &self,
        sender_seq_num: i64,
        target_seq_num: i64,
    ) -> anyhow::Result<()> {
        sqlx::query(
            r#"
            UPDATE broker_sessions
            SET sender_seq_num = $2, target_seq_num = $3, updated_at = NOW()
            WHERE session_id = $1
            "#,
        )
        .bind(&self.session_id)
        .bind(sender_seq_num)
        .bind(target_seq_num)
        .execute(&self.db_pool)
        .await?;

        Ok(())
    }

    /// Disconnect session (graceful shutdown)
    pub async fn disconnect(&self) -> anyhow::Result<()> {
        info!("Disconnecting FIX session: {}", self.session_id);

        {
            let mut state = self.session_state.write().await;
            *state = SessionState::Disconnected;
        }

        sqlx::query(
            r#"
            UPDATE broker_sessions
            SET session_state = 'DISCONNECTED', disconnected_at = NOW(), updated_at = NOW()
            WHERE session_id = $1
            "#,
        )
        .bind(&self.session_id)
        .execute(&self.db_pool)
        .await?;

        Ok(())
    }
}

/// Row type for session info query
#[derive(sqlx::FromRow)]
struct SessionInfoRow {
    sender_seq_num: i64,
    target_seq_num: i64,
    session_state: String,
    last_heartbeat_sent: Option<chrono::DateTime<chrono::Utc>>,
    last_heartbeat_received: Option<chrono::DateTime<chrono::Utc>>,
    connected_at: Option<chrono::DateTime<chrono::Utc>>,
}

/// Session information recovered from database
#[derive(Debug, Clone)]
pub struct SessionInfo {
    /// Outgoing sequence number
    pub sender_seq_num: i64,
    /// Expected incoming sequence number
    pub target_seq_num: i64,
    /// Session state
    pub session_state: String,
    /// Last heartbeat sent timestamp
    pub last_heartbeat_sent: Option<chrono::DateTime<chrono::Utc>>,
    /// Last heartbeat received timestamp
    pub last_heartbeat_received: Option<chrono::DateTime<chrono::Utc>>,
    /// Connection timestamp
    pub connected_at: Option<chrono::DateTime<chrono::Utc>>,
}

/// Order recovery manager for replaying unsent orders
pub struct OrderRecovery {
    /// Database connection pool
    db_pool: PgPool,
}

/// Row type for unsent order query
#[derive(sqlx::FromRow)]
struct UnsentOrderRow {
    client_order_id: String,
    account_id: String,
    symbol: String,
    side: String,
    order_type: String,
    quantity: rust_decimal::Decimal,
    price: Option<rust_decimal::Decimal>,
    stop_price: Option<rust_decimal::Decimal>,
    metadata: Option<serde_json::Value>,
    submitted_at: Option<chrono::DateTime<chrono::Utc>>,
}

impl OrderRecovery {
    /// Create a new order recovery manager
    pub fn new(db_pool: PgPool) -> Self {
        Self { db_pool }
    }

    /// Recover unsent orders from database (PENDING_SUBMIT status)
    pub async fn recover_unsent_orders(&self) -> anyhow::Result<Vec<UnsentOrder>> {
        info!("Recovering unsent orders from database");

        let orders: Vec<UnsentOrderRow> = sqlx::query_as(
            r#"
            SELECT
                client_order_id,
                account_id,
                symbol,
                side,
                order_type,
                quantity,
                price,
                stop_price,
                metadata,
                submitted_at
            FROM broker_orders
            WHERE status = 'PENDING_SUBMIT'
            ORDER BY created_at ASC
            "#,
        )
        .fetch_all(&self.db_pool)
        .await?;

        let unsent_orders: Vec<UnsentOrder> = orders
            .into_iter()
            .map(|row| UnsentOrder {
                client_order_id: row.client_order_id,
                account_id: row.account_id,
                symbol: row.symbol,
                side: row.side,
                order_type: row.order_type,
                quantity: row.quantity,
                price: row.price,
                stop_price: row.stop_price,
                metadata: row.metadata.unwrap_or_default(),
                submitted_at: row.submitted_at,
            })
            .collect();

        info!(
            "Recovered {} unsent orders from database",
            unsent_orders.len()
        );

        Ok(unsent_orders)
    }

    /// Mark order as submitted after successful send
    pub async fn mark_order_submitted(&self, client_order_id: &str) -> anyhow::Result<()> {
        sqlx::query(
            r#"
            UPDATE broker_orders
            SET status = 'SUBMITTED', updated_at = NOW()
            WHERE client_order_id = $1
            "#,
        )
        .bind(client_order_id)
        .execute(&self.db_pool)
        .await?;

        info!("Order {} marked as SUBMITTED", client_order_id);
        Ok(())
    }

    /// Mark order as failed after retry exhaustion
    pub async fn mark_order_failed(
        &self,
        client_order_id: &str,
        error_msg: &str,
    ) -> anyhow::Result<()> {
        sqlx::query(
            r#"
            UPDATE broker_orders
            SET status = 'REJECTED',
                metadata = jsonb_set(
                    COALESCE(metadata, '{}'::jsonb),
                    '{reject_reason}',
                    to_jsonb($2::text)
                ),
                updated_at = NOW()
            WHERE client_order_id = $1
            "#,
        )
        .bind(client_order_id)
        .bind(error_msg)
        .execute(&self.db_pool)
        .await?;

        error!("Order {} marked as REJECTED: {}", client_order_id, error_msg);
        Ok(())
    }
}

/// Unsent order recovered from database
#[derive(Debug, Clone)]
pub struct UnsentOrder {
    /// Client order ID
    pub client_order_id: String,
    /// Account ID
    pub account_id: String,
    /// Trading symbol
    pub symbol: String,
    /// Order side (BUY, SELL)
    pub side: String,
    /// Order type (MARKET, LIMIT, etc.)
    pub order_type: String,
    /// Order quantity
    pub quantity: rust_decimal::Decimal,
    /// Limit price (optional)
    pub price: Option<rust_decimal::Decimal>,
    /// Stop price (optional)
    pub stop_price: Option<rust_decimal::Decimal>,
    /// Order metadata
    pub metadata: serde_json::Value,
    /// Submission timestamp
    pub submitted_at: Option<chrono::DateTime<chrono::Utc>>,
}

/// Position recovery manager for reconciling positions with broker
pub struct PositionRecovery {
    /// Database connection pool
    db_pool: PgPool,
}

/// Row type for position query
#[derive(sqlx::FromRow)]
struct PositionRow {
    symbol: String,
    quantity: rust_decimal::Decimal,
    avg_entry_price: Option<rust_decimal::Decimal>,
    market_value: Option<rust_decimal::Decimal>,
    unrealized_pnl: Option<rust_decimal::Decimal>,
    realized_pnl: Option<rust_decimal::Decimal>,
    last_updated: chrono::DateTime<chrono::Utc>,
}

impl PositionRecovery {
    /// Create a new position recovery manager
    pub fn new(db_pool: PgPool) -> Self {
        Self { db_pool }
    }

    /// Reconcile positions from broker (MVP: query database)
    pub async fn reconcile_positions(&self, account_id: &str) -> anyhow::Result<Vec<Position>> {
        info!("Reconciling positions for account: {}", account_id);

        let positions: Vec<PositionRow> = sqlx::query_as(
            r#"
            SELECT
                symbol,
                quantity,
                avg_entry_price,
                market_value,
                unrealized_pnl,
                realized_pnl,
                last_updated
            FROM broker_positions
            WHERE account_id = $1
            "#,
        )
        .bind(account_id)
        .fetch_all(&self.db_pool)
        .await?;

        let reconciled_positions: Vec<Position> = positions
            .into_iter()
            .map(|row| Position {
                symbol: row.symbol,
                quantity: row.quantity,
                avg_entry_price: row.avg_entry_price,
                market_value: row.market_value,
                unrealized_pnl: row.unrealized_pnl,
                realized_pnl: row.realized_pnl,
                last_updated: row.last_updated,
            })
            .collect();

        info!(
            "Reconciled {} positions for account {}",
            reconciled_positions.len(),
            account_id
        );

        Ok(reconciled_positions)
    }

    /// Update position in database
    pub async fn update_position(&self, account_id: &str, position: &Position) -> anyhow::Result<()> {
        sqlx::query(
            r#"
            INSERT INTO broker_positions
            (account_id, symbol, quantity, avg_entry_price, market_value, unrealized_pnl, realized_pnl, last_updated, created_at)
            VALUES ($1, $2, $3, $4, $5, $6, $7, NOW(), NOW())
            ON CONFLICT (account_id, symbol) DO UPDATE SET
                quantity = EXCLUDED.quantity,
                avg_entry_price = EXCLUDED.avg_entry_price,
                market_value = EXCLUDED.market_value,
                unrealized_pnl = EXCLUDED.unrealized_pnl,
                realized_pnl = EXCLUDED.realized_pnl,
                last_updated = NOW()
            "#,
        )
        .bind(account_id)
        .bind(&position.symbol)
        .bind(position.quantity)
        .bind(position.avg_entry_price)
        .bind(position.market_value)
        .bind(position.unrealized_pnl)
        .bind(position.realized_pnl)
        .execute(&self.db_pool)
        .await?;

        Ok(())
    }
}

/// Position data for reconciliation
#[derive(Debug, Clone)]
pub struct Position {
    /// Trading symbol
    pub symbol: String,
    /// Position quantity (signed: positive=long, negative=short)
    pub quantity: rust_decimal::Decimal,
    /// Average entry price
    pub avg_entry_price: Option<rust_decimal::Decimal>,
    /// Current market value
    pub market_value: Option<rust_decimal::Decimal>,
    /// Unrealized P&L
    pub unrealized_pnl: Option<rust_decimal::Decimal>,
    /// Realized P&L
    pub realized_pnl: Option<rust_decimal::Decimal>,
    /// Last update timestamp
    pub last_updated: chrono::DateTime<chrono::Utc>,
}

/// Health monitor for service degradation detection
pub struct HealthMonitor {
    /// Session state
    session_state: Arc<RwLock<SessionState>>,
    /// Database pool
    db_pool: PgPool,
}

impl HealthMonitor {
    /// Create a new health monitor
    pub fn new(session_state: Arc<RwLock<SessionState>>, db_pool: PgPool) -> Self {
        Self {
            session_state,
            db_pool,
        }
    }

    /// Check service health and return status
    pub async fn check_health(&self) -> HealthStatus {
        let session_state = *self.session_state.read().await;

        // Check database connection
        let db_healthy = sqlx::query("SELECT 1")
            .fetch_optional(&self.db_pool)
            .await
            .is_ok();

        // Check FIX session state
        let session_healthy = session_state == SessionState::Active;

        let overall_status = if !db_healthy {
            HealthStatusLevel::Unhealthy
        } else if !session_healthy {
            HealthStatusLevel::Degraded
        } else {
            HealthStatusLevel::Healthy
        };

        HealthStatus {
            level: overall_status,
            database_healthy: db_healthy,
            session_connected: session_healthy,
            session_state,
        }
    }
}

/// Health status levels
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum HealthStatusLevel {
    /// Service is fully operational
    Healthy,
    /// Service is operational but degraded (e.g., session disconnected)
    Degraded,
    /// Service is unhealthy (e.g., database unavailable)
    Unhealthy,
}

/// Health status details
#[derive(Debug, Clone)]
pub struct HealthStatus {
    /// Overall health level
    pub level: HealthStatusLevel,
    /// Database connectivity status
    pub database_healthy: bool,
    /// FIX session connectivity status
    pub session_connected: bool,
    /// Current session state
    pub session_state: SessionState,
}

#[cfg(test)]
mod tests {
    use super::*;

    // Note: Database tests require a running PostgreSQL instance
    // These are integration tests and should be run with `cargo test --features database`

    #[test]
    fn test_health_status_levels() {
        // Test that health status levels are properly defined
        assert_eq!(
            std::mem::discriminant(&HealthStatusLevel::Healthy),
            std::mem::discriminant(&HealthStatusLevel::Healthy)
        );
        assert_ne!(
            std::mem::discriminant(&HealthStatusLevel::Healthy),
            std::mem::discriminant(&HealthStatusLevel::Degraded)
        );
    }
}
