use crate::error::Result;
use crate::node::{Inspectable, NodeId, NodeKind, NodeMetadata, NodeState};
use crate::tree::NodeTree;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{broadcast, RwLock};

/// Events emitted by the registry when the tree changes.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub enum InspectorEvent {
    NodeRegistered {
        id: NodeId,
        parent: NodeId,
        kind: NodeKind,
    },
    NodeRemoved {
        id: NodeId,
    },
    StateChanged {
        id: NodeId,
        old: NodeState,
        new: NodeState,
    },
    PropertiesUpdated {
        id: NodeId,
        keys: Vec<String>,
    },
}

/// Central registry for the inspector tree.
///
/// Thread-safe: all operations acquire the internal `RwLock` as needed.
/// Emits `InspectorEvent`s on a broadcast channel for reactive consumers.
pub struct InspectorRegistry {
    tree: Arc<RwLock<NodeTree>>,
    event_tx: broadcast::Sender<InspectorEvent>,
}

impl InspectorRegistry {
    /// Create a new registry with a root node.
    ///
    /// The root node ID is `"dirigent"` by default, with `NodeKind::Root`.
    pub fn new() -> Self {
        let root_id = NodeId::new("dirigent");
        let root_meta =
            NodeMetadata::new(NodeKind::Root, "Dirigent").with_state(NodeState::Running);
        let tree = NodeTree::new(root_id, root_meta);
        let (event_tx, _) = broadcast::channel(500);

        Self {
            tree: Arc::new(RwLock::new(tree)),
            event_tx,
        }
    }

    /// Create a registry with a custom root node.
    pub fn with_root(root_id: NodeId, root_metadata: NodeMetadata) -> Self {
        let tree = NodeTree::new(root_id, root_metadata);
        let (event_tx, _) = broadcast::channel(500);

        Self {
            tree: Arc::new(RwLock::new(tree)),
            event_tx,
        }
    }

    /// Get the root node ID.
    pub async fn root_id(&self) -> NodeId {
        let tree = self.tree.read().await;
        tree.root_id().clone()
    }

    /// Register a new node under the given parent.
    ///
    /// Returns a `NodeHandle` that the producer can use to update this node.
    pub async fn register(
        self: &Arc<Self>,
        id: NodeId,
        parent: &NodeId,
        metadata: NodeMetadata,
        inspectable: Option<Arc<dyn Inspectable>>,
    ) -> Result<crate::handle::NodeHandle> {
        let kind = metadata.kind.clone();
        let parent_clone = parent.clone();

        {
            let mut tree = self.tree.write().await;
            tree.insert(id.clone(), parent, metadata, inspectable)?;
        }

        let _ = self.event_tx.send(InspectorEvent::NodeRegistered {
            id: id.clone(),
            parent: parent_clone,
            kind,
        });

        Ok(crate::handle::NodeHandle::new(id, Arc::clone(self)))
    }

    /// Deregister a node (and reparent its children to its parent).
    pub async fn deregister(&self, id: &NodeId) -> Result<()> {
        {
            let mut tree = self.tree.write().await;
            tree.remove(id)?;
        }

        let _ = self
            .event_tx
            .send(InspectorEvent::NodeRemoved { id: id.clone() });
        Ok(())
    }

    /// Deregister a node and all its descendants.
    pub async fn deregister_subtree(&self, id: &NodeId) -> Result<()> {
        {
            let mut tree = self.tree.write().await;
            tree.remove_subtree(id)?;
        }

        let _ = self
            .event_tx
            .send(InspectorEvent::NodeRemoved { id: id.clone() });
        Ok(())
    }

    /// Update a node's lifecycle state.
    pub async fn update_state(&self, id: &NodeId, state: NodeState) -> Result<()> {
        let old = {
            let mut tree = self.tree.write().await;
            tree.update_state(id, state.clone())?
        };

        if old != state {
            let _ = self.event_tx.send(InspectorEvent::StateChanged {
                id: id.clone(),
                old,
                new: state,
            });
        }

        Ok(())
    }

    /// Update or insert properties on a node.
    pub async fn update_properties(
        &self,
        id: &NodeId,
        props: HashMap<String, serde_json::Value>,
    ) -> Result<()> {
        let keys = {
            let mut tree = self.tree.write().await;
            tree.update_properties(id, props)?
        };

        if !keys.is_empty() {
            let _ = self.event_tx.send(InspectorEvent::PropertiesUpdated {
                id: id.clone(),
                keys,
            });
        }

        Ok(())
    }

