use serde::Serialize;

use crate::EventKind;
use config::rocketmq::RocketMqConfig;
use queue::{
    EVENT_PREFIX, Envelope, Labels, Publisher, QueueError, QueueResult, QueueRuntime, RocketMqConn,
};

#[derive(Clone, Debug)]
pub struct EventHandler {
    publisher: Publisher,
}

impl EventHandler {
    pub fn new(mq: RocketMqConn) -> Self {
        Self {
            publisher: Publisher::new(mq),
        }
    }

    pub async fn connect(config: &RocketMqConfig, runtime: &QueueRuntime) -> QueueResult<Self> {
        let conn = RocketMqConn::init(config, runtime).await?;
        Ok(Self::new(conn))
    }

    pub async fn optional(
        config: Option<&RocketMqConfig>,
        runtime: &QueueRuntime,
    ) -> QueueResult<Option<Self>> {
        match config {
            Some(config) => Self::connect(config, runtime).await.map(Some),
            None => Ok(None),
        }
    }

    pub fn envelope<T>(&self, kind: EventKind, payload: T) -> Envelope<T> {
        Envelope::new(kind.as_str(), payload)
    }

    #[tracing::instrument(skip_all, fields(kind = %kind, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn send_labeled<T>(
        &self,
        kind: EventKind,
        labels: Labels,
        payload: T,
        request_id: Option<&str>,
    ) -> QueueResult<()>
    where
        T: Serialize,
    {
        self.send_keyed(kind, labels, None, payload, request_id)
            .await
    }

    #[tracing::instrument(skip_all, fields(kind = %kind, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn send_keyed<T>(
        &self,
        kind: EventKind,
        labels: Labels,
        idempotency_key: Option<String>,
        payload: T,
        request_id: Option<&str>,
    ) -> QueueResult<()>
    where
        T: Serialize,
    {
        let mut envelope = self
            .envelope(kind, payload)
            .with_source(kind.domain())
            .with_labels(labels);
        if let Some(key) = idempotency_key {
            envelope = envelope.with_idempotency_key(key);
        }
        if let Some(trace_id) = envelope::context::request_id().or_else(|| {
            request_id
                .filter(|request_id| !request_id.is_empty())
                .map(str::to_owned)
        }) {
            envelope = envelope.with_trace_id(trace_id);
        }
        self.send_envelope(&envelope, request_id).await
    }

    #[tracing::instrument(skip_all, fields(request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn send_envelope<T>(
        &self,
        envelope: &Envelope<T>,
        request_id: Option<&str>,
    ) -> QueueResult<()>
    where
        T: Serialize,
    {
        let subject = format!("{EVENT_PREFIX}{}", envelope.kind);
        envelope.verify_subject(EVENT_PREFIX, &subject)?;
        self.dispatch(envelope, request_id).await
    }

    #[tracing::instrument(skip_all, fields(subject = %subject, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn publish_durable<T: Serialize>(
        &self,
        subject: &str,
        envelope: &Envelope<T>,
        request_id: Option<&str>,
    ) -> QueueResult<()> {
        envelope.verify_event_subject(subject)?;
        self.dispatch(envelope, request_id).await
    }

    pub async fn shutdown(&self) -> QueueResult<()> {
        self.publisher.conn().shutdown().await;
        Ok(())
    }

    #[tracing::instrument(skip_all, fields(kind = %envelope.kind, request_id = %envelope::context::resolve_request_id(request_id)))]
    async fn dispatch<T: Serialize>(
        &self,
        envelope: &Envelope<T>,
        request_id: Option<&str>,
    ) -> QueueResult<()> {
        let kind = concrete_kind(&envelope.kind)?;
        let payload = serde_json::to_value(&envelope.payload).map_err(QueueError::from)?;
        let tag = kind.tag(Some(&payload));
        let request_id = envelope::context::request_id().or_else(|| {
            request_id
                .filter(|request_id| !request_id.is_empty())
                .map(str::to_owned)
        });
        let properties = envelope.properties(Some(&payload), request_id.as_deref());
        let topic = self.publisher.topic(kind.topic());
        let body = envelope.encode()?;

        self.publisher
            .publish(&topic, tag, vec![envelope.message_id()], properties, body)
            .await
    }
}

pub(crate) fn concrete_kind(kind: &str) -> QueueResult<EventKind> {
    EventKind::parse(kind)
        .ok_or_else(|| QueueError::payload(format!("event kind `{kind}` is not a declared event")))
}