use crate::error::{InspectorError, Result};
use serde::{Deserialize, Serialize};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use tokio::sync::{mpsc, oneshot};

/// A command that can be sent to a node.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct NodeCommand {
    /// Unique command ID for correlation with responses.
    pub id: String,
    /// What kind of command this is.
    pub kind: CommandKind,
    /// Arbitrary payload data.
    pub payload: serde_json::Value,
}

/// The type of command being sent to a node.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub enum CommandKind {
    /// Request the node to report its internal state (introspective).
    Introspect,
    /// Execute a named operation.
    Execute(String),
    /// Custom extension command.
    Custom(String),
}

/// Response from a node to a command.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct CommandResponse {
    /// The command ID this is responding to.
    pub command_id: String,
    /// Whether the command was handled successfully.
    pub success: bool,
    /// Response data.
    pub data: serde_json::Value,
}

impl CommandResponse {
    /// Create a success response.
    pub fn ok(command_id: impl Into<String>, data: serde_json::Value) -> Self {
        Self {
            command_id: command_id.into(),
            success: true,
            data,
        }
    }

    /// Create an error response.
    pub fn err(command_id: impl Into<String>, message: impl Into<String>) -> Self {
        Self {
            command_id: command_id.into(),
            success: false,
            data: serde_json::json!({ "error": message.into() }),
        }
    }
}

type CommandPayload = (NodeCommand, oneshot::Sender<CommandResponse>);

/// Create a new inspector channel pair.
///
/// Returns `(sender, receiver)` where:
/// - The **sender** is used by callers (UI, API) to send commands to the node.
/// - The **receiver** is used by the node's loop to receive and respond to commands.
///
/// `capacity` controls the bounded channel size.
pub fn channel(capacity: usize) -> (InspectorChannelSender, InspectorChannelReceiver) {
    let (tx, rx) = mpsc::channel(capacity);
    let pending = Arc::new(AtomicUsize::new(0));

    (
        InspectorChannelSender {
            tx,
            pending: Arc::clone(&pending),
        },
        InspectorChannelReceiver { rx, pending },
    )
}

/// Caller-side of the inspector channel: send commands, check queue depth.
pub struct InspectorChannelSender {
    tx: mpsc::Sender<CommandPayload>,
    pending: Arc<AtomicUsize>,
}

impl InspectorChannelSender {
    /// Send a command and wait for the response.
    pub async fn send(&self, cmd: NodeCommand) -> Result<CommandResponse> {
        let (resp_tx, resp_rx) = oneshot::channel();
        self.pending.fetch_add(1, Ordering::Relaxed);

        self.tx
            .send((cmd, resp_tx))
            .await
            .map_err(|_| InspectorError::ChannelClosed)?;

        resp_rx.await.map_err(|_| InspectorError::ChannelClosed)
    }

    /// Try to send a command without waiting, returning a receiver for the response.
    ///
    /// This is useful when you want to fire off a command and collect the
    /// response later, or in a `select!` branch.
    pub fn try_send(&self, cmd: NodeCommand) -> Result<oneshot::Receiver<CommandResponse>> {
        let (resp_tx, resp_rx) = oneshot::channel();
        self.pending.fetch_add(1, Ordering::Relaxed);

        self.tx.try_send((cmd, resp_tx)).map_err(|e| match e {
            mpsc::error::TrySendError::Full(_) => {
                self.pending.fetch_sub(1, Ordering::Relaxed);
                InspectorError::ChannelFull
            }
            mpsc::error::TrySendError::Closed(_) => {
                self.pending.fetch_sub(1, Ordering::Relaxed);
                InspectorError::ChannelClosed
            }
        })?;

        Ok(resp_rx)
    }

    /// Number of commands currently in the queue (approximate).
    pub fn pending_count(&self) -> usize {
        self.pending.load(Ordering::Relaxed)
    }
}

impl Clone for InspectorChannelSender {
    fn clone(&self) -> Self {
        Self {
            tx: self.tx.clone(),
            pending: Arc::clone(&self.pending),
        }
    }
}

/// Node-side of the inspector channel: receive commands, send responses.
pub struct InspectorChannelReceiver {
    rx: mpsc::Receiver<CommandPayload>,
    pending: Arc<AtomicUsize>,
}

impl InspectorChannelReceiver {
    /// Receive the next command. Returns `None` when all senders are dropped.
    ///
    /// The returned `oneshot::Sender` must be used to send back a response.
    pub async fn recv(&mut self) -> Option<(NodeCommand, oneshot::Sender<CommandResponse>)> {
        let result = self.rx.recv().await;
        if result.is_some() {
            self.pending.fetch_sub(1, Ordering::Relaxed);
        }
        result
    }

    /// Try to receive without blocking. Returns `None` if the channel is empty.
    pub fn try_recv(&mut self) -> Option<(NodeCommand, oneshot::Sender<CommandResponse>)> {
        match self.rx.try_recv() {
            Ok(payload) => {
                self.pending.fetch_sub(1, Ordering::Relaxed);
                Some(payload)
            }
            Err(_) => None,
        }
    }

