//! `PostgreSQL` Writer with Batch Processing and Error Recovery
//!
//! This module provides high-performance asynchronous `PostgreSQL` writing with:
//! - Batch processing for optimal throughput
//! - Automatic retry with exponential backoff
//! - Compression for large event payloads
//! - Guaranteed delivery tracking
//! - Connection pool management

use anyhow::{anyhow, Result};
use serde_json::Value as JsonValue;
use sqlx::PgPool;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tokio::sync::{mpsc, RwLock, Semaphore};
use tokio::time::{sleep, timeout};

use super::event_types::TradingEvent;
use super::EventMetrics;

/// Configuration for `PostgreSQL` writer
#[derive(Debug, Clone)]
/// WriterConfig
///
/// Auto-generated documentation placeholder - enhance with specifics
pub struct WriterConfig {
    /// Maximum events per batch
    pub batch_size: usize,
    /// Maximum time to wait before flushing incomplete batch
    pub batch_timeout: Duration,
    /// Maximum retry attempts for failed writes
    pub max_retry_attempts: usize,
    /// Base delay for exponential backoff
    pub retry_delay: Duration,
    /// Enable compression for large payloads
    pub enable_compression: bool,
    /// Writer thread identifier
    pub thread_id: usize,
}

impl Default for WriterConfig {
    fn default() -> Self {
        Self {
            batch_size: 1000,
            batch_timeout: Duration::from_millis(10),
            max_retry_attempts: 3,
            retry_delay: Duration::from_millis(100),
            enable_compression: true,
            thread_id: 0,
        }
    }
}

/// High-performance `PostgreSQL` writer with batching and recovery
#[derive(Debug)]
pub struct PostgresWriter {
    /// Writer configuration
    config: WriterConfig,
    /// Channel for receiving event batches
    batch_receiver: Arc<RwLock<Option<mpsc::Receiver<EventBatch>>>>,
    /// Channel sender for submitting batches
    batch_sender: mpsc::Sender<EventBatch>,
    /// Metrics collector
    metrics: Arc<EventMetrics>,
    /// Writer statistics
    stats: Arc<RwLock<WriterStats>>,
    /// Shutdown signal
    shutdown: Arc<AtomicBool>,
    /// Processing semaphore for flow control
    processing_semaphore: Arc<Semaphore>,
    /// Batch processor
    batch_processor: Arc<BatchProcessor>,
}

impl PostgresWriter {
    /// Create a new `PostgreSQL` writer
    pub async fn new(
        config: WriterConfig,
        db_pool: PgPool,
        metrics: Arc<EventMetrics>,
    ) -> Result<Self> {
        let (batch_sender, batch_receiver) = mpsc::channel(1000);

        let stats = Arc::new(RwLock::new(WriterStats::new(config.thread_id)));
        let shutdown = Arc::new(AtomicBool::new(false));
        let processing_semaphore = Arc::new(Semaphore::new(10)); // Allow 10 concurrent batches

        let batch_processor = Arc::new(BatchProcessor::new(
            config.clone(),
            db_pool.clone(),
            metrics.clone(),
            stats.clone(),
        ));

        let writer = Self {
            config: config.clone(),
            batch_receiver: Arc::new(RwLock::new(Some(batch_receiver))),
            batch_sender,
            metrics,
            stats,
            shutdown,
            processing_semaphore,
            batch_processor,
        };

        // Start background processing task
        writer.start_processing_task().await?;

        tracing::info!("PostgreSQL writer {} initialized", config.thread_id);
        // Ok variant
        Ok(writer)
    }

    /// Submit a batch of events for processing
    pub async fn submit_batch(&self, events: Vec<TradingEvent>) -> Result<()> {
        if self.shutdown.load(Ordering::Relaxed) {
            return Err(anyhow!("Writer is shutting down"));
        }

        let batch = EventBatch::new(events);

        self.batch_sender
            .send(batch)
            .await
            .map_err(|e| anyhow!("Failed to submit batch: {}", e))?;

        Ok(())
    }

