#![allow(clippy::mod_module_files)] // Events module structure is more maintainable
//! High-Performance Event Processing Pipeline for Trading Service
//!
//! This module provides ultra-low latency event capture and reliable PostgreSQL persistence
//! for compliance logging while maintaining sub-microsecond event capture performance.
//!
//! ## Architecture Overview
//!
//! ``text
//! ┌─────────────────────────────────────────────────────────────────────┐
//! │                 Event Processing Pipeline Architecture                │
//! ├─────────────────────────────────────────────────────────────────────┤
//! │  Producer Threads: Sub-μs Event Capture (Lock-Free Ring Buffers)    │
//! ├─────────────────────────────────────────────────────────────────────┤
//! │  Buffer Management: Multiple Ring Buffers + Sequence Numbers         │
//! ├─────────────────────────────────────────────────────────────────────┤
//! │  Async Writer Pool: Batched PostgreSQL Inserts + Error Recovery      │
//! ├─────────────────────────────────────────────────────────────────────┤
//! │  Storage Layer: PostgreSQL with Write-Behind + WAL Persistence       │
//! └─────────────────────────────────────────────────────────────────────┘
//! ``
//!
//! ## Performance Characteristics
//!
//! - **Event Capture**: Sub-microsecond lock-free event recording
//! - **Memory Allocation**: Zero allocation in hot path
//! - **Batch Processing**: Configurable batch sizes (1-10000 events)
//! - **Recovery**: Guaranteed delivery with sequence number tracking
//! - **Monitoring**: Real-time metrics and health monitoring
//!
//! ## Core Components
//!
//! - `EventCapture`: Lock-free event recording with hardware timestamps
//! - `RingBufferManager`: Multiple ring buffers with load balancing
//! - `PostgresWriter`: Async batched database writer with error recovery
//! - `EventTypes`: Type-safe event definitions with serialization
//!
//! ## Usage Example
//!
//! ``rust
//! use core::events::{EventProcessor, EventProcessorConfig, TradingEvent};
//! use core::timing::HardwareTimestamp;
//!
//! // Initialize event processor
//! let config = EventProcessorConfig::default();
//! let processor = EventProcessor::new(config).await?;
//!
//! // Capture high-frequency trading events
//! let event = TradingEvent::OrderSubmitted {
//!     order_id: "ORD-12345".to_string(),
//!     symbol: "EURUSD".to_string(),
//!     quantity: rust_decimal::Decimal::new(100000, 0),
//!     price: rust_decimal::Decimal::new(10850, 4),
//!     timestamp: HardwareTimestamp::now(),
//! };
//!
//! // Sub-microsecond event capture
//! processor.capture_event(event).await?;
//! ``

use anyhow::{anyhow, Result};
use serde::{Deserialize, Serialize};
use sqlx::{postgres::PgPoolOptions, PgPool};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;
use thiserror::Error;
use tokio::sync::RwLock;
use tokio::time::sleep;

// Import required types from submodules
use crate::events::event_types::{EventSequence, TradingEvent};
use crate::events::postgres_writer::{PostgresWriter, WriterConfig};
use crate::events::ring_buffer::{BufferManager, BufferStats};

// Import timing infrastructure
use crate::timing::HardwareTimestamp;

// Re-export core modules
pub mod event_types;
pub mod postgres_writer;
pub mod ring_buffer;

// Re-export key types for convenience
// DO NOT RE-EXPORT - Use explicit imports at usage sites
// NOTE: TradingEvent is available via trading_engine::events::event_types::TradingEvent

