oxirs-stream 0.2.4

//! # Visual Stream Designer and Debugger
//!
//! Comprehensive visual tool for designing, debugging, and optimizing stream processing pipelines.
//! Provides a graph-based interface for building complex stream processing flows with real-time
//! debugging, performance profiling, and automatic optimization suggestions.
//!
//! ## Features
//! - Visual pipeline designer with drag-and-drop interface
//! - Real-time debugging with event visualization
//! - Performance profiling and bottleneck detection
//! - Automatic pipeline validation and optimization
//! - Export/import pipeline definitions (JSON, YAML, DOT)
//! - Integration with existing stream processing operators
//! - Live monitoring and metrics dashboard
//! - Breakpoint support for debugging
//! - Time-travel debugging for historical analysis

use anyhow::{anyhow, Result};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::Arc;
use tokio::sync::RwLock;
use tracing::{debug, info};
use uuid::Uuid;

use crate::event::StreamEvent;

/// Visual pipeline designer configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VisualDesignerConfig {
    pub enable_auto_layout: bool,
    pub enable_validation: bool,
    pub enable_optimization: bool,
    pub max_nodes: usize,
    pub max_edges: usize,
    pub enable_real_time_debug: bool,
    pub debug_buffer_size: usize,
    pub enable_profiling: bool,
    pub export_formats: Vec<ExportFormat>,
}

/// Supported export formats
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum ExportFormat {
    Json,
    Yaml,
    Dot,     // GraphViz DOT format
    Mermaid, // Mermaid diagram format
    Svg,     // SVG image
    Png,     // PNG image
}

/// Pipeline node representing a stream operation
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PipelineNode {
    pub id: String,
    pub name: String,
    pub node_type: NodeType,
    pub position: Position,
    pub config: NodeConfig,
    pub metadata: NodeMetadata,
    pub status: NodeStatus,
}

/// Node types for different stream operations
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum NodeType {
    // Source nodes
    Source(SourceType),
    // Processing nodes
    Map,
    Filter,
    FlatMap,
    Reduce,
    Aggregate,
    Join,
    Window,
    // Transformation nodes
    Transform(TransformType),
    // ML nodes
    MLModel(MLModelType),
    // Output nodes
    Sink(SinkType),
    // Control nodes
    Router,
    Splitter,
    Merger,
    // Debug nodes
    Breakpoint,
    Logger,
    Profiler,
}

/// Source types
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum SourceType {
    Kafka,
    Nats,
    Redis,
    Memory,
    File,
    WebSocket,
    Http,
    Custom(String),
}

/// Transform types
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum TransformType {
    RdfTransform,
    SparqlQuery,
    GraphPattern,
    Custom(String),
}

/// ML model types
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum MLModelType {
    OnlineLearning,
    AnomalyDetection,
    Prediction,
    Classification,
    Clustering,
    Custom(String),
}

/// Sink types
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum SinkType {
    Kafka,
    Nats,
    Redis,
    Database,
    File,
    WebSocket,
    Http,
    Custom(String),
}

/// Node position in visual canvas
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Position {
    pub x: f64,
    pub y: f64,
    pub z: Option<f64>, // For 3D visualization
}

/// Node configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NodeConfig {
    pub parameters: HashMap<String, ConfigValue>,
    pub input_ports: Vec<Port>,
    pub output_ports: Vec<Port>,
    pub resource_limits: ResourceLimits,
}

/// Configuration value types
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(untagged)]
pub enum ConfigValue {
    String(String),
    Number(f64),
    Boolean(bool),
    Array(Vec<ConfigValue>),
    Object(HashMap<String, ConfigValue>),
}

/// Port for node connections
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Port {
    pub id: String,
    pub name: String,
    pub port_type: PortType,
    pub data_type: DataType,
    pub required: bool,
}

/// Port types
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum PortType {
    Input,
    Output,
}

/// Data types flowing through ports
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum DataType {
    StreamEvent,
    RdfTriple,
    SparqlResult,
    Json,
    Binary,
    Custom(String),
}

/// Resource limits for nodes
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResourceLimits {
    pub max_memory_mb: Option<u64>,
    pub max_cpu_percent: Option<f64>,
    pub max_execution_time_ms: Option<u64>,
    pub max_events_per_second: Option<u64>,
}

/// Node metadata
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NodeMetadata {
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
    pub version: String,
    pub author: Option<String>,
    pub description: Option<String>,
    pub tags: Vec<String>,
}

/// Node status for monitoring
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum NodeStatus {
    Idle,
    Running,
    Paused,
    Error(String),
    Completed,
}

/// Pipeline edge connecting nodes
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PipelineEdge {
    pub id: String,
    pub source_node_id: String,
    pub source_port_id: String,
    pub target_node_id: String,
    pub target_port_id: String,
    pub edge_type: EdgeType,
    pub config: EdgeConfig,
    pub metadata: EdgeMetadata,
}

/// Edge types
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum EdgeType {
    DataFlow,
    ControlFlow,
    Conditional(Condition),
}

/// Condition for conditional edges
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct Condition {
    pub expression: String,
    pub predicate_type: PredicateType,
}

