//! Cluster state monitoring with reactive event streams.
//!
//! The [`ServiceDiscovery`] watches etcd for cluster membership and leadership changes,
//! providing both pull-based state access and push-based event notifications.
//!
//! # Usage Patterns
//!
//! ## Event-driven (recommended for reacting to changes)
//! ```ignore
//! let mut events = discovery.subscribe();
//! while let Ok(event) = events.recv().await {
//!     match event {
//!         ClusterEvent::NodeJoined(n) => update_backends(),
//!         ClusterEvent::LeaderElected(n) => route_writes_to(n),
//!         _ => {}
//!     }
//! }
//! ```
//!
//! ## Watch-based (recommended for periodic state checks)
//! ```ignore
//! let mut watch = discovery.watch_nodes();
//! loop {
//!     watch.changed().await?;
//!     let nodes = watch.borrow();
//!     metrics.set(nodes.len());
//! }
//! ```
//!
//! ## Direct access (for one-off queries)
//! ```ignore
//! let nodes = discovery.nodes();
//! let leader = discovery.leader();
//! ```

mod handlers;
mod state;
mod watch_loop;

use std::sync::Arc;

use tokio::sync::{broadcast, watch};

use crate::error::ClusterError;
use crate::event::ClusterEvent;
use crate::types::Node;

use state::ServiceDiscoveryInner;

/// Discovers and monitors cluster nodes via etcd watches.
///
/// Provides three ways to observe cluster state:
/// - **Events** via [`subscribe()`](Self::subscribe): Push-based notifications for all changes
/// - **Watch** via [`watch_nodes()`](Self::watch_nodes): Efficient state observation
/// - **Direct** via [`nodes()`](Self::nodes): Immediate state access
#[derive(Clone)]
pub struct ServiceDiscovery {
    inner: Arc<ServiceDiscoveryInner>,
}

impl ServiceDiscovery {
    /// Creates a new ServiceDiscovery for the specified etcd cluster and group.
    ///
    /// # Arguments
    /// * `etcd_endpoints` - List of etcd endpoints (e.g., `["http://localhost:2379"]`)
    /// * `group_name` - Logical group name for node registration (e.g., `"workers"`)
    ///
    /// # Panics
    /// Panics if `etcd_endpoints` is empty or contains invalid URLs.
    ///
    /// # Example
    /// ```ignore
    /// let discovery = ServiceDiscovery::new(
    ///     vec!["http://localhost:2379".to_string()],
    ///     "my-service".to_string(),
    /// );
    /// ```
    pub fn new(etcd_endpoints: Vec<String>, group_name: String) -> Self {
        assert!(!etcd_endpoints.is_empty(), "etcd_endpoints cannot be empty");
        for endpoint in &etcd_endpoints {
            assert!(
                endpoint.starts_with("http://") || endpoint.starts_with("https://"),
                "Invalid etcd endpoint '{}': must start with http:// or https://",
                endpoint
            );
        }

        ServiceDiscovery {
            inner: Arc::new(ServiceDiscoveryInner::new(etcd_endpoints, group_name)),
        }
    }

    /// Returns the group name this discovery is watching.
    pub fn group_name(&self) -> &str {
        &self.inner.group_name
    }

    // =========================================================================
    // Reactive API - Event Subscription
    // =========================================================================

    /// Subscribe to cluster events.
    ///
    /// Returns a broadcast receiver that yields [`ClusterEvent`]s as they occur.
    /// Use this for event-driven architectures where you need to react to
    /// specific changes (node joins/leaves, leader elections).
    ///
    /// # Buffering
    ///
    /// Events are buffered up to 256 messages. If a receiver falls behind,
    /// it will receive a `RecvError::Lagged` error indicating missed events.
    ///
    /// # Example
    /// ```ignore
    /// let mut events = discovery.subscribe();
    /// loop {
    ///     match events.recv().await {
    ///         Ok(ClusterEvent::NodeJoined(n)) => {
    ///             println!("New node: {}", n.id);
    ///             update_load_balancer();
    ///         }
    ///         Ok(ClusterEvent::LeaderElected(n)) => {
    ///             println!("New leader: {}", n.id);
    ///         }
    ///         Err(broadcast::error::RecvError::Lagged(n)) => {
    ///             println!("Missed {} events, refreshing state", n);
    ///         }
    ///         Err(broadcast::error::RecvError::Closed) => break,
    ///         _ => {}
    ///     }
    /// }
    /// ```
    pub fn subscribe(&self) -> broadcast::Receiver<ClusterEvent> {
        self.inner.event_tx.subscribe()
    }

