oxirs-vec 0.3.2

//! Main real-time embedding pipeline implementation

use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, RwLock};
use tracing::debug;
use uuid::Uuid;

use crate::real_time_embedding_pipeline::{
    config::PipelineConfig,
    consistency::ConsistencyManager,
    coordination::UpdateCoordinator,
    monitoring::{ConsoleAlertHandler, PipelinePerformanceMonitor},
    streaming::{StreamConfig, StreamProcessor},
    traits::{EmbeddingGenerator, IncrementalVectorIndex},
    types::PipelineStatistics,
    versioning::VersionManager,
    PipelineError, PipelineResult,
};

/// Real-time embedding pipeline for streaming updates
pub struct RealTimeEmbeddingPipeline {
    /// Pipeline configuration
    config: PipelineConfig,
    /// Embedding generators
    embedding_generators: Arc<RwLock<HashMap<String, Box<dyn EmbeddingGenerator>>>>,
    /// Vector indices for incremental updates
    indices: Arc<RwLock<HashMap<String, Box<dyn IncrementalVectorIndex>>>>,
    /// Stream processors
    stream_processors: Arc<RwLock<HashMap<String, StreamProcessor>>>,
    /// Update coordinator
    update_coordinator: Arc<UpdateCoordinator>,
    /// Performance monitor
    performance_monitor: Arc<PipelinePerformanceMonitor>,
    /// Version manager
    version_manager: Arc<VersionManager>,
    /// Consistency manager
    consistency_manager: Arc<ConsistencyManager>,
    /// Running flag
    is_running: AtomicBool,
    /// Statistics
    stats: Arc<PipelineStatistics>,
}

impl RealTimeEmbeddingPipeline {
    /// Create a new real-time embedding pipeline
    pub fn new(config: PipelineConfig) -> PipelineResult<Self> {
        let embedding_generators = Arc::new(RwLock::new(HashMap::new()));
        let indices = Arc::new(RwLock::new(HashMap::new()));
        let stream_processors = Arc::new(RwLock::new(HashMap::new()));

        let update_coordinator = Arc::new(UpdateCoordinator::new(&config).map_err(|e| {
            PipelineError::ConfigurationError {
                message: format!("Failed to create update coordinator: {}", e),
            }
        })?);

        let alert_handler = Arc::new(ConsoleAlertHandler);
        let performance_monitor = Arc::new(PipelinePerformanceMonitor::new(
            config.monitoring_config.clone(),
            alert_handler,
        ));

        let version_manager = Arc::new(
            VersionManager::new(config.version_control.clone()).map_err(|e| {
                PipelineError::VersionError {
                    message: format!("Failed to create version manager: {}", e),
                }
            })?,
        );

        let consistency_manager = Arc::new(
            ConsistencyManager::new(config.consistency_level.clone()).map_err(|e| {
                PipelineError::ConsistencyError {
                    message: format!("Failed to create consistency manager: {}", e),
                }
            })?,
        );

        let stats = Arc::new(PipelineStatistics::default());

        Ok(Self {
            config,
            embedding_generators,
            indices,
            stream_processors,
            update_coordinator,
            performance_monitor,
            version_manager,
            consistency_manager,
            is_running: AtomicBool::new(false),
            stats,
        })
    }

    /// Start the pipeline
    pub async fn start(&self) -> PipelineResult<()> {
        if self.is_running.load(Ordering::Acquire) {
            return Err(PipelineError::AlreadyRunning);
        }

        self.is_running.store(true, Ordering::Release);

        // Start performance monitoring
        self.performance_monitor
            .start()
            .await
            .map_err(|e| PipelineError::MonitoringError {
                message: format!("Failed to start performance monitor: {}", e),
            })?;

        // Start update coordinator
        self.start_update_coordinator().await?;

        // Start stream processors
        self.start_stream_processors().await?;

        // Start consistency checking
        self.consistency_manager
            .start_consistency_checking()
            .await
            .map_err(|e| PipelineError::ConsistencyError {
                message: format!("Failed to start consistency checking: {}", e),
            })?;

        // Start version manager
        self.version_manager
            .start()
            .await
            .map_err(|e| PipelineError::VersionError {
                message: format!("Failed to start version manager: {}", e),
            })?;

        Ok(())
    }