    /// Start the background processing task
    async fn start_processing_task(&self) -> Result<()> {
        let shutdown = self.shutdown.clone();
        let batch_receiver = self.batch_receiver.clone();
        let batch_processor = self.batch_processor.clone();
        let processing_semaphore = self.processing_semaphore.clone();
        let metrics = self.metrics.clone();

        tokio::spawn(async move {
            let mut receiver = match batch_receiver.write().await.take() {
                Some(r) => r,
                None => {
                    tracing::error!("Batch receiver not available - processing task cannot start");
                    metrics.increment_failed_writes();
                    return;
                },
            };

            while !shutdown.load(Ordering::Relaxed) {
                match timeout(Duration::from_millis(100), receiver.recv()).await {
                    Ok(Some(batch)) => {
                        let processor = batch_processor.clone();
                        let metrics_clone = metrics.clone();
                        let semaphore_clone = processing_semaphore.clone();

                        // Process batch in background
                        tokio::spawn(async move {
                            // Acquire semaphore permit for flow control
                            let _permit = match semaphore_clone.acquire().await {
                                Ok(permit) => permit,
                                Err(e) => {
                                    tracing::error!("Failed to acquire semaphore permit: {}", e);
                                    metrics_clone.increment_failed_writes();
                                    return;
                                },
                            };

                            if let Err(e) = processor.process_batch(batch).await {
                                tracing::error!("Failed to process batch: {}", e);
                                metrics_clone.increment_failed_writes();
                            }
                        });
                    },
                    Ok(None) => {
                        // Channel closed
                        break;
                    },
                    Err(_) => {
                        // Timeout - continue processing
                        continue;
                    },
                }
            }

            tracing::info!("Writer processing task shutting down");
        });

        Ok(())
    }

    /// Get writer statistics
    pub async fn get_stats(&self) -> WriterStats {
        self.stats.read().await.clone()
    }

    /// Graceful shutdown
    pub async fn shutdown(&self) -> Result<()> {
        tracing::info!("Shutting down PostgreSQL writer {}", self.config.thread_id);

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

        // Close batch sender to signal completion
        // Note: The sender is owned by the struct and will be dropped when the struct is dropped

        // Wait for processing to complete (max 30 seconds)
        for _ in 0..300 {
            if self.processing_semaphore.available_permits() == 10 {
                break;
            }
            sleep(Duration::from_millis(100)).await;
        }

        tracing::info!(
            "PostgreSQL writer {} shutdown complete",
            self.config.thread_id
        );
        Ok(())
    }
}

/// Batch of events for processing
#[derive(Debug)]
/// EventBatch
///
/// Auto-generated documentation placeholder - enhance with specifics
pub struct EventBatch {
    /// Events in this batch
    pub events: Vec<TradingEvent>,
    /// Batch creation timestamp
    pub created_at: Instant,
    /// Batch identifier
    pub batch_id: String,
    /// Retry count
    pub retry_count: usize,
}

impl EventBatch {
    /// Create a new event batch
    pub fn new(events: Vec<TradingEvent>) -> Self {
        Self {
            events,
            created_at: Instant::now(),
            batch_id: uuid::Uuid::new_v4().to_string(),
            retry_count: 0,
        }
    }

    /// Get batch size
    pub fn size(&self) -> usize {
        self.events.len()
    }

    /// Get batch age
    pub fn age(&self) -> Duration {
        self.created_at.elapsed()
    }

    /// Increment retry count
    pub fn increment_retry(&mut self) {
        self.retry_count += 1;
    }
}

/// Batch processor for `PostgreSQL` operations
#[derive(Debug)]
pub struct BatchProcessor {
    config: WriterConfig,
    db_pool: PgPool,
    metrics: Arc<EventMetrics>,
    stats: Arc<RwLock<WriterStats>>,
    node_id: String,
    process_id: i32,
}