    /// Number of commands currently in the queue (approximate).
    pub fn pending_count(&self) -> usize {
        self.pending.load(Ordering::Relaxed)
    }
}

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

    fn introspect_cmd(id: &str) -> NodeCommand {
        NodeCommand {
            id: id.to_string(),
            kind: CommandKind::Introspect,
            payload: serde_json::Value::Null,
        }
    }

    fn exec_cmd(id: &str, name: &str, payload: serde_json::Value) -> NodeCommand {
        NodeCommand {
            id: id.to_string(),
            kind: CommandKind::Execute(name.to_string()),
            payload,
        }
    }

    #[tokio::test]
    async fn test_send_recv_response() {
        let (sender, mut receiver) = channel(10);

        // Simulate a node loop in a background task
        let node_task = tokio::spawn(async move {
            if let Some((cmd, resp_tx)) = receiver.recv().await {
                assert_eq!(cmd.id, "cmd-1");
                assert!(matches!(cmd.kind, CommandKind::Introspect));
                let _ = resp_tx.send(CommandResponse::ok(
                    &cmd.id,
                    serde_json::json!({ "queue_len": 5 }),
                ));
            }
        });

        let response = sender.send(introspect_cmd("cmd-1")).await.unwrap();
        assert!(response.success);
        assert_eq!(response.command_id, "cmd-1");
        assert_eq!(response.data["queue_len"], 5);

        node_task.await.unwrap();
    }

    #[tokio::test]
    async fn test_try_send() {
        let (sender, mut receiver) = channel(10);

        let resp_rx = sender
            .try_send(exec_cmd("cmd-2", "restart", serde_json::json!({})))
            .unwrap();

        // Node responds
        let (cmd, resp_tx) = receiver.recv().await.unwrap();
        assert_eq!(cmd.id, "cmd-2");
        let _ = resp_tx.send(CommandResponse::ok(&cmd.id, serde_json::json!("restarted")));

        let response = resp_rx.await.unwrap();
        assert!(response.success);
    }

    #[tokio::test]
    async fn test_pending_count() {
        let (sender, mut receiver) = channel(10);

        assert_eq!(sender.pending_count(), 0);

        // Send 3 commands without receiving
        let _r1 = sender.try_send(introspect_cmd("a")).unwrap();
        let _r2 = sender.try_send(introspect_cmd("b")).unwrap();
        let _r3 = sender.try_send(introspect_cmd("c")).unwrap();

        assert_eq!(sender.pending_count(), 3);
        assert_eq!(receiver.pending_count(), 3);

        // Receive one
        let (cmd, resp_tx) = receiver.recv().await.unwrap();
        let _ = resp_tx.send(CommandResponse::ok(&cmd.id, serde_json::Value::Null));

        assert_eq!(receiver.pending_count(), 2);
    }

    #[tokio::test]
    async fn test_try_send_full() {
        let (sender, _receiver) = channel(1);

        // Fill the channel
        let _r1 = sender.try_send(introspect_cmd("a")).unwrap();

        // Should fail with ChannelFull
        let result = sender.try_send(introspect_cmd("b"));
        assert!(matches!(result, Err(InspectorError::ChannelFull)));
    }

    #[tokio::test]
    async fn test_send_after_receiver_dropped() {
        let (sender, receiver) = channel(10);
        drop(receiver);

        let result = sender.send(introspect_cmd("orphan")).await;
        assert!(matches!(result, Err(InspectorError::ChannelClosed)));
    }

    #[tokio::test]
    async fn test_recv_after_sender_dropped() {
        let (sender, mut receiver) = channel(10);
        drop(sender);

        let result = receiver.recv().await;
        assert!(result.is_none());
    }

    #[tokio::test]
    async fn test_try_recv_empty() {
        let (_sender, mut receiver) = channel(10);
        assert!(receiver.try_recv().is_none());
    }

    #[tokio::test]
    async fn test_error_response() {
        let (sender, mut receiver) = channel(10);

        let node_task = tokio::spawn(async move {
            if let Some((cmd, resp_tx)) = receiver.recv().await {
                let _ = resp_tx.send(CommandResponse::err(&cmd.id, "not supported"));
            }
        });

        let response = sender
            .send(exec_cmd("cmd-x", "unknown", serde_json::Value::Null))
            .await
            .unwrap();
        assert!(!response.success);
        assert_eq!(response.data["error"], "not supported");

        node_task.await.unwrap();
    }

    #[tokio::test]
    async fn test_multiple_senders() {
        let (sender, mut receiver) = channel(10);
        let sender2 = sender.clone();

        let _r1 = sender.try_send(introspect_cmd("from-1")).unwrap();
        let _r2 = sender2.try_send(introspect_cmd("from-2")).unwrap();

        let (cmd1, resp_tx1) = receiver.recv().await.unwrap();
        let _ = resp_tx1.send(CommandResponse::ok(&cmd1.id, serde_json::Value::Null));

        let (cmd2, resp_tx2) = receiver.recv().await.unwrap();
        let _ = resp_tx2.send(CommandResponse::ok(&cmd2.id, serde_json::Value::Null));

        // Both commands were received (order may vary since mpsc is FIFO)
        let ids = vec![cmd1.id, cmd2.id];
        assert!(ids.contains(&"from-1".to_string()));
        assert!(ids.contains(&"from-2".to_string()));
    }

    #[test]
    fn test_command_serialization() {
        let cmd = exec_cmd("cmd-1", "restart", serde_json::json!({"force": true}));
        let json = serde_json::to_string(&cmd).unwrap();
        let deserialized: NodeCommand = serde_json::from_str(&json).unwrap();
        assert_eq!(deserialized.id, "cmd-1");
        assert!(matches!(deserialized.kind, CommandKind::Execute(ref name) if name == "restart"));
    }

    #[test]
    fn test_response_serialization() {
        let resp = CommandResponse::ok("cmd-1", serde_json::json!({"status": "done"}));
        let json = serde_json::to_string(&resp).unwrap();
        let deserialized: CommandResponse = serde_json::from_str(&json).unwrap();
        assert!(deserialized.success);
        assert_eq!(deserialized.data["status"], "done");
    }
}
