//! Event handlers for node and election changes.

use std::collections::HashSet;
use std::time::Duration;

use etcd_client::{Client, EventType, GetOptions, SortOrder, SortTarget};
use tracing::{debug, info, warn};

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

use super::ServiceDiscovery;

/// Default timeout for etcd operations.
const ETCD_OPERATION_TIMEOUT: Duration = Duration::from_secs(5);

/// Emits a cluster event to all subscribers.
///
/// Ignores send errors - they just mean no active subscribers.
pub(super) fn emit(discovery: &ServiceDiscovery, event: ClusterEvent) {
    let _ = discovery.inner.event_tx.send(event);
}

/// Emits events for actual differences between old and new node lists.
///
/// This is used during reconnection to emit proper events for changes
/// that occurred while disconnected.
pub(super) fn emit_diff_events(discovery: &ServiceDiscovery, old_nodes: &[Node], new_nodes: &[Node]) {
    let old_ids: HashSet<_> = old_nodes.iter().map(|n| &n.id).collect();
    let new_ids: HashSet<_> = new_nodes.iter().map(|n| &n.id).collect();

    // Nodes that left
    for node in old_nodes {
        if !new_ids.contains(&node.id) {
            info!("Node left (detected on reconnect): {}", node.id);
            emit(discovery, ClusterEvent::NodeLeft(node.clone()));
        }
    }

    // Nodes that joined
    for node in new_nodes {
        if !old_ids.contains(&node.id) {
            info!("Node joined (detected on reconnect): {}", node.id);
            emit(discovery, ClusterEvent::NodeJoined(node.clone()));
        }
    }

    // Nodes that were updated (IP change)
    for new_node in new_nodes {
        if let Some(old_node) = old_nodes.iter().find(|n| n.id == new_node.id) {
            if old_node.ip != new_node.ip {
                info!(
                    "Node updated (detected on reconnect): {} {} -> {}",
                    new_node.id, old_node.ip, new_node.ip
                );
                emit(
                    discovery,
                    ClusterEvent::NodeUpdated {
                        old: old_node.clone(),
                        new: new_node.clone(),
                    },
                );
            }
        }
    }
}

/// Handles node registry events from etcd watch.
pub(super) fn handle_node_events(
    discovery: &ServiceDiscovery,
    events: &[etcd_client::Event],
    prefix: &str,
) {
    for event in events {
        match event.event_type() {
            EventType::Put => {
                if let Some(kv) = event.kv() {
                    match serde_json::from_slice::<Node>(kv.value()) {
                        Ok(new_node) => {
                            let nodes = discovery.nodes();
                            let existing = nodes.iter().position(|n| n.id == new_node.id);

                            match existing {
                                Some(pos) => {
                                    let old_node = &nodes[pos];
                                    // Check if something meaningful changed (IP or metadata)
                                    let ip_changed = old_node.ip != new_node.ip;
                                    let metadata_changed = old_node.metadata != new_node.metadata;

                                    if ip_changed || metadata_changed {
                                        if ip_changed {
                                            info!(
                                                "Node {} updated: {} -> {}",
                                                new_node.id, old_node.ip, new_node.ip
                                            );
                                        } else {
                                            debug!("Node {} metadata updated", new_node.id);
                                        }
                                        emit(
                                            discovery,
                                            ClusterEvent::NodeUpdated {
                                                old: old_node.clone(),
                                                new: new_node.clone(),
                                            },
                                        );
                                    } else {
                                        debug!("Node {} heartbeat (no changes)", new_node.id);
                                    }
                                    // Update node in place (always update to refresh last_seen)
                                    let mut current = (*nodes).clone();
                                    current[pos] = new_node;
                                    discovery.inner.nodes_tx.send_replace(std::sync::Arc::new(current));
                                }
                                None => {
                                    info!("Node joined: {}", new_node.id);
                                    emit(discovery, ClusterEvent::NodeJoined(new_node.clone()));
                                    let mut current = (*nodes).clone();
                                    current.push(new_node);
                                    discovery.inner.nodes_tx.send_replace(std::sync::Arc::new(current));
                                }
                            }
                        }
                        Err(e) => {
                            let key = String::from_utf8_lossy(kv.key());
                            warn!("Malformed node at {}: {}", key, e);
                        }
                    }
                }
            }
            EventType::Delete => {
                if let Some(kv) = event.kv() {
                    let key = match kv.key_str() {
                        Ok(k) => k,
                        Err(e) => {
                            warn!("Invalid UTF-8 in key: {}", e);
                            continue;
                        }
                    };
                    let node_id = key.strip_prefix(prefix).unwrap_or(key);

                    let nodes = discovery.nodes();
                    if let Some(pos) = nodes.iter().position(|n| n.id == node_id) {
                        let removed = nodes[pos].clone();
                        info!("Node left: {}", removed.id);
                        emit(discovery, ClusterEvent::NodeLeft(removed.clone()));

                        let mut current = (*nodes).clone();
                        current.remove(pos);
                        discovery.inner.nodes_tx.send_replace(std::sync::Arc::new(current));

                        // If removed node was leader, clear leader
                        if let Some(leader) = discovery.leader() {
                            if leader.id == node_id {
                                discovery.inner.leader_tx.send_replace(None);
                                emit(discovery, ClusterEvent::LeaderLost);
                            }
                        }
                    }
                }
            }
        }
    }
}