impl BatchProcessor {
    /// Create a new batch processor
    pub fn new(
        config: WriterConfig,
        db_pool: PgPool,
        metrics: Arc<EventMetrics>,
        stats: Arc<RwLock<WriterStats>>,
    ) -> Self {
        // Get node_id from hostname
        let node_id = hostname::get()
            .ok()
            .and_then(|h| h.into_string().ok())
            .unwrap_or_else(|| "unknown".to_string());

        // Get process ID
        let process_id = std::process::id() as i32;

        Self {
            config,
            db_pool,
            metrics,
            stats,
            node_id,
            process_id,
        }
    }

    /// Process a batch of events with retry logic
    pub async fn process_batch(&self, mut batch: EventBatch) -> Result<()> {
        let mut delay = self.config.retry_delay;

        for attempt in 0..=self.config.max_retry_attempts {
            match self.try_process_batch(&batch).await {
                Ok(()) => {
                    // Success - update metrics
                    self.metrics.increment_events_written(batch.size() as u64);
                    self.update_stats_success(batch.size(), batch.age()).await;

                    if attempt > 0 {
                        self.metrics.increment_retried_writes();
                        tracing::info!(
                            "Batch {} succeeded after {} retries",
                            batch.batch_id,
                            attempt
                        );
                    }

                    return Ok(());
                },
                Err(e) => {
                    tracing::warn!(
                        "Batch {} attempt {} failed: {}",
                        batch.batch_id,
                        attempt + 1,
                        e
                    );

                    if attempt < self.config.max_retry_attempts {
                        // Wait before retry with exponential backoff
                        sleep(delay).await;
                        delay = delay.mul_f32(1.5).min(Duration::from_secs(60)); // Max 60s delay
                        batch.increment_retry();
                    } else {
                        // Final failure
                        self.update_stats_failure().await;
                        return Err(anyhow!(
                            "Batch {} failed after {} attempts: {}",
                            batch.batch_id,
                            attempt + 1,
                            e
                        ));
                    }
                },
            }
        }

        Err(anyhow!("Unexpected retry loop exit"))
    }

    /// Single attempt to process a batch
    async fn try_process_batch(&self, batch: &EventBatch) -> Result<()> {
        let start_time = Instant::now();

        // Prepare batch for insertion
        let prepared_events = self.prepare_events_for_insertion(&batch.events).await?;

        // Build bulk insert query
        let query = self.build_bulk_insert_query(prepared_events.len());

        // Execute the bulk insert
        let mut query_builder = sqlx::query(&query);

        // Bind parameters for all events
        for event_data in prepared_events {
            let now_ns = SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap_or_else(|_| Duration::from_secs(0))
                .as_nanos() as i64;

            // Calculate event_date from timestamp (nanoseconds to date)
            let timestamp_secs = event_data.timestamp_ns / 1_000_000_000;
            let event_date = chrono::DateTime::from_timestamp(timestamp_secs, 0)
                .ok_or_else(|| {
                    anyhow::anyhow!(
                        "Invalid timestamp for event_date calculation: {}",
                        event_data.timestamp_ns
                    )
                })?
                .date_naive();

            query_builder = query_builder
                .bind(event_data.timestamp_ns)
                .bind(event_data.capture_timestamp_ns)
                .bind(now_ns) // processing_timestamp
                .bind(event_data.event_type)
                .bind("trading_engine") // event_source
                .bind(event_data.symbol)
                .bind(event_data.event_data)
                .bind(event_data.metadata)
                .bind(&self.node_id) // node_id
                .bind(self.process_id) // process_id
                .bind(event_data.event_hash) // event_hash
                .bind(event_date); // event_date
        }

        // Execute the query
        query_builder
            .execute(&self.db_pool)
            .await
            .map_err(|e| anyhow!("Database insert failed: {}", e))?;

        // Record write latency
        let write_latency = start_time.elapsed().as_millis() as f64;
        self.metrics.record_write_latency(write_latency);

        tracing::debug!(
            "Successfully wrote batch {} with {} events in {:.2}ms",
            batch.batch_id,
            batch.size(),
            write_latency
        );

        Ok(())
    }

