use std::sync::Arc;

use rocketmq_client_rust::{
    ConsumeConcurrentlyContext, ConsumeConcurrentlyStatus, DefaultMQPushConsumer, MQPushConsumer,
};
use rocketmq_model::common::message::message_ext::MessageExt;

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

pub type MessageHandler = Arc<dyn Fn(&[&MessageExt]) -> bool + Send + Sync>;

#[derive(Clone, Debug)]
pub struct ConsumerConfig {
    pub name_server: String,
    pub consumer_group: String,
    pub topic: String,
    pub tag: String,
    pub workers: u32,
    pub max_reconsume: i32,
}

pub struct Consumer {
    inner: DefaultMQPushConsumer,
}

impl Consumer {
    pub async fn start(
        config: ConsumerConfig,
        runtime: &QueueRuntime,
        handler: MessageHandler,
    ) -> QueueResult<Self> {
        if config.name_server.trim().is_empty() {
            return Err(QueueError::config("rocketmq.name_server must not be blank"));
        }
        if config.consumer_group.trim().is_empty() {
            return Err(QueueError::config("consumer group must not be blank"));
        }
        if config.topic.trim().is_empty() {
            return Err(QueueError::config("consumer topic must not be blank"));
        }
        let mut consumer = DefaultMQPushConsumer::builder(runtime.client())
            .consumer_group(config.consumer_group)
            .name_server_addr(config.name_server.replace(',', ";"))
            .consume_thread_min(config.workers.max(1))
            .consume_thread_max(config.workers.max(1))
            .consume_message_batch_max_size(1)
            .max_reconsume_times(config.max_reconsume.max(1))
            .build();
        consumer
            .subscribe(config.topic, config.tag)
            .await
            .map_err(|error| {
                QueueError::connect(format!("subscribing to rocketmq failed: {error}"))
            })?;
        let listener = move |messages: &[&MessageExt], _context: &ConsumeConcurrentlyContext| {
            if handler(messages) {
                Ok(ConsumeConcurrentlyStatus::ConsumeSuccess)
            } else {
                Ok(ConsumeConcurrentlyStatus::ReconsumeLater)
            }
        };
        consumer.register_message_listener_concurrently(listener);
        consumer.start().await.map_err(|error| {
            QueueError::connect(format!("starting the rocketmq consumer failed: {error}"))
        })?;
        Ok(Self { inner: consumer })
    }

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