use bytes::Bytes;
use rocketmq_model::common::message::message_single::Message;

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

#[derive(Clone, Debug)]
pub struct Publisher {
    conn: RocketMqConn,
}

impl Publisher {
    pub fn new(conn: RocketMqConn) -> Self {
        Self { conn }
    }

    pub fn conn(&self) -> &RocketMqConn {
        &self.conn
    }

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

    pub fn topic(&self, area: &str) -> String {
        self.conn.config().topic(area)
    }

    pub async fn publish(
        &self,
        topic: &str,
        tag: &str,
        keys: Vec<String>,
        properties: Vec<(String, String)>,
        body: Bytes,
    ) -> QueueResult<()> {
        let max_payload = self.max_payload();
        if body.len() > max_payload {
            return Err(QueueError::payload(format!(
                "event body for `{topic}` is {} bytes, exceeding the {max_payload}-byte limit",
                body.len()
            )));
        }

        let mut builder = Message::builder()
            .topic(topic)
            .tags(tag)
            .keys(keys)
            .body(body.to_vec());
        for (key, value) in properties {
            builder = builder.raw_property(key.clone(), value).map_err(|error| {
                QueueError::publish(format!(
                    "invalid message property `{key}` for `{topic}`: {error}"
                ))
            })?;
        }
        let message = builder.build().map_err(|error| {
            QueueError::publish(format!(
                "failed to build the message for `{topic}`: {error}"
            ))
        })?;

        self.conn.send(topic, message).await
    }
}