use std::sync::Arc;

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

use crate::error::{QueueError, QueueResult};

#[derive(Clone)]
pub struct RocketMqConn {
    producer: Arc<Mutex<DefaultMQProducer>>,
    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.is_valid().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(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<()> {
        let mut producer = self.producer.lock().await;
        let result = producer
            .send_with_timeout(message, self.config.send_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(())
    }

    pub async fn shutdown(&self) {
        let mut producer = self.producer.lock().await;
        producer.shutdown().await;
    }
}