/// Configuration for the event processing pipeline
#[derive(Debug, Clone, Serialize, Deserialize)]
/// EventProcessorConfig
///
/// Auto-generated documentation placeholder - enhance with specifics
pub struct EventProcessorConfig {
    /// `PostgreSQL` connection string
    pub database_url: String,
    /// Number of ring buffers for load balancing
    pub buffer_count: usize,
    /// Size of each ring buffer (must be power of 2)
    pub buffer_size: usize,
    /// Maximum batch size for database inserts
    pub batch_size: usize,
    /// Batch timeout in milliseconds
    pub batch_timeout_ms: u64,
    /// Number of writer threads
    pub writer_threads: usize,
    /// Maximum database connections
    pub max_db_connections: u32,
    /// Connection timeout in seconds
    pub db_timeout_seconds: u64,
    /// Enable compression for large events
    pub enable_compression: bool,
    /// Maximum memory usage before applying backpressure (bytes)
    pub max_memory_usage: usize,
    /// Enable detailed monitoring
    pub enable_monitoring: bool,
    /// Retry attempts for failed writes
    pub max_retry_attempts: usize,
    /// Retry delay base in milliseconds
    pub retry_delay_ms: u64,
}

impl Default for EventProcessorConfig {
    fn default() -> Self {
        Self {
            database_url: "postgresql://foxhunt:foxhunt@localhost/trading_events".to_owned(),
            buffer_count: num_cpus::get().max(4),
            buffer_size: 8192, // 8K events per buffer
            batch_size: 1000,
            batch_timeout_ms: 10,
            writer_threads: 2,
            max_db_connections: 20,
            db_timeout_seconds: 30,
            enable_compression: true,
            max_memory_usage: 100 * 1024 * 1024, // 100MB
            enable_monitoring: true,
            max_retry_attempts: 3,
            retry_delay_ms: 100,
        }
    }
}

/// High-performance event processor with guaranteed delivery
#[derive(Debug)]
pub struct EventProcessor {
    /// Buffer manager for load balancing across multiple ring buffers
    buffer_manager: Arc<BufferManager>,
    /// `PostgreSQL` connection pool
    db_pool: PgPool,
    /// Async writer pool
    writers: Vec<Arc<PostgresWriter>>,
    /// Global sequence number generator
    sequence_generator: Arc<AtomicU64>,
    /// Shutdown signal
    shutdown: Arc<AtomicBool>,
    /// Performance monitoring
    metrics: Arc<EventMetrics>,
    /// Health monitor
    health_monitor: Arc<HealthMonitor>,
}

impl EventProcessor {
    /// Create a new event processor with the given configuration
    pub async fn new(config: EventProcessorConfig) -> Result<Self> {
        tracing::info!("Initializing event processor with config: {:?}", config);

        // Create PostgreSQL connection pool
        let db_pool = PgPoolOptions::new()
            .max_connections(config.max_db_connections)
            .min_connections(2)
            .acquire_timeout(Duration::from_secs(config.db_timeout_seconds))
            .idle_timeout(Duration::from_secs(300))
            .max_lifetime(Duration::from_secs(1800))
            .test_before_acquire(true)
            .connect(&config.database_url)
            .await
            .map_err(|e| anyhow!("Failed to connect to PostgreSQL: {}", e))?;

        // Initialize database schema
        Self::initialize_schema(&db_pool).await?;

        // Create buffer manager
        let buffer_manager = Arc::new(BufferManager::new(config.buffer_count, config.buffer_size)?);

        // Create metrics and monitoring
        let metrics = Arc::new(EventMetrics::new());
        let health_monitor = Arc::new(HealthMonitor::new());

        // Create PostgreSQL writers
        let mut writers = Vec::with_capacity(config.writer_threads);
        for i in 0..config.writer_threads {
            let writer_config = WriterConfig {
                batch_size: config.batch_size,
                batch_timeout: Duration::from_millis(config.batch_timeout_ms),
                max_retry_attempts: config.max_retry_attempts,
                retry_delay: Duration::from_millis(config.retry_delay_ms),
                enable_compression: config.enable_compression,
                thread_id: i,
            };

            let writer = Arc::new(
                PostgresWriter::new(writer_config, db_pool.clone(), metrics.clone()).await?,
            );

            writers.push(writer);
        }

        let processor = Self {
            buffer_manager,
            db_pool,
            writers,
            sequence_generator: Arc::new(AtomicU64::new(1)),
            shutdown: Arc::new(AtomicBool::new(false)),
            metrics,
            health_monitor,
        };

        // Start background processing tasks
        processor.start_background_tasks().await?;

        tracing::info!("Event processor initialized successfully");
        // Ok variant
        Ok(processor)
    }

