//! Command listener for remote node control
//!
//! Subscribes to `system/cli/command` and handles node lifecycle commands
//! (start, stop, restart) from the dashboard or other clients.
//!
//! The CommandListener runs as a background task during dev mode and shares
//! the ProcessService with the main DevSession to control nodes. Status updates
//! are published to Redis so the dashboard can track node state.

use crate::services::ProcessService;
use crate::ui::LogBuffer;
use anyhow::{Context, Result};
use futures_util::StreamExt;
use mecha10_core::lifecycle::{get_timestamp_ms, NodeLifecycleStatus, NodeStatusEvent};
use mecha10_core::messages::system::{Command, SystemCommand};
use std::collections::HashMap;
use std::sync::Arc;

/// Type alias for the shared ProcessService (uses std::sync::Mutex for sync access)
type SharedProcessService = Arc<std::sync::Mutex<ProcessService>>;
/// Type alias for the Redis connection (uses tokio::sync::Mutex for async access)
type SharedRedisConn = Arc<tokio::sync::Mutex<redis::aio::MultiplexedConnection>>;

/// Topic for CLI commands
pub const TOPIC_CLI_COMMAND: &str = "system/cli/command";

/// Configuration needed by CommandListener
#[derive(Clone)]
pub struct CommandListenerConfig {
    pub redis_url: String,
    pub project_name: String,
    pub robot_id: String,
    pub control_plane_url: String,
    pub relay_url: String,
    /// Optional log buffer for TUI logging
    pub log_buffer: Option<LogBuffer>,
}

/// Command listener for remote node control
///
/// Subscribes to Redis pub/sub and handles node lifecycle commands.
/// Runs as a background task during dev mode.
///
/// This listener shares the ProcessService with the main DevSession
/// to control nodes that were started by the session.
pub struct CommandListener {
    config: CommandListenerConfig,
    /// Shared ProcessService from the CliContext
    process_service: SharedProcessService,
}

impl CommandListener {
    /// Create a new command listener with a shared ProcessService
    pub fn new(config: CommandListenerConfig, process_service: SharedProcessService) -> Self {
        Self {
            config,
            process_service,
        }
    }

    /// Debug logging to file (since TUI captures stderr)
    fn log_debug(msg: &str) {
        let _ = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open("/tmp/command-listener.log")
            .and_then(|mut f| {
                use std::io::Write;
                writeln!(f, "[{}] {}", chrono::Utc::now(), msg)
            });
    }

    /// Push a message to the TUI log buffer if available
    fn log_to_tui(&self, msg: String) {
        if let Some(ref buffer) = self.config.log_buffer {
            buffer.push(msg);
        }
    }

    /// Resolve a target name to the full identifier used in ProcessService
    ///
    /// Commands from dashboard use short names (e.g., "speaker"), but ProcessService
    /// registers nodes with full identifiers (e.g., "@mecha10/speaker").
    /// This method finds the matching full identifier.
    fn resolve_node_identifier(&self, target: &str) -> Option<String> {
        let mut ps = self.process_service.lock().unwrap();

        // First, check if target is already a full identifier
        if ps.is_running(target) {
            return Some(target.to_string());
        }

        // Try bundled node format: @mecha10/<name>
        let bundled = format!("@mecha10/{}", target);
        if ps.is_running(&bundled) {
            return Some(bundled);
        }

        // Try local project node format: @<project>/<name>
        let local = format!("@{}/{}", self.config.project_name, target);
        if ps.is_running(&local) {
            return Some(local);
        }

        None
    }