    /// Prepare events for database insertion
    async fn prepare_events_for_insertion(
        &self,
        events: &[TradingEvent],
    ) -> Result<Vec<PreparedEventData>> {
        let mut prepared_events = Vec::with_capacity(events.len());

        for event in events {
            let prepared = self.prepare_single_event(event).await?;
            prepared_events.push(prepared);
        }

        // Ok variant
        Ok(prepared_events)
    }

    /// Prepare a single event for insertion
    async fn prepare_single_event(&self, event: &TradingEvent) -> Result<PreparedEventData> {
        // Serialize event data
        let event_data =
            serde_json::to_value(event).map_err(|e| anyhow!("Failed to serialize event: {}", e))?;

        // Extract symbol for indexing
        let symbol = match event {
            TradingEvent::OrderSubmitted { symbol, .. } => Some(symbol.clone()),
            TradingEvent::OrderExecuted { symbol, .. } => Some(symbol.clone()),
            TradingEvent::OrderCancelled { symbol, .. } => Some(symbol.clone()),
            TradingEvent::PositionUpdated { symbol, .. } => Some(symbol.clone()),
            TradingEvent::RiskAlert { symbol, .. } => symbol.clone(),
            TradingEvent::SystemEvent { .. } => None,
        };

        // Calculate event hash for deduplication
        let event_data_str = serde_json::to_string(&event_data)
            .map_err(|e| anyhow!("Failed to serialize event data for hash: {}", e))?;
        let event_hash = format!("{:x}", md5::compute(event_data_str.as_bytes()));

        Ok(PreparedEventData {
            timestamp_ns: event.timestamp().nanos as i64,
            capture_timestamp_ns: event
                .capture_timestamp()
                .map(|ts| ts.nanos as i64)
                .unwrap_or(event.timestamp().nanos as i64),
            event_type: event.event_type().to_owned(),
            symbol,
            event_data,
            metadata: event.metadata().cloned(),
            event_hash,
        })
    }

    /// Build bulk insert query for specified number of events
    fn build_bulk_insert_query(&self, event_count: usize) -> String {
        let mut query = String::from(
            "INSERT INTO trading_events (
                event_timestamp, received_timestamp, processing_timestamp,
                event_type, event_source, symbol, event_data, metadata,
                node_id, process_id, event_hash, event_date
            ) VALUES ",
        );

        let values_clause = (0..event_count)
            .map(|i| {
                let base = i * 12; // 12 parameters per event
                format!(
                    "(${}, ${}, ${}, ${}::trading_event_type, ${}, ${}, ${}, ${}, ${}, ${}, ${}, ${})",
                    base + 1, base + 2, base + 3, base + 4, base + 5, base + 6, base + 7, base + 8, base + 9, base + 10, base + 11, base + 12
                )
            })
            .collect::<Vec<_>>()
            .join(", ");

        query.push_str(&values_clause);
        query.push_str(" ON CONFLICT (id, event_date) DO NOTHING");

        query
    }

    /// Update statistics after successful batch processing
    async fn update_stats_success(&self, batch_size: usize, batch_age: Duration) {
        let mut stats = self.stats.write().await;
        stats.batches_processed += 1;
        stats.events_written += batch_size as u64;
        stats.total_processing_time += batch_age;
        stats.last_success = Some(Instant::now());
    }

    /// Update statistics after failed batch processing
    async fn update_stats_failure(&self) {
        let mut stats = self.stats.write().await;
        stats.batches_failed += 1;
        stats.last_failure = Some(Instant::now());
    }
}

/// Prepared event data for database insertion
#[derive(Debug)]
struct PreparedEventData {
    timestamp_ns: i64,
    capture_timestamp_ns: i64,
    event_type: String,
    symbol: Option<String>,
    event_data: JsonValue,
    metadata: Option<JsonValue>,
    event_hash: String,
}