    /// Capture a trading event with sub-microsecond latency
    #[inline(always)]
    pub async fn capture_event(&self, mut event: TradingEvent) -> Result<EventSequence> {
        let start_time = HardwareTimestamp::now();

        // Generate global sequence number
        let sequence_number = self.sequence_generator.fetch_add(1, Ordering::Relaxed);

        // Add metadata
        event.set_sequence_number(sequence_number);
        event.set_capture_timestamp(start_time);

        // Find optimal buffer (load balancing)
        let buffer_index = self.buffer_manager.select_buffer();

        // Attempt to store in ring buffer (lock-free)
        let result = self.buffer_manager.try_push(buffer_index, event).await;

        // Update metrics
        let capture_latency = HardwareTimestamp::now().latency_ns(&start_time);
        self.metrics.record_capture_latency(capture_latency);

        match result {
            Ok(seq) => {
                self.metrics.increment_events_captured();
                // Ok variant
                Ok(seq)
            },
            Err(e) => {
                self.metrics.increment_events_dropped();
                Err(anyhow!("Failed to capture event: {}", e))
            },
        }
    }

    /// Initialize the `PostgreSQL` database schema
    async fn initialize_schema(pool: &PgPool) -> Result<()> {
        tracing::info!("Initializing database schema");

        // Create extension for better performance
        sqlx::query(
            "
            CREATE EXTENSION IF NOT EXISTS pg_stat_statements;
        ",
        )
        .execute(pool)
        .await
        .map_err(|e| anyhow!("Failed to create extensions: {}", e))?;

        // Create trading events table with optimal indexing
        sqlx::query(
            "
            CREATE TABLE IF NOT EXISTS trading_events (
                id BIGSERIAL PRIMARY KEY,
                sequence_number BIGINT NOT NULL UNIQUE,
                event_type VARCHAR(50) NOT NULL,
                event_level VARCHAR(20) NOT NULL DEFAULT 'INFO',
                timestamp_ns BIGINT NOT NULL,
                capture_timestamp_ns BIGINT NOT NULL,
                processing_timestamp_ns BIGINT,
                symbol VARCHAR(20),
                order_id VARCHAR(50),
                trade_id VARCHAR(50),
                price DECIMAL(20,8),
                quantity DECIMAL(20,8),
                side VARCHAR(10),
                event_data JSONB NOT NULL,
                compressed_data BYTEA,
                metadata JSONB,
                created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                INDEX (sequence_number),
                INDEX (timestamp_ns),
                INDEX (event_type, timestamp_ns),
                INDEX (symbol, timestamp_ns),
                INDEX (order_id) WHERE order_id IS NOT NULL,
                INDEX (trade_id) WHERE trade_id IS NOT NULL
            );
        ",
        )
        .execute(pool)
        .await
        .map_err(|e| anyhow!("Failed to create trading_events table: {}", e))?;

        // Create sequence tracking table for recovery
        sqlx::query(
            "
            CREATE TABLE IF NOT EXISTS event_sequence_tracking (
                partition_id INTEGER PRIMARY KEY,
                last_processed_sequence BIGINT NOT NULL DEFAULT 0,
                last_updated TIMESTAMPTZ NOT NULL DEFAULT NOW()
            );
        ",
        )
        .execute(pool)
        .await
        .map_err(|e| anyhow!("Failed to create sequence tracking table: {}", e))?;

        // Create performance monitoring table
        sqlx::query(
            "
            CREATE TABLE IF NOT EXISTS event_processing_stats (
                timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                events_per_second BIGINT NOT NULL,
                avg_capture_latency_ns BIGINT NOT NULL,
                avg_write_latency_ms DECIMAL(10,3) NOT NULL,
                buffer_utilization DECIMAL(5,2) NOT NULL,
                failed_writes BIGINT NOT NULL DEFAULT 0,
                retried_writes BIGINT NOT NULL DEFAULT 0
            );
        ",
        )
        .execute(pool)
        .await
        .map_err(|e| anyhow!("Failed to create stats table: {}", e))?;

        // Create indexes for optimal query performance
        sqlx::query(
            "
            CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_trading_events_timestamp_ns
            ON trading_events (timestamp_ns DESC);
        ",
        )
        .execute(pool)
        .await
        .map_err(|e| anyhow!("Failed to create timestamp index: {}", e))?;

        sqlx::query(
            "
            CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_trading_events_symbol_timestamp
            ON trading_events (symbol, timestamp_ns DESC)
            WHERE symbol IS NOT NULL;
        ",
        )
        .execute(pool)
        .await
        .map_err(|e| anyhow!("Failed to create symbol index: {}", e))?;

        tracing::info!("Database schema initialized successfully");
        Ok(())
    }