    /// Start listening for commands
    ///
    /// This runs until the Redis connection is closed or an error occurs.
    /// Uses a shutdown receiver to allow graceful termination.
    pub async fn run(self, mut shutdown: tokio::sync::broadcast::Receiver<()>) -> Result<()> {
        Self::log_debug(&format!(
            "CommandListener::run starting, redis_url={}",
            self.config.redis_url
        ));

        // Connect to Redis for pub/sub
        let client = redis::Client::open(self.config.redis_url.as_str())
            .context("Failed to create Redis client for command listener")?;

        Self::log_debug("Redis client created, getting pub/sub connection...");

        let mut pubsub = client
            .get_async_pubsub()
            .await
            .context("Failed to get Redis pub/sub connection")?;

        Self::log_debug("Got pub/sub connection, subscribing...");

        // Subscribe to command topic
        pubsub
            .subscribe(TOPIC_CLI_COMMAND)
            .await
            .context("Failed to subscribe to command topic")?;

        Self::log_debug(&format!("Subscribed to {}, entering message loop", TOPIC_CLI_COMMAND));

        // Create a separate connection for publishing status updates
        let publish_conn = client
            .get_multiplexed_async_connection()
            .await
            .context("Failed to get Redis connection for publishing")?;
        let publish_conn: SharedRedisConn = Arc::new(tokio::sync::Mutex::new(publish_conn));

        // Get message stream
        let mut msg_stream = pubsub.on_message();

        loop {
            tokio::select! {
                // Handle shutdown signal
                _ = shutdown.recv() => {
                    break;
                }

                // Handle incoming commands from Redis
                msg = msg_stream.next() => {
                    match msg {
                        Some(msg) => {
                            let payload: String = msg.get_payload().unwrap_or_default();
                            // Debug: write to file since TUI captures stderr
                            let _ = std::fs::OpenOptions::new()
                                .create(true).append(true)
                                .open("/tmp/command-listener.log")
                                .and_then(|mut f| {
                                    use std::io::Write;
                                    writeln!(f, "[{}] Received: {}", chrono::Utc::now(), payload)
                                });
                            if let Err(e) = self.handle_message(&payload, Arc::clone(&publish_conn)).await {
                                let _ = std::fs::OpenOptions::new()
                                    .create(true).append(true)
                                    .open("/tmp/command-listener.log")
                                    .and_then(|mut f| {
                                        use std::io::Write;
                                        writeln!(f, "[{}] Error: {}", chrono::Utc::now(), e)
                                    });
                                tracing::error!("Error handling command: {}", e);
                            }
                        }
                        None => {
                            tracing::warn!("Redis pub/sub connection closed");
                            break;
                        }
                    }
                }
            }
        }

        // Note: We don't cleanup here - the shared ProcessService is managed by DevSession
        Ok(())
    }

    /// Handle an incoming command message
    async fn handle_message(&self, payload: &str, publish_conn: SharedRedisConn) -> Result<()> {
        Self::log_debug(&format!("Parsing command: {}", payload));
        let cmd: SystemCommand =
            serde_json::from_str(payload).with_context(|| format!("Failed to parse command: {}", payload))?;

        Self::log_debug(&format!(
            "Parsed command: {:?} for target '{}'",
            cmd.command, cmd.target
        ));

        match cmd.command {
            Command::Start => {
                self.handle_start(&cmd.target, publish_conn).await?;
            }
            Command::Stop => {
                self.handle_stop(&cmd.target, publish_conn).await?;
            }
            Command::Restart => {
                self.handle_restart(&cmd.target, publish_conn).await?;
            }
            _ => {
                tracing::debug!("Ignoring unsupported command: {:?}", cmd.command);
            }
        }

        Ok(())
    }

    /// Handle start command
    async fn handle_start(&self, target: &str, publish_conn: SharedRedisConn) -> Result<()> {
        // Check if already running using resolver
        if self.resolve_node_identifier(target).is_some() {
            let msg = format!("⚠️  [Dashboard] Node '{}' is already running", target);
            self.log_to_tui(msg);
            return Ok(());
        }

        let msg = format!("▶️  [Dashboard] Starting node '{}'...", target);
        self.log_to_tui(msg);

        // Publish starting status
        self.publish_status(target, NodeLifecycleStatus::Starting, None, None, &publish_conn)
            .await?;

        // Build environment variables
        let project_env = self.build_project_env();

        // Start the node (spawn_node_direct handles identifier creation internally)
        // Local project nodes: @local/* (legacy) or plain names (no @ prefix)
        // Framework nodes: @mecha10/*
        let is_local_node = target.starts_with("@local/") || !target.starts_with('@');
        let use_node_runner = ProcessService::is_node_runner_available() && is_local_node;

        let result = {
            let mut ps = self.process_service.lock().unwrap();
            if use_node_runner {
                ps.spawn_node_runner(target, Some(project_env))
            } else {
                ps.spawn_node_direct(target, &self.config.project_name, Some(project_env))
            }
        };

        match result {
            Ok(pid) => {
                self.publish_status(target, NodeLifecycleStatus::Started, Some(pid), None, &publish_conn)
                    .await?;
                let msg = format!("✅ [Dashboard] Started node '{}' (PID: {})", target, pid);
                self.log_to_tui(msg);
            }
            Err(e) => {
                self.publish_status(
                    target,
                    NodeLifecycleStatus::Failed,
                    None,
                    Some(e.to_string()),
                    &publish_conn,
                )
                .await?;
                let msg = format!("❌ [Dashboard] Failed to start node '{}': {}", target, e);
                self.log_to_tui(msg);
            }
        }

        Ok(())
    }

