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 tokio_util::sync::CancellationToken;
use tracing::{info, warn};
use uuid::Uuid;

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

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;
                    }
                };
                match handle.block_on(handler_consumer.deliver(request.clone())) {
                    Ok(sent) => handle.block_on(handler_consumer.emit_sent(&sent)),
                    Err(error) => {
                        success = false;
                        handle.block_on(handler_consumer.emit_failed(&request, &error));
                    }
                }
            }
            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) {
        let Some(events) = self.events.as_ref() else {
            return;
        };
        let request_id = sent.request_id.to_string();
        let labels = Labels::new().with(label::REQUEST_ID, &request_id);
        if let Err(error) = events
            .send_labeled(
                label::EMAIL_DELIVERY_CREATE,
                labels,
                sent,
                Some(request_id.as_str()),
            )
            .await
        {
            warn!(error = %error, "failed to publish the email delivery event");
        }
    }

    async fn emit_failed(&self, request: &SendEmailRequest, error: &EmailError) {
        let Some(events) = self.events.as_ref() else {
            return;
        };
        let request_id = request.request_id.to_string();
        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);
        if let Err(publish_error) = events
            .send_labeled(
                label::EMAIL_FAILURE_CREATE,
                labels,
                &failed,
                Some(request_id.as_str()),
            )
            .await
        {
            warn!(error = %publish_error, "failed to publish the email failure event");
        }
    }
}