/// Predicate types
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum PredicateType {
    Equals,
    NotEquals,
    GreaterThan,
    LessThan,
    Contains,
    Matches,
    Custom(String),
}

/// Edge configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EdgeConfig {
    pub buffer_size: usize,
    pub backpressure_strategy: BackpressureStrategy,
    pub error_handling: ErrorHandling,
}

/// Backpressure strategies
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum BackpressureStrategy {
    Drop,
    Buffer,
    Block,
    Exponential,
    Adaptive,
}

/// Error handling strategies
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ErrorHandling {
    Propagate,
    Ignore,
    Retry { max_attempts: u32 },
    DeadLetter,
    Custom(String),
}

/// Edge metadata
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EdgeMetadata {
    pub created_at: DateTime<Utc>,
    pub label: Option<String>,
    pub style: EdgeStyle,
}

/// Edge visual style
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EdgeStyle {
    pub color: String,
    pub thickness: f64,
    pub line_type: LineType,
}

/// Line types for edges
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum LineType {
    Solid,
    Dashed,
    Dotted,
    Curved,
}

/// Visual pipeline definition
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VisualPipeline {
    pub id: String,
    pub name: String,
    pub description: Option<String>,
    pub version: String,
    pub nodes: HashMap<String, PipelineNode>,
    pub edges: HashMap<String, PipelineEdge>,
    pub metadata: PipelineMetadata,
    pub validation_result: Option<ValidationResult>,
}

/// Pipeline metadata
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PipelineMetadata {
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
    pub author: Option<String>,
    pub tags: Vec<String>,
    pub properties: HashMap<String, String>,
}

/// Validation result
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ValidationResult {
    pub is_valid: bool,
    pub errors: Vec<ValidationError>,
    pub warnings: Vec<ValidationWarning>,
    pub validated_at: DateTime<Utc>,
}

/// Validation error
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ValidationError {
    pub error_type: ValidationErrorType,
    pub message: String,
    pub node_id: Option<String>,
    pub edge_id: Option<String>,
}

/// Validation error types
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum ValidationErrorType {
    MissingRequiredPort,
    IncompatibleDataTypes,
    CyclicDependency,
    DisconnectedNode,
    InvalidConfiguration,
    ResourceLimitExceeded,
    DuplicateNodeId,
}

/// Validation warning
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ValidationWarning {
    pub warning_type: ValidationWarningType,
    pub message: String,
    pub node_id: Option<String>,
    pub suggestion: Option<String>,
}

/// Validation warning types
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum ValidationWarningType {
    UnusedPort,
    SuboptimalConfiguration,
    PerformanceBottleneck,
    MemoryPressure,
    DeprecatedNode,
}

/// Pipeline debugger for real-time debugging
#[derive(Debug)]
pub struct PipelineDebugger {
    pub pipeline: VisualPipeline,
    pub config: DebuggerConfig,
    pub state: Arc<RwLock<DebuggerState>>,
    pub breakpoints: Arc<RwLock<HashMap<String, Breakpoint>>>,
    pub event_history: Arc<RwLock<VecDeque<DebugEvent>>>,
}

/// Debugger configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DebuggerConfig {
    pub enable_breakpoints: bool,
    pub enable_event_capture: bool,
    pub max_event_history: usize,
    pub enable_time_travel: bool,
    pub enable_profiling: bool,
    pub capture_intermediate_results: bool,
}

/// Debugger state
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DebuggerState {
    pub is_running: bool,
    pub is_paused: bool,
    pub current_node_id: Option<String>,
    pub execution_stack: Vec<String>,
    pub variables: HashMap<String, DebugVariable>,
    pub metrics: DebugMetrics,
}

/// Debug variable
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DebugVariable {
    pub name: String,
    pub value: String,
    pub var_type: String,
    pub scope: String,
}

/// Debug metrics
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct DebugMetrics {
    pub events_processed: u64,
    pub events_dropped: u64,
    pub average_latency_ms: f64,
    pub throughput_per_second: f64,
    pub memory_usage_mb: f64,
    pub cpu_usage_percent: f64,
}

/// Breakpoint for debugging
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Breakpoint {
    pub id: String,
    pub node_id: String,
    pub condition: Option<String>,
    pub enabled: bool,
    pub hit_count: u64,
    pub max_hits: Option<u64>,
}

/// Debug event captured during execution
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DebugEvent {
    pub timestamp: DateTime<Utc>,
    pub node_id: String,
    pub event_type: DebugEventType,
    pub data: StreamEvent,
    pub metadata: HashMap<String, String>,
}

/// Debug event types
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum DebugEventType {
    NodeEnter,
    NodeExit,
    BreakpointHit,
    Error,
    Warning,
}

/// Visual designer main struct
pub struct VisualStreamDesigner {
    config: VisualDesignerConfig,
    pipelines: Arc<RwLock<HashMap<String, VisualPipeline>>>,
    debuggers: Arc<RwLock<HashMap<String, PipelineDebugger>>>,
    validator: Arc<PipelineValidator>,
    optimizer: Arc<PipelineOptimizer>,
}