    /// Handle stop command
    async fn handle_stop(&self, target: &str, publish_conn: SharedRedisConn) -> Result<()> {
        Self::log_debug(&format!("handle_stop called for '{}'", target));

        // Resolve short name to full identifier
        let node_id = match self.resolve_node_identifier(target) {
            Some(id) => {
                Self::log_debug(&format!("Resolved '{}' -> '{}'", target, id));
                id
            }
            None => {
                Self::log_debug(&format!("Node '{}' NOT FOUND in ProcessService!", target));
                let msg = format!("⚠️  [Dashboard] Node '{}' is not running", target);
                self.log_to_tui(msg);
                return Ok(());
            }
        };

        let msg = format!("🛑 [Dashboard] Stopping node '{}'...", target);
        self.log_to_tui(msg);

        // Publish stopping status (don't fail the stop if status publish fails)
        let _ = self
            .publish_status(target, NodeLifecycleStatus::Stopping, None, None, &publish_conn)
            .await;

        // Stop the node using the resolved identifier
        let result = {
            let mut ps = self.process_service.lock().unwrap();
            ps.stop(&node_id)
        };

        match result {
            Ok(()) => {
                self.publish_status(target, NodeLifecycleStatus::Stopped, None, None, &publish_conn)
                    .await?;
                let msg = format!("✅ [Dashboard] Stopped node '{}'", target);
                self.log_to_tui(msg);
            }
            Err(e) => {
                self.publish_status(
                    target,
                    NodeLifecycleStatus::Failed,
                    None,
                    Some(e.to_string()),
                    &publish_conn,
                )
                .await?;
                let msg = format!("❌ [Dashboard] Failed to stop node '{}': {}", target, e);
                self.log_to_tui(msg);
            }
        }

        Ok(())
    }

    /// Handle restart command
    async fn handle_restart(&self, target: &str, publish_conn: SharedRedisConn) -> Result<()> {
        let msg = format!("🔄 [Dashboard] Restarting node '{}'...", target);
        self.log_to_tui(msg);

        // Stop first if running (resolve_node_identifier is called inside handle_stop)
        let is_running = self.resolve_node_identifier(target).is_some();

        if is_running {
            self.handle_stop(target, Arc::clone(&publish_conn)).await?;
            // Brief delay for cleanup
            tokio::time::sleep(std::time::Duration::from_millis(500)).await;
        }

        // Then start
        self.handle_start(target, publish_conn).await?;

        Ok(())
    }

    /// Publish node status event
    async fn publish_status(
        &self,
        node_name: &str,
        status: NodeLifecycleStatus,
        pid: Option<u32>,
        error: Option<String>,
        conn: &SharedRedisConn,
    ) -> Result<()> {
        let event = NodeStatusEvent {
            node: node_name.to_string(),
            status,
            pid,
            error,
            timestamp: get_timestamp_ms(),
        };

        let topic = format!("system/node/{}/status", node_name);
        let payload = serde_json::to_string(&event)?;

        let mut conn = conn.lock().await;
        redis::cmd("PUBLISH")
            .arg(&topic)
            .arg(&payload)
            .query_async::<i64>(&mut *conn)
            .await
            .with_context(|| format!("Failed to publish status for node '{}'", node_name))?;

        tracing::debug!("Published status for '{}': {:?}", node_name, event.status);

        Ok(())
    }

    /// Build project environment variables
    fn build_project_env(&self) -> HashMap<String, String> {
        use crate::services::CredentialsService;

        let mut env = HashMap::new();

        // Control plane URL
        env.insert("CONTROL_PLANE_URL".to_string(), self.config.control_plane_url.clone());

        // Robot info
        env.insert("ROBOT_ID".to_string(), self.config.robot_id.clone());

        // API key from credentials
        let credentials_service = CredentialsService::new();
        if let Ok(Some(api_key)) = credentials_service.get_api_key() {
            env.insert("ROBOT_API_KEY".to_string(), api_key);
        }

        // WebRTC relay URL
        env.insert("WEBRTC_RELAY_URL".to_string(), self.config.relay_url.clone());

        // Redis URL
        env.insert("REDIS_URL".to_string(), self.config.redis_url.clone());

        env
    }
}
