d5939db
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());
}
}