use std::path::PathBuf;
use std::sync::Arc;

use queue::{AutoOffsetReset, Queue, SubscribeHandle, SubscriptionMode};
use tracing::{debug, error, warn};

use crate::config::AppEmailSettings;
use crate::error::EmailError;
use crate::events::{
    EmailFailed, EmailSent, RenderAndSendEmailRequest, SendBatchEmailRequest, SendEmailRequest,
    now_ms,
};
use crate::mime::{MimeBuildOptions, build_message};
use crate::template::TemplateRegistry;
use crate::transport::EmailTransport;

pub struct EmailService {
    inner: Arc<EmailServiceInner>,
}

struct EmailServiceInner {
    transport: EmailTransport,
    templates: tokio::sync::RwLock<TemplateRegistry>,
    cfg: Arc<AppEmailSettings>,
    queue: Arc<Queue>,
}

#[derive(Debug)]
pub struct EmailSubscribers {
    pub single: SubscribeHandle,
    pub batch: SubscribeHandle,
    pub render: SubscribeHandle,
}

impl EmailSubscribers {
    pub async fn shutdown(self) -> Result<(), EmailError> {
        let (a, b, c) = tokio::join!(
            self.single.shutdown(),
            self.batch.shutdown(),
            self.render.shutdown(),
        );
        a?;
        b?;
        c?;
        Ok(())
    }
}

impl Clone for EmailService {
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone(),
        }
    }
}

impl EmailService {
    pub fn transport(&self) -> &EmailTransport {
        &self.inner.transport
    }

    pub async fn render_html(
        &self,
        id: &str,
        locale: Option<&str>,
        vars: &serde_json::Value,
    ) -> Result<String, EmailError> {
        self.inner
            .templates
            .read()
            .await
            .render_html(id, locale, vars)
    }

    pub async fn render_text(
        &self,
        id: &str,
        locale: Option<&str>,
        vars: &serde_json::Value,
    ) -> Result<String, EmailError> {
        self.inner
            .templates
            .read()
            .await
            .render_text(id, locale, vars)
    }

    pub async fn send(&self, mut req: SendEmailRequest) -> Result<EmailSent, EmailError> {
        if req.to.is_empty() {
            return Err(EmailError::Missing("to recipients".to_string()));
        }
        let opts = MimeBuildOptions {
            max_attachment_bytes: self.inner.cfg.max_attachment_bytes,
        };
        let message = build_message(&req, self.inner.transport.from_mailbox(), &opts)?;
        let message_id = self.inner.transport.send(message).await?;
        let total_recipients = req.to.len() + req.cc.len() + req.bcc.len();
        let sent = EmailSent {
            request_id: req.request_id.take(),
            message_id,
            recipient_count: total_recipients,
            metadata: req.metadata.clone(),
            sent_at_ms: now_ms(),
        };
        Ok(sent)
    }

    pub async fn send_batch(
        &self,
        req: SendBatchEmailRequest,
    ) -> Result<Vec<EmailSent>, EmailError> {
        if req.recipients.is_empty() {
            return Err(EmailError::Missing("recipients".to_string()));
        }
        let mut out = Vec::with_capacity(req.recipients.len());
        for recipient in req.recipients {
            let single = SendEmailRequest {
                to: vec![recipient.to],
                cc: req.cc.clone(),
                bcc: req.bcc.clone(),
                reply_to: req.reply_to.clone(),
                subject: req.subject.clone(),
                text: req.text.clone(),
                html: req.html.clone(),
                attachments: req.attachments.clone(),
                headers: req.headers.clone(),
                locale: req.locale.clone(),
                request_id: req.request_id.clone(),
                metadata: recipient.metadata,
            };
            match self.send(single).await {
                Ok(sent) => out.push(sent),
                Err(e) => {
                    error!(error = %e, "batch recipient send failed");
                    let failed = EmailFailed {
                        request_id: req.request_id.clone(),
                        error: e.to_string(),
                        recipient_count: 1,
                        metadata: std::collections::HashMap::new(),
                        failed_at_ms: now_ms(),
                    };
                    self.publish_failed(failed).await?;
                }
            }
        }
        Ok(out)
    }

    pub async fn render_and_send(
        &self,
        req: RenderAndSendEmailRequest,
    ) -> Result<EmailSent, EmailError> {
        let locale_str = req.locale.clone();
        let locale_borrow = locale_str.as_deref();
        let templates = self.inner.templates.read().await;
        let html = templates
            .render_html(&req.template_id, locale_borrow, &req.vars)
            .ok();
        let text = templates
            .render_text(&req.template_id, locale_borrow, &req.vars)
            .ok();
        drop(templates);
        if html.is_none() && text.is_none() {
            return Err(EmailError::TemplateNotFound(req.template_id));
        }
        let single = SendEmailRequest {
            to: vec![req.to],
            cc: req.cc,
            bcc: req.bcc,
            reply_to: req.reply_to,
            subject: req.subject,
            text,
            html,
            attachments: req.attachments,
            headers: vec![],
            locale: locale_str,
            request_id: req.request_id,
            metadata: req.metadata,
        };
        self.send(single).await
    }

    pub async fn publish_sent(&self, sent: EmailSent) -> Result<(), EmailError> {
        self.inner.queue.typed::<EmailSent>().publish(&sent).await?;
        Ok(())
    }

    pub async fn publish_failed(&self, failed: EmailFailed) -> Result<(), EmailError> {
        self.inner
            .queue
            .typed::<EmailFailed>()
            .publish(&failed)
            .await?;
        Ok(())
    }