    // =========================================================================
    // Reactive API - Watch Channels
    // =========================================================================

    /// Watch for node list changes.
    ///
    /// Returns a watch receiver that always holds the latest node list.
    /// More efficient than polling - the receiver only wakes when state changes.
    ///
    /// # Example
    /// ```ignore
    /// let mut watch = discovery.watch_nodes();
    /// loop {
    ///     // Blocks until nodes change
    ///     watch.changed().await?;
    ///     let nodes = watch.borrow();
    ///     NODE_COUNT_METRIC.set(nodes.len() as i64);
    /// }
    /// ```
    pub fn watch_nodes(&self) -> watch::Receiver<Arc<Vec<Node>>> {
        self.inner.nodes_rx.clone()
    }

    /// Watch for leader changes.
    ///
    /// Returns a watch receiver that holds the current leader (if any).
    ///
    /// # Example
    /// ```ignore
    /// let mut watch = discovery.watch_leader();
    /// loop {
    ///     watch.changed().await?;
    ///     match &*watch.borrow() {
    ///         Some(leader) => println!("Leader: {}", leader.id),
    ///         None => println!("No leader"),
    ///     }
    /// }
    /// ```
    pub fn watch_leader(&self) -> watch::Receiver<Option<Node>> {
        self.inner.leader_rx.clone()
    }

    // =========================================================================
    // Direct State Access
    // =========================================================================

    /// Returns a snapshot of current cluster nodes.
    ///
    /// This is a cheap clone of an `Arc<Vec<Node>>`.
    /// For repeated access, consider using [`watch_nodes()`](Self::watch_nodes) instead.
    ///
    /// # Example
    /// ```ignore
    /// for node in discovery.nodes().iter() {
    ///     println!("Node {} at {}", node.id, node.ip);
    /// }
    /// ```
    #[inline]
    pub fn nodes(&self) -> Arc<Vec<Node>> {
        self.inner.nodes_rx.borrow().clone()
    }

    /// Returns the current leader, if any.
    ///
    /// # Example
    /// ```ignore
    /// if let Some(leader) = discovery.leader() {
    ///     route_write_request_to(leader.ip);
    /// }
    /// ```
    #[inline]
    pub fn leader(&self) -> Option<Node> {
        self.inner.leader_rx.borrow().clone()
    }

    /// Returns true if the discovery has completed initial sync with etcd.
    #[inline]
    pub fn is_ready(&self) -> bool {
        *self.inner.ready_rx.borrow()
    }

    // =========================================================================
    // Convenience Helpers
    // =========================================================================

    /// Wait until at least `min_count` nodes are registered.
    ///
    /// Useful for startup sequences where you need backends before serving traffic.
    ///
    /// # Example
    /// ```ignore
    /// // Wait for at least one backend before starting the load balancer
    /// discovery.wait_for_nodes(1).await;
    /// start_accepting_requests();
    /// ```
    pub async fn wait_for_nodes(&self, min_count: usize) {
        let mut watch = self.watch_nodes();
        loop {
            if watch.borrow().len() >= min_count {
                return;
            }
            if watch.changed().await.is_err() {
                return; // Channel closed
            }
        }
    }

    /// Wait until a leader is elected.
    ///
    /// Returns immediately if a leader already exists.
    ///
    /// # Example
    /// ```ignore
    /// let leader = discovery.wait_for_leader().await;
    /// println!("Leader elected: {}", leader.id);
    /// ```
    pub async fn wait_for_leader(&self) -> Option<Node> {
        let mut watch = self.watch_leader();
        loop {
            if let Some(leader) = watch.borrow().clone() {
                return Some(leader);
            }
            if watch.changed().await.is_err() {
                return None; // Channel closed
            }
        }
    }

