gitdataai/libs/email/lib.rs

use config::AppConfig;
use lettre::message::Mailbox;
use lettre::transport::smtp::{PoolConfig, SmtpTransport};
use lettre::Transport;
use metrics::counter;
use regex::Regex;
use serde::{Deserialize, Serialize};
use std::sync::LazyLock;
use std::time::Duration;
use tokio::sync::mpsc;

#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct EmailMessage {
    pub to: String,
    pub subject: String,
    pub body: String,
}

static EMAIL_REGEX: LazyLock<Regex> =
    LazyLock::new(|| Regex::new(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$").unwrap());

#[derive(Clone)]
pub struct AppEmail {
    sender: mpsc::Sender<EmailMessage>,
}

impl AppEmail {
    pub async fn init(cfg: &AppConfig) -> anyhow::Result<Self> {
        let smtp_host = cfg.smtp_host()?;
        let smtp_port = cfg.smtp_port()?;
        let smtp_username = cfg.smtp_username()?;
        let smtp_password = cfg.smtp_password()?;
        let smtp_from = cfg.smtp_from()?;
        let smtp_tls = cfg.smtp_tls()?;
        let smtp_timeout = cfg.smtp_timeout()?;

        // Port 465 = SMTPS (implicit TLS via smtps://), others = STARTTLS via smtp://
        let url = if smtp_port == 465 && smtp_tls {
            format!(
                "smtps://{}:{}@{}:{}",
                smtp_username, smtp_password, smtp_host, smtp_port
            )
        } else {
            let tls_mode = if smtp_tls {
                "required"
            } else {
                "opportunistic"
            };
            format!(
                "smtp://{}:{}@{}:{}?tls={}",
                smtp_username, smtp_password, smtp_host, smtp_port, tls_mode
            )
        };

        let mailer = SmtpTransport::from_url(&url)
            .map_err(|e| anyhow::anyhow!("SMTP transport build error: {}", e))?
            .timeout(Some(Duration::from_secs(smtp_timeout)))
            .pool_config(PoolConfig::new().min_idle(0).max_size(10))
            .build();

        let from: Mailbox = smtp_from.parse()?;

        let (tx, mut rx) = mpsc::channel::<EmailMessage>(100);

        tokio::spawn(async move {
            while let Some(msg) = rx.recv().await {
                if !EMAIL_REGEX.is_match(&msg.clone().to) {
                    counter!("email_validation_skipped_total").increment(1);
                    continue;
                }

                let email = match lettre::Message::builder()
                    .from(from.clone())
                    .to(msg.clone().to.parse().unwrap())
                    .subject(msg.clone().subject)
                    .body(msg.clone().body)
                {
                    Ok(e) => e,
                    Err(_) => {
                        counter!("email_build_errors_total").increment(1);
                        tracing::warn!(to = %msg.to, "Email build error");
                        continue;
                    }
                };
                let mut success = false;
                for i in 0..3 {
                    counter!("email_send_attempts_total").increment(1);
                    let mailer = mailer.clone();
                    let email = email.clone();
                    let result = tokio::task::spawn_blocking(move || mailer.send(&email)).await;

                    match result {
                        Ok(Ok(_)) => {
                            success = true;
                            break;
                        }
                        Ok(Err(e)) => {
                            if i == 2 {
                                counter!("email_send_failures_total").increment(1);
                                tracing::error!(to = %msg.to, error = %e, "Email send failed after retries");
                            }
                            tokio::time::sleep(Duration::from_secs((1 << i) as u64)).await;
                        }
                        Err(e) => {
                            tracing::error!(to = %msg.to, error = %e, "Email spawn error");
                            break;
                        }
                    }
                }

                if success {
                    counter!("email_sent_total").increment(1);
                }
            }
        });

        Ok(Self { sender: tx })
    }

    pub async fn send(&self, msg: EmailMessage) -> anyhow::Result<()> {
        self.sender
            .send(msg)
            .await
            .map_err(|e| anyhow::anyhow!("queue send error: {}", e))
    }
}