use std::sync::Arc;
use anyhow::{Result, Context};
use chrono::{Utc, Datelike};
use tokio::sync::mpsc;
use tracing::{info, error};
use uuid::Uuid;

use crate::abstraction::DataProviderLogic;
use crate::dtos::{SessionDataDto, ProviderCacheDto};
use crate::messages::{FetchCompanyDataCommand, CompanyProfilePersistedEvent, FinancialsPersistedEvent, DataFetchFailedEvent};
use crate::observability::ObservabilityTaskStatus;
use crate::persistence_client::PersistenceClient;

/// 任务状态管理器接口，用于解耦 AppState
#[async_trait::async_trait]
pub trait TaskState: Send + Sync {
    fn update_status(&self, task_id: Uuid, status: ObservabilityTaskStatus, progress: u8, details: String);
    fn fail_task(&self, task_id: Uuid, error: String);
    fn complete_task(&self, task_id: Uuid, details: String);
    fn get_nats_addr(&self) -> String;
    fn get_persistence_url(&self) -> String;
}

/// 通用工作流引擎
pub struct StandardFetchWorkflow;

impl StandardFetchWorkflow {
    /// 执行完整的数据获取工作流
    pub async fn run<L, S>(
        state: Arc<S>,
        logic: Arc<L>,
        command: FetchCompanyDataCommand,
        completion_tx: Option<mpsc::Sender<()>>,
    ) -> Result<()> 
    where 
        L: DataProviderLogic + 'static,
        S: TaskState + 'static,
    {
        let task_id = command.request_id;
        let symbol = command.symbol.to_string();
        let provider_id = logic.provider_id();

        // 1. Market Check
        if !logic.supports_market(&command.market) {
            let msg = format!("Skipping: Market '{}' not supported by {}", command.market, provider_id);
            info!("{}", msg);
            // 如果不支持，认为是正常结束（忽略），或者也可以视具体需求报错
            // 这里我们选择静默忽略或标记为 Skipped (如果 TaskStatus 支持)
            // 暂时标记为 Completed 但带有说明
            state.complete_task(task_id, msg);
            if let Some(tx) = completion_tx { let _ = tx.send(()).await; }
            return Ok(());
        }

        state.update_status(task_id, ObservabilityTaskStatus::InProgress, 10, "Checking cache...".to_string());

        // 2. Persistence Client
        let persistence = PersistenceClient::new(state.get_persistence_url());
        
        // 3. Cache Check
        let cache_key = format!("{}:{}:all", provider_id, symbol);
        
        // Wrap in a block to catch errors and update task status
        let result: Result<_, anyhow::Error> = async {
            let (profile, financials) = match persistence.get_cache(&cache_key).await? {
                Some(cache_entry) => {
                    info!("Cache HIT for {}", cache_key);
                    state.update_status(task_id, ObservabilityTaskStatus::InProgress, 50, "Data retrieved from cache".to_string());
                    serde_json::from_value(cache_entry.data_payload)
                        .context("Failed to deserialize cache")?
                },
                None => {
                    info!("Cache MISS for {}", cache_key);
                    state.update_status(task_id, ObservabilityTaskStatus::InProgress, 20, format!("Fetching from {}...", provider_id));
                    
                    let (p, f) = logic.fetch_data(&symbol).await?;
                    
                    // Write Back to Cache
                    let payload = serde_json::json!((&p, &f));
                    persistence.set_cache(&ProviderCacheDto {
                        cache_key: cache_key.clone(),
                        data_payload: payload,
                        expires_at: Utc::now() + chrono::Duration::hours(24),
                        updated_at: None,
                    }).await?;
                    
                    (p, f)
                }
            };

            state.update_status(task_id, ObservabilityTaskStatus::InProgress, 80, "Snapshotting data...".to_string());

            // 4. Snapshot Session Data
            persistence.insert_session_data(&SessionDataDto {
                request_id: task_id,
                symbol: symbol.clone(),
                provider: provider_id.to_string(),
                data_type: "company_profile".to_string(),
                data_payload: serde_json::to_value(&profile)?,
                created_at: None,
            }).await?;

            persistence.insert_session_data(&SessionDataDto {
                request_id: task_id,
                symbol: symbol.clone(),
                provider: provider_id.to_string(),
                data_type: "financial_statements".to_string(),
                data_payload: serde_json::to_value(&financials)?,
                created_at: None,
            }).await?;

            // 5. Publish Events
            let nats_addr = state.get_nats_addr();
            // Connect to NATS (Scoped connection)
            let nats = async_nats::connect(&nats_addr).await
                .context("Failed to connect to NATS")?;

            // Event 1: Profile Persisted
            let profile_event = CompanyProfilePersistedEvent {
                request_id: task_id,
                symbol: command.symbol.clone(),
            };
            nats.publish("events.data.company_profile_persisted", serde_json::to_vec(&profile_event)?.into()).await?;

            // Event 2: Financials Persisted
            let years: std::collections::BTreeSet<u16> = financials.iter()
                .map(|f| f.period_date.year() as u16)
                .collect();
            let summary = format!("Fetched {} years of data", years.len());
            
            let financials_event = FinancialsPersistedEvent {
                request_id: task_id,
                symbol: command.symbol.clone(),
                years_updated: years.into_iter().collect(),
                template_id: command.template_id.clone(),
                provider_id: Some(provider_id.to_string()),
                data_summary: Some(summary),
            };
            nats.publish("events.data.financials_persisted", serde_json::to_vec(&financials_event)?.into()).await?;

            // CRITICAL: Flush to ensure messages are sent before drop
            nats.flush().await.context("Failed to flush NATS")?;
            
            Ok(())
        }.await;

        match result {
            Ok(_) => {
                info!("Workflow for {}/{} completed successfully.", provider_id, symbol);
                state.complete_task(task_id, "Workflow finished successfully".to_string());
                if let Some(tx) = completion_tx { let _ = tx.send(()).await; }
                Ok(())
            },
            Err(e) => {
                error!("Workflow for {}/{} failed: {}", provider_id, symbol, e);
                state.fail_task(task_id, e.to_string());
                
                // Try to report failure via NATS
                if let Ok(nats) = async_nats::connect(&state.get_nats_addr()).await {
                    let fail_event = DataFetchFailedEvent {
                        request_id: task_id,
                        symbol: command.symbol.clone(),
                        error: e.to_string(),
                        provider_id: Some(provider_id.to_string()),
                    };
                    let _ = nats.publish("events.data.fetch_failed", serde_json::to_vec(&fail_event).unwrap().into()).await;
                    let _ = nats.flush().await;
                }

                if let Some(tx) = completion_tx { let _ = tx.send(()).await; }
                Err(e)
            }
        }
    }
}