impl VisualStreamDesigner {
    /// Create a new visual stream designer
    pub fn new(config: VisualDesignerConfig) -> Self {
        Self {
            config: config.clone(),
            pipelines: Arc::new(RwLock::new(HashMap::new())),
            debuggers: Arc::new(RwLock::new(HashMap::new())),
            validator: Arc::new(PipelineValidator::new(config.clone())),
            optimizer: Arc::new(PipelineOptimizer::new(config)),
        }
    }

    /// Create a new empty pipeline
    pub async fn create_pipeline(
        &self,
        name: String,
        description: Option<String>,
    ) -> Result<String> {
        let pipeline = VisualPipeline {
            id: Uuid::new_v4().to_string(),
            name,
            description,
            version: "1.0.0".to_string(),
            nodes: HashMap::new(),
            edges: HashMap::new(),
            metadata: PipelineMetadata {
                created_at: Utc::now(),
                updated_at: Utc::now(),
                author: None,
                tags: Vec::new(),
                properties: HashMap::new(),
            },
            validation_result: None,
        };

        let id = pipeline.id.clone();
        self.pipelines.write().await.insert(id.clone(), pipeline);

        info!("Created new pipeline: {}", id);
        Ok(id)
    }

    /// Add a node to the pipeline
    pub async fn add_node(
        &self,
        pipeline_id: &str,
        name: String,
        node_type: NodeType,
        position: Position,
    ) -> Result<String> {
        let mut pipelines = self.pipelines.write().await;
        let pipeline = pipelines
            .get_mut(pipeline_id)
            .ok_or_else(|| anyhow!("Pipeline not found"))?;

        let node = PipelineNode {
            id: Uuid::new_v4().to_string(),
            name,
            node_type: node_type.clone(),
            position,
            config: Self::default_node_config(&node_type),
            metadata: NodeMetadata {
                created_at: Utc::now(),
                updated_at: Utc::now(),
                version: "1.0.0".to_string(),
                author: None,
                description: None,
                tags: Vec::new(),
            },
            status: NodeStatus::Idle,
        };

        let node_id = node.id.clone();
        pipeline.nodes.insert(node_id.clone(), node);
        pipeline.metadata.updated_at = Utc::now();

        debug!("Added node {} to pipeline {}", node_id, pipeline_id);
        Ok(node_id)
    }

    /// Add an edge connecting two nodes
    pub async fn add_edge(
        &self,
        pipeline_id: &str,
        source_node_id: String,
        source_port_id: String,
        target_node_id: String,
        target_port_id: String,
    ) -> Result<String> {
        let mut pipelines = self.pipelines.write().await;
        let pipeline = pipelines
            .get_mut(pipeline_id)
            .ok_or_else(|| anyhow!("Pipeline not found"))?;

        // Validate nodes exist
        if !pipeline.nodes.contains_key(&source_node_id) {
            return Err(anyhow!("Source node not found"));
        }
        if !pipeline.nodes.contains_key(&target_node_id) {
            return Err(anyhow!("Target node not found"));
        }

        let edge = PipelineEdge {
            id: Uuid::new_v4().to_string(),
            source_node_id,
            source_port_id,
            target_node_id,
            target_port_id,
            edge_type: EdgeType::DataFlow,
            config: EdgeConfig {
                buffer_size: 1000,
                backpressure_strategy: BackpressureStrategy::Block,
                error_handling: ErrorHandling::Propagate,
            },
            metadata: EdgeMetadata {
                created_at: Utc::now(),
                label: None,
                style: EdgeStyle {
                    color: "#000000".to_string(),
                    thickness: 2.0,
                    line_type: LineType::Solid,
                },
            },
        };

        let edge_id = edge.id.clone();
        pipeline.edges.insert(edge_id.clone(), edge);
        pipeline.metadata.updated_at = Utc::now();

        debug!("Added edge {} to pipeline {}", edge_id, pipeline_id);
        Ok(edge_id)
    }

    /// Validate a pipeline
    pub async fn validate_pipeline(&self, pipeline_id: &str) -> Result<ValidationResult> {
        let pipelines = self.pipelines.read().await;
        let pipeline = pipelines
            .get(pipeline_id)
            .ok_or_else(|| anyhow!("Pipeline not found"))?;

        self.validator.validate(pipeline).await
    }

    /// Optimize a pipeline
    pub async fn optimize_pipeline(&self, pipeline_id: &str) -> Result<OptimizationResult> {
        let pipelines = self.pipelines.read().await;
        let pipeline = pipelines
            .get(pipeline_id)
            .ok_or_else(|| anyhow!("Pipeline not found"))?;

        self.optimizer.optimize(pipeline).await
    }

