// Authors: Robert Lopez
// License: MIT (See `LICENSE.md`)

pub use lapin;
mod test;

use lapin::{
    options::{
        BasicAckOptions, BasicGetOptions, BasicNackOptions, BasicPublishOptions,
        BasicRecoverAsyncOptions, ExchangeDeclareOptions, QueueBindOptions, QueueDeclareOptions,
        QueueDeleteOptions, QueuePurgeOptions,
    },
    types::{FieldTable, MessageCount},
    BasicProperties, Channel, Connection, ConnectionProperties, Error, ExchangeKind,
};
use std::collections::HashMap;

/// Result for getting a single message from a `Queue`
pub struct GetMessageResult {
    pub message: Vec<u8>,
    pub delivery_tag: u64,
}

/// Represents a single `Queue`
#[derive(Clone)]
pub struct Queue {
    pub name: String,
    pub channel: Channel,
    pub exchange: String,
    pub routing_key: String,
}

impl Queue {
    /// Construct a `Queue`
    ///
    /// Exchange is: `"{name}_exchange"`
    ///
    /// Routing key is: `"{name}_routing_key"`
    ///
    /// ---
    /// Example Usage:
    /// ```
    ///
    /// let channel: Channel = ...;
    ///
    /// let queue = Queue::new(
    ///     "my_queue",
    ///     channel,
    ///     None,
    ///     None,
    /// ).await?;
    /// ```
    pub async fn new(
        name: &str,
        channel: Channel,
        options: Option<QueueDeclareOptions>,
        arguments: Option<FieldTable>,
    ) -> Result<Self, Error> {
        channel
            .queue_declare(
                name,
                options.unwrap_or_default(),
                arguments.unwrap_or_default(),
            )
            .await?;

        let exchange = format!("{}_exchange", name);
        let routing_key = format!("{}_routing_key", name);

        channel
            .exchange_declare(
                &exchange,
                ExchangeKind::Direct,
                ExchangeDeclareOptions {
                    durable: true,
                    ..Default::default()
                },
                FieldTable::default(),
            )
            .await?;

        channel
            .queue_bind(
                name,
                &exchange,
                &routing_key,
                QueueBindOptions::default(),
                FieldTable::default(),
            )
            .await?;

        Ok(Self {
            name: name.to_string(),
            channel,
            exchange,
            routing_key,
        })
    }

    /// Purge the `Queue`, returns the amount of
    /// messages deleted.
    ///
    /// ---
    /// Example Usage:
    /// ```
    ///
    /// let queue: Queue = ...;
    ///
    /// let deleted_count: MessageCount = queue.purge(false).await?;
    /// ```
    pub async fn purge(&self, no_wait: bool) -> Result<MessageCount, Error> {
        self.channel
            .queue_purge(&self.name, QueuePurgeOptions { nowait: no_wait })
            .await
    }

    /// Recover the `Queue`.
    ///
    /// ---
    /// Example Usage:
    /// ```
    ///
    /// let queue: Queue = ...;
    ///
    /// queue.recover(false).await?;
    /// ```
    pub async fn recover(&self, requeue: bool) -> Result<(), Error> {
        self.channel
            .basic_recover_async(BasicRecoverAsyncOptions { requeue })
            .await
    }

    /// Delete the `Queue`, returns the amount of
    /// messages deleted.
    ///
    /// ---
    /// Example Usage:
    /// ```
    ///
    /// let queue: Queue = ...;
    ///
    /// let deleted_count: MessageCount = queue.purge(None).await?;
    /// ```
    pub async fn delete(&self, options: Option<QueueDeleteOptions>) -> Result<MessageCount, Error> {
        self.channel
            .queue_delete(&self.name, options.unwrap_or_default())
            .await
    }

    /// Publish a `message` to the `Queue`.
    ///
    /// ---
    /// Example Usage:
    /// ```
    ///
    /// let queue: Queue = ...;
    ///
    /// let message = b"1337".to_vec();
    ///
    /// queue.publish_message(message, None).await?;
    /// ```
    pub async fn publish_message(
        &self,
        message: Vec<u8>,
        properties: Option<BasicProperties>,
    ) -> Result<(), Error> {
        self.channel
            .basic_publish(
                &self.exchange,
                &self.routing_key,
                BasicPublishOptions {
                    mandatory: false,
                    immediate: false,
                },
                &message,
                properties.unwrap_or_default(),
            )
            .await?
            .await?;

        Ok(())
    }

    /// Get a single message from the `Queue`.
    ///
    /// Returns `None` if there was no messages.
    ///
    /// ---
    /// Example Usage:
    /// ```
    ///
    /// let queue: Queue = ...;
    ///
    /// while let Some(
    ///     GetMessageResult {
    ///        message,
    ///        delivery_tag,
    ///     }
    /// ) = queue.get_message.await? {
    ///     ...
    /// }
    /// ```
    pub async fn get_message(&self) -> Result<Option<GetMessageResult>, Error> {
        if let Some(delivery) = self
            .channel
            .basic_get(&self.name, BasicGetOptions { no_ack: false })
            .await?
        {
            return Ok(Some(GetMessageResult {
                message: delivery.data.to_vec(),
                delivery_tag: delivery.delivery_tag,
            }));
        }

        Ok(None)
    }