    pub fn start_subscribers(&self) -> Result<EmailSubscribers, EmailError> {
        let mode = SubscriptionMode::Group {
            group_id: self.inner.cfg.subscriber_group_id.clone(),
            auto_offset_reset: parse_offset_reset(&self.inner.cfg.auto_offset_reset)?,
        };
        let svc_send = self.clone();
        let svc_batch = self.clone();
        let svc_render = self.clone();
        let single = self
            .inner
            .queue
            .typed::<SendEmailRequest>()
            .subscribe(mode.clone(), move |msg| {
                let svc = svc_send.clone();
                let req = msg.value;
                async move {
                    let request_id = req.request_id.clone();
                    let metadata = req.metadata.clone();
                    let recipient_count = req.to.len() + req.cc.len() + req.bcc.len();
                    match svc.send(req).await {
                        Ok(sent) => {
                            if let Err(e) = svc.publish_sent(sent).await {
                                warn!(error = %e, "publish EmailSent failed");
                            }
                        }
                        Err(e) => {
                            error!(error = %e, request_id = ?request_id, "email send failed after retries");
                            let failed = EmailFailed {
                                request_id,
                                error: e.to_string(),
                                recipient_count,
                                metadata,
                                failed_at_ms: now_ms(),
                            };
                            if let Err(pub_err) = svc.publish_failed(failed).await {
                                error!(error = %pub_err, "publish EmailFailed failed");
                            }
                        }
                    }
                    Ok(())
                }
            })?;
        let batch = self
            .inner
            .queue
            .typed::<SendBatchEmailRequest>()
            .subscribe(mode.clone(), move |msg| {
                let svc = svc_batch.clone();
                let req = msg.value;
                async move {
                    if let Err(e) = svc.send_batch(req).await {
                        error!(error = %e, "batch send failed after retries");
                    }
                    Ok(())
                }
            })?;
        let render = self
            .inner
            .queue
            .typed::<RenderAndSendEmailRequest>()
            .subscribe(mode, move |msg| {
                let svc = svc_render.clone();
                let req = msg.value;
                async move {
                    let request_id = req.request_id.clone();
                    let metadata = req.metadata.clone();
                    let recipient_count = 1 + req.cc.len() + req.bcc.len();
                    match svc.render_and_send(req).await {
                        Ok(sent) => {
                            if let Err(e) = svc.publish_sent(sent).await {
                                warn!(error = %e, "publish EmailSent failed");
                            }
                        }
                        Err(e) => {
                            error!(error = %e, request_id = ?request_id, "render_and_send failed after retries");
                            let failed = EmailFailed {
                                request_id,
                                error: e.to_string(),
                                recipient_count,
                                metadata,
                                failed_at_ms: now_ms(),
                            };
                            if let Err(pub_err) = svc.publish_failed(failed).await {
                                error!(error = %pub_err, "publish EmailFailed failed");
                            }
                        }
                    }
                    Ok(())
                }
            })?;
        debug!(group = %self.inner.cfg.subscriber_group_id, "email service subscribers started");
        Ok(EmailSubscribers {
            single,
            batch,
            render,
        })
    }

    pub async fn shutdown(&self) -> Result<(), EmailError> {
        Ok(())
    }
}

fn parse_offset_reset(value: &str) -> Result<AutoOffsetReset, EmailError> {
    match value {
        "earliest" => Ok(AutoOffsetReset::Earliest),
        "latest" => Ok(AutoOffsetReset::Latest),
        "none" => Ok(AutoOffsetReset::None),
        other => Err(EmailError::Config(format!(
            "unknown auto_offset_reset: {other}"
        ))),
    }
}

pub struct EmailServiceBuilder {
    cfg: AppEmailSettings,
    queue: Arc<Queue>,
    template_dir: Option<PathBuf>,
}

impl EmailServiceBuilder {
    pub fn new(cfg: AppEmailSettings, queue: Arc<Queue>) -> Self {
        Self {
            cfg,
            queue,
            template_dir: None,
        }
    }

    pub fn with_template_dir(mut self, dir: PathBuf) -> Self {
        self.template_dir = Some(dir);
        self
    }

    pub async fn build(self) -> Result<Option<EmailService>, EmailError> {
        if !self.cfg.enabled {
            return Ok(None);
        }
        self.cfg.is_valid().map_err(EmailError::Config)?;
        let mut templates = TemplateRegistry::new(self.cfg.default_locale.clone());
        if let Some(dir) = &self.template_dir {
            let loaded = templates.load_dir(dir)?;
            debug!(loaded, dir = %dir.display(), "email templates loaded");
        }
        let transport = EmailTransport::from_config(&self.cfg)?;
        Ok(Some(EmailService {
            inner: Arc::new(EmailServiceInner {
                transport,
                templates: tokio::sync::RwLock::new(templates),
                cfg: Arc::new(self.cfg),
                queue: self.queue,
            }),
        }))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::AppEmailSettings;

    fn cfg() -> AppEmailSettings {
        AppEmailSettings {
            enabled: true,
            smtp_port: 587,
            smtp_host: "smtp.example.com".into(),
            ..Default::default()
        }
    }

    #[test]
    fn parse_offset_reset_known() {
        assert!(matches!(
            parse_offset_reset("earliest"),
            Ok(AutoOffsetReset::Earliest)
        ));
        assert!(matches!(
            parse_offset_reset("latest"),
            Ok(AutoOffsetReset::Latest)
        ));
        assert!(parse_offset_reset("foo").is_err());
    }

    #[test]
    fn builder_returns_none_when_disabled() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let mut c = cfg();
        c.enabled = false;
        let queue_handle = rt
            .block_on(queue::QueueBuilder::from_config(queue::AppQueueSettings::default()).build())
            .unwrap()
            .expect("queue must build for test");
        let queue_arc = Arc::new(queue_handle);
        let res = rt
            .block_on(EmailServiceBuilder::new(c, queue_arc).build())
            .unwrap();
        assert!(res.is_none());
    }
}