use serde::Serialize;
use serde_json::Value;

use crate::event::{EVENT_PREFIX, RuntimeEvent};
use config::identity::rocketmq::RocketMqConfig;
use queue::identity::{Envelope, Labels, Publisher, QueueError, QueueResult, RocketMqConn};

pub const DURABLE_DOMAINS: &[&str] = &[
    "auth",
    "email",
    "guild",
    "chat",
    "git",
    "issue",
    "organization",
    "pull_request",
];

#[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) -> QueueResult<Self> {
        let conn = RocketMqConn::init(config).await?;
        Ok(Self::new(conn))
    }

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

    pub fn publisher(&self) -> &Publisher {
        &self.publisher
    }

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

    pub fn envelope<T>(&self, event: RuntimeEvent, payload: T) -> Envelope<T> {
        Envelope::new(event.kind(), payload)
            .with_source(event.domain())
            .with_delivery(event.delivery())
    }

    #[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
    pub async fn send<T>(
        &self,
        event: RuntimeEvent,
        payload: T,
        request_id: Option<&str>,
    ) -> QueueResult<()>
    where
        T: Serialize,
    {
        self.send_labeled(event, Labels::default(), payload, request_id)
            .await
    }

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

    #[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
    pub async fn send_keyed<T>(
        &self,
        event: RuntimeEvent,
        labels: Labels,
        idempotency_key: Option<String>,
        payload: T,
        request_id: Option<&str>,
    ) -> QueueResult<()>
    where
        T: Serialize,
    {
        let mut envelope = self.envelope(event, payload).with_labels(labels);
        if let Some(key) = idempotency_key {
            envelope = envelope.with_idempotency_key(key);
        }
        self.send_envelope(event, &envelope, request_id).await
    }

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

    #[tracing::instrument(skip_all, fields(subject = %subject, request_id = %request_id.unwrap_or_default()))]
    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 health(&self) -> QueueResult<()> {
        let topic = self.publisher.topic("login");
        self.publisher.conn().health(&topic).await
    }

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

    #[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
    async fn dispatch<T: Serialize>(
        &self,
        envelope: &Envelope<T>,
        request_id: Option<&str>,
    ) -> QueueResult<()> {
        let payload = serde_json::to_value(&envelope.payload).map_err(QueueError::from)?;
        let action = payload.get("action").and_then(Value::as_str);
        let route = crate::route::route(&envelope.kind, action, Some(&payload))?;
        let request_id = envelope::context::request_id();
        let properties = envelope.properties(Some(&payload), request_id.as_deref());
        let topic = self.publisher.topic(&route.area);
        let body = envelope.summary_body(&route.tag)?;

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

const PUBLISH_ATTEMPTS: usize = 3;
const PUBLISH_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
const PUBLISH_BACKOFF: std::time::Duration = std::time::Duration::from_millis(50);

/// 统一的事件转发重试：单次尝试带超时，失败按退避重试，
/// 最终仍失败打 ERROR，便于告警发现事件链断了。
/// `context` 仅用于日志文案（如 "auth" / "email"）。
#[tracing::instrument(skip_all, fields(event = %event, context = context, request_id = %_request_id.unwrap_or_default()))]
pub async fn retry_publish<T>(
    events: &EventHandler,
    event: RuntimeEvent,
    labels: Labels,
    payload: T,
    context: &str,
    _request_id: Option<&str>,
) where
    T: Serialize + Clone,
{
    for attempt in 1..=PUBLISH_ATTEMPTS {
        let published = events.send_labeled(event, labels.clone(), payload.clone(), _request_id);
        match tokio::time::timeout(PUBLISH_TIMEOUT, published).await {
            Ok(Ok(())) => {
                if attempt > 1 {
                    tracing::info!(event = %event, attempt, "{context} event forwarded after retry");
                }
                return;
            }
            Ok(Err(error)) => {
                tracing::warn!(
                    event = %event,
                    subject = %event.subject(),
                    attempt,
                    %error,
                    "failed to publish {context} event"
                );
            }
            Err(_) => {
                tracing::warn!(event = %event, attempt, "publishing {context} event timed out");
            }
        }
        if attempt < PUBLISH_ATTEMPTS {
            tokio::time::sleep(PUBLISH_BACKOFF * attempt as u32).await;
        }
    }
    tracing::error!(
        event = %event,
        subject = %event.subject(),
        attempts = PUBLISH_ATTEMPTS,
        "{context} event could not be forwarded to rocketmq"
    );
}