//! RabbitMQ command execution via lapin.
//!
//! Implements the [`CommandExecutor`] trait for RabbitMQ, translating dashboard
//! commands into AMQP operations:
//!
//! | Command  | AMQP Operation                          |
//! |----------|-----------------------------------------|
//! | `retry`  | `basic_publish` to the default exchange  |
//! | `pause`  | `basic_qos(prefetch_count=0)`            |
//! | `resume` | `basic_qos(original_prefetch)`           |
//! | `purge`  | `queue_purge`                            |

use std::sync::atomic::{AtomicU16, Ordering};
use std::sync::Arc;

use async_trait::async_trait;
use lapin::options::{BasicPublishOptions, BasicQosOptions, QueuePurgeOptions};
use lapin::BasicProperties;
use queuerious::agent::traits::{CommandExecutor, CommandResult, PendingCommand};
use tokio::sync::RwLock;

use crate::channel_factory::ChannelFactory;

/// Executes queue management commands against RabbitMQ via a lapin [`Channel`](lapin::Channel).
///
/// # Channel Durability
///
/// The executor caches an AMQP channel and automatically recreates it via the
/// [`ChannelFactory`] when the channel becomes disconnected (e.g. after a
/// RabbitMQ restart).
///
/// # Pause / Resume
///
/// The executor tracks the original `prefetch_count` set at construction time.
/// A `pause` command sets `basic_qos(prefetch_count=0)` to stop message delivery,
/// and `resume` restores the original prefetch count.
///
/// # Example
///
/// ```no_run
/// use std::sync::Arc;
/// use queuerious_lapin::{RabbitMqCommandExecutor, StaticConnectionFactory};
///
/// # async fn example(conn: lapin::Connection) {
/// let factory = Arc::new(StaticConnectionFactory::new(conn));
/// let executor = RabbitMqCommandExecutor::new(factory, 10);
/// // Use with an agent that polls for pending commands.
/// # }
/// ```
pub struct RabbitMqCommandExecutor {
    factory: Arc<dyn ChannelFactory>,
    channel: RwLock<Option<lapin::Channel>>,
    original_prefetch: AtomicU16,
}

impl RabbitMqCommandExecutor {
    /// Create a new executor backed by a channel factory.
    ///
    /// `prefetch_count` is the current QoS prefetch count, saved so that
    /// `resume` can restore it after a `pause`.
    pub fn new(factory: Arc<dyn ChannelFactory>, prefetch_count: u16) -> Self {
        Self {
            factory,
            channel: RwLock::new(None),
            original_prefetch: AtomicU16::new(prefetch_count),
        }
    }

    /// Returns the stored original prefetch count.
    pub fn original_prefetch(&self) -> u16 {
        self.original_prefetch.load(Ordering::SeqCst)
    }

    /// Updates the stored original prefetch count.
    ///
    /// Call this if the application changes its QoS settings outside of
    /// the pause/resume cycle.
    pub fn set_original_prefetch(&self, prefetch_count: u16) {
        self.original_prefetch
            .store(prefetch_count, Ordering::SeqCst);
    }

    /// Get a healthy cached channel, creating a new one if needed.
    async fn get_channel(&self) -> Result<lapin::Channel, lapin::Error> {
        // Fast path: cached channel is still connected
        {
            let guard = self.channel.read().await;
            if let Some(ch) = guard.as_ref() {
                if ch.status().connected() {
                    return Ok(ch.clone());
                }
            }
        }

        // Slow path: create new channel
        let mut guard = self.channel.write().await;
        // Double-check after acquiring write lock
        if let Some(ch) = guard.as_ref() {
            if ch.status().connected() {
                return Ok(ch.clone());
            }
        }

        tracing::debug!("Creating new AMQP channel for command execution");
        let new_ch = self.factory.create_channel().await?;
        *guard = Some(new_ch.clone());
        Ok(new_ch)
    }