    /// Get a clone of a node's metadata.
    pub async fn get_node(&self, id: &NodeId) -> Option<NodeMetadata> {
        let tree = self.tree.read().await;
        tree.get(id).map(|n| n.metadata.clone())
    }

    /// Get metadata for all direct children of a node.
    pub async fn get_children(&self, id: &NodeId) -> Vec<(NodeId, NodeMetadata)> {
        let tree = self.tree.read().await;
        tree.children(id)
            .into_iter()
            .map(|n| (n.id.clone(), n.metadata.clone()))
            .collect()
    }

    /// Check if a node exists.
    pub async fn contains(&self, id: &NodeId) -> bool {
        let tree = self.tree.read().await;
        tree.contains(id)
    }

    /// Get the total number of nodes.
    pub async fn node_count(&self) -> usize {
        let tree = self.tree.read().await;
        tree.len()
    }

    /// Call `Inspectable::inspect()` on a node, if it implements the trait.
    ///
    /// Returns `None` if the node doesn't exist or doesn't have an `Inspectable` impl.
    pub async fn inspect_node(&self, id: &NodeId) -> Option<serde_json::Value> {
        let inspectable = {
            let tree = self.tree.read().await;
            tree.get(id)
                .and_then(|n| n.inspectable.as_ref().map(Arc::clone))
        };

        match inspectable {
            Some(i) => Some(i.inspect().await),
            None => None,
        }
    }

    /// Subscribe to tree change events.
    pub fn subscribe(&self) -> broadcast::Receiver<InspectorEvent> {
        self.event_tx.subscribe()
    }