    /// Export pipeline to specified format
    pub async fn export_pipeline(&self, pipeline_id: &str, format: ExportFormat) -> Result<String> {
        let pipelines = self.pipelines.read().await;
        let pipeline = pipelines
            .get(pipeline_id)
            .ok_or_else(|| anyhow!("Pipeline not found"))?;

        match format {
            ExportFormat::Json => serde_json::to_string_pretty(pipeline)
                .map_err(|e| anyhow!("JSON export failed: {}", e)),
            ExportFormat::Yaml => {
                serde_yaml::to_string(pipeline).map_err(|e| anyhow!("YAML export failed: {}", e))
            }
            ExportFormat::Dot => self.export_dot(pipeline),
            ExportFormat::Mermaid => self.export_mermaid(pipeline),
            ExportFormat::Svg => Err(anyhow!("SVG export not yet implemented")),
            ExportFormat::Png => Err(anyhow!("PNG export not yet implemented")),
        }
    }

    /// Import pipeline from string
    pub async fn import_pipeline(&self, data: &str, format: ExportFormat) -> Result<String> {
        let pipeline: VisualPipeline = match format {
            ExportFormat::Json => {
                serde_json::from_str(data).map_err(|e| anyhow!("JSON import failed: {}", e))?
            }
            ExportFormat::Yaml => {
                serde_yaml::from_str(data).map_err(|e| anyhow!("YAML import failed: {}", e))?
            }
            _ => return Err(anyhow!("Import format not supported")),
        };

        let id = pipeline.id.clone();
        self.pipelines.write().await.insert(id.clone(), pipeline);

        info!("Imported pipeline: {}", id);
        Ok(id)
    }

    /// Create a debugger for a pipeline
    pub async fn create_debugger(
        &self,
        pipeline_id: &str,
        config: DebuggerConfig,
    ) -> Result<String> {
        let pipelines = self.pipelines.read().await;
        let pipeline = pipelines
            .get(pipeline_id)
            .ok_or_else(|| anyhow!("Pipeline not found"))?;

        let max_event_history = config.max_event_history;
        let debugger = PipelineDebugger {
            pipeline: pipeline.clone(),
            config,
            state: Arc::new(RwLock::new(DebuggerState {
                is_running: false,
                is_paused: false,
                current_node_id: None,
                execution_stack: Vec::new(),
                variables: HashMap::new(),
                metrics: DebugMetrics::default(),
            })),
            breakpoints: Arc::new(RwLock::new(HashMap::new())),
            event_history: Arc::new(RwLock::new(VecDeque::with_capacity(max_event_history))),
        };

        let debugger_id = Uuid::new_v4().to_string();
        self.debuggers
            .write()
            .await
            .insert(debugger_id.clone(), debugger);

        info!(
            "Created debugger {} for pipeline {}",
            debugger_id, pipeline_id
        );
        Ok(debugger_id)
    }

    /// Add a breakpoint to a debugger
    pub async fn add_breakpoint(
        &self,
        debugger_id: &str,
        node_id: String,
        condition: Option<String>,
    ) -> Result<String> {
        let debuggers = self.debuggers.read().await;
        let debugger = debuggers
            .get(debugger_id)
            .ok_or_else(|| anyhow!("Debugger not found"))?;

        let breakpoint = Breakpoint {
            id: Uuid::new_v4().to_string(),
            node_id,
            condition,
            enabled: true,
            hit_count: 0,
            max_hits: None,
        };

        let breakpoint_id = breakpoint.id.clone();
        debugger
            .breakpoints
            .write()
            .await
            .insert(breakpoint_id.clone(), breakpoint);

        debug!(
            "Added breakpoint {} to debugger {}",
            breakpoint_id, debugger_id
        );
        Ok(breakpoint_id)
    }

    /// Get debugger state
    pub async fn get_debugger_state(&self, debugger_id: &str) -> Result<DebuggerState> {
        let debuggers = self.debuggers.read().await;
        let debugger = debuggers
            .get(debugger_id)
            .ok_or_else(|| anyhow!("Debugger not found"))?;

        let state = debugger.state.read().await.clone();
        drop(debuggers); // Explicitly drop to avoid borrow issues
        Ok(state)
    }

    /// Get event history from debugger
    pub async fn get_event_history(
        &self,
        debugger_id: &str,
        limit: Option<usize>,
    ) -> Result<Vec<DebugEvent>> {
        let debuggers = self.debuggers.read().await;
        let debugger = debuggers
            .get(debugger_id)
            .ok_or_else(|| anyhow!("Debugger not found"))?;

        let history = debugger.event_history.read().await;
        let limit = limit.unwrap_or(history.len());

        Ok(history.iter().rev().take(limit).cloned().collect())
    }

    /// Export pipeline to DOT format (GraphViz)
    fn export_dot(&self, pipeline: &VisualPipeline) -> Result<String> {
        let mut dot = String::new();
        dot.push_str(&format!("digraph \"{}\" {{\n", pipeline.name));
        dot.push_str("  rankdir=LR;\n");
        dot.push_str("  node [shape=box];\n\n");

        // Add nodes
        for (id, node) in &pipeline.nodes {
            let label = format!("{}\\n{:?}", node.name, node.node_type);
            dot.push_str(&format!("  \"{}\" [label=\"{}\"];\n", id, label));
        }

        dot.push('\n');

        // Add edges
        for edge in pipeline.edges.values() {
            dot.push_str(&format!(
                "  \"{}\" -> \"{}\";\n",
                edge.source_node_id, edge.target_node_id
            ));
        }

        dot.push_str("}\n");
        Ok(dot)
    }

