use std::sync::Arc;

use rocketmq_client_rust::producer::default_mq_producer::DefaultMQProducer;
use rocketmq_common::common::message::message_single::Message;
use tokio::sync::Mutex;

use crate::identity::error::{QueueError, QueueResult};
use config::identity::rocketmq::RocketMqConfig;

pub const DEFAULT_MAX_MESSAGE_BYTES: usize = 4 * 1024 * 1024;

#[derive(Clone)]
pub struct RocketMqConn {
    producer: Arc<dyn Producer>,
    config: RocketMqConfig,
}

impl std::fmt::Debug for RocketMqConn {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("RocketMqConn")
            .field("name_server", &self.config.name_server)
            .field("producer_group", &self.config.producer_group)
            .field("topic_prefix", &self.config.topic_prefix)
            .finish()
    }
}

impl RocketMqConn {
    pub async fn init(config: &RocketMqConfig) -> QueueResult<Self> {
        config.validate().map_err(QueueError::config)?;

        let mut producer = DefaultMQProducer::builder()
            .producer_group(config.producer_group.clone())
            .name_server_addr(config.name_server.replace(',', ";"))
            .build();
        producer.start().await.map_err(|error| {
            QueueError::connect(format!("failed to start the rocketmq producer: {error}"))
        })?;

        Ok(Self {
            producer: Arc::new(RocketProducer(Mutex::new(producer))),
            config: config.clone(),
        })
    }

    pub fn config(&self) -> &RocketMqConfig {
        &self.config
    }

    pub fn max_payload(&self) -> usize {
        self.config.max_message_bytes
    }

    pub async fn send(&self, topic: &str, message: Message) -> QueueResult<()> {
        self.producer
            .send(topic, message, self.config.send_timeout_ms)
            .await
    }

    pub async fn health(&self, topic: &str) -> QueueResult<()> {
        self.producer.health(topic).await
    }

    pub async fn shutdown(&self) {
        self.producer.shutdown().await;
    }
}

#[async_trait::async_trait]
trait Producer: Send + Sync {
    async fn send(&self, topic: &str, message: Message, timeout_ms: u64) -> QueueResult<()>;
    async fn health(&self, topic: &str) -> QueueResult<()>;
    async fn shutdown(&self);
}

struct RocketProducer(Mutex<DefaultMQProducer>);

#[async_trait::async_trait]
impl Producer for RocketProducer {
    async fn send(&self, topic: &str, message: Message, timeout_ms: u64) -> QueueResult<()> {
        let mut producer = self.0.lock().await;
        let result = producer
            .send_with_timeout(message, timeout_ms)
            .await
            .map_err(|error| {
                QueueError::publish(format!("failed to publish to `{topic}`: {error}"))
            })?;
        if result.is_none() {
            return Err(QueueError::publish(format!(
                "producer returned no send result for `{topic}`"
            )));
        }
        Ok(())
    }

    async fn health(&self, topic: &str) -> QueueResult<()> {
        let mut producer = self.0.lock().await;
        producer
            .fetch_publish_message_queues(topic)
            .await
            .map(|_| ())
            .map_err(|error| {
                QueueError::connect(format!(
                    "rocketmq route lookup for `{topic}` failed: {error}"
                ))
            })
    }

    async fn shutdown(&self) {
        self.0.lock().await.shutdown().await;
    }
}