//! Event monitoring component for the Transform Orchestrator
//!
//! Handles FieldValueSet event monitoring and transform discovery,
//! extracted from the main TransformOrchestrator for better separation of concerns.

use super::persistence_manager::PersistenceManager;
use crate::fold_db_core::infrastructure::message_bus::{
    schema_events::{TransformExecuted, TransformRegistrationRequest, TransformTriggered, TransformRegistered},
    MessageBus,
};
use crate::transform::manager::{TransformManager, types::TransformRunner};
use crate::fold_db_core::infrastructure::message_bus::events::schema_events::TransformRegistrationResponse;
use crate::schema::SchemaError;
use log::{error, info};
use std::sync::Arc;
use std::thread;
use std::time::Duration;

/// Handles monitoring of MutationExecuted and TransformTriggered events for automatic transform execution
pub struct TransformEventMonitor {
    /// Single monitoring thread for all events
    _monitoring_thread: Option<thread::JoinHandle<()>>,
}

impl TransformEventMonitor {
    /// Create a new TransformEventMonitor and start monitoring
    pub fn new(
        message_bus: Arc<MessageBus>,
        manager: Arc<TransformManager>,
        _persistence: PersistenceManager,
    ) -> Self {
        let monitoring_thread = Self::start_unified_monitoring(
            Arc::clone(&message_bus),
            Arc::clone(&manager),
        );

        Self {
            _monitoring_thread: Some(monitoring_thread),
        }
    }

    /// Start unified monitoring for MutationExecuted and TransformTriggered events
    fn start_unified_monitoring(
        message_bus: Arc<MessageBus>,
        manager: Arc<TransformManager>,
    ) -> thread::JoinHandle<()> {
        let mut mutation_executed_consumer = message_bus.subscribe::<crate::fold_db_core::infrastructure::message_bus::query_events::MutationExecuted>();
        let mut triggered_consumer = message_bus.subscribe::<TransformTriggered>();
        let mut registration_consumer = message_bus.subscribe::<TransformRegistrationRequest>();
        let manager_clone = Arc::clone(&manager);
        thread::spawn(move || {
            info!("TransformEventMonitor: Starting unified monitoring of MutationExecuted and TransformTriggered events");

            loop {
                // Check MutationExecuted events - trigger transforms after mutation completion
                if let Ok(event) = mutation_executed_consumer.try_recv() {
                    if let Err(e) =
                        Self::handle_mutation_executed_event(&event, &manager_clone, &message_bus)
                    {
                        error!("Error handling mutation executed event: {}", e);
                    }
                }

                // Check TransformTriggered events
                if let Ok(event) = triggered_consumer.try_recv() {
                    if let Err(e) =
                        Self::handle_transform_triggered_event(&event, &manager_clone, &message_bus)
                    {
                        error!("Error handling TransformTriggered event: {}", e);
                    }
                }

                // Check TransformRegistrationRequest events
                if let Ok(event) = registration_consumer.try_recv() {
                    if let Err(e) =
                        Self::handle_transform_registration_request_event(&event, &manager_clone, &message_bus)
                    {
                        error!("Error handling TransformRegistrationRequest event: {}", e);
                    }
                }

                // Small sleep to prevent busy waiting
                thread::sleep(Duration::from_millis(10));
            }
        })
    }

    /// Handle a TransformTriggered event by executing the transform
    fn handle_transform_triggered_event(
        event: &TransformTriggered,
        manager: &Arc<TransformManager>,
        message_bus: &Arc<MessageBus>,
    ) -> Result<(), SchemaError> {
        let result = manager.execute_transform_with_context(&event.transform_id, &event.mutation_context);

        match result {
            Ok(transform_result) => {
                info!("Transform {} executed successfully: {} records", event.transform_id, transform_result.records.len());

                // Publish TransformExecuted event
                Self::publish_transform_executed(
                    message_bus,
                    &event.transform_id,
                    &format!("{} records produced", transform_result.records.len()),
                )?;

                Ok(())
            }
            Err(e) => {
                error!("Transform {} execution failed: {}", event.transform_id, e);

                // Publish TransformExecuted event with error
                Self::publish_transform_executed(
                    message_bus,
                    &event.transform_id,
                    &format!("error: {}", e),
                )?;

                Err(e)
            }
        }
    }

    /// Publish TransformExecuted event
    fn publish_transform_executed(
        message_bus: &Arc<MessageBus>,
        transform_id: &str,
        result: &str,
    ) -> Result<(), SchemaError> {
        let executed_event = TransformExecuted {
            transform_id: transform_id.to_string(),
            result: result.to_string(),
        };

        message_bus.publish(executed_event).map_err(|e| {
            error!("Failed to publish TransformExecuted event for {}: {}", transform_id, e);
            SchemaError::InvalidData(format!("Failed to publish TransformExecuted event: {}", e))
        })?;

        Ok(())
    }

    /// Handle a MutationExecuted event by triggering transforms for the schema
    /// @tomtang: this is the entry point for the transform orchestration
    fn handle_mutation_executed_event(
        event: &crate::fold_db_core::infrastructure::message_bus::query_events::MutationExecuted,
        manager: &Arc<TransformManager>,
        message_bus: &Arc<MessageBus>,
    ) -> Result<(), SchemaError> {
        let mut unique_transform_ids = std::collections::HashSet::new();
        for field_name in event.fields_affected.clone() {
            match manager.get_transforms_for_field(&event.schema, &field_name) {
                Ok(transform_ids) => {
                    unique_transform_ids.extend(transform_ids);
                }
                Err(e) => {
                    error!("Failed to get transforms for field {}.{}: {}", event.schema, field_name, e);
                    return Err(e);
                }
            }
        }
        Self::add_transforms_to_queue(
            &unique_transform_ids,
            manager,
            message_bus,
            &event.mutation_context,
        )?;
        Ok(())
    }


