//! Multi-listener coordination for high-availability setup.
//!
//! Manages multiple listeners with shared checkpoint store,
//! providing leader election, health monitoring, and failover coordination.

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

use dashmap::DashMap;
use parking_lot::Mutex;

use super::state::{ListenerState, ListenerStateMachine};
use crate::error::{ObserverError, Result};

/// Health status of a listener
#[derive(Debug, Clone)]
pub struct ListenerHealth {
    /// Unique identifier for the listener
    pub listener_id:     String,
    /// Whether the listener is currently healthy
    pub is_healthy:      bool,
    /// Last processed checkpoint ID
    pub last_checkpoint: i64,
    /// Current state of the listener
    pub state:           ListenerState,
    /// Timestamp of last heartbeat
    pub last_heartbeat:  Instant,
}

/// Handle to a registered listener
pub struct ListenerHandle {
    /// Unique identifier for the listener
    pub listener_id:    String,
    /// State machine managing listener lifecycle
    pub state_machine:  ListenerStateMachine,
    /// Current checkpoint (last processed event ID)
    pub checkpoint:     Arc<AtomicU64>,
    /// Last heartbeat timestamp.
    ///
    /// Stored in a `parking_lot::Mutex` because the critical section is a single
    /// `Instant` read/write with no `.await` inside (see F019).
    pub last_heartbeat: Arc<Mutex<Instant>>,
}

/// Coordinates multiple listeners
/// This is a lightweight coordinator that manages listener state and checkpoints
/// without requiring a specific checkpoint store implementation.
#[derive(Clone)]
pub struct MultiListenerCoordinator {
    listeners: Arc<DashMap<String, Arc<ListenerHandle>>>,
    leader_id: Arc<tokio::sync::Mutex<Option<String>>>,
}

impl MultiListenerCoordinator {
    /// Create new coordinator
    #[must_use]
    pub fn new() -> Self {
        Self {
            listeners: Arc::new(DashMap::new()),
            leader_id: Arc::new(tokio::sync::Mutex::new(None)),
        }
    }

    /// Register a listener
    ///
    /// # Errors
    ///
    /// This function currently always returns `Ok`; the `Result` return type is
    /// reserved for future validation.
    #[allow(clippy::unused_async)] // Reason: trait/interface requires async signature
    pub async fn register_listener(&self, listener_id: String) -> Result<()> {
        let handle = Arc::new(ListenerHandle {
            listener_id:    listener_id.clone(),
            state_machine:  ListenerStateMachine::new(listener_id.clone()),
            checkpoint:     Arc::new(AtomicU64::new(0)),
            last_heartbeat: Arc::new(Mutex::new(Instant::now())),
        });

        self.listeners.insert(listener_id, handle);
        Ok(())
    }

    /// Deregister a listener
    ///
    /// # Errors
    ///
    /// This function currently always returns `Ok`; the `Result` return type is
    /// reserved for future validation.
    pub fn deregister_listener(&self, listener_id: &str) -> Result<()> {
        self.listeners.remove(listener_id);
        Ok(())
    }

    /// Get listener state
    ///
    /// # Errors
    ///
    /// Returns [`ObserverError::InvalidConfig`] if `listener_id` is not registered.
    pub async fn get_listener_state(&self, listener_id: &str) -> Result<ListenerState> {
        let handle = self.listeners.get(listener_id).ok_or(ObserverError::InvalidConfig {
            message: format!("Listener {listener_id} not found"),
        })?;

        Ok(handle.state_machine.get_state().await)
    }

    /// Update listener heartbeat
    ///
    /// # Errors
    ///
    /// Returns [`ObserverError::InvalidConfig`] if `listener_id` is not registered.
    ///
    /// Synchronous because the heartbeat update is a `parking_lot::Mutex<Instant>`
    /// write — no `.await` occurs inside the critical section (see F019).
    pub fn update_heartbeat(&self, listener_id: &str) -> Result<()> {
        let handle = self.listeners.get(listener_id).ok_or(ObserverError::InvalidConfig {
            message: format!("Listener {listener_id} not found"),
        })?;

        *handle.last_heartbeat.lock() = Instant::now();
        Ok(())
    }