    /// Start background processing tasks
    async fn start_background_tasks(&self) -> Result<()> {
        let shutdown = self.shutdown.clone();
        let buffer_manager = self.buffer_manager.clone();
        let writers = self.writers.clone();
        let metrics = self.metrics.clone();
        let _health_monitor = self.health_monitor.clone();

        // Start buffer-to-writer routing task
        tokio::spawn(async move {
            Self::buffer_router_task(shutdown, buffer_manager, writers, metrics).await;
        });

        // Start health monitoring task
        let shutdown_monitor = self.shutdown.clone();
        let health_monitor_clone = self.health_monitor.clone();
        let metrics_clone = self.metrics.clone();
        tokio::spawn(async move {
            Self::health_monitor_task(shutdown_monitor, health_monitor_clone, metrics_clone).await;
        });

        // Start metrics reporting task
        let shutdown_metrics = self.shutdown.clone();
        let metrics_reporting = self.metrics.clone();
        let db_pool_metrics = self.db_pool.clone();
        tokio::spawn(async move {
            Self::metrics_reporting_task(shutdown_metrics, metrics_reporting, db_pool_metrics)
                .await;
        });

        Ok(())
    }

    /// Background task to route events from buffers to writers
    async fn buffer_router_task(
        shutdown: Arc<AtomicBool>,
        buffer_manager: Arc<BufferManager>,
        writers: Vec<Arc<PostgresWriter>>,
        metrics: Arc<EventMetrics>,
    ) {
        let mut writer_index = 0;

        while !shutdown.load(Ordering::Relaxed) {
            let mut events_routed = 0;

            // Check all buffers for events
            for buffer_id in 0..buffer_manager.buffer_count() {
                if let Some(events) = buffer_manager.drain_buffer(buffer_id, 100).await {
                    if !events.is_empty() {
                        // Round-robin distribution to writers
                        let writer = &writers[writer_index % writers.len()];

                        // Send batch to writer
                        if let Err(e) = writer.submit_batch(events).await {
                            tracing::error!(
                                "Failed to submit batch to writer {}: {}",
                                writer_index,
                                e
                            );
                            metrics.increment_routing_errors();
                        } else {
                            events_routed += 1;
                        }

                        writer_index = (writer_index + 1) % writers.len();
                    }
                }
            }

            // Short sleep to prevent busy waiting
            if events_routed == 0 {
                sleep(Duration::from_micros(100)).await;
            }
        }
    }