    /// Export pipeline to Mermaid format
    fn export_mermaid(&self, pipeline: &VisualPipeline) -> Result<String> {
        let mut mermaid = String::new();
        mermaid.push_str("graph LR\n");

        // Add nodes
        for (id, node) in &pipeline.nodes {
            let label = format!("{}[{}]", id, node.name);
            mermaid.push_str(&format!("  {}\n", label));
        }

        // Add edges
        for edge in pipeline.edges.values() {
            mermaid.push_str(&format!(
                "  {} --> {}\n",
                edge.source_node_id, edge.target_node_id
            ));
        }

        Ok(mermaid)
    }

    /// Get default node configuration based on type
    fn default_node_config(_node_type: &NodeType) -> NodeConfig {
        NodeConfig {
            parameters: HashMap::new(),
            input_ports: vec![Port {
                id: "input".to_string(),
                name: "Input".to_string(),
                port_type: PortType::Input,
                data_type: DataType::StreamEvent,
                required: true,
            }],
            output_ports: vec![Port {
                id: "output".to_string(),
                name: "Output".to_string(),
                port_type: PortType::Output,
                data_type: DataType::StreamEvent,
                required: false,
            }],
            resource_limits: ResourceLimits {
                max_memory_mb: Some(1024),
                max_cpu_percent: Some(50.0),
                max_execution_time_ms: Some(5000),
                max_events_per_second: Some(10000),
            },
        }
    }

    /// List all pipelines
    pub async fn list_pipelines(&self) -> Vec<PipelineInfo> {
        let pipelines = self.pipelines.read().await;
        pipelines
            .values()
            .map(|p| PipelineInfo {
                id: p.id.clone(),
                name: p.name.clone(),
                version: p.version.clone(),
                node_count: p.nodes.len(),
                edge_count: p.edges.len(),
                created_at: p.metadata.created_at,
                updated_at: p.metadata.updated_at,
            })
            .collect()
    }

    /// Get pipeline details
    pub async fn get_pipeline(&self, pipeline_id: &str) -> Result<VisualPipeline> {
        let pipelines = self.pipelines.read().await;
        pipelines
            .get(pipeline_id)
            .cloned()
            .ok_or_else(|| anyhow!("Pipeline not found"))
    }

    /// Delete a pipeline
    pub async fn delete_pipeline(&self, pipeline_id: &str) -> Result<()> {
        let mut pipelines = self.pipelines.write().await;
        pipelines
            .remove(pipeline_id)
            .ok_or_else(|| anyhow!("Pipeline not found"))?;

        info!("Deleted pipeline: {}", pipeline_id);
        Ok(())
    }
}

/// Pipeline information summary
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PipelineInfo {
    pub id: String,
    pub name: String,
    pub version: String,
    pub node_count: usize,
    pub edge_count: usize,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
}

/// Pipeline validator
pub struct PipelineValidator {
    config: VisualDesignerConfig,
}

impl PipelineValidator {
    pub fn new(config: VisualDesignerConfig) -> Self {
        Self { config }
    }

    pub async fn validate(&self, pipeline: &VisualPipeline) -> Result<ValidationResult> {
        let mut errors = Vec::new();
        let mut warnings = Vec::new();

        // Check for cyclic dependencies
        if self.has_cycle(pipeline) {
            errors.push(ValidationError {
                error_type: ValidationErrorType::CyclicDependency,
                message: "Pipeline contains cyclic dependencies".to_string(),
                node_id: None,
                edge_id: None,
            });
        }

        // Check for disconnected nodes
        let disconnected = self.find_disconnected_nodes(pipeline);
        for node_id in disconnected {
            warnings.push(ValidationWarning {
                warning_type: ValidationWarningType::UnusedPort,
                message: format!("Node {} is disconnected", node_id),
                node_id: Some(node_id),
                suggestion: Some("Connect this node to the pipeline".to_string()),
            });
        }

        // Check data type compatibility
        for edge in pipeline.edges.values() {
            if let Err(e) = self.check_data_type_compatibility(pipeline, edge) {
                errors.push(ValidationError {
                    error_type: ValidationErrorType::IncompatibleDataTypes,
                    message: e.to_string(),
                    node_id: None,
                    edge_id: Some(edge.id.clone()),
                });
            }
        }

        // Check resource limits
        if pipeline.nodes.len() > self.config.max_nodes {
            errors.push(ValidationError {
                error_type: ValidationErrorType::ResourceLimitExceeded,
                message: format!(
                    "Pipeline exceeds maximum node count: {} > {}",
                    pipeline.nodes.len(),
                    self.config.max_nodes
                ),
                node_id: None,
                edge_id: None,
            });
        }

        Ok(ValidationResult {
            is_valid: errors.is_empty(),
            errors,
            warnings,
            validated_at: Utc::now(),
        })
    }

    fn has_cycle(&self, pipeline: &VisualPipeline) -> bool {
        let mut visited = HashSet::new();
        let mut rec_stack = HashSet::new();

        for node_id in pipeline.nodes.keys() {
            if Self::detect_cycle_util(pipeline, node_id, &mut visited, &mut rec_stack) {
                return true;
            }
        }

        false
    }