    /// Re-publish a message to the queue via the default exchange.
    async fn handle_retry(&self, command: &PendingCommand) -> CommandResult {
        let channel = match self.get_channel().await {
            Ok(ch) => ch,
            Err(e) => {
                return CommandResult {
                    success: false,
                    message: Some(format!("Failed to get AMQP channel: {}", e)),
                    data: None,
                };
            }
        };

        let payload = command
            .parameters
            .get("payload")
            .and_then(|v| v.as_str())
            .map(|s| s.as_bytes().to_vec())
            .unwrap_or_default();

        match channel
            .basic_publish(
                "",
                &command.queue_name,
                BasicPublishOptions::default(),
                &payload,
                BasicProperties::default(),
            )
            .await
        {
            Ok(confirm) => match confirm.await {
                Ok(_) => CommandResult {
                    success: true,
                    message: Some(format!(
                        "Message re-published to queue '{}'",
                        command.queue_name
                    )),
                    data: None,
                },
                Err(e) => CommandResult {
                    success: false,
                    message: Some(format!(
                        "Publish confirmation failed for queue '{}': {}",
                        command.queue_name, e
                    )),
                    data: None,
                },
            },
            Err(e) => CommandResult {
                success: false,
                message: Some(format!(
                    "Failed to publish to queue '{}': {}",
                    command.queue_name, e
                )),
                data: None,
            },
        }
    }

    /// Stop message delivery by setting `prefetch_count` to 0.
    async fn handle_pause(&self, command: &PendingCommand) -> CommandResult {
        let channel = match self.get_channel().await {
            Ok(ch) => ch,
            Err(e) => {
                return CommandResult {
                    success: false,
                    message: Some(format!("Failed to get AMQP channel: {}", e)),
                    data: None,
                };
            }
        };

        match channel.basic_qos(0, BasicQosOptions::default()).await {
            Ok(_) => CommandResult {
                success: true,
                message: Some(format!(
                    "Queue '{}' paused (prefetch set to 0)",
                    command.queue_name
                )),
                data: None,
            },
            Err(e) => CommandResult {
                success: false,
                message: Some(format!(
                    "Failed to pause queue '{}': {}",
                    command.queue_name, e
                )),
                data: None,
            },
        }
    }

    /// Restore the original prefetch count to resume message delivery.
    async fn handle_resume(&self, command: &PendingCommand) -> CommandResult {
        let channel = match self.get_channel().await {
            Ok(ch) => ch,
            Err(e) => {
                return CommandResult {
                    success: false,
                    message: Some(format!("Failed to get AMQP channel: {}", e)),
                    data: None,
                };
            }
        };

        let prefetch = self.original_prefetch.load(Ordering::SeqCst);
        match channel
            .basic_qos(prefetch, BasicQosOptions::default())
            .await
        {
            Ok(_) => CommandResult {
                success: true,
                message: Some(format!(
                    "Queue '{}' resumed (prefetch restored to {})",
                    command.queue_name, prefetch
                )),
                data: None,
            },
            Err(e) => CommandResult {
                success: false,
                message: Some(format!(
                    "Failed to resume queue '{}': {}",
                    command.queue_name, e
                )),
                data: None,
            },
        }
    }

    /// Remove all messages from the queue.
    async fn handle_purge(&self, command: &PendingCommand) -> CommandResult {
        let channel = match self.get_channel().await {
            Ok(ch) => ch,
            Err(e) => {
                return CommandResult {
                    success: false,
                    message: Some(format!("Failed to get AMQP channel: {}", e)),
                    data: None,
                };
            }
        };

        match channel
            .queue_purge(&command.queue_name, QueuePurgeOptions::default())
            .await
        {
            Ok(count) => CommandResult {
                success: true,
                message: Some(format!(
                    "Purged {} messages from queue '{}'",
                    count, command.queue_name
                )),
                data: None,
            },
            Err(e) => CommandResult {
                success: false,
                message: Some(format!(
                    "Failed to purge queue '{}': {}",
                    command.queue_name, e
                )),
                data: None,
            },
        }
    }
}