    /// Update listener checkpoint.
    ///
    /// The checkpoint is stored as its raw bit pattern in an `AtomicU64` so that
    /// negative sentinel values (e.g. `-1` for "no checkpoint yet") round-trip
    /// correctly without silent truncation.
    ///
    /// # Errors
    ///
    /// Returns [`ObserverError::InvalidConfig`] if `listener_id` is not registered.
    pub fn update_checkpoint(&self, listener_id: &str, checkpoint: i64) -> Result<()> {
        let handle = self.listeners.get(listener_id).ok_or(ObserverError::InvalidConfig {
            message: format!("Listener {listener_id} not found"),
        })?;

        // `i64 as u64` in Rust is a two's-complement bit-preserving cast:
        // -1i64 becomes u64::MAX and round-trips correctly when read back as `as i64`.
        // This preserves negative sentinel values (e.g. -1 = "no checkpoint yet").
        #[allow(clippy::cast_sign_loss)] // Reason: intentional bit-preserving reinterpretation
        handle.checkpoint.store(checkpoint as u64, Ordering::SeqCst);
        Ok(())
    }

    /// Get health status of all listeners
    ///
    /// # Errors
    ///
    /// This function currently always returns `Ok`; the `Result` return type is
    /// reserved for future validation.
    pub async fn check_listener_health(&self) -> Result<Vec<ListenerHealth>> {
        let mut health_statuses = Vec::new();

        for entry in self.listeners.iter() {
            let handle = entry.value();
            let state = handle.state_machine.get_state().await;
            let last_heartbeat = *handle.last_heartbeat.lock();
            #[allow(clippy::cast_possible_wrap)]
            // Reason: value is non-negative; wrap cannot occur in practice
            let checkpoint = handle.checkpoint.load(Ordering::SeqCst) as i64;

            // Healthy if Running and heartbeat within 60s
            let is_healthy =
                state == ListenerState::Running && last_heartbeat.elapsed().as_secs() < 60;

            health_statuses.push(ListenerHealth {
                listener_id: handle.listener_id.clone(),
                is_healthy,
                last_checkpoint: checkpoint,
                state,
                last_heartbeat,
            });
        }

        Ok(health_statuses)
    }

    /// Elect leader among listeners
    ///
    /// # Errors
    ///
    /// Returns [`ObserverError::InvalidConfig`] if no healthy listener is available
    /// to become leader. Propagates errors from `check_listener_health`.
    pub async fn elect_leader(&self) -> Result<String> {
        let mut leader = self.leader_id.lock().await;

        // Check if current leader is still healthy
        if let Some(current_leader) = leader.as_ref() {
            if let Ok(state) = self.get_listener_state(current_leader).await {
                if state == ListenerState::Running {
                    return Ok(current_leader.clone());
                }
            }
        }

        // Find healthy listeners; sort by ID for deterministic election
        // (DashMap iteration order is unspecified).
        let health = self.check_listener_health().await?;
        let mut healthy: Vec<_> = health.iter().filter(|h| h.is_healthy).collect();
        healthy.sort_by(|a, b| a.listener_id.cmp(&b.listener_id));

        if healthy.is_empty() {
            return Err(ObserverError::InvalidConfig {
                message: "No healthy listeners available for election".to_string(),
            });
        }

        // Select lowest-ID healthy listener (deterministic across nodes).
        let new_leader = healthy[0].listener_id.clone();
        *leader = Some(new_leader.clone());

        Ok(new_leader)
    }

    /// Transition a listener to a new state.
    ///
    /// # Errors
    ///
    /// Returns an error if the listener is not registered or if the transition
    /// is not permitted by the listener's state machine.
    pub async fn transition_listener_state(
        &self,
        listener_id: &str,
        next_state: ListenerState,
    ) -> Result<()> {
        let handle = self.listeners.get(listener_id).ok_or(ObserverError::InvalidConfig {
            message: format!("Listener {listener_id} not found"),
        })?;
        handle.state_machine.transition(next_state).await
    }

    /// Get number of listeners
    #[must_use]
    pub fn listener_count(&self) -> usize {
        self.listeners.len()
    }

    /// Get current leader
    pub async fn get_leader(&self) -> Option<String> {
        self.leader_id.lock().await.clone()
    }
}

impl Default for MultiListenerCoordinator {
    fn default() -> Self {
        Self::new()
    }
}