    /// Wait until the discovery receives its first state sync from etcd.
    ///
    /// After this returns, [`nodes()`](Self::nodes) and [`leader()`](Self::leader)
    /// will return accurate data.
    ///
    /// # Example
    /// ```ignore
    /// discovery.wait_ready().await;
    /// // Now safe to query state
    /// ```
    pub async fn wait_ready(&self) {
        // Check current state first (handles case where Ready was already emitted)
        if *self.inner.ready_rx.borrow() {
            return;
        }

        // Wait for ready signal via watch channel
        let mut watch = self.inner.ready_rx.clone();
        loop {
            if *watch.borrow() {
                return;
            }
            if watch.changed().await.is_err() {
                return; // Channel closed
            }
        }
    }

    // =========================================================================
    // Main Run Loop
    // =========================================================================

    /// Starts the discovery's watch loop.
    ///
    /// This method runs indefinitely, watching etcd for changes and emitting
    /// events. It automatically reconnects on connection failures.
    ///
    /// # Arguments
    /// * `shutdown` - Optional shutdown signal. When received, the discovery will exit gracefully.
    ///
    /// # Returns
    /// Returns `Ok(())` on graceful shutdown, `Err` if unable to connect to etcd after retries.
    ///
    /// # Example
    /// ```ignore
    /// let discovery = ServiceDiscovery::new(endpoints, group);
    /// let (shutdown_tx, _) = broadcast::channel(1);
    /// let mut shutdown_rx = shutdown_tx.subscribe();
    /// tokio::spawn(async move { discovery.run(Some(&mut shutdown_rx)).await });
    /// ```
    pub async fn run(
        &self,
        shutdown: Option<&mut broadcast::Receiver<()>>,
    ) -> Result<(), ClusterError> {
        watch_loop::run(self, shutdown).await
    }

    // =========================================================================
    // Test Helpers (cfg(test) only)
    // =========================================================================

    #[cfg(test)]
    fn emit_event(&self, event: ClusterEvent) {
        // Check if Ready event before moving
        let is_ready = matches!(event, ClusterEvent::Ready);
        handlers::emit(self, event);
        // Also set ready state if Ready event
        if is_ready {
            self.inner.ready_tx.send_replace(true);
        }
    }

