//! Real-time chat system implementation

use serde::{Deserialize, Serialize};
use std::pin::Pin;
use std::time::Duration;
use tokio::sync::broadcast;
use tokio_stream::Stream;

use super::{
    connection::ConnectionManager, events::RealTimeEvent, streaming::LiveMessageStreamer,
    typing::{AggregateTypingStatistics, TypingIndicator, TypingSessionStatistics},
};

/// Real-time chat system configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RealtimeConfig {
    pub heartbeat_timeout: u64,
    pub health_check_interval: u64,
    pub max_message_size: usize,
    pub message_queue_limit: usize,
    pub backpressure_threshold: usize,
    pub processing_rate: u64,
}

impl Default for RealtimeConfig {
    fn default() -> Self {
        Self {
            heartbeat_timeout: 30,
            health_check_interval: 5,
            max_message_size: 1024 * 1024,
            message_queue_limit: 10_000,
            backpressure_threshold: 1_000,
            processing_rate: 100,
        }
    }
}

/// Real-time chat system with unified event aggregation
///
/// This struct aggregates events from three child components:
/// - `ConnectionManager`: connection status changes, heartbeats
/// - `TypingIndicator`: typing started/stopped events
/// - `LiveMessageStreamer`: message received/updated events
///
/// External code can subscribe to all events via `subscribe_to_events()`
/// which provides a unified event bus for all realtime chat activity.
pub struct RealtimeChat {
    connection_manager: ConnectionManager,
    message_streamer: LiveMessageStreamer,
    typing_indicator: TypingIndicator,
    config: RealtimeConfig,
    event_sender: broadcast::Sender<RealTimeEvent>,
    is_running: bool,
}

impl RealtimeChat {
    /// Create a new real-time chat system
    #[must_use]
    pub fn new(config: RealtimeConfig) -> Self {
        let (event_sender, _) = broadcast::channel(1000);

        Self {
            connection_manager: ConnectionManager::new(
                config.heartbeat_timeout,
                config.health_check_interval,
            ),
            message_streamer: LiveMessageStreamer::new(
                config.message_queue_limit,
                config.backpressure_threshold,
                config.processing_rate,
            ),
            typing_indicator: TypingIndicator::new(5, 60),
            config,
            event_sender,
            is_running: false,
        }
    }

    /// Start the real-time chat system
    pub fn start(
        &mut self,
    ) -> Pin<Box<dyn Stream<Item = crate::domain::context::chunks::CandleUnitChunk> + Send>> {
        if self.is_running {
            return Box::pin(crate::async_stream::spawn_stream(|tx| async move {
                // Already running - send success immediately
                let _ = tx.send(crate::domain::context::chunks::CandleUnitChunk::Success);
            }));
        }
        self.is_running = true;
        self.connection_manager.start_health_check();
        let _message_processing = self.message_streamer.start_processing();
        let _typing_cleanup = self.typing_indicator.start_cleanup_task();

        // Event Aggregation: Forward child component events to central event bus
        
        // Task 1: Forward connection manager events
        let event_sender_conn = self.event_sender.clone();
        let mut conn_events = self.connection_manager.subscribe();
        tokio::spawn(async move {
            while let Ok(event) = conn_events.recv().await {
                // Ignore send errors (no active receivers is normal)
                let _ = event_sender_conn.send(event);
            }
        });

        // Task 2: Forward typing indicator events
        let event_sender_typing = self.event_sender.clone();
        let mut typing_events = self.typing_indicator.subscribe();
        tokio::spawn(async move {
            while let Ok(event) = typing_events.recv().await {
                let _ = event_sender_typing.send(event);
            }
        });

        // Task 3: Forward message streamer events
        let event_sender_messages = self.event_sender.clone();
        let mut message_events = self.message_streamer.subscribe_to_events();
        tokio::spawn(async move {
            while let Ok(event) = message_events.recv().await {
                let _ = event_sender_messages.send(event);
            }
        });

        // Merge streams manually
        Box::pin(crate::async_stream::spawn_stream(move |tx| async move {
            // Start both streams concurrently
            // This is a simplified merge - in a full implementation you'd handle both streams properly
            // Send success to indicate startup completed
            let _ = tx.send(crate::domain::context::chunks::CandleUnitChunk::Success);
        }))
    }

    /// Get the current configuration
    pub fn get_config(&self) -> &RealtimeConfig {
        &self.config
    }

    /// Update configuration dynamically (selected fields)
    pub fn update_config(&mut self, new_config: RealtimeConfig) {
        self.config = new_config;
        // Update child components with new config
        self.message_streamer = LiveMessageStreamer::new(
            self.config.message_queue_limit,
            self.config.backpressure_threshold,
            self.config.processing_rate,
        );
    }

    /// Broadcast an event to all subscribers
    ///
    /// # Errors
    ///
    /// Returns `broadcast::error::SendError` if there are no active receivers
    pub fn broadcast_event(
        &self,
        event: RealTimeEvent,
    ) -> Result<usize, broadcast::error::SendError<RealTimeEvent>> {
        self.event_sender.send(event)
    }

    /// Create a receiver for real-time events
    pub fn subscribe_to_events(&self) -> broadcast::Receiver<RealTimeEvent> {
        self.event_sender.subscribe()
    }

    /// Get the number of active event subscribers
    pub fn get_event_subscriber_count(&self) -> usize {
        self.event_sender.receiver_count()
    }

    /// Get typing session statistics for a specific user
    ///
    /// Returns detailed statistics about a user's typing behavior including
    /// total typing duration, event count, and session information.
    pub fn get_user_typing_statistics(&self, user_id: &str, session_id: &str) -> Option<TypingSessionStatistics> {
        self.typing_indicator
            .typing_states
            .get(&format!("{user_id}:{session_id}"))
            .map(|entry| entry.value().get_session_statistics())
    }

    /// Get aggregate typing statistics across all active sessions
    pub fn get_aggregate_typing_statistics(&self) -> AggregateTypingStatistics {
        let mut total_users = 0;
        let mut total_typing_events = 0;
        let mut total_typing_duration_nanos = 0u64;
        
        for entry in self.typing_indicator.typing_states.iter() {
            let state = entry.value();
            total_users += 1;
            total_typing_events += state.event_count();
            total_typing_duration_nanos += state.total_typing_duration_nanos();
        }
        
        AggregateTypingStatistics {
            total_users,
            total_typing_events,
            total_typing_duration: Duration::from_nanos(total_typing_duration_nanos),
            indicator_stats: self.typing_indicator.get_statistics(),
        }
    }
}

/// Type alias for backwards compatibility
pub type RealTimeSystem = RealtimeChat;
