scirs2-metrics 0.4.4

//! Data source management for real-time visualization
//!
//! This module provides data source abstractions for streaming real-time data
//! to interactive dashboard widgets.

#![allow(clippy::too_many_arguments)]
#![allow(dead_code)]

use crate::error::{MetricsError, Result};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::{HashMap, VecDeque};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

/// Data source trait for real-time data
pub trait DataSource: std::fmt::Debug + Send + Sync {
    /// Get data source ID
    fn id(&self) -> &str;

    /// Get latest data
    fn get_data(&self) -> Result<Value>;

    /// Subscribe to data updates
    fn subscribe(&mut self, callback: Box<dyn Fn(Value) + Send + Sync>) -> Result<String>;

    /// Unsubscribe from updates
    fn unsubscribe(&mut self, subscription_id: &str) -> Result<()>;

    /// Connect to data source
    fn connect(&mut self) -> Result<()>;

    /// Disconnect from data source
    fn disconnect(&mut self) -> Result<()>;

    /// Check if connected
    fn is_connected(&self) -> bool;

    /// Get configuration
    fn config(&self) -> &DataSourceConfig;

    /// Update configuration
    fn update_config(&mut self, config: DataSourceConfig) -> Result<()>;

    /// Get data history
    fn get_history(&self, start: Instant, end: Instant) -> Result<Vec<(Instant, Value)>>;
}

/// Data source configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DataSourceConfig {
    /// Data source ID
    pub id: String,
    /// Data source type
    pub source_type: DataSourceType,
    /// Connection configuration
    pub connection: ConnectionConfig,
    /// Data format configuration
    pub format: DataFormatConfig,
    /// Caching configuration
    pub cache: CacheConfig,
    /// Error handling configuration
    pub error_handling: ErrorHandlingConfig,
}

/// Data source type enumeration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum DataSourceType {
    /// WebSocket data source
    WebSocket,
    /// HTTP polling data source
    HttpPolling,
    /// Server-sent events
    ServerSentEvents,
    /// File-based data source
    File,
    /// Database connection
    Database,
    /// Custom data source
    Custom(String),
}

/// Connection configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConnectionConfig {
    /// Connection URL
    pub url: String,
    /// Connection headers
    pub headers: HashMap<String, String>,
    /// Authentication configuration
    pub auth: Option<AuthConfig>,
    /// Retry configuration
    pub retry: RetryConfig,
    /// Connection pooling
    pub pooling: ConnectionPoolConfig,
    /// Timeout settings
    pub timeout: Duration,
}

/// Authentication configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AuthConfig {
    /// Authentication type
    pub auth_type: AuthType,
    /// Credentials
    pub credentials: HashMap<String, String>,
    /// Token refresh URL
    pub refresh_url: Option<String>,
    /// Token expiration
    pub expires_in: Option<Duration>,
}

/// Authentication type enumeration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum AuthType {
    /// Basic authentication
    Basic,
    /// Bearer token
    Bearer,
    /// API key
    ApiKey,
    /// OAuth 2.0
    OAuth2,
    /// Custom authentication
    Custom(String),
}

/// Retry configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RetryConfig {
    /// Maximum retry attempts
    pub max_attempts: u32,
    /// Initial retry delay
    pub initial_delay: Duration,
    /// Maximum retry delay
    pub max_delay: Duration,
    /// Backoff multiplier
    pub backoff_multiplier: f64,
    /// Jitter enabled
    pub jitter: bool,
}

/// Connection pooling configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConnectionPoolConfig {
    /// Enable connection pooling
    pub enabled: bool,
    /// Maximum pool size
    pub max_size: u32,
    /// Minimum pool size
    pub min_size: u32,
    /// Connection timeout
    pub connection_timeout: Duration,
    /// Idle timeout
    pub idle_timeout: Duration,
}

/// Data format configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DataFormatConfig {
    /// Data format
    pub format: DataFormat,
    /// Schema configuration
    pub schema: Option<Value>,
    /// Field mappings
    pub field_mappings: HashMap<String, String>,
    /// Data validation
    pub validation: ValidationConfig,
}

/// Data format enumeration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum DataFormat {
    /// JSON format
    Json,
    /// CSV format
    Csv,
    /// XML format
    Xml,
    /// Protocol Buffers
    Protobuf,
    /// MessagePack
    MessagePack,
    /// Custom format
    Custom(String),
}

/// Validation configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ValidationConfig {
    /// Enable validation
    pub enabled: bool,
    /// Validation rules
    pub rules: Vec<ValidationRule>,
    /// Action on validation failure
    pub on_failure: ValidationAction,
}

/// Validation rule
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ValidationRule {
    /// Field path
    pub field: String,
    /// Rule type
    pub rule_type: ValidationRuleType,
    /// Rule parameters
    pub parameters: HashMap<String, Value>,
}