    #[cfg(test)]
    fn set_nodes(&self, nodes: Vec<Node>) {
        self.inner.nodes_tx.send_replace(Arc::new(nodes));
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::time::{Duration, Instant};

    #[test]
    fn test_nodes_method_performance() {
        let discovery = ServiceDiscovery::new(vec!["http://dummy:2379".into()], "test".into());

        // Populate with 1000 nodes
        let nodes: Vec<Node> = (0..1000)
            .map(|i| Node {
                id: format!("node-{}", i),
                ip: "127.0.0.1".parse().unwrap(),
                last_seen: 0,
                metadata: serde_json::Value::Null,
            })
            .collect();
        discovery.set_nodes(nodes);

        // Warm-up
        for _ in 0..1000 {
            let _ = discovery.nodes();
        }

        // Benchmark
        let start = Instant::now();
        for _ in 0..10_000_000 {
            let _ = discovery.nodes();
        }
        let duration = start.elapsed();

        println!(
            "10M nodes() calls in {:.2}s ({:.0} ns/call)",
            duration.as_secs_f64(),
            duration.as_nanos() as f64 / 10_000_000.0
        );

        assert!(duration.as_secs() < 2, "Performance regression");
    }

    #[tokio::test]
    async fn test_watch_nodes_returns_receiver() {
        let discovery = ServiceDiscovery::new(vec!["http://dummy:2379".into()], "test".into());
        let rx = discovery.watch_nodes();
        assert!(rx.borrow().is_empty());
    }

    /// Test for race condition in wait_ready()
    #[tokio::test]
    async fn test_wait_ready_race_condition() {
        let discovery = ServiceDiscovery::new(vec!["http://dummy:2379".into()], "test".into());

        // Simulate run() emitting Ready event BEFORE wait_ready() is called
        discovery.emit_event(ClusterEvent::Ready);

        // Small yield to ensure the event is fully processed
        tokio::task::yield_now().await;

        // Now call wait_ready() - with the fix, it checks the ready flag first
        let result = tokio::time::timeout(Duration::from_millis(100), discovery.wait_ready()).await;

        // This should pass now because wait_ready() checks the ready flag
        assert!(
            result.is_ok(),
            "wait_ready() should return immediately when Ready was already emitted"
        );
    }

    /// Test that watch channel preserves state properly
    #[tokio::test]
    async fn test_watch_preserves_state() {
        let discovery = ServiceDiscovery::new(vec!["http://dummy:2379".into()], "test".into());

        // Set up initial nodes
        let nodes = vec![
            Node {
                id: "node-1".into(),
                ip: "10.0.0.1".parse().unwrap(),
                last_seen: 1000,
                metadata: serde_json::Value::Null,
            },
            Node {
                id: "node-2".into(),
                ip: "10.0.0.2".parse().unwrap(),
                last_seen: 1000,
                metadata: serde_json::Value::Null,
            },
        ];
        discovery.set_nodes(nodes.clone());

        // Get a watch receiver
        let watch = discovery.watch_nodes();

        // Verify state
        let current = watch.borrow();
        assert_eq!(current.len(), 2, "Watch should see 2 nodes");
    }

    /// Test emit_diff_events correctly identifies changes
    #[tokio::test]
    async fn test_emit_diff_events() {
        let discovery = ServiceDiscovery::new(vec!["http://dummy:2379".into()], "test".into());

        let old_nodes = vec![
            Node {
                id: "node-1".into(),
                ip: "10.0.0.1".parse().unwrap(),
                last_seen: 1000,
                metadata: serde_json::Value::Null,
            },
            Node {
                id: "node-2".into(),
                ip: "10.0.0.2".parse().unwrap(),
                last_seen: 1000,
                metadata: serde_json::Value::Null,
            },
        ];

        let new_nodes = vec![
            Node {
                id: "node-1".into(),
                ip: "10.0.0.100".parse().unwrap(), // IP changed
                last_seen: 2000,
                metadata: serde_json::Value::Null,
            },
            // node-2 removed
            Node {
                id: "node-3".into(), // New node
                ip: "10.0.0.3".parse().unwrap(),
                last_seen: 2000,
                metadata: serde_json::Value::Null,
            },
        ];

        // Subscribe before emitting
        let mut events = discovery.subscribe();

        // Emit diff events
        handlers::emit_diff_events(&discovery, &old_nodes, &new_nodes);

        // Collect events
        let mut received = Vec::new();
        while let Ok(result) =
            tokio::time::timeout(Duration::from_millis(50), events.recv()).await
        {
            if let Ok(event) = result {
                received.push(event);
            }
        }

        // Should have: NodeLeft(node-2), NodeJoined(node-3), NodeUpdated(node-1)
        assert_eq!(received.len(), 3, "Expected 3 events, got {:?}", received);

        let has_left = received
            .iter()
            .any(|e| matches!(e, ClusterEvent::NodeLeft(n) if n.id == "node-2"));
        let has_joined = received
            .iter()
            .any(|e| matches!(e, ClusterEvent::NodeJoined(n) if n.id == "node-3"));
        let has_updated = received
            .iter()
            .any(|e| matches!(e, ClusterEvent::NodeUpdated { old, new } if old.id == "node-1" && new.ip.to_string() == "10.0.0.100"));

        assert!(has_left, "Should emit NodeLeft for node-2");
        assert!(has_joined, "Should emit NodeJoined for node-3");
        assert!(has_updated, "Should emit NodeUpdated for node-1");
    }

    /// Test endpoint validation
    #[test]
    #[should_panic(expected = "etcd_endpoints cannot be empty")]
    fn test_empty_endpoints_panics() {
        let _ = ServiceDiscovery::new(vec![], "test".into());
    }

    #[test]
    #[should_panic(expected = "Invalid etcd endpoint")]
    fn test_invalid_endpoint_panics() {
        let _ = ServiceDiscovery::new(vec!["localhost:2379".into()], "test".into());
    }
}