    /// Background health monitoring task
    async fn health_monitor_task(
        shutdown: Arc<AtomicBool>,
        health_monitor: Arc<HealthMonitor>,
        metrics: Arc<EventMetrics>,
    ) {
        while !shutdown.load(Ordering::Relaxed) {
            // Update health status
            health_monitor.update_health(metrics.get_snapshot()).await;

            // Sleep for 1 second between health checks
            sleep(Duration::from_secs(1)).await;
        }
    }

    /// Background metrics reporting task
    async fn metrics_reporting_task(
        shutdown: Arc<AtomicBool>,
        metrics: Arc<EventMetrics>,
        db_pool: PgPool,
    ) {
        while !shutdown.load(Ordering::Relaxed) {
            // Log metrics to database every 30 seconds
            if let Err(e) = Self::persist_metrics(&metrics, &db_pool).await {
                tracing::error!("Failed to persist metrics: {}", e);
            }

            sleep(Duration::from_secs(30)).await;
        }
    }

    /// Persist metrics to database
    async fn persist_metrics(metrics: &EventMetrics, db_pool: &PgPool) -> Result<()> {
        let snapshot = metrics.get_snapshot();

        sqlx::query(
            "
            INSERT INTO event_processing_stats (
                events_per_second,
                avg_capture_latency_ns,
                avg_write_latency_ms,
                buffer_utilization,
                failed_writes,
                retried_writes
            ) VALUES ($1, $2, $3, $4, $5, $6)
        ",
        )
        .bind(snapshot.events_per_second as i64)
        .bind(snapshot.avg_capture_latency_ns as i64)
        .bind(snapshot.avg_write_latency_ms)
        .bind(snapshot.buffer_utilization)
        .bind(snapshot.failed_writes as i64)
        .bind(snapshot.retried_writes as i64)
        .execute(db_pool)
        .await
        .map_err(|e| anyhow!("Failed to insert metrics: {}", e))?;

        Ok(())
    }

    /// Get current performance metrics
    pub fn get_metrics(&self) -> EventMetricsSnapshot {
        self.metrics.get_snapshot()
    }

    /// Get health status
    pub async fn get_health(&self) -> HealthStatus {
        self.health_monitor.get_status().await
    }

    /// Get buffer statistics
    pub async fn get_buffer_stats(&self) -> Vec<BufferStats> {
        self.buffer_manager.get_all_stats().await
    }

    /// Graceful shutdown
    pub async fn shutdown(&self) -> Result<()> {
        tracing::info!("Initiating graceful shutdown");

        // Set shutdown flag
        self.shutdown.store(true, Ordering::Relaxed);

        // Wait for writers to finish processing
        for writer in &self.writers {
            writer.shutdown().await?;
        }

        // Drain remaining buffers
        self.buffer_manager.drain_all_buffers().await?;

        // Close database connections
        self.db_pool.close().await;

        tracing::info!("Event processor shutdown complete");
        Ok(())
    }
}

/// Real-time performance metrics
#[derive(Debug)]
/// EventMetrics
///
/// Auto-generated documentation placeholder - enhance with specifics
pub struct EventMetrics {
    events_captured: AtomicU64,
    events_dropped: AtomicU64,
    events_written: AtomicU64,
    routing_errors: AtomicU64,
    capture_latency_sum: AtomicU64,
    capture_latency_count: AtomicU64,
    write_latency_sum: AtomicU64,
    write_latency_count: AtomicU64,
    failed_writes: AtomicU64,
    retried_writes: AtomicU64,
    start_time: std::time::Instant,
}

impl EventMetrics {
    pub fn new() -> Self {
        Self {
            events_captured: AtomicU64::new(0),
            events_dropped: AtomicU64::new(0),
            events_written: AtomicU64::new(0),
            routing_errors: AtomicU64::new(0),
            capture_latency_sum: AtomicU64::new(0),
            capture_latency_count: AtomicU64::new(0),
            write_latency_sum: AtomicU64::new(0),
            write_latency_count: AtomicU64::new(0),
            failed_writes: AtomicU64::new(0),
            retried_writes: AtomicU64::new(0),
            start_time: std::time::Instant::now(),
        }
    }