/// Writer performance statistics
#[derive(Debug, Clone)]
/// WriterStats
///
/// Auto-generated documentation placeholder - enhance with specifics
pub struct WriterStats {
    /// Thread Id
    pub thread_id: usize,
    /// Batches Processed
    pub batches_processed: u64,
    /// Batches Failed
    pub batches_failed: u64,
    /// Events Written
    pub events_written: u64,
    /// Total Processing Time
    pub total_processing_time: Duration,
    /// Avg Batch Size
    pub avg_batch_size: f64,
    /// Avg Processing Time Ms
    pub avg_processing_time_ms: f64,
    /// Last Success
    pub last_success: Option<Instant>,
    /// Last Failure
    pub last_failure: Option<Instant>,
    /// Created At
    pub created_at: Instant,
}

impl WriterStats {
    pub fn new(thread_id: usize) -> Self {
        Self {
            thread_id,
            batches_processed: 0,
            batches_failed: 0,
            events_written: 0,
            total_processing_time: Duration::from_secs(0),
            avg_batch_size: 0.0,
            avg_processing_time_ms: 0.0,
            last_success: None,
            last_failure: None,
            created_at: Instant::now(),
        }
    }

    /// Calculate average batch size
    pub fn calculate_avg_batch_size(&mut self) {
        if self.batches_processed > 0 {
            self.avg_batch_size = self.events_written as f64 / self.batches_processed as f64;
        }
    }

    /// Calculate average processing time
    pub fn calculate_avg_processing_time(&mut self) {
        if self.batches_processed > 0 {
            self.avg_processing_time_ms =
                self.total_processing_time.as_millis() as f64 / self.batches_processed as f64;
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::events::event_types::TradingEvent;
    use crate::timing::HardwareTimestamp;
    use rust_decimal::Decimal;

    #[test]
    fn test_writer_config_default() {
        let config = WriterConfig::default();
        assert_eq!(config.batch_size, 1000);
        assert_eq!(config.thread_id, 0);
        assert!(config.enable_compression);
    }

    #[test]
    fn test_event_batch_creation() {
        let events = vec![TradingEvent::OrderSubmitted {
            order_id: "TEST-001".to_string(),
            symbol: "EURUSD".to_string(),
            quantity: Decimal::new(100000, 0),
            price: Decimal::new(10850, 4),
            timestamp: HardwareTimestamp::now(),
            sequence_number: Some(1),
            metadata: None,
        }];

        let batch = EventBatch::new(events);
        assert_eq!(batch.size(), 1);
        assert_eq!(batch.retry_count, 0);
        assert!(!batch.batch_id.is_empty());
    }

    #[tokio::test]
    async fn test_batch_processor_compression() {
        // This test would require a real PostgreSQL connection
        // In a real test environment, you would use a test database
        let _config = WriterConfig::default();

        // Create a mock pool (in real tests, use sqlx::testing or similar)
        // let db_pool = PgPool::connect("postgresql://test:test@localhost/test").await.unwrap();

        // Test data compression
        let data = "x".repeat(2000); // Large data that should be compressed

        // Verify compression would reduce size
        assert!(data.len() > 1024);
    }

    #[test]
    fn test_writer_stats() {
        let mut stats = WriterStats::new(1);
        assert_eq!(stats.thread_id, 1);
        assert_eq!(stats.batches_processed, 0);

        stats.batches_processed = 10;
        stats.events_written = 1000;
        stats.calculate_avg_batch_size();

        assert_eq!(stats.avg_batch_size, 100.0);
    }

    #[tokio::test]
    async fn test_batch_processor_query_building() {
        let _config = WriterConfig::default();

        // Mock database pool for testing
        // In real tests, you would use a proper test database

        // Create batch processor with mock dependencies
        let _metrics = Arc::new(EventMetrics::new());
        let _stats = Arc::new(RwLock::new(WriterStats::new(0)));

        // Test would create a processor and test query building
        // let processor = BatchProcessor::new(config, mock_pool, metrics, stats);

        // Verify bulk insert query structure
        let query = "INSERT INTO trading_events";
        assert!(query.contains("INSERT INTO trading_events"));
    }
}