    fn detect_cycle_util(
        pipeline: &VisualPipeline,
        node_id: &str,
        visited: &mut HashSet<String>,
        rec_stack: &mut HashSet<String>,
    ) -> bool {
        if !visited.contains(node_id) {
            visited.insert(node_id.to_string());
            rec_stack.insert(node_id.to_string());

            // Get all neighbors
            let neighbors: Vec<String> = pipeline
                .edges
                .values()
                .filter(|e| e.source_node_id == node_id)
                .map(|e| e.target_node_id.clone())
                .collect();

            for neighbor in neighbors {
                if !visited.contains(&neighbor)
                    && Self::detect_cycle_util(pipeline, &neighbor, visited, rec_stack)
                {
                    return true;
                }
                if rec_stack.contains(&neighbor) {
                    return true;
                }
            }
        }

        rec_stack.remove(node_id);
        false
    }

    fn find_disconnected_nodes(&self, pipeline: &VisualPipeline) -> Vec<String> {
        let mut connected = HashSet::new();

        for edge in pipeline.edges.values() {
            connected.insert(edge.source_node_id.clone());
            connected.insert(edge.target_node_id.clone());
        }

        pipeline
            .nodes
            .keys()
            .filter(|id| !connected.contains(*id))
            .cloned()
            .collect()
    }

    fn check_data_type_compatibility(
        &self,
        _pipeline: &VisualPipeline,
        _edge: &PipelineEdge,
    ) -> Result<()> {
        // Simplified check - in real implementation, would verify port data types match
        Ok(())
    }
}

/// Pipeline optimizer
pub struct PipelineOptimizer {
    config: VisualDesignerConfig,
}

impl PipelineOptimizer {
    pub fn new(config: VisualDesignerConfig) -> Self {
        Self { config }
    }

    pub async fn optimize(&self, pipeline: &VisualPipeline) -> Result<OptimizationResult> {
        let mut suggestions = Vec::new();

        // Analyze pipeline structure
        let metrics = self.analyze_structure(pipeline);

        // Generate optimization suggestions
        if metrics.avg_chain_length > 10.0 {
            suggestions.push(OptimizationSuggestion {
                suggestion_type: OptimizationType::ReduceChainLength,
                impact: ImpactLevel::High,
                description: "Consider breaking long chains into parallel branches".to_string(),
                estimated_improvement: 30.0,
            });
        }

        if metrics.parallel_opportunities > 0 {
            suggestions.push(OptimizationSuggestion {
                suggestion_type: OptimizationType::IncreaseParallelism,
                impact: ImpactLevel::Medium,
                description: format!(
                    "Found {} opportunities for parallel processing",
                    metrics.parallel_opportunities
                ),
                estimated_improvement: 15.0 * metrics.parallel_opportunities as f64,
            });
        }

        Ok(OptimizationResult {
            original_metrics: metrics,
            suggestions,
            optimized_at: Utc::now(),
        })
    }

    fn analyze_structure(&self, pipeline: &VisualPipeline) -> PipelineMetrics {
        let mut total_chain_length = 0;
        let mut chain_count = 0;

        // Simple chain length calculation
        for node_id in pipeline.nodes.keys() {
            let depth = self.calculate_depth(pipeline, node_id);
            total_chain_length += depth;
            chain_count += 1;
        }

        PipelineMetrics {
            node_count: pipeline.nodes.len(),
            edge_count: pipeline.edges.len(),
            avg_chain_length: if chain_count > 0 {
                total_chain_length as f64 / chain_count as f64
            } else {
                0.0
            },
            max_chain_length: total_chain_length,
            parallel_opportunities: self.count_parallel_opportunities(pipeline),
            bottleneck_nodes: Vec::new(),
        }
    }

    fn calculate_depth(&self, pipeline: &VisualPipeline, node_id: &str) -> usize {
        let mut depth = 0;
        let mut current = node_id.to_string();

        while let Some(edge) = pipeline
            .edges
            .values()
            .find(|e| e.target_node_id == current)
        {
            depth += 1;
            current = edge.source_node_id.clone();
        }

        depth
    }

    fn count_parallel_opportunities(&self, pipeline: &VisualPipeline) -> usize {
        let mut count = 0;

        for node_id in pipeline.nodes.keys() {
            let outgoing: Vec<_> = pipeline
                .edges
                .values()
                .filter(|e| e.source_node_id == *node_id)
                .collect();

            if outgoing.len() > 1 {
                count += 1;
            }
        }

        count
    }
}

/// Optimization result
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OptimizationResult {
    pub original_metrics: PipelineMetrics,
    pub suggestions: Vec<OptimizationSuggestion>,
    pub optimized_at: DateTime<Utc>,
}

/// Pipeline metrics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PipelineMetrics {
    pub node_count: usize,
    pub edge_count: usize,
    pub avg_chain_length: f64,
    pub max_chain_length: usize,
    pub parallel_opportunities: usize,
    pub bottleneck_nodes: Vec<String>,
}