/// Handles election events from etcd watch.
pub(super) async fn handle_election_events(
    discovery: &ServiceDiscovery,
    client: &mut Client,
    events: &[etcd_client::Event],
    election_key: &str,
) {
    for event in events {
        match event.event_type() {
            EventType::Put => {
                // Check event value first to avoid unnecessary GET
                // on leader heartbeats (proclaim). Only fetch from etcd if leader
                // actually changed.
                let event_leader_id = event
                    .kv()
                    .and_then(|kv| String::from_utf8(kv.value().to_vec()).ok());

                let Some(leader_id) = event_leader_id else {
                    warn!("Election event with invalid leader ID");
                    continue;
                };

                // Quick check: if this matches current leader, it's just a heartbeat
                let current_leader = discovery.leader();
                if current_leader.as_ref().map(|l| &l.id) == Some(&leader_id) {
                    debug!("Leader heartbeat (no change)");
                    continue;
                }

                // Leader may have changed. For correctness, we still fetch from etcd
                // to ensure we have the true leader (earliest create_revision wins).
                // This handles edge cases with multiple concurrent candidates.
                let resp = match tokio::time::timeout(ETCD_OPERATION_TIMEOUT, async {
                    client
                        .get(
                            election_key,
                            Some(
                                GetOptions::new()
                                    .with_prefix()
                                    .with_sort(SortTarget::Create, SortOrder::Ascend)
                                    .with_limit(1),
                            ),
                        )
                        .await
                })
                .await
                {
                    Ok(Ok(r)) => r,
                    Ok(Err(e)) => {
                        warn!("Failed to fetch leader: {}", e);
                        continue;
                    }
                    Err(_) => {
                        warn!("Timeout fetching leader");
                        continue;
                    }
                };

                if let Some(kv) = resp.kvs().first() {
                    if let Ok(confirmed_leader_id) = String::from_utf8(kv.value().to_vec()) {
                        // Double-check in case fetched leader differs from event
                        if current_leader.as_ref().map(|l| &l.id) == Some(&confirmed_leader_id) {
                            debug!("Leader confirmed unchanged after fetch");
                            continue;
                        }

                        // Find the node
                        let nodes = discovery.nodes();
                        if let Some(leader) =
                            nodes.iter().find(|n| n.id == confirmed_leader_id).cloned()
                        {
                            info!("New leader elected: {}", leader.id);
                            discovery.inner.leader_tx.send_replace(Some(leader.clone()));
                            emit(discovery, ClusterEvent::LeaderElected(leader));
                        }
                    }
                }
            }
            EventType::Delete => {
                if discovery.leader().is_some() {
                    info!("Leader lost");
                    discovery.inner.leader_tx.send_replace(None);
                    emit(discovery, ClusterEvent::LeaderLost);
                }
            }
        }
    }
}
