use std::sync::Arc;

use config::email::EmailPoolConfig;
use config::rocketmq::RocketMqConfig;
use config::smtp::SmtpConfig;
use events::{EventHandler, Labels, label};
use lettre::message::Mailbox;
use queue::{Consumer, ConsumerConfig, MessageHandler};
use serde::Serialize;
use tokio_util::sync::CancellationToken;
use tracing::{Instrument, info, warn};
use uuid::Uuid;

use crate::error::EmailError;
use crate::events::{EmailFailed, EmailSent, SendEmailRequest, now_ms};
use crate::transport::{EmailTransport, EmailTransportBuilder, build_message};

const EVENT_PUBLISH_ATTEMPTS: usize = 3;

async fn publish_event<T: Serialize>(
    events: &EventHandler,
    kind: &str,
    labels: Labels,
    payload: &T,
    request_id: &str,
) -> bool {
    let mut attempt = 1_usize;
    loop {
        match events
            .send_labeled(kind, labels.clone(), payload, Some(request_id))
            .await
        {
            Ok(()) => return true,
            Err(error) if attempt >= EVENT_PUBLISH_ATTEMPTS => {
                tracing::error!(kind, attempt, %error, "failed to publish the email event");
                return false;
            }
            Err(error) => {
                tracing::warn!(kind, attempt, %error, "retrying the email event publish");
                tokio::time::sleep(std::time::Duration::from_millis(50) * attempt as u32).await;
                attempt += 1;
            }
        }
    }
}

pub struct EmailConsumer {
    transport: EmailTransport,
    from: Mailbox,
    max_attachment_bytes: usize,
    pool: EmailPoolConfig,
    events: Option<EventHandler>,
}

impl EmailConsumer {
    pub fn new(
        smtp: SmtpConfig,
        pool: EmailPoolConfig,
        events: Option<EventHandler>,
    ) -> Result<Self, EmailError> {
        smtp.is_valid().map_err(EmailError::InvalidConfig)?;
        pool.is_valid().map_err(EmailError::InvalidConfig)?;
        let from = smtp.from.trim().parse::<Mailbox>()?;
        let transport = EmailTransportBuilder::new(
            smtp.host.clone(),
            smtp.port,
            smtp.security,
            smtp.username.clone(),
            smtp.password.clone(),
        )
        .connection_timeout(smtp.connect_timeout())
        .build()?;
        Ok(Self {
            transport,
            from,
            max_attachment_bytes: smtp.max_attachment_bytes,
            pool,
            events,
        })
    }

    pub async fn run(
        self,
        rocketmq: RocketMqConfig,
        cancel: CancellationToken,
    ) -> Result<(), EmailError> {
        let topic = self.pool.mail_topic.clone();
        let group = self.pool.consumer_group.clone();
        let workers = self.pool.workers.max(1) as u32;
        let max_reconsume = self.pool.max_reconsume.min(16) as i32;
        let consumer = Arc::new(self);
        let handler_consumer = consumer.clone();
        let handle = tokio::runtime::Handle::current();
        let handler: MessageHandler = Arc::new(move |messages| {
            let mut success = true;
            for message in messages {
                let request: SendEmailRequest = match message
                    .body()
                    .ok_or_else(|| EmailError::Missing("message body is empty".into()))
                    .and_then(|body| Ok(serde_json::from_slice(&body)?))
                {
                    Ok(request) => request,
                    Err(error) => {
                        warn!(error = %error, "mail message rejected");
                        success = false;
                        continue;
                    }
                };
                let request_id = request
                    .trace
                    .as_deref()
                    .filter(|value| !value.is_empty())
                    .map(str::to_owned)
                    .unwrap_or_else(|| request.request_id.to_string());
                let span = tracing::info_span!("email.delivery", request_id = %request_id);
                match handle.block_on(handler_consumer.deliver(request.clone()).instrument(span)) {
                    Ok(sent) => handle.block_on(handler_consumer.emit_sent(&sent, &request_id)),
                    Err(error) => {
                        success = false;
                        handle.block_on(handler_consumer.emit_failed(
                            &request,
                            &error,
                            &request_id,
                        ));
                    }
                }
            }
            success
        });
        let mut consumer = Consumer::start(
            ConsumerConfig {
                name_server: rocketmq.name_server,
                consumer_group: group.clone(),
                topic: topic.clone(),
                tag: "send".to_owned(),
                workers,
                max_reconsume,
            },
            handler,
        )
        .await
        .map_err(EmailError::from)?;
        info!(
            topic,
            group, workers, max_reconsume, "mail consumer started"
        );
        cancel.cancelled().await;
        consumer.shutdown().await;
        info!("mail consumer stopped");
        Ok(())
    }

    async fn deliver(&self, mut request: SendEmailRequest) -> Result<EmailSent, EmailError> {
        if request.request_id.is_nil() {
            request.request_id = Uuid::now_v7();
        }
        let recipient_count = request.to.len() + request.cc.len() + request.bcc.len();
        if recipient_count == 0 {
            return Err(EmailError::Missing("email has no recipients".into()));
        }
        let message = build_message(&request, self.from.clone(), self.max_attachment_bytes)?;
        let request_id = request.request_id.to_string();
        let smtp_message_id = self.transport.send(message, Some(&request_id)).await?;
        Ok(EmailSent {
            request_id: request.request_id,
            recipient_count,
            smtp_message_id,
            sent_at_ms: now_ms(),
            metadata: request.metadata,
        })
    }

    async fn emit_sent(&self, sent: &EmailSent, request_id: &str) {
        let Some(events) = self.events.as_ref() else {
            return;
        };
        let labels = Labels::new().with(label::REQUEST_ID, request_id);
        let _ = publish_event(
            events,
            label::EMAIL_DELIVERY_CREATE,
            labels,
            sent,
            request_id,
        )
        .await;
    }

    async fn emit_failed(&self, request: &SendEmailRequest, error: &EmailError, request_id: &str) {
        let Some(events) = self.events.as_ref() else {
            return;
        };
        let failed = EmailFailed {
            request_id: request.request_id,
            error: error.to_string(),
            recipient_count: request.to.len() + request.cc.len() + request.bcc.len(),
            failed_at_ms: now_ms(),
            metadata: request.metadata.clone(),
        };
        let labels = Labels::new().with(label::REQUEST_ID, request_id);
        let _ = publish_event(
            events,
            label::EMAIL_FAILURE_CREATE,
            labels,
            &failed,
            request_id,
        )
        .await;
    }
}