//! HTTP session management for MCP Streamable HTTP transport
//!
//! Manages HTTP-level sessions (separate from MCP protocol sessions) with:
//! - SSE broadcast channels for server→client messages
//! - Event buffering for resumability (Last-Event-ID support)
//! - Session lifecycle management

use std::collections::VecDeque;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Instant;

use dashmap::DashMap;
use parking_lot::RwLock;
use tokio::sync::broadcast;
use uuid::Uuid;

use super::types::{HttpTransportConfig, SseEvent};

/// HTTP Session state (not the MCP protocol session)
///
/// Each HTTP session has:
/// - Unique session ID (returned via Mcp-Session-Id header)
/// - SSE broadcast channel for server→client push
/// - Event buffer for resumability
/// - Activity tracking for cleanup
pub struct HttpSession {
    /// Session ID (Mcp-Session-Id header value)
    pub id: String,

    /// Created timestamp
    pub created_at: Instant,

    /// Last activity timestamp (atomic for lock-free updates)
    last_activity: Arc<AtomicU64>,

    /// SSE event broadcaster for server→client messages
    ///
    /// Multiple GET /mcp requests can subscribe to this channel
    pub sse_tx: broadcast::Sender<SseEvent>,

    /// Event ID counter for resumability (monotonically increasing)
    event_counter: AtomicU64,

    /// Buffered events for resumability (circular buffer)
    ///
    /// Stores recent events with their IDs for Last-Event-ID replay
    pub event_buffer: RwLock<VecDeque<(u64, SseEvent)>>,

    /// Maximum buffer size
    buffer_size: usize,

    /// Active SSE connection count
    active_connections: AtomicU64,
}

impl HttpSession {
    /// Create new HTTP session
    pub fn new(buffer_size: usize) -> Self {
        let (sse_tx, _) = broadcast::channel(1024);

        Self {
            id: Uuid::new_v4().to_string(),
            created_at: Instant::now(),
            last_activity: Arc::new(AtomicU64::new(Instant::now().elapsed().as_secs())),
            sse_tx,
            event_counter: AtomicU64::new(0),
            event_buffer: RwLock::new(VecDeque::with_capacity(buffer_size)),
            buffer_size,
            active_connections: AtomicU64::new(0),
        }
    }

    /// Update last activity timestamp
    pub fn touch(&self) {
        let elapsed = Instant::now().duration_since(self.created_at).as_secs();
        self.last_activity.store(elapsed, Ordering::Relaxed);
    }

    /// Get seconds since last activity
    pub fn seconds_since_activity(&self) -> u64 {
        let now = Instant::now().duration_since(self.created_at).as_secs();
        let last = self.last_activity.load(Ordering::Relaxed);
        now.saturating_sub(last)
    }

    /// Generate next event ID
    pub fn next_event_id(&self) -> String {
        let id = self.event_counter.fetch_add(1, Ordering::SeqCst);
        format!("{}-{}", self.id, id)
    }

    /// Parse event ID to extract sequence number
    fn parse_event_id(&self, event_id: &str) -> Option<u64> {
        // Format: "{session_id}-{seq}"
        let parts: Vec<&str> = event_id.split('-').collect();
        if parts.len() >= 6 {
            // UUID has 5 parts, seq is last
            parts.last()?.parse().ok()
        } else {
            None
        }
    }

    /// Buffer an event for resumability
    pub fn buffer_event(&self, seq: u64, event: SseEvent) {
        let mut buffer = self.event_buffer.write();

        // Add to buffer
        buffer.push_back((seq, event));

        // Maintain circular buffer size
        while buffer.len() > self.buffer_size {
            buffer.pop_front();
        }
    }

    /// Get events after a given event ID (for resumability)
    ///
    /// Returns events with sequence numbers > last_seq
    pub fn get_events_after(&self, last_event_id: &str) -> Vec<SseEvent> {
        let Some(last_seq) = self.parse_event_id(last_event_id) else {
            tracing::warn!("Invalid event ID format: {}", last_event_id);
            return vec![];
        };

        let buffer = self.event_buffer.read();

        buffer
            .iter()
            .filter(|(seq, _)| *seq > last_seq)
            .map(|(_, event)| event.clone())
            .collect()
    }

    /// Increment active connection count
    pub fn inc_connections(&self) -> u64 {
        self.active_connections.fetch_add(1, Ordering::SeqCst) + 1
    }

    /// Decrement active connection count
    pub fn dec_connections(&self) -> u64 {
        self.active_connections.fetch_sub(1, Ordering::SeqCst) - 1
    }

    /// Get active connection count
    pub fn connection_count(&self) -> u64 {
        self.active_connections.load(Ordering::Relaxed)
    }
}

/// Manages HTTP sessions across requests
///
/// Provides thread-safe session creation, lookup, and cleanup
pub struct HttpSessionManager {
    sessions: Arc<DashMap<String, Arc<HttpSession>>>,
    config: HttpTransportConfig,
}

impl HttpSessionManager {
    /// Create new session manager
    pub fn new(config: HttpTransportConfig) -> Self {
        Self {
            sessions: Arc::new(DashMap::new()),
            config,
        }
    }

    /// Create new session, returns session handle
    pub fn create_session(&self) -> Arc<HttpSession> {
        let session = Arc::new(HttpSession::new(self.config.event_buffer_size));
        self.sessions
            .insert(session.id.clone(), Arc::clone(&session));

        tracing::info!(session_id = %session.id, "Created new HTTP session");

        session
    }

