gitdataai/libs/email/lib.rs

127 lines
4.2 KiB
Rust

use config::AppConfig;
use lettre::message::Mailbox;
use lettre::transport::smtp::{PoolConfig, SmtpTransport};
use lettre::Transport;
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, logs: slog::Logger) -> 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) {
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(_) => {
slog::warn!(logs, "Email build error: to={}", msg.to);
continue;
}
};
let mut success = false;
for i in 0..3 {
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 {
slog::error!(
logs,
"Email send failed after retries: to={}, error={}",
msg.to,
e
);
}
tokio::time::sleep(Duration::from_secs((1 << i) as u64)).await;
}
Err(e) => {
slog::error!(logs, "Email spawn error: to={}, err={}", msg.to, e);
break;
}
}
}
if !success {
slog::warn!(logs, "Email send permanently failed: to={}", msg.to);
}
}
});
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))
}
}