    /// REMOVED: add_transforms_to_queue - TransformEventMonitor should not manage persistence directly
    /// This responsibility belongs to PersistenceManager through TransformOrchestrator
    /// 
    /// Filter out unapproved transforms before publishing TransformTriggered events
    fn add_transforms_to_queue(
        transform_ids: &std::collections::HashSet<String>,
        manager: &Arc<TransformManager>,
        message_bus: &Arc<MessageBus>,
        mutation_context: &Option<
            crate::fold_db_core::infrastructure::message_bus::atom_events::MutationContext,
        >,
    ) -> Result<(), SchemaError> {
        use crate::schema::SchemaState;
        
        // Publish TransformTriggered events only for approved transforms
        for transform_id in transform_ids {
            // Check if the transform's schema is approved before publishing event
            let schema_state = manager.get_schema_state(transform_id)?
                .unwrap_or(SchemaState::Available);
            
            if schema_state != SchemaState::Approved {
                // Skip unapproved transforms - don't emit TransformTriggered event
                continue;
            }
            
            let triggered_event = if let Some(ref context) = mutation_context {
                if context.incremental {
                    TransformTriggered::with_context(transform_id.clone(), context.clone())
                } else {
                    TransformTriggered::new(transform_id.clone())
                }
            } else {
                TransformTriggered::new(transform_id.clone())
            };

            match message_bus.publish(triggered_event) {
                Ok(()) => {
                    // TransformTriggered event published successfully
                }
                Err(e) => {
                    error!("Failed to publish TransformTriggered event for {}: {}", transform_id, e);
                    return Err(SchemaError::InvalidData(format!(
                        "Failed to publish TransformTriggered event for {}: {}",
                        transform_id, e
                    )));
                }
            }
        }

        Ok(())
    }


    /// Handle a TransformRegistrationRequest event by registering the transform
    fn handle_transform_registration_request_event(
        event: &TransformRegistrationRequest,
        manager: &Arc<TransformManager>,
        message_bus: &Arc<MessageBus>,
    ) -> Result<(), SchemaError> {
        info!("🔧 TransformEventMonitor: Handling TransformRegistrationRequest for '{}'", event.registration.transform_id);
        
        // Check if transform already exists
        let transform_exists = manager.transform_exists(&event.registration.transform_id)?;
        if transform_exists {
            info!("ℹ️  TransformEventMonitor: Transform '{}' already registered, skipping re-registration", event.registration.transform_id);
            return Ok(());
        }
        
        // Handle the transform registration using the TransformManager instance
        match manager.handle_transform_registration(&event.registration) {
            Ok(()) => {
                info!("✅ TransformEventMonitor: Successfully registered transform '{}'", event.registration.transform_id);
                
                // Publish success response
                let response = TransformRegistrationResponse {
                    correlation_id: event.correlation_id.clone(),
                    success: true,
                    error: None,
                };
                
                if let Err(e) = message_bus.publish(response) {
                    error!("Failed to publish TransformRegistrationResponse: {}", e);
                    return Err(SchemaError::InvalidData(format!(
                        "Failed to publish TransformRegistrationResponse: {}", e
                    )));
                }

                // Publish TransformRegistered event to trigger backfill
                // Extract source schema name from the transform's input fields
                // Look up the transform's schema from the database
                let transform_schema = tokio::runtime::Handle::current().block_on(manager.db_ops.get_schema(event.registration.transform.get_schema_name()))?.ok_or_else(|| {
                    SchemaError::InvalidData(format!("Transform schema '{}' not found", event.registration.transform.get_schema_name()))
                })?;
                let source_schema_name = transform_schema
                    .get_inputs()
                    .first()
                    .ok_or_else(|| SchemaError::InvalidData("Transform has no input schema fields".to_string()))?
                    .split('.')
                    .next()
                    .ok_or_else(|| SchemaError::InvalidData("Invalid input schema field format".to_string()))?
                    .to_string();
                
                let transform_registered = TransformRegistered {
                    transform_id: event.registration.transform_id.clone(),
                    source_schema_name,
                    correlation_id: event.correlation_id.clone(),
                };
                
                if let Err(e) = message_bus.publish(transform_registered) {
                    error!("Failed to publish TransformRegistered event: {}", e);
                    return Err(SchemaError::InvalidData(format!(
                        "Failed to publish TransformRegistered event: {}", e
                    )));
                }
                
                Ok(())
            }
            Err(e) => {
                error!("❌ TransformEventMonitor: Failed to register transform '{}': {}", event.registration.transform_id, e);
                
                // Publish error response
                let response = TransformRegistrationResponse {
                    correlation_id: event.correlation_id.clone(),
                    success: false,
                    error: Some(e.to_string()),
                };
                
                if let Err(publish_err) = message_bus.publish(response) {
                    error!("Failed to publish TransformRegistrationResponse: {}", publish_err);
                }
                
                Err(e)
            }
        }
    }

    /// Stop monitoring (the thread will be stopped when the TransformEventMonitor is dropped)
    pub fn stop_monitoring(&mut self) {
        if let Some(_handle) = self._monitoring_thread.take() {
            // In a real implementation, we would send a shutdown signal
            // For now, the thread will be stopped when the handle is dropped
            info!("TransformEventMonitor: Stopping unified event monitoring");
            // Note: In a production system, you would want to implement
            // a proper shutdown mechanism using channels or atomic flags
        }
    }
}

impl Drop for TransformEventMonitor {
    fn drop(&mut self) {
        self.stop_monitoring();
    }
}