/// Validation rule type enumeration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ValidationRuleType {
    /// Required field
    Required,
    /// Type check
    Type(String),
    /// Range check
    Range { min: Option<f64>, max: Option<f64> },
    /// Pattern match
    Pattern(String),
    /// Custom validation
    Custom(String),
}

/// Validation action enumeration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ValidationAction {
    /// Reject invalid data
    Reject,
    /// Log warning and proceed
    Warn,
    /// Apply default values
    DefaultValues,
    /// Custom action
    Custom(String),
}

/// Cache configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CacheConfig {
    /// Enable caching
    pub enabled: bool,
    /// Cache size (number of entries)
    pub size: usize,
    /// Cache TTL
    pub ttl: Duration,
    /// Cache strategy
    pub strategy: CacheStrategy,
}

/// Cache strategy enumeration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum CacheStrategy {
    /// Least Recently Used
    LRU,
    /// First In, First Out
    FIFO,
    /// Time-based expiration
    TTL,
    /// Custom strategy
    Custom(String),
}

/// Error handling configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ErrorHandlingConfig {
    /// Retry on error
    pub retry_on_error: bool,
    /// Circuit breaker configuration
    pub circuit_breaker: Option<CircuitBreakerConfig>,
    /// Fallback data source
    pub fallback_source: Option<String>,
    /// Error notification
    pub notify_on_error: bool,
}

/// Circuit breaker configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CircuitBreakerConfig {
    /// Failure threshold
    pub failure_threshold: u32,
    /// Success threshold
    pub success_threshold: u32,
    /// Timeout duration
    pub timeout: Duration,
    /// Half-open retry timeout
    pub half_open_timeout: Duration,
}

/// Data update notification
#[derive(Debug, Clone)]
pub struct DataUpdate {
    /// Source ID
    pub source_id: String,
    /// Timestamp
    pub timestamp: Instant,
    /// Updated data
    pub data: Value,
    /// Change type
    pub change_type: ChangeType,
    /// Affected fields
    pub affected_fields: Vec<String>,
}

/// Change type enumeration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ChangeType {
    /// New data added
    Insert,
    /// Existing data updated
    Update,
    /// Data deleted
    Delete,
    /// Data replaced
    Replace,
    /// Full refresh
    Refresh,
}

/// Data source manager
pub struct DataSourceManager {
    /// Registered data sources
    sources: Arc<Mutex<HashMap<String, Box<dyn DataSource>>>>,
    /// Subscriptions
    subscriptions: Arc<Mutex<HashMap<String, Vec<Box<dyn Fn(DataUpdate) + Send + Sync>>>>>,
    /// Change detector
    change_detector: Arc<Mutex<ChangeDetector>>,
}

/// Change detector for data updates
#[derive(Debug)]
pub struct ChangeDetector {
    /// Change detection configuration
    config: ChangeDetectionConfig,
    /// Previous data states
    previous_states: HashMap<String, Value>,
    /// Change history
    history: VecDeque<DataUpdate>,
}

/// Change detection configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ChangeDetectionConfig {
    /// Enable change detection
    pub enabled: bool,
    /// Detection strategy
    pub strategy: ChangeDetectionStrategy,
    /// Comparison depth
    pub depth: u32,
    /// Ignore fields
    pub ignore_fields: Vec<String>,
    /// Notification configuration
    pub notification: ChangeNotificationConfig,
}

/// Change detection strategy enumeration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ChangeDetectionStrategy {
    /// Deep comparison
    Deep,
    /// Shallow comparison
    Shallow,
    /// Hash-based comparison
    Hash,
    /// Timestamp-based
    Timestamp,
    /// Custom strategy
    Custom(String),
}

/// Change notification configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ChangeNotificationConfig {
    /// Batch notifications
    pub batch_notifications: bool,
    /// Batch size
    pub batch_size: usize,
    /// Batch timeout
    pub batch_timeout: Duration,
    /// Notification filters
    pub filters: Vec<NotificationFilter>,
}

/// Notification filter
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NotificationFilter {
    /// Filter type
    pub filter_type: FilterType,
    /// Filter parameters
    pub parameters: HashMap<String, Value>,
}

/// Filter type enumeration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum FilterType {
    /// Field-based filter
    Field(String),
    /// Value-based filter
    Value(Value),
    /// Change type filter
    ChangeType(ChangeType),
    /// Custom filter
    Custom(String),
}

impl DataSourceManager {
    /// Create new data source manager
    pub fn new() -> Self {
        Self {
            sources: Arc::new(Mutex::new(HashMap::new())),
            subscriptions: Arc::new(Mutex::new(HashMap::new())),
            change_detector: Arc::new(Mutex::new(ChangeDetector::new())),
        }
    }