/// Optimization suggestion
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OptimizationSuggestion {
    pub suggestion_type: OptimizationType,
    pub impact: ImpactLevel,
    pub description: String,
    pub estimated_improvement: f64,
}

/// Optimization types
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum OptimizationType {
    ReduceChainLength,
    IncreaseParallelism,
    OptimizeBufferSize,
    ReduceMemoryUsage,
    ImproveLocality,
}

/// Impact levels
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum ImpactLevel {
    Low,
    Medium,
    High,
    Critical,
}

impl Default for VisualDesignerConfig {
    fn default() -> Self {
        Self {
            enable_auto_layout: true,
            enable_validation: true,
            enable_optimization: true,
            max_nodes: 1000,
            max_edges: 5000,
            enable_real_time_debug: true,
            debug_buffer_size: 10000,
            enable_profiling: true,
            export_formats: vec![
                ExportFormat::Json,
                ExportFormat::Yaml,
                ExportFormat::Dot,
                ExportFormat::Mermaid,
            ],
        }
    }
}

impl Default for DebuggerConfig {
    fn default() -> Self {
        Self {
            enable_breakpoints: true,
            enable_event_capture: true,
            max_event_history: 10000,
            enable_time_travel: true,
            enable_profiling: true,
            capture_intermediate_results: true,
        }
    }
}

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

    #[tokio::test]
    async fn test_create_pipeline() {
        let designer = VisualStreamDesigner::new(VisualDesignerConfig::default());
        let pipeline_id = designer
            .create_pipeline(
                "Test Pipeline".to_string(),
                Some("Test description".to_string()),
            )
            .await
            .unwrap();

        assert!(!pipeline_id.is_empty());

        let pipelines = designer.list_pipelines().await;
        assert_eq!(pipelines.len(), 1);
        assert_eq!(pipelines[0].name, "Test Pipeline");
    }

    #[tokio::test]
    async fn test_add_nodes() {
        let designer = VisualStreamDesigner::new(VisualDesignerConfig::default());
        let pipeline_id = designer
            .create_pipeline("Test".to_string(), None)
            .await
            .unwrap();

        let source_id = designer
            .add_node(
                &pipeline_id,
                "Source".to_string(),
                NodeType::Source(SourceType::Memory),
                Position {
                    x: 0.0,
                    y: 0.0,
                    z: None,
                },
            )
            .await
            .unwrap();

        let sink_id = designer
            .add_node(
                &pipeline_id,
                "Sink".to_string(),
                NodeType::Sink(SinkType::Memory),
                Position {
                    x: 100.0,
                    y: 0.0,
                    z: None,
                },
            )
            .await
            .unwrap();

        assert!(!source_id.is_empty());
        assert!(!sink_id.is_empty());

        let pipeline = designer.get_pipeline(&pipeline_id).await.unwrap();
        assert_eq!(pipeline.nodes.len(), 2);
    }

    #[tokio::test]
    async fn test_add_edge() {
        let designer = VisualStreamDesigner::new(VisualDesignerConfig::default());
        let pipeline_id = designer
            .create_pipeline("Test".to_string(), None)
            .await
            .unwrap();

        let source_id = designer
            .add_node(
                &pipeline_id,
                "Source".to_string(),
                NodeType::Source(SourceType::Memory),
                Position {
                    x: 0.0,
                    y: 0.0,
                    z: None,
                },
            )
            .await
            .unwrap();

        let sink_id = designer
            .add_node(
                &pipeline_id,
                "Sink".to_string(),
                NodeType::Sink(SinkType::Memory),
                Position {
                    x: 100.0,
                    y: 0.0,
                    z: None,
                },
            )
            .await
            .unwrap();

        let edge_id = designer
            .add_edge(
                &pipeline_id,
                source_id,
                "output".to_string(),
                sink_id,
                "input".to_string(),
            )
            .await
            .unwrap();

        assert!(!edge_id.is_empty());

        let pipeline = designer.get_pipeline(&pipeline_id).await.unwrap();
        assert_eq!(pipeline.edges.len(), 1);
    }

    #[tokio::test]
    async fn test_validate_pipeline() {
        let designer = VisualStreamDesigner::new(VisualDesignerConfig::default());
        let pipeline_id = designer
            .create_pipeline("Test".to_string(), None)
            .await
            .unwrap();

        let source_id = designer
            .add_node(
                &pipeline_id,
                "Source".to_string(),
                NodeType::Source(SourceType::Memory),
                Position {
                    x: 0.0,
                    y: 0.0,
                    z: None,
                },
            )
            .await
            .unwrap();

        let sink_id = designer
            .add_node(
                &pipeline_id,
                "Sink".to_string(),
                NodeType::Sink(SinkType::Memory),
                Position {
                    x: 100.0,
                    y: 0.0,
                    z: None,
                },
            )
            .await
            .unwrap();

        designer
            .add_edge(
                &pipeline_id,
                source_id,
                "output".to_string(),
                sink_id,
                "input".to_string(),
            )
            .await
            .unwrap();

        let validation = designer.validate_pipeline(&pipeline_id).await.unwrap();
        assert!(validation.is_valid);
        assert!(validation.errors.is_empty());
    }

    #[tokio::test]
    async fn test_export_import_json() {
        let designer = VisualStreamDesigner::new(VisualDesignerConfig::default());
        let pipeline_id = designer
            .create_pipeline("Test".to_string(), None)
            .await
            .unwrap();

        designer
            .add_node(
                &pipeline_id,
                "Source".to_string(),
                NodeType::Source(SourceType::Memory),
                Position {
                    x: 0.0,
                    y: 0.0,
                    z: None,
                },
            )
            .await
            .unwrap();

        let json = designer
            .export_pipeline(&pipeline_id, ExportFormat::Json)
            .await
            .unwrap();

        assert!(!json.is_empty());

        let new_id = designer
            .import_pipeline(&json, ExportFormat::Json)
            .await
            .unwrap();

        assert!(!new_id.is_empty());

        let imported = designer.get_pipeline(&new_id).await.unwrap();
        assert_eq!(imported.nodes.len(), 1);
    }

    #[tokio::test]
    async fn test_export_dot() {
        let designer = VisualStreamDesigner::new(VisualDesignerConfig::default());
        let pipeline_id = designer
            .create_pipeline("Test".to_string(), None)
            .await
            .unwrap();

        let source_id = designer
            .add_node(
                &pipeline_id,
                "Source".to_string(),
                NodeType::Source(SourceType::Memory),
                Position {
                    x: 0.0,
                    y: 0.0,
                    z: None,
                },
            )
            .await
            .unwrap();

        let sink_id = designer
            .add_node(
                &pipeline_id,
                "Sink".to_string(),
                NodeType::Sink(SinkType::Memory),
                Position {
                    x: 100.0,
                    y: 0.0,
                    z: None,
                },
            )
            .await
            .unwrap();

        designer
            .add_edge(
                &pipeline_id,
                source_id,
                "output".to_string(),
                sink_id,
                "input".to_string(),
            )
            .await
            .unwrap();

        let dot = designer
            .export_pipeline(&pipeline_id, ExportFormat::Dot)
            .await
            .unwrap();

        assert!(dot.contains("digraph"));
        assert!(dot.contains("Source"));
        assert!(dot.contains("Sink"));
        assert!(dot.contains("->"));
    }

    #[tokio::test]
    async fn test_create_debugger() {
        let designer = VisualStreamDesigner::new(VisualDesignerConfig::default());
        let pipeline_id = designer
            .create_pipeline("Test".to_string(), None)
            .await
            .unwrap();

        let debugger_id = designer
            .create_debugger(&pipeline_id, DebuggerConfig::default())
            .await
            .unwrap();

        assert!(!debugger_id.is_empty());

        let state = designer.get_debugger_state(&debugger_id).await.unwrap();
        assert!(!state.is_running);
        assert!(!state.is_paused);
    }

    #[tokio::test]
    async fn test_add_breakpoint() {
        let designer = VisualStreamDesigner::new(VisualDesignerConfig::default());
        let pipeline_id = designer
            .create_pipeline("Test".to_string(), None)
            .await
            .unwrap();

        let node_id = designer
            .add_node(
                &pipeline_id,
                "Filter".to_string(),
                NodeType::Filter,
                Position {
                    x: 0.0,
                    y: 0.0,
                    z: None,
                },
            )
            .await
            .unwrap();

        let debugger_id = designer
            .create_debugger(&pipeline_id, DebuggerConfig::default())
            .await
            .unwrap();

        let breakpoint_id = designer
            .add_breakpoint(&debugger_id, node_id, None)
            .await
            .unwrap();

        assert!(!breakpoint_id.is_empty());
    }

    #[tokio::test]
    async fn test_optimize_pipeline() {
        let designer = VisualStreamDesigner::new(VisualDesignerConfig::default());
        let pipeline_id = designer
            .create_pipeline("Test".to_string(), None)
            .await
            .unwrap();

        // Create a chain of nodes
        let mut prev_id = None;
        for i in 0..12 {
            let node_id = designer
                .add_node(
                    &pipeline_id,
                    format!("Node{}", i),
                    NodeType::Map,
                    Position {
                        x: i as f64 * 100.0,
                        y: 0.0,
                        z: None,
                    },
                )
                .await
                .unwrap();

            if let Some(prev) = prev_id {
                designer
                    .add_edge(
                        &pipeline_id,
                        prev,
                        "output".to_string(),
                        node_id.clone(),
                        "input".to_string(),
                    )
                    .await
                    .unwrap();
            }

            prev_id = Some(node_id);
        }

        let optimization = designer.optimize_pipeline(&pipeline_id).await.unwrap();
        assert!(!optimization.suggestions.is_empty());
    }

    #[tokio::test]
    async fn test_delete_pipeline() {
        let designer = VisualStreamDesigner::new(VisualDesignerConfig::default());
        let pipeline_id = designer
            .create_pipeline("Test".to_string(), None)
            .await
            .unwrap();

        assert_eq!(designer.list_pipelines().await.len(), 1);

        designer.delete_pipeline(&pipeline_id).await.unwrap();
        assert_eq!(designer.list_pipelines().await.len(), 0);
    }
}