d3d11f4
use std::sync::Arc;
use lettre::address::Address;
use lettre::message::Mailbox;
use tracing::{debug, info, trace, warn};
use uuid::Uuid;
use crate::error::EmailError;
use crate::events::{
CHANNEL_BATCH, CHANNEL_FAILED, CHANNEL_RENDER, CHANNEL_SEND, CHANNEL_SENT, EmailAddress,
EmailFailed, EmailSent, RenderAndSendEmailRequest, SendBatchEmailRequest, SendEmailRequest,
now_ms,
};
use crate::queue::EmailQueue;
use crate::queue::QueueSettings;
use crate::template::TemplateRegistry;
use crate::templates;
use crate::transport::{EmailTransport, EmailTransportBuilder, build_message};
#[derive(Clone)]
pub struct EmailService {
inner: Arc<EmailServiceInner>,
}
struct EmailServiceInner {
config: Arc<config::email::EmailConfig>,
transport: Option<EmailTransport>,
from: Option<Mailbox>,
queue: Option<EmailQueue>,
templates: TemplateRegistry,
max_attachment_bytes: usize,
}
impl EmailService {
pub fn enabled(&self) -> bool {
self.inner.transport.is_some()
}
pub fn config(&self) -> &config::email::EmailConfig {
&self.inner.config
}
pub fn templates(&self) -> &TemplateRegistry {
&self.inner.templates
}
pub fn queue(&self) -> Option<&EmailQueue> {
self.inner.queue.as_ref()
}
pub async fn render(&self, name: &str, vars: &serde_json::Value) -> Result<String, EmailError> {
self.inner.templates.render(name, vars).await
}
pub async fn send(&self, mut req: SendEmailRequest) -> Result<EmailSent, EmailError> {
let transport = self.inner.transport.as_ref().ok_or(EmailError::Disabled)?;
let from = self
.inner
.from
.clone()
.ok_or_else(|| EmailError::Config("from address not configured".into()))?;
if req.request_id.is_nil() {
req.request_id = Uuid::now_v7();
}
let recipient_count = req.to.len() + req.cc.len() + req.bcc.len();
if recipient_count == 0 {
return Err(EmailError::Missing("email has no recipients".into()));
}
debug!(
request_id = %req.request_id,
recipients = recipient_count,
subject_len = req.subject.len(),
attachments = req.attachments.len(),
"service::send dispatch"
);
let message = build_message(&req, from, self.inner.max_attachment_bytes)?;
let smtp_message_id = match transport.send(message).await {
Ok(message_id) => {
crate::metrics::email_sent("ok");
message_id
}
Err(error) => {
crate::metrics::email_sent("error");
return Err(error);
}
};
let sent = EmailSent {
request_id: req.request_id,
recipient_count,
smtp_message_id,
sent_at_ms: now_ms(),
metadata: req.metadata,
};
if let Some(q) = &self.inner.queue
&& let Err(err) = q.publish_to(CHANNEL_SENT, &sent).await
{
warn!(error = %err, "failed to publish EmailSent event");
}
info!(
request_id = %sent.request_id,
recipients = sent.recipient_count,
"email sent"
);
Ok(sent)
}
pub async fn send_mail(
&self,
to: &str,
subject: &str,
text: &str,
html: &str,
) -> Result<(), EmailError> {
trace!(to, subject_len = subject.len(), "service::send_mail");
let req = SendEmailRequest {
request_id: Uuid::now_v7(),
to: vec![EmailAddress::new(to)],
cc: vec![],
bcc: vec![],
subject: subject.to_string(),
text: if text.is_empty() {
None
} else {
Some(text.to_string())
},
html: if html.is_empty() {
None
} else {
Some(html.to_string())
},
reply_to: None,
attachments: vec![],
metadata: Default::default(),
queued_at_ms: now_ms(),
trace: crate::trace::capture(),
};
let _ = self.send(req).await?;
Ok(())
}
pub async fn send_batch(
&self,
req: SendBatchEmailRequest,
) -> Result<Vec<EmailSent>, EmailError> {
let mut sent_list = Vec::with_capacity(req.recipients.len());
let default_meta = req.metadata.clone();
let total = req.recipients.len();
let batch_id = req.request_id;
debug!(batch = %batch_id, total, "service::send_batch dispatch");
for (idx, recipient) in req.recipients.into_iter().enumerate() {
if recipient.to.address.is_empty() {
warn!(batch = %batch_id, index = idx, "skipping batch recipient without address");
continue;
}
debug!(batch = %batch_id, index = idx, address = %recipient.to.address, "batch recipient processing");
let mut meta = default_meta.clone();
meta.insert("batch_request_id".into(), batch_id.to_string());
meta.insert("batch_index".into(), idx.to_string());
for (k, v) in &recipient.metadata {
meta.insert(k.clone(), v.clone());
}
let send_req = SendEmailRequest {
request_id: if recipient.request_id.is_nil() {
Uuid::now_v7()
} else {
recipient.request_id
},
to: vec![recipient.to.clone()],
cc: vec![],
bcc: vec![],
subject: if recipient.subject.is_empty() {
req.default_subject.clone()
} else {
recipient.subject
},
text: recipient.text.or_else(|| req.default_text.clone()),
html: recipient.html.or_else(|| req.default_html.clone()),
reply_to: None,
attachments: vec![],
metadata: meta,
queued_at_ms: now_ms(),
trace: crate::trace::capture(),
};
match self.send(send_req).await {
Ok(sent) => sent_list.push(sent),
Err(err) => {
warn!(
batch = %batch_id,
index = idx,
error = %err,
"batch recipient failed"
);
let failed = EmailFailed {
request_id: Uuid::nil(),
error: err.to_string(),
recipient_count: 1,
failed_at_ms: now_ms(),
metadata: Default::default(),
};
if let Some(q) = &self.inner.queue
&& let Err(err) = q.publish_to(CHANNEL_FAILED, &failed).await
{
warn!(error = %err, "failed to publish EmailFailed event");
}
}
}
}
debug!(batch = %batch_id, sent = sent_list.len(), total, "batch processed");
Ok(sent_list)
}
pub async fn render_and_send(
&self,
req: RenderAndSendEmailRequest,
) -> Result<EmailSent, EmailError> {
let recipient_count = req.to.len() + req.cc.len() + req.bcc.len();
debug!(
request_id = %req.request_id,
template = %req.template,
recipients = recipient_count,
"service::render_and_send dispatch"
);
// The locale travels inside the template vars: registered templates are
// `Fn(&Value) -> Markup`, and every one of them reads `locale` to pick
// its catalogue. Injecting here keeps the `render_and_send` contract
// (and the queue payload) unchanged for callers.
let mut vars = req.vars.clone();
if let Some(object) = vars.as_object_mut()
&& !object.contains_key("locale")
{
let tag = req
.locale
.as_deref()
.and_then(i18n::UiLocale::parse)
.unwrap_or(i18n::UiLocale::En);
object.insert(
"locale".to_string(),
serde_json::Value::String(tag.tag().to_string()),
);
}
let html = self.inner.templates.render(&req.template, &vars).await?;
debug!(
request_id = %req.request_id,
template = %req.template,
bytes = html.len(),
"template rendered"
);
let send_req = SendEmailRequest {
request_id: req.request_id,
to: req.to,
cc: req.cc,
bcc: req.bcc,
subject: req.subject,
text: req.text,
html: Some(html),
reply_to: None,
attachments: vec![],
metadata: req.metadata,
queued_at_ms: now_ms(),
trace: crate::trace::capture(),
};
self.send(send_req).await
}
pub async fn publish_send(&self, req: &SendEmailRequest) -> Result<(), EmailError> {
let q = self.inner.queue.as_ref().ok_or(EmailError::Disabled)?;
q.publish_to(CHANNEL_SEND, req).await
}
pub async fn publish_batch(&self, req: &SendBatchEmailRequest) -> Result<(), EmailError> {
let q = self.inner.queue.as_ref().ok_or(EmailError::Disabled)?;
q.publish_to(CHANNEL_BATCH, req).await
}
pub async fn publish_render(&self, req: &RenderAndSendEmailRequest) -> Result<(), EmailError> {
let q = self.inner.queue.as_ref().ok_or(EmailError::Disabled)?;
q.publish_to(CHANNEL_RENDER, req).await
}
pub async fn publish_sent(&self, sent: &EmailSent) -> Result<(), EmailError> {
let q = self.inner.queue.as_ref().ok_or(EmailError::Disabled)?;
q.publish_to(CHANNEL_SENT, sent).await
}
pub async fn publish_failed(&self, failed: &EmailFailed) -> Result<(), EmailError> {
let q = self.inner.queue.as_ref().ok_or(EmailError::Disabled)?;
q.publish_to(CHANNEL_FAILED, failed).await
}
pub async fn start_subscribers(
&self,
) -> Result<crate::subscriber::EmailSubscribers, EmailError> {
crate::subscriber::start_subscribers(self.clone()).await
}
}
pub struct EmailServiceBuilder {
email_config: config::email::EmailConfig,
queue_config: Option<config::email_queue::EmailQueueConfig>,
redis_url: Option<String>,
templates: TemplateRegistry,
}
impl EmailServiceBuilder {
pub fn new(email_config: config::email::EmailConfig) -> Self {
Self {
email_config,
queue_config: None,
redis_url: None,
templates: TemplateRegistry::new(),
}
}
pub fn with_queue(
mut self,
redis_url: impl Into<String>,
queue_config: config::email_queue::EmailQueueConfig,
) -> Self {
self.redis_url = Some(redis_url.into());
self.queue_config = Some(queue_config);
self
}
pub fn with_templates(mut self, templates: TemplateRegistry) -> Self {
self.templates = templates;
self
}
pub async fn register_default_templates(self) -> Self {
self.templates
.register("verify_code", templates::verify_code::render)
.await;
self.templates
.register("reset_link", templates::reset_link::render)
.await;
self.templates
.register(
"email_change_confirm",
templates::email_change_confirm::render,
)
.await;
self
}
pub async fn build(self) -> Result<EmailService, EmailError> {
if !self.email_config.enabled() {
info!("email disabled at config; transport=None, queue=None");
return Ok(EmailService {
inner: Arc::new(EmailServiceInner {
config: Arc::new(self.email_config),
transport: None,
from: None,
queue: None,
templates: self.templates,
max_attachment_bytes: 0,
}),
});
}
self.email_config
.is_valid()
.map_err(EmailError::InvalidConfig)?;
let from_addr = self
.email_config
.from_address()
.ok_or_else(|| EmailError::Config("email.from_address missing".into()))?;
let parsed: Address = from_addr.parse()?;
let from = match self.email_config.from_name() {
Some(name) => Mailbox::new(Some(name.to_string()), parsed),
None => Mailbox::new(None, parsed),
};
let smtp_host = self.email_config.smtp_host().to_string();
let smtp_port = self.email_config.smtp_port();
let tls_mode = match smtp_port {
465 => "implicit-tls",
587 => "starttls",
_ => "opportunistic",
};
let transport = EmailTransportBuilder::new(
smtp_host.clone(),
smtp_port,
self.email_config.smtp_username().map(|s| s.to_string()),
self.email_config.smtp_password().map(|s| s.to_string()),
)
.connection_timeout(std::time::Duration::from_millis(
self.email_config.connection_timeout_ms(),
))
.build()?;
info!(host = %smtp_host, port = smtp_port, tls = tls_mode, "smtp transport ready");
let queue = match (&self.redis_url, &self.queue_config) {
(Some(url), Some(qcfg)) if qcfg.enabled() => {
let queue = EmailQueue::connect(
url,
QueueSettings {
channel_prefix: qcfg.channel_prefix().to_string(),
max_retries: qcfg.max_retries(),
retry_backoff: qcfg.retry_backoff(),
enabled: true,
},
)
.await?;
info!(
prefix = %qcfg.channel_prefix(),
max_retries = qcfg.max_retries(),
"redis queue connected"
);
Some(queue)
}
_ => {
info!("redis queue disabled; pubsub subscribers will not start");
None
}
};
let max_attachment_bytes = self.email_config.max_attachment_bytes();
Ok(EmailService {
inner: Arc::new(EmailServiceInner {
config: Arc::new(self.email_config),
transport: Some(transport),
from: Some(from),
queue,
templates: self.templates,
max_attachment_bytes,
}),
})
}
}
impl Default for EmailService {
fn default() -> Self {
EmailService {
inner: Arc::new(EmailServiceInner {
config: Arc::new(config::email::EmailConfig::default()),
transport: None,
from: None,
queue: None,
templates: TemplateRegistry::new(),
max_attachment_bytes: 0,
}),
}
}
}