    /// Stop the pipeline
    pub async fn stop(&self) -> PipelineResult<()> {
        self.is_running.store(false, Ordering::Release);

        // Stop performance monitoring (ignore NotRunning — monitor may not have been started)
        let _ = self.performance_monitor.stop().await;

        // Stop consistency and version managers (ignore NotRunning)
        let _ = self.consistency_manager.stop().await;
        let _ = self.version_manager.stop().await;

        // Stop stream processors - simplified approach to avoid cloning issues
        debug!("Stream processor stopping not yet implemented to avoid cloning issues");

        Ok(())
    }

    /// Add an embedding generator
    pub fn add_embedding_generator(
        &self,
        name: String,
        generator: Box<dyn EmbeddingGenerator>,
    ) -> PipelineResult<()> {
        let mut generators =
            self.embedding_generators
                .write()
                .map_err(|_| PipelineError::CoordinationError {
                    message: "Failed to acquire generators lock".to_string(),
                })?;

        generators.insert(name, generator);
        Ok(())
    }

    /// Add a vector index
    pub fn add_vector_index(
        &self,
        name: String,
        index: Box<dyn IncrementalVectorIndex>,
    ) -> PipelineResult<()> {
        let mut indices = self
            .indices
            .write()
            .map_err(|_| PipelineError::CoordinationError {
                message: "Failed to acquire indices lock".to_string(),
            })?;

        indices.insert(name, index);
        Ok(())
    }

    /// Create a new stream processor
    pub async fn create_stream(&self, config: StreamConfig) -> PipelineResult<String> {
        let stream_id = Uuid::new_v4().to_string();
        let processor = StreamProcessor::new(stream_id.clone(), config).map_err(|e| {
            PipelineError::StreamProcessingError {
                message: format!("Failed to create stream processor: {e}"),
            }
        })?;

        {
            let mut processors =
                self.stream_processors
                    .write()
                    .map_err(|_| PipelineError::CoordinationError {
                        message: "Failed to acquire stream processors lock".to_string(),
                    })?;

            processors.insert(stream_id.clone(), processor);
        }

        Ok(stream_id)
    }

    /// Remove a stream processor
    pub async fn remove_stream(&self, stream_id: &str) -> PipelineResult<()> {
        let processor = {
            let mut processors =
                self.stream_processors
                    .write()
                    .map_err(|_| PipelineError::CoordinationError {
                        message: "Failed to acquire stream processors lock".to_string(),
                    })?;
            processors.remove(stream_id)
        };

        if let Some(processor) = processor {
            processor
                .stop()
                .await
                .map_err(|e| PipelineError::StreamProcessingError {
                    message: format!("Failed to stop stream processor: {e}"),
                })?;
        }

        Ok(())
    }

    /// Get pipeline statistics
    pub fn get_statistics(&self) -> Arc<PipelineStatistics> {
        self.stats.clone()
    }

    /// Get pipeline health status
    pub async fn health_check(
        &self,
    ) -> PipelineResult<crate::real_time_embedding_pipeline::types::HealthCheckResult> {
        let mut components = HashMap::new();

        // Check performance monitor health
        let monitor_health = self.performance_monitor.get_health_status().await?;
        components.insert("performance_monitor".to_string(), monitor_health);

        // Check consistency manager health
        let consistency_health = self.consistency_manager.health_check().await.map_err(|e| {
            PipelineError::ConsistencyError {
                message: format!("Consistency manager health check failed: {}", e),
            }
        })?;
        components.insert("consistency_manager".to_string(), consistency_health);

        // Check version manager health
        let version_health =
            self.version_manager
                .health_check()
                .await
                .map_err(|e| PipelineError::VersionError {
                    message: format!("Version manager health check failed: {}", e),
                })?;
        components.insert("version_manager".to_string(), version_health);

        // Check stream processors health
        // Collect processor names first to avoid holding lock during async calls
        let processor_names: Vec<String> = {
            let processors =
                self.stream_processors
                    .read()
                    .map_err(|_| PipelineError::CoordinationError {
                        message: "Failed to acquire stream processors lock".to_string(),
                    })?;
            processors.keys().cloned().collect()
        };

        // Assume healthy for stream processors (they don't have async health check)
        for name in processor_names {
            components.insert(
                format!("stream_processor_{name}"),
                crate::real_time_embedding_pipeline::traits::HealthStatus::Healthy,
            );
        }

        // Determine overall health status
        let overall_status = if components.values().all(|status| {
            matches!(
                status,
                crate::real_time_embedding_pipeline::traits::HealthStatus::Healthy
            )
        }) {
            crate::real_time_embedding_pipeline::traits::HealthStatus::Healthy
        } else if components.values().any(|status| {
            matches!(
                status,
                crate::real_time_embedding_pipeline::traits::HealthStatus::Unhealthy { .. }
            )
        }) {
            crate::real_time_embedding_pipeline::traits::HealthStatus::Unhealthy {
                message: "One or more components are unhealthy".to_string(),
            }
        } else {
            crate::real_time_embedding_pipeline::traits::HealthStatus::Warning {
                message: "Some components have warnings".to_string(),
            }
        };

        Ok(
            crate::real_time_embedding_pipeline::types::HealthCheckResult {
                status: overall_status,
                components,
                timestamp: std::time::SystemTime::now(),
                details: HashMap::new(),
            },
        )
    }