    /// Get a snapshot of the entire tree (see snapshot module).
    pub async fn snapshot(&self) -> crate::snapshot::TreeSnapshot {
        let tree = self.tree.read().await;
        crate::snapshot::TreeSnapshot::from_tree(&tree)
    }
}

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

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

    #[tokio::test]
    async fn test_registry_new_has_root() {
        let registry = Arc::new(InspectorRegistry::new());
        let root = registry.root_id().await;
        assert_eq!(root.as_str(), "dirigent");
        assert!(registry.contains(&root).await);
        assert_eq!(registry.node_count().await, 1);
    }

    #[tokio::test]
    async fn test_register_and_get() {
        let registry = Arc::new(InspectorRegistry::new());
        let root = registry.root_id().await;

        let id = NodeId::new("dirigent/connectors");
        let meta = NodeMetadata::new(NodeKind::Connector, "Connectors");

        let handle = registry
            .register(id.clone(), &root, meta, None)
            .await
            .unwrap();
        assert_eq!(handle.id().as_str(), "dirigent/connectors");

        let retrieved = registry.get_node(&id).await.unwrap();
        assert_eq!(retrieved.label, "Connectors");
        assert_eq!(registry.node_count().await, 2);
    }

    #[tokio::test]
    async fn test_register_emits_event() {
        let registry = Arc::new(InspectorRegistry::new());
        let root = registry.root_id().await;
        let mut rx = registry.subscribe();

        let id = NodeId::new("dirigent/test");
        let meta = NodeMetadata::new(NodeKind::Service, "Test");

        let _handle = registry
            .register(id.clone(), &root, meta, None)
            .await
            .unwrap();

        let event = rx.recv().await.unwrap();
        match event {
            InspectorEvent::NodeRegistered {
                id: event_id,
                parent,
                kind,
            } => {
                assert_eq!(event_id.as_str(), "dirigent/test");
                assert_eq!(parent.as_str(), "dirigent");
                assert_eq!(kind, NodeKind::Service);
            }
            _ => panic!("Expected NodeRegistered event"),
        }
    }

    #[tokio::test]
    async fn test_deregister() {
        let registry = Arc::new(InspectorRegistry::new());
        let root = registry.root_id().await;

        let id = NodeId::new("dirigent/temp");
        let meta = NodeMetadata::new(NodeKind::AsyncTask, "Temp");
        let _handle = registry
            .register(id.clone(), &root, meta, None)
            .await
            .unwrap();

        assert!(registry.contains(&id).await);
        registry.deregister(&id).await.unwrap();
        assert!(!registry.contains(&id).await);
    }

    #[tokio::test]
    async fn test_update_state_emits_event() {
        let registry = Arc::new(InspectorRegistry::new());
        let root = registry.root_id().await;

        let id = NodeId::new("dirigent/svc");
        let meta = NodeMetadata::new(NodeKind::Service, "Service");
        let _handle = registry
            .register(id.clone(), &root, meta, None)
            .await
            .unwrap();

        let mut rx = registry.subscribe();

        registry
            .update_state(&id, NodeState::Error("crash".into()))
            .await
            .unwrap();

        let event = rx.recv().await.unwrap();
        match event {
            InspectorEvent::StateChanged { id: eid, old, new } => {
                assert_eq!(eid.as_str(), "dirigent/svc");
                assert_eq!(old, NodeState::Initializing);
                assert_eq!(new, NodeState::Error("crash".into()));
            }
            _ => panic!("Expected StateChanged event"),
        }
    }

    #[tokio::test]
    async fn test_no_event_on_same_state() {
        let registry = Arc::new(InspectorRegistry::new());
        let root = registry.root_id().await;

        let id = NodeId::new("dirigent/svc");
        let meta = NodeMetadata::new(NodeKind::Service, "Service").with_state(NodeState::Running);
        let _handle = registry
            .register(id.clone(), &root, meta, None)
            .await
            .unwrap();

        let mut rx = registry.subscribe();

        // Update to the same state
        registry
            .update_state(&id, NodeState::Running)
            .await
            .unwrap();

        // Should not receive a StateChanged event
        let result = tokio::time::timeout(std::time::Duration::from_millis(50), rx.recv()).await;
        assert!(result.is_err(), "Should not emit event for same state");
    }

    #[tokio::test]
    async fn test_update_properties() {
        let registry = Arc::new(InspectorRegistry::new());
        let root = registry.root_id().await;

        let id = NodeId::new("dirigent/proc");
        let meta = NodeMetadata::new(NodeKind::Process, "Process");
        let _handle = registry
            .register(id.clone(), &root, meta, None)
            .await
            .unwrap();

        let mut props = HashMap::new();
        props.insert("pid".to_string(), serde_json::json!(1234));
        props.insert("cpu".to_string(), serde_json::json!(50.0));

        registry.update_properties(&id, props).await.unwrap();

        let node = registry.get_node(&id).await.unwrap();
        assert_eq!(node.properties["pid"], serde_json::json!(1234));
        assert_eq!(node.properties["cpu"], serde_json::json!(50.0));
    }

    #[tokio::test]
    async fn test_get_children() {
        let registry = Arc::new(InspectorRegistry::new());
        let root = registry.root_id().await;

        let _h1 = registry
            .register(
                NodeId::new("dirigent/a"),
                &root,
                NodeMetadata::new(NodeKind::Service, "A"),
                None,
            )
            .await
            .unwrap();
        let _h2 = registry
            .register(
                NodeId::new("dirigent/b"),
                &root,
                NodeMetadata::new(NodeKind::Service, "B"),
                None,
            )
            .await
            .unwrap();

        let children = registry.get_children(&root).await;
        assert_eq!(children.len(), 2);
        let labels: Vec<&str> = children.iter().map(|(_, m)| m.label.as_str()).collect();
        assert!(labels.contains(&"A"));
        assert!(labels.contains(&"B"));
    }

    #[tokio::test]
    async fn test_inspect_node() {
        use async_trait::async_trait;

        struct MockInspectable;

        #[async_trait]
        impl Inspectable for MockInspectable {
            async fn inspect(&self) -> serde_json::Value {
                serde_json::json!({ "internal_queue_len": 42 })
            }
            fn current_state(&self) -> NodeState {
                NodeState::Running
            }
        }

        let registry = Arc::new(InspectorRegistry::new());
        let root = registry.root_id().await;

        let id = NodeId::new("dirigent/inspectable");
        let meta = NodeMetadata::new(NodeKind::Service, "Inspectable Service");
        let _handle = registry
            .register(id.clone(), &root, meta, Some(Arc::new(MockInspectable)))
            .await
            .unwrap();

        let result = registry.inspect_node(&id).await;
        assert_eq!(
            result.unwrap(),
            serde_json::json!({ "internal_queue_len": 42 })
        );

        // Non-inspectable node returns None
        let id2 = NodeId::new("dirigent/plain");
        let _h2 = registry
            .register(
                id2.clone(),
                &root,
                NodeMetadata::new(NodeKind::Service, "Plain"),
                None,
            )
            .await
            .unwrap();
        assert!(registry.inspect_node(&id2).await.is_none());
    }
}