    /// Get existing session by ID
    pub fn get_session(&self, id: &str) -> Option<Arc<HttpSession>> {
        self.sessions.get(id).map(|entry| Arc::clone(&entry))
    }

    /// Remove session (DELETE /mcp)
    pub fn remove_session(&self, id: &str) -> bool {
        if let Some((_, session)) = self.sessions.remove(id) {
            tracing::info!(session_id = id, "Removed HTTP session");

            // Ensure SSE channels are dropped to close streams
            drop(session);
            true
        } else {
            false
        }
    }

    /// Cleanup expired sessions (background task)
    ///
    /// Removes sessions that have been inactive for longer than the configured timeout
    pub async fn cleanup_expired(&self) {
        let timeout_secs = self.config.session_timeout.as_secs();
        let mut expired = vec![];

        for entry in self.sessions.iter() {
            let session = entry.value();
            if session.seconds_since_activity() > timeout_secs {
                expired.push(entry.key().clone());
            }
        }

        for id in expired {
            tracing::info!(
                session_id = %id,
                "Cleaning up expired HTTP session"
            );
            self.remove_session(&id);
        }
    }

    /// Get number of active sessions
    pub fn session_count(&self) -> usize {
        self.sessions.len()
    }
}

impl Clone for HttpSessionManager {
    fn clone(&self) -> Self {
        Self {
            sessions: Arc::clone(&self.sessions),
            config: self.config.clone(),
        }
    }
}

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

    #[test]
    fn test_session_creation() {
        let session = HttpSession::new(100);
        assert!(!session.id.is_empty());
        assert_eq!(session.buffer_size, 100);
        assert_eq!(session.connection_count(), 0);
    }

    #[test]
    fn test_event_id_generation() {
        let session = HttpSession::new(100);

        let id1 = session.next_event_id();
        let id2 = session.next_event_id();

        assert_ne!(id1, id2);
        assert!(id1.contains(&session.id));
        assert!(id2.contains(&session.id));
    }

    #[test]
    fn test_event_buffering() {
        let session = HttpSession::new(3);

        let event1 = SseEvent::new("data1".to_string());
        let event2 = SseEvent::new("data2".to_string());
        let event3 = SseEvent::new("data3".to_string());
        let event4 = SseEvent::new("data4".to_string());

        session.buffer_event(0, event1);
        session.buffer_event(1, event2);
        session.buffer_event(2, event3);
        session.buffer_event(3, event4); // Should evict event1

        let buffer = session.event_buffer.read();
        assert_eq!(buffer.len(), 3);
        assert_eq!(buffer[0].0, 1); // event1 evicted
        assert_eq!(buffer[2].0, 3);
    }

    #[test]
    fn test_get_events_after() {
        let session = HttpSession::new(10);

        session.buffer_event(
            0,
            SseEvent::new("data0".to_string()).with_id(format!("{}-0", session.id)),
        );
        session.buffer_event(
            1,
            SseEvent::new("data1".to_string()).with_id(format!("{}-1", session.id)),
        );
        session.buffer_event(
            2,
            SseEvent::new("data2".to_string()).with_id(format!("{}-2", session.id)),
        );

        let last_event_id = format!("{}-0", session.id);
        let events = session.get_events_after(&last_event_id);

        assert_eq!(events.len(), 2);
        assert_eq!(events[0].data, "data1");
        assert_eq!(events[1].data, "data2");
    }

    #[test]
    fn test_connection_counting() {
        let session = HttpSession::new(100);

        assert_eq!(session.inc_connections(), 1);
        assert_eq!(session.inc_connections(), 2);
        assert_eq!(session.connection_count(), 2);

        assert_eq!(session.dec_connections(), 1);
        assert_eq!(session.connection_count(), 1);
    }

    #[test]
    fn test_activity_tracking() {
        let session = HttpSession::new(100);

        std::thread::sleep(std::time::Duration::from_millis(100));

        let before_touch = session.seconds_since_activity();
        assert!(before_touch > 0);

        session.touch();

        let after_touch = session.seconds_since_activity();
        assert_eq!(after_touch, 0);
    }

    #[tokio::test]
    async fn test_manager_create_and_get() {
        let config = HttpTransportConfig::default();
        let manager = HttpSessionManager::new(config);

        let session = manager.create_session();
        let id = session.id.clone();

        let retrieved = manager.get_session(&id);
        assert!(retrieved.is_some());
        assert_eq!(retrieved.unwrap().id, id);
    }

    #[tokio::test]
    async fn test_manager_remove() {
        let config = HttpTransportConfig::default();
        let manager = HttpSessionManager::new(config);

        let session = manager.create_session();
        let id = session.id.clone();

        assert!(manager.remove_session(&id));
        assert!(manager.get_session(&id).is_none());
        assert!(!manager.remove_session(&id)); // Already removed
    }

    #[tokio::test]
    async fn test_manager_cleanup() {
        let mut config = HttpTransportConfig::default();
        config.session_timeout = std::time::Duration::from_millis(50);

        let manager = HttpSessionManager::new(config);

        let session = manager.create_session();
        let id = session.id.clone();

        // Wait for timeout
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;

        manager.cleanup_expired().await;

        assert!(manager.get_session(&id).is_none());
    }
}