    /// Register data source
    pub fn register_source(&self, source: Box<dyn DataSource>) -> Result<()> {
        let id = source.id().to_string();
        self.sources
            .lock()
            .expect("Operation failed")
            .insert(id, source);
        Ok(())
    }

    /// Unregister data source
    pub fn unregister_source(&self, source_id: &str) -> Result<()> {
        self.sources
            .lock()
            .expect("Operation failed")
            .remove(source_id);
        self.subscriptions
            .lock()
            .expect("Operation failed")
            .remove(source_id);
        Ok(())
    }

    /// Get data source
    pub fn get_source(&self, source_id: &str) -> Option<String> {
        // Simplified - would return actual source reference
        self.sources
            .lock()
            .expect("Operation failed")
            .get(source_id)
            .map(|_| source_id.to_string())
    }

    /// Subscribe to data updates
    pub fn subscribe<F>(&self, source_id: &str, callback: F) -> Result<String>
    where
        F: Fn(DataUpdate) + Send + Sync + 'static,
    {
        let subscription_id = format!("{}_{}", source_id, scirs2_core::random::random::<u64>());
        self.subscriptions
            .lock()
            .expect("Operation failed")
            .entry(source_id.to_string())
            .or_default()
            .push(Box::new(callback));
        Ok(subscription_id)
    }
}

impl ChangeDetector {
    /// Create new change detector
    pub fn new() -> Self {
        Self {
            config: ChangeDetectionConfig::default(),
            previous_states: HashMap::new(),
            history: VecDeque::new(),
        }
    }

    /// Detect changes in data
    pub fn detect_changes(&mut self, source_id: &str, data: &Value) -> Vec<DataUpdate> {
        // Simplified change detection logic
        let mut updates = Vec::new();

        if let Some(previous) = self.previous_states.get(source_id) {
            if previous != data {
                updates.push(DataUpdate {
                    source_id: source_id.to_string(),
                    timestamp: Instant::now(),
                    data: data.clone(),
                    change_type: ChangeType::Update,
                    affected_fields: vec!["*".to_string()],
                });
            }
        } else {
            updates.push(DataUpdate {
                source_id: source_id.to_string(),
                timestamp: Instant::now(),
                data: data.clone(),
                change_type: ChangeType::Insert,
                affected_fields: vec!["*".to_string()],
            });
        }

        self.previous_states
            .insert(source_id.to_string(), data.clone());
        updates
    }
}

impl Default for ChangeDetectionConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            strategy: ChangeDetectionStrategy::Deep,
            depth: 10,
            ignore_fields: Vec::new(),
            notification: ChangeNotificationConfig::default(),
        }
    }
}

impl Default for ChangeNotificationConfig {
    fn default() -> Self {
        Self {
            batch_notifications: false,
            batch_size: 10,
            batch_timeout: Duration::from_millis(100),
            filters: Vec::new(),
        }
    }
}

impl Default for DataSourceConfig {
    fn default() -> Self {
        Self {
            id: "default".to_string(),
            source_type: DataSourceType::WebSocket,
            connection: ConnectionConfig::default(),
            format: DataFormatConfig::default(),
            cache: CacheConfig::default(),
            error_handling: ErrorHandlingConfig::default(),
        }
    }
}

impl Default for ConnectionConfig {
    fn default() -> Self {
        Self {
            url: "ws://localhost:8080".to_string(),
            headers: HashMap::new(),
            auth: None,
            retry: RetryConfig::default(),
            pooling: ConnectionPoolConfig::default(),
            timeout: Duration::from_secs(30),
        }
    }
}

impl Default for RetryConfig {
    fn default() -> Self {
        Self {
            max_attempts: 3,
            initial_delay: Duration::from_millis(100),
            max_delay: Duration::from_secs(30),
            backoff_multiplier: 2.0,
            jitter: true,
        }
    }
}

impl Default for ConnectionPoolConfig {
    fn default() -> Self {
        Self {
            enabled: false,
            max_size: 10,
            min_size: 1,
            connection_timeout: Duration::from_secs(10),
            idle_timeout: Duration::from_secs(300),
        }
    }
}

impl Default for DataFormatConfig {
    fn default() -> Self {
        Self {
            format: DataFormat::Json,
            schema: None,
            field_mappings: HashMap::new(),
            validation: ValidationConfig::default(),
        }
    }
}

impl Default for ValidationConfig {
    fn default() -> Self {
        Self {
            enabled: false,
            rules: Vec::new(),
            on_failure: ValidationAction::Warn,
        }
    }
}

impl Default for CacheConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            size: 1000,
            ttl: Duration::from_secs(300),
            strategy: CacheStrategy::LRU,
        }
    }
}

impl Default for ErrorHandlingConfig {
    fn default() -> Self {
        Self {
            retry_on_error: true,
            circuit_breaker: None,
            fallback_source: None,
            notify_on_error: true,
        }
    }
}