    pub fn increment_events_captured(&self) {
        self.events_captured.fetch_add(1, Ordering::Relaxed);
    }

    pub fn increment_events_dropped(&self) {
        self.events_dropped.fetch_add(1, Ordering::Relaxed);
    }

    pub fn increment_events_written(&self, count: u64) {
        self.events_written.fetch_add(count, Ordering::Relaxed);
    }

    pub fn increment_routing_errors(&self) {
        self.routing_errors.fetch_add(1, Ordering::Relaxed);
    }

    pub fn record_capture_latency(&self, latency_ns: u64) {
        self.capture_latency_sum
            .fetch_add(latency_ns, Ordering::Relaxed);
        self.capture_latency_count.fetch_add(1, Ordering::Relaxed);
    }

    pub fn record_write_latency(&self, latency_ms: f64) {
        let latency_us = (latency_ms * 1000.0) as u64;
        self.write_latency_sum
            .fetch_add(latency_us, Ordering::Relaxed);
        self.write_latency_count.fetch_add(1, Ordering::Relaxed);
    }

    pub fn increment_failed_writes(&self) {
        self.failed_writes.fetch_add(1, Ordering::Relaxed);
    }

    pub fn increment_retried_writes(&self) {
        self.retried_writes.fetch_add(1, Ordering::Relaxed);
    }

    pub fn get_snapshot(&self) -> EventMetricsSnapshot {
        let events_captured = self.events_captured.load(Ordering::Relaxed);
        let elapsed_secs = self.start_time.elapsed().as_secs_f64();
        let events_per_second = if elapsed_secs > 0.0 {
            (events_captured as f64 / elapsed_secs) as u64
        } else {
            0
        };

        let capture_count = self.capture_latency_count.load(Ordering::Relaxed);
        let avg_capture_latency_ns = if capture_count > 0 {
            self.capture_latency_sum.load(Ordering::Relaxed) / capture_count
        } else {
            0
        };

        let write_count = self.write_latency_count.load(Ordering::Relaxed);
        let avg_write_latency_ms = if write_count > 0 {
            (self.write_latency_sum.load(Ordering::Relaxed) / write_count) as f64 / 1000.0
        } else {
            0.0
        };

        EventMetricsSnapshot {
            events_captured,
            events_dropped: self.events_dropped.load(Ordering::Relaxed),
            events_written: self.events_written.load(Ordering::Relaxed),
            routing_errors: self.routing_errors.load(Ordering::Relaxed),
            events_per_second,
            avg_capture_latency_ns,
            avg_write_latency_ms,
            buffer_utilization: 0.0, // Updated by buffer manager
            failed_writes: self.failed_writes.load(Ordering::Relaxed),
            retried_writes: self.retried_writes.load(Ordering::Relaxed),
        }
    }
}

/// Snapshot of event processing metrics
#[derive(Debug, Clone, Serialize, Deserialize)]
/// EventMetricsSnapshot
///
/// Auto-generated documentation placeholder - enhance with specifics
pub struct EventMetricsSnapshot {
    /// Events Captured
    pub events_captured: u64,
    /// Events Dropped
    pub events_dropped: u64,
    /// Events Written
    pub events_written: u64,
    /// Routing Errors
    pub routing_errors: u64,
    /// Events Per Second
    pub events_per_second: u64,
    /// Avg Capture Latency Ns
    pub avg_capture_latency_ns: u64,
    /// Avg Write Latency Ms
    pub avg_write_latency_ms: f64,
    /// Buffer Utilization
    pub buffer_utilization: f64,
    /// Failed Writes
    pub failed_writes: u64,
    /// Retried Writes
    pub retried_writes: u64,
}