#[async_trait]
impl CommandExecutor for RabbitMqCommandExecutor {
    async fn execute(&self, command: &PendingCommand) -> CommandResult {
        match command.command_type.as_str() {
            "retry" => self.handle_retry(command).await,
            "pause" => self.handle_pause(command).await,
            "resume" => self.handle_resume(command).await,
            "purge" => self.handle_purge(command).await,
            other => CommandResult {
                success: false,
                message: Some(format!(
                    "Unknown command type '{}' is not supported by the RabbitMQ adapter",
                    other
                )),
                data: None,
            },
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::Utc;
    use uuid::Uuid;

    // --- Prefetch save/restore logic tests ---

    #[test]
    fn stores_original_prefetch_on_construction() {
        let prefetch = AtomicU16::new(10);
        assert_eq!(prefetch.load(Ordering::SeqCst), 10);
    }

    #[test]
    fn prefetch_survives_pause_resume_cycle() {
        let original = 25u16;
        let prefetch = AtomicU16::new(original);

        // Pause: we'd set QoS to 0, but the saved value stays
        let saved = prefetch.load(Ordering::SeqCst);
        assert_eq!(saved, original);

        // Resume: restore from saved
        let restored = prefetch.load(Ordering::SeqCst);
        assert_eq!(restored, original);
    }

    #[test]
    fn set_prefetch_updates_stored_value() {
        let prefetch = AtomicU16::new(10);
        prefetch.store(50, Ordering::SeqCst);
        assert_eq!(prefetch.load(Ordering::SeqCst), 50);
    }

    #[test]
    fn prefetch_zero_is_valid() {
        let prefetch = AtomicU16::new(0);
        assert_eq!(prefetch.load(Ordering::SeqCst), 0);
    }

    #[test]
    fn prefetch_max_value() {
        let prefetch = AtomicU16::new(u16::MAX);
        assert_eq!(prefetch.load(Ordering::SeqCst), u16::MAX);
    }

    // --- Integration tests requiring RabbitMQ ---

    /// A test channel factory that connects on demand.
    struct TestConnectionFactory {
        uri: String,
    }

    #[async_trait]
    impl ChannelFactory for TestConnectionFactory {
        async fn create_channel(&self) -> Result<lapin::Channel, lapin::Error> {
            let conn =
                lapin::Connection::connect(&self.uri, lapin::ConnectionProperties::default())
                    .await?;
            conn.create_channel().await
        }
    }

    /// Helper to connect to RabbitMQ for integration tests.
    /// Returns None if RabbitMQ is not available.
    async fn test_setup() -> Option<(lapin::Connection, Arc<dyn ChannelFactory>)> {
        let uri = std::env::var("RABBITMQ_URL").unwrap_or_else(|_| "amqp://localhost:5672".into());
        let conn = lapin::Connection::connect(&uri, lapin::ConnectionProperties::default())
            .await
            .ok()?;
        let factory: Arc<dyn ChannelFactory> =
            Arc::new(TestConnectionFactory { uri });
        Some((conn, factory))
    }

    fn test_command(command_type: &str, queue_name: &str) -> PendingCommand {
        PendingCommand {
            command_id: Uuid::new_v4(),
            command_type: command_type.into(),
            queue_name: queue_name.into(),
            parameters: serde_json::Value::Object(serde_json::Map::new()),
            issued_at: Utc::now(),
        }
    }

    #[tokio::test]
    async fn unknown_command_returns_error() {
        let Some((_conn, factory)) = test_setup().await else {
            eprintln!("Skipping: RabbitMQ not available");
            return;
        };

        let executor = RabbitMqCommandExecutor::new(factory, 10);
        let command = test_command("nonexistent", "test-queue");
        let result = executor.execute(&command).await;

        assert!(!result.success);
        assert!(result.message.as_ref().unwrap().contains("not supported"));
    }

    #[tokio::test]
    async fn executor_original_prefetch_getter() {
        let Some((_conn, factory)) = test_setup().await else {
            eprintln!("Skipping: RabbitMQ not available");
            return;
        };

        let executor = RabbitMqCommandExecutor::new(factory, 42);
        assert_eq!(executor.original_prefetch(), 42);
    }

    #[tokio::test]
    async fn executor_set_original_prefetch() {
        let Some((_conn, factory)) = test_setup().await else {
            eprintln!("Skipping: RabbitMQ not available");
            return;
        };

        let executor = RabbitMqCommandExecutor::new(factory, 10);
        executor.set_original_prefetch(30);
        assert_eq!(executor.original_prefetch(), 30);
    }

    #[tokio::test]
    async fn pause_sets_qos_to_zero() {
        let Some((_conn, factory)) = test_setup().await else {
            eprintln!("Skipping: RabbitMQ not available");
            return;
        };

        let executor = RabbitMqCommandExecutor::new(factory, 10);
        let command = test_command("pause", "test-queue");
        let result = executor.execute(&command).await;

        assert!(result.success);
        assert!(result.message.as_ref().unwrap().contains("paused"));
        // Original prefetch should still be saved
        assert_eq!(executor.original_prefetch(), 10);
    }

    #[tokio::test]
    async fn resume_restores_original_prefetch() {
        let Some((_conn, factory)) = test_setup().await else {
            eprintln!("Skipping: RabbitMQ not available");
            return;
        };

        let executor = RabbitMqCommandExecutor::new(factory, 15);

        // Pause first
        let pause_cmd = test_command("pause", "test-queue");
        let pause_result = executor.execute(&pause_cmd).await;
        assert!(pause_result.success);

        // Resume
        let resume_cmd = test_command("resume", "test-queue");
        let resume_result = executor.execute(&resume_cmd).await;
        assert!(resume_result.success);
        assert!(resume_result
            .message
            .as_ref()
            .unwrap()
            .contains("restored to 15"));
    }

    #[tokio::test]
    async fn purge_on_declared_queue() {
        let Some((conn, factory)) = test_setup().await else {
            eprintln!("Skipping: RabbitMQ not available");
            return;
        };

        // Declare a temporary queue so purge has a valid target
        let queue_name = format!("queuerious-test-purge-{}", Uuid::new_v4());
        let cleanup_channel = conn.create_channel().await.expect("Failed to create channel");
        cleanup_channel
            .queue_declare(
                &queue_name,
                lapin::options::QueueDeclareOptions {
                    auto_delete: true,
                    ..Default::default()
                },
                lapin::types::FieldTable::default(),
            )
            .await
            .expect("Failed to declare test queue");

        let executor = RabbitMqCommandExecutor::new(factory, 10);
        let command = test_command("purge", &queue_name);
        let result = executor.execute(&command).await;

        assert!(result.success);
        assert!(result.message.as_ref().unwrap().contains("Purged"));

        // Clean up
        let _ = cleanup_channel
            .queue_delete(&queue_name, lapin::options::QueueDeleteOptions::default())
            .await;
    }

    #[tokio::test]
    async fn retry_publishes_message() {
        let Some((conn, factory)) = test_setup().await else {
            eprintln!("Skipping: RabbitMQ not available");
            return;
        };

        // Declare a temporary queue
        let queue_name = format!("queuerious-test-retry-{}", Uuid::new_v4());
        let cleanup_channel = conn.create_channel().await.expect("Failed to create channel");
        cleanup_channel
            .queue_declare(
                &queue_name,
                lapin::options::QueueDeclareOptions {
                    auto_delete: true,
                    ..Default::default()
                },
                lapin::types::FieldTable::default(),
            )
            .await
            .expect("Failed to declare test queue");

        let executor = RabbitMqCommandExecutor::new(factory, 10);
        let mut command = test_command("retry", &queue_name);
        command.parameters = serde_json::json!({ "payload": "retry body" });

        let result = executor.execute(&command).await;
        assert!(result.success);
        assert!(result.message.as_ref().unwrap().contains("re-published"));

        // Clean up
        let _ = cleanup_channel
            .queue_delete(&queue_name, lapin::options::QueueDeleteOptions::default())
            .await;
    }

    #[tokio::test]
    async fn purge_nonexistent_queue_returns_error() {
        let Some((_conn, factory)) = test_setup().await else {
            eprintln!("Skipping: RabbitMQ not available");
            return;
        };

        let executor = RabbitMqCommandExecutor::new(factory, 10);
        let command = test_command("purge", "nonexistent-queue-that-does-not-exist");
        let result = executor.execute(&command).await;

        assert!(!result.success);
        assert!(result.message.is_some());
    }

    #[tokio::test]
    async fn retry_with_empty_payload() {
        let Some((conn, factory)) = test_setup().await else {
            eprintln!("Skipping: RabbitMQ not available");
            return;
        };

        // Declare a temporary queue
        let queue_name = format!("queuerious-test-retry-empty-{}", Uuid::new_v4());
        let cleanup_channel = conn.create_channel().await.expect("Failed to create channel");
        cleanup_channel
            .queue_declare(
                &queue_name,
                lapin::options::QueueDeclareOptions {
                    auto_delete: true,
                    ..Default::default()
                },
                lapin::types::FieldTable::default(),
            )
            .await
            .expect("Failed to declare test queue");

        let executor = RabbitMqCommandExecutor::new(factory, 10);
        // No payload in parameters
        let command = test_command("retry", &queue_name);
        let result = executor.execute(&command).await;

        assert!(result.success);

        // Clean up
        let _ = cleanup_channel
            .queue_delete(&queue_name, lapin::options::QueueDeleteOptions::default())
            .await;
    }
}