    /// Check if the pipeline is running
    pub fn is_running(&self) -> bool {
        self.is_running.load(Ordering::Acquire)
    }

    /// Get pipeline configuration
    pub fn get_config(&self) -> &PipelineConfig {
        &self.config
    }

    /// Update pipeline configuration (for some settings)
    pub async fn update_config(&mut self, new_config: PipelineConfig) -> PipelineResult<()> {
        if self.is_running() {
            return Err(PipelineError::ConfigurationError {
                message: "Cannot update configuration while pipeline is running".to_string(),
            });
        }

        self.config = new_config;
        Ok(())
    }

    /// Get list of available embedding generators
    pub fn list_embedding_generators(&self) -> PipelineResult<Vec<String>> {
        let generators =
            self.embedding_generators
                .read()
                .map_err(|_| PipelineError::CoordinationError {
                    message: "Failed to acquire generators lock".to_string(),
                })?;

        Ok(generators.keys().cloned().collect())
    }

    /// Get list of available vector indices
    pub fn list_vector_indices(&self) -> PipelineResult<Vec<String>> {
        let indices = self
            .indices
            .read()
            .map_err(|_| PipelineError::CoordinationError {
                message: "Failed to acquire indices lock".to_string(),
            })?;

        Ok(indices.keys().cloned().collect())
    }

    /// Get list of active streams
    pub fn list_streams(&self) -> PipelineResult<Vec<String>> {
        let processors =
            self.stream_processors
                .read()
                .map_err(|_| PipelineError::CoordinationError {
                    message: "Failed to acquire stream processors lock".to_string(),
                })?;

        Ok(processors.keys().cloned().collect())
    }

    // Private helper methods

    async fn start_update_coordinator(&self) -> PipelineResult<()> {
        self.update_coordinator
            .start()
            .await
            .map_err(|e| PipelineError::CoordinationError {
                message: format!("Failed to start update coordinator: {}", e),
            })
    }

    async fn start_stream_processors(&self) -> PipelineResult<()> {
        // For now, skip starting processors to avoid mutex await issue
        // TODO: Implement proper async processor starting mechanism
        debug!("Stream processors start not yet implemented to avoid mutex await issue");
        Ok(())
    }
}

impl Drop for RealTimeEmbeddingPipeline {
    fn drop(&mut self) {
        if self.is_running.load(Ordering::Acquire) {
            // Best effort to stop the pipeline
            self.is_running.store(false, Ordering::Release);
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::real_time_embedding_pipeline::config::ConsistencyLevel;
    type Result<T> = std::result::Result<T, Box<dyn std::error::Error>>;

    #[tokio::test]
    async fn test_pipeline_creation() {
        let config = PipelineConfig::default();
        let pipeline = RealTimeEmbeddingPipeline::new(config);
        assert!(pipeline.is_ok());
    }

    #[tokio::test]
    async fn test_pipeline_start_stop() -> Result<()> {
        let config = PipelineConfig::default();
        let pipeline = RealTimeEmbeddingPipeline::new(config)?;

        assert!(!pipeline.is_running());

        // Start pipeline
        let start_result = pipeline.start().await;
        // May fail due to missing dependencies in test environment
        // but should not panic
        let _ = start_result;

        // Stop pipeline
        let stop_result = pipeline.stop().await;
        let _ = stop_result;
        Ok(())
    }

    #[test]
    fn test_pipeline_configuration() -> Result<()> {
        let config = PipelineConfig {
            consistency_level: ConsistencyLevel::Strong,
            max_batch_size: 500,
            ..Default::default()
        };

        let pipeline = RealTimeEmbeddingPipeline::new(config)?;
        assert_eq!(pipeline.get_config().max_batch_size, 500);
        assert_eq!(
            pipeline.get_config().consistency_level,
            ConsistencyLevel::Strong
        );
        Ok(())
    }
}