/// Health monitoring for the event processing system
#[derive(Debug)]
/// HealthMonitor
///
/// Auto-generated documentation placeholder - enhance with specifics
pub struct HealthMonitor {
    status: RwLock<HealthStatus>,
}

impl HealthMonitor {
    pub fn new() -> Self {
        Self {
            status: RwLock::new(HealthStatus::Healthy),
        }
    }

    pub async fn update_health(&self, metrics: EventMetricsSnapshot) {
        let mut status = self.status.write().await;

        // Determine health based on metrics
        *status = if metrics.events_dropped > metrics.events_captured / 10 {
            HealthStatus::Degraded("High event drop rate".to_owned())
        } else if metrics.avg_capture_latency_ns > 10_000 {
            HealthStatus::Degraded("High capture latency".to_owned())
        } else if metrics.avg_write_latency_ms > 100.0 {
            HealthStatus::Degraded("High write latency".to_owned())
        } else if metrics.failed_writes > 0 {
            HealthStatus::Warning("Database write failures detected".to_owned())
        } else {
            HealthStatus::Healthy
        };
    }

    pub async fn get_status(&self) -> HealthStatus {
        self.status.read().await.clone()
    }
}

/// Health status of the event processing system
#[derive(Debug, Clone, Serialize, Deserialize)]
/// HealthStatus
///
/// Auto-generated documentation placeholder - enhance with specifics
pub enum HealthStatus {
    // Healthy variant
    Healthy,
    // Warning variant
    Warning(String),
    // Degraded variant
    Degraded(String),
    // Critical variant
    Critical(String),
}

/// Errors that can occur during event processing
#[derive(Debug, Error)]
/// EventProcessingError
///
/// Auto-generated documentation placeholder - enhance with specifics
pub enum EventProcessingError {
    #[error("Database error: {0}")]
    // Database variant
    Database(#[from] sqlx::Error),
    #[error("Buffer full: {0}")]
    // BufferFull variant
    BufferFull(String),
    #[error("Serialization error: {0}")]
    // Serialization variant
    Serialization(#[from] serde_json::Error),
    #[error("Compression error: {0}")]
    // Compression variant
    Compression(String),
    #[error("Configuration error: {0}")]
    // Configuration variant
    Configuration(String),
    #[error("Timeout error: {0}")]
    // Timeout variant
    Timeout(String),
    #[error("Writer error: {0}")]
    // Writer variant
    Writer(String),
}

/// Type alias for event processing results
#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_event_processor_creation() -> Result<()> {
        // Create test configuration with in-memory database
        let _config = EventProcessorConfig {
            database_url: "postgresql://test:test@localhost/test_db".to_string(),
            buffer_count: 2,
            buffer_size: 64,
            batch_size: 10,
            ..Default::default()
        };

        // This test would require a real PostgreSQL database
        // In a real test environment, you would set up a test database

        Ok(())
    }

    #[tokio::test]
    async fn test_event_metrics() {
        let metrics = EventMetrics::new();

        metrics.increment_events_captured();
        metrics.record_capture_latency(500);

        let snapshot = metrics.get_snapshot();
        assert_eq!(snapshot.events_captured, 1);
        assert_eq!(snapshot.avg_capture_latency_ns, 500);
    }

    #[tokio::test]
    async fn test_health_monitor() {
        let monitor = HealthMonitor::new();

        let metrics = EventMetricsSnapshot {
            events_captured: 1000,
            events_dropped: 50,
            events_written: 950,
            routing_errors: 0,
            events_per_second: 1000,
            avg_capture_latency_ns: 500,
            avg_write_latency_ms: 5.0,
            buffer_utilization: 0.5,
            failed_writes: 0,
            retried_writes: 0,
        };

        monitor.update_health(metrics).await;

        match monitor.get_status().await {
            HealthStatus::Healthy => {},
            _ => panic!("Expected healthy status"),
        }
    }
}