    /// Acknowledge a gotten message.
    ///
    /// ---
    /// Example Usage:
    /// ```
    ///
    /// let queue: Queue = ...;
    ///
    /// while let Some(
    ///     GetMessageResult {
    ///        message,
    ///        delivery_tag,
    ///     }
    /// ) = queue.get_message.await? {
    ///     queue.acknowledge_message(delivery_tag).await?;
    /// }
    /// ```
    pub async fn acknowledge_message(&self, delivery_tag: u64) -> Result<(), Error> {
        self.channel
            .basic_ack(delivery_tag, BasicAckOptions { multiple: false })
            .await
    }

    /// No Acknowledge a gotten message.
    ///
    /// Add `requeue` to `true`, to requeue
    /// the message.
    ///
    /// ---
    /// Example Usage:
    /// ```
    ///
    /// let queue: Queue = ...;
    ///
    /// while let Some(
    ///     GetMessageResult {
    ///        message,
    ///        delivery_tag,
    ///     }
    /// ) = queue.get_message.await? {
    ///     ... Some issue processing the message
    ///
    ///     queue.no_acknowledge_message(true, delivery_tag).await?;
    /// }
    /// ```
    pub async fn no_acknowledge_message(
        &self,
        requeue: bool,
        delivery_tag: u64,
    ) -> Result<(), Error> {
        self.channel
            .basic_nack(
                delivery_tag,
                BasicNackOptions {
                    requeue,
                    multiple: false,
                },
            )
            .await
    }
}

/// Simplified RabbitMQ client
///
/// ---
/// Example Usage:
/// ```
///
/// let mut rabbit_mq = RabbitMQ::new("amqp://guest:guest@localhost:5672/%2f", None).await?;
///
/// rabbit_mq.register_queue("test_queue", None, None).await?;
///
/// rabbit_mq.publish_message("test_queue", b"1337".to_vec(), None).await?;
///
/// if let Some(test_queue) = rabbit_mq.get_queue("test_queue") {
///    while let Some(GetMessageResult {
///         message,
///        delivery_tag,
///     }) = test_queue.get_message().await? {
///     ...
/// }
/// ```
pub struct RabbitMQ {
    pub queues: HashMap<String, Queue>,
    pub connection: Connection,
}

impl RabbitMQ {
    /// Construct the `RabbitMQ` client.
    ///
    /// ---
    /// Example Usage:
    /// ```
    ///
    /// let rabbit_mq = RabbitMQ::new("amqp://guest:guest@localhost:5672/", None).await?;
    /// ```
    pub async fn new(uri: &str, options: Option<ConnectionProperties>) -> Result<Self, Error> {
        let connection = Connection::connect(uri, options.unwrap_or_default()).await?;

        Ok(Self {
            queues: HashMap::new(),
            connection,
        })
    }

    /// Register a `Queue`.
    ///
    /// ---
    /// Example Usage:
    /// ```
    ///
    /// let mut rabbit_mq = ...;
    ///
    /// rabbit_mq.register_queue("my_queue", None, None).await?;
    /// ```
    pub async fn register_queue(
        &mut self,
        queue_name: &str,
        options: Option<QueueDeclareOptions>,
        arguments: Option<FieldTable>,
    ) -> Result<(), Error> {
        let channel = self.connection.create_channel().await?;

        let queue = Queue::new(queue_name, channel, options, arguments).await?;

        self.queues.insert(queue_name.to_string(), queue);

        Ok(())
    }

    /// Get a `&Queue`.
    ///
    /// Returns `None` if the `Queue` was not found.
    ///
    /// ---
    /// Example Usage:
    /// ```
    ///
    /// let mut rabbit_mq = ...;
    ///
    /// let queue: Option<&Queue> = rabbit_mq.get_queue("my_queue").await?;
    /// ```
    pub fn get_queue(&self, queue_name: &str) -> Option<&Queue> {
        if let Some(queue) = self.queues.get(queue_name) {
            return Some(queue);
        }

        None
    }

    /// Publish a message to a `Queue`.
    ///
    /// Returns `None` if the `Queue` was not found.
    ///
    /// ---
    /// Example Usage:
    /// ```
    ///
    /// let mut rabbit_mq = ...;
    ///
    /// let message = b"1337".to_vec();
    ///
    /// rabbit_mq.publish_message(
    ///     "my_queue",
    ///     &message,
    ///     None
    /// ).await?;
    /// ```
    pub async fn publish_message(
        &self,
        queue_name: &str,
        message: Vec<u8>,
        properties: Option<BasicProperties>,
    ) -> Result<Option<()>, Error> {
        if let Some(queue) = self.queues.get(queue_name) {
            queue.publish_message(message, properties).await?;

            return Ok(Some(()));
        }

        Ok(None)
    }

    /// Publish a batch messages to a `Queue`, unchecked.
    ///
    /// Meaning, does not handle errors, just attempts
    /// to post all messages in the list to the `Queue`.
    ///
    /// Returns `None` if the `Queue` was not found.
    ///
    /// ---
    /// Example Usage:
    /// ```
    ///
    /// let mut rabbit_mq = ...;
    ///
    /// let messages = vec![
    ///     ...
    /// ];
    ///
    /// rabbit_mq.publish_messages_unchecked(
    ///     "my_queue",
    ///     messages,
    ///     None
    /// ).await;
    /// ```
    pub async fn publish_messages_unchecked(
        &self,
        queue_name: &str,
        messages: Vec<Vec<u8>>,
        properties: Option<BasicProperties>,
    ) -> Option<()> {
        if let Some(queue) = self.queues.get(queue_name) {
            for message in messages {
                let _ = queue.publish_message(message, properties.clone()).await;
            }

            return Some(());
        }

        None
    }
}