use crate::error::Result;
use crate::state::{AppState, ServiceOperationalStatus};
use common_contracts::messages::FetchCompanyDataCommand;
use common_contracts::observability::ObservabilityTaskStatus;
use common_contracts::subjects::NatsSubject;
use futures_util::StreamExt;
use std::sync::Arc;
use std::time::Duration;
use tracing::{error, info, warn};

pub async fn run(state: AppState) -> Result<()> {
    info!("Starting NATS message consumer...");

    loop {
        let status = state.status.read().await.clone();
        if let ServiceOperationalStatus::Degraded { reason } = status {
            warn!(
                "Service is in degraded state (reason: {}). Pausing message consumption for 30s.",
                reason
            );
            tokio::time::sleep(Duration::from_secs(30)).await;
            continue;
        }

        info!("Service is Active. Connecting to NATS...");
        match async_nats::connect(&state.config.nats_addr).await {
            Ok(client) => {
                info!("Successfully connected to NATS.");
                if let Err(e) = subscribe_and_process(state.clone(), client).await {
                    error!("NATS subscription error: {}. Reconnecting in 10s...", e);
                }
            }
            Err(e) => {
                error!("Failed to connect to NATS: {}. Retrying in 10s...", e);
            }
        }
        tokio::time::sleep(Duration::from_secs(10)).await;
    }
}

async fn subscribe_and_process(state: AppState, client: async_nats::Client) -> Result<()> {
    let subject = NatsSubject::DataFetchCommands.to_string();
    let mut subscriber = client.subscribe(subject.clone()).await?;
    info!(
        "Consumer started, waiting for messages on subject '{}'",
        subject
    );

    while let Some(message) = subscriber.next().await {
        let current_status = state.status.read().await.clone();
        if matches!(current_status, ServiceOperationalStatus::Degraded {..}) {
            warn!("Service became degraded. Disconnecting from NATS and pausing consumption.");
            subscriber.unsubscribe().await?;
            return Ok(());
        }

        info!("Received NATS message.");
        let state_clone = state.clone();
        let publisher_clone = client.clone();

        tokio::spawn(async move {
            match serde_json::from_slice::<FetchCompanyDataCommand>(&message.payload) {
                Ok(command) => {
                    info!("Deserialized command for symbol: {}", command.symbol);

                    // Skip processing if market is 'CN'
                    if command.market.to_uppercase() == "CN" {
                        info!(
                            "Skipping command for symbol '{}' as its market ('{}') is 'CN'.",
                            command.symbol, command.market
                        );
                        return;
                    }

                    if let Err(e) =
                        crate::worker::handle_fetch_command(state_clone, command, publisher_clone)
                            .await
                    {
                        error!("Error handling fetch command: {:?}", e);
                    }
                }
                Err(e) => {
                    error!("Failed to deserialize message: {}", e);
                }
            }
        });
    }

    Ok(())
}

pub async fn subscribe_to_data_commands(app_state: Arc<AppState>, nats_client: async_nats::Client) -> Result<()> {
    let mut subscriber = nats_client.subscribe(NatsSubject::DataFetchCommands.to_string()).await?;

    while let Some(message) = subscriber.next().await {
        let command: FetchCompanyDataCommand = match serde_json::from_slice(&message.payload) {
            Ok(c) => c,
            Err(e) => {
                error!("Failed to deserialize message: {}", e);
                continue;
            }
        };
        let task_id = command.request_id;

        if command.market.to_uppercase() == "CN" {
            info!(
                "Skipping command for symbol '{}' as its market ('{}') is 'CN'.",
                command.symbol, command.market
            );
            continue;
        }

        app_state.tasks.insert(task_id, common_contracts::observability::TaskProgress {
            request_id: task_id,
            task_name: format!("finnhub:{}", command.symbol),
            status: ObservabilityTaskStatus::Queued,
            progress_percent: 0,
            details: "Command received".to_string(),
            started_at: chrono::Utc::now(),
        });

        // Spawn the workflow in a separate task
        let workflow_state = app_state.clone();
        let publisher_clone = nats_client.clone();
        tokio::spawn(async move {
            let state_owned = (*workflow_state).clone();
            let result = crate::worker::handle_fetch_command(state_owned, command, publisher_clone).await;
            if let Err(e) = result {
                error!(
                    "Error executing Finnhub workflow for task {}: {:?}",
                    task_id, e
                );
            }
        });
    }
    Ok(())
}
