b7cd2b1
use std::sync::Arc;
use std::time::Duration;
use lettre::message::Mailbox;
use serde::Serialize;
use tracing::{debug, info, trace, warn};
use uuid::Uuid;
use crate::error::EmailError;
use crate::events::{
EmailAddress, EmailFailed, EmailSent, RenderAndSendEmailRequest, SendBatchEmailRequest,
SendEmailRequest, now_ms,
};
use crate::queue::{EmailQueue, Job, JobHandler, QueueSettings};
use crate::template::TemplateRegistry;
use crate::templates;
use crate::transport::{EmailTransport, EmailTransportBuilder, build_message};
use ::events::event::{EmailEvent, RestOperate};
use ::events::{EventHandler, Labels, RuntimeEvent};
use db::DbConn;
/// 邮件服务：对外只做入队，实际投递由本地队列的 worker 完成。
#[derive(Clone)]
pub struct EmailService {
worker: Arc<EmailWorker>,
queue: EmailQueue,
}
/// 真正持有 SMTP 连接和模板的一方。
struct EmailWorker {
transport: Option<EmailTransport>,
from: Option<Mailbox>,
templates: TemplateRegistry,
max_attachment_bytes: usize,
events: Option<EventHandler>,
/// 投递结果先写 `event_outbox`；为空时退回直接投递。
database: Option<DbConn>,
}
impl EmailService {
pub fn enabled(&self) -> bool {
self.worker.transport.is_some()
}
pub fn templates(&self) -> &TemplateRegistry {
&self.worker.templates
}
pub fn queue(&self) -> &EmailQueue {
&self.queue
}
/// 当前积压条数。
pub fn depth(&self) -> usize {
self.queue.depth()
}
pub fn capacity(&self) -> usize {
self.queue.capacity()
}
pub async fn render(&self, name: &str, vars: &serde_json::Value) -> Result<String, EmailError> {
self.worker.templates.render(name, vars).await
}
/// 入队一封纯文本/HTML 邮件，返回入队后的积压条数。
#[tracing::instrument(skip_all, fields(request_id = %_request_id.unwrap_or_default()))]
pub async fn enqueue(
&self,
mut req: SendEmailRequest,
_request_id: Option<&str>,
) -> Result<usize, EmailError> {
if req.request_id.is_nil() {
req.request_id = Uuid::now_v7();
}
if req.queued_at_ms == 0 {
req.queued_at_ms = now_ms();
}
self.queue.enqueue(Job::Send(req), _request_id).await
}
#[tracing::instrument(skip_all, fields(request_id = %_request_id.unwrap_or_default()))]
pub async fn enqueue_batch(
&self,
mut req: SendBatchEmailRequest,
_request_id: Option<&str>,
) -> Result<usize, EmailError> {
if req.request_id.is_nil() {
req.request_id = Uuid::now_v7();
}
if req.queued_at_ms == 0 {
req.queued_at_ms = now_ms();
}
self.queue.enqueue(Job::Batch(req), _request_id).await
}
#[tracing::instrument(skip_all, fields(request_id = %_request_id.unwrap_or_default()))]
pub async fn enqueue_render(
&self,
mut req: RenderAndSendEmailRequest,
_request_id: Option<&str>,
) -> Result<usize, EmailError> {
if req.request_id.is_nil() {
req.request_id = Uuid::now_v7();
}
if req.queued_at_ms == 0 {
req.queued_at_ms = now_ms();
}
self.queue.enqueue(Job::Render(req), _request_id).await
}
/// 认证中心内部发信入口：直接给一段文本 + 一段 HTML。
#[tracing::instrument(skip_all, fields(request_id = %_request_id.unwrap_or_default()))]
pub async fn enqueue_mail(
&self,
to: &str,
subject: &str,
text: &str,
html: &str,
_request_id: Option<&str>,
) -> Result<(), EmailError> {
trace!(to, subject_len = subject.len(), "service::enqueue_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: None,
};
self.enqueue(req, _request_id).await.map(|_| ())
}
/// 优雅退出：先排空队列，超时后强制停 worker。
pub async fn shutdown(&self, grace: Duration) -> Result<(), EmailError> {
self.queue.shutdown(grace).await
}
}
impl EmailWorker {
#[tracing::instrument(skip_all, fields(request_id = %_request_id.unwrap_or_default()))]
async fn deliver(
&self,
mut req: SendEmailRequest,
_request_id: Option<&str>,
) -> Result<EmailSent, EmailError> {
let transport = self.transport.as_ref().ok_or(EmailError::Disabled)?;
let from = self
.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::deliver"
);
let message = build_message(&req, from, self.max_attachment_bytes)?;
let smtp_message_id = match transport.send(message, _request_id).await {
Ok(message_id) => {
crate::metrics::email_sent(crate::metrics::OUTCOME_OK);
message_id
}
Err(error) => {
crate::metrics::email_sent(crate::metrics::OUTCOME_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,
};
info!(
request_id = %sent.request_id,
recipients = sent.recipient_count,
"email sent"
);
Ok(sent)
}
#[tracing::instrument(skip_all, fields(request_id = %_request_id.unwrap_or_default()))]
async fn deliver_batch(
&self,
req: SendBatchEmailRequest,
_request_id: Option<&str>,
) -> Result<Vec<EmailSent>, EmailError> {
let default_meta = req.metadata.clone();
let total = req.recipients.len();
let batch_id = req.request_id;
debug!(batch = %batch_id, total, "service::deliver_batch");
// 收集有效收件人后并发投递；单收件人失败仅记录，失败明细不阻断其余收件人。
let mut send_reqs = Vec::with_capacity(total);
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;
}
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: None,
};
send_reqs.push(send_req);
}
let results = futures_util::future::join_all(
send_reqs
.into_iter()
.map(|send_req| async move { self.deliver(send_req, _request_id).await }),
)
.await;
let mut sent_list = Vec::with_capacity(results.len());
for (idx, result) in results.into_iter().enumerate() {
match result {
Ok(sent) => {
self.emit(
RuntimeEvent::Email(EmailEvent::Delivery {
operate: RestOperate::Create,
}),
Labels::new().with("request_id", sent.request_id),
&sent,
None,
)
.await;
sent_list.push(sent);
}
Err(err) => {
warn!(
batch = %batch_id,
index = idx,
error = %err,
"batch recipient failed"
);
let failed = EmailFailed {
request_id: batch_id,
error: err.to_string(),
recipient_count: 1,
failed_at_ms: now_ms(),
metadata: Default::default(),
};
self.emit(
RuntimeEvent::Email(EmailEvent::Failure {
operate: RestOperate::Create,
}),
Labels::new().with("request_id", batch_id),
&failed,
None,
)
.await;
}
}
}
debug!(batch = %batch_id, sent = sent_list.len(), total, "batch processed");
Ok(sent_list)
}
#[tracing::instrument(skip_all, fields(request_id = %_request_id.unwrap_or_default()))]
async fn deliver_render(
&self,
req: RenderAndSendEmailRequest,
_request_id: Option<&str>,
) -> 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::deliver_render"
);
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(crate::i18n::UiLocale::parse)
.unwrap_or(crate::i18n::UiLocale::default_locale());
object.insert(
"locale".to_string(),
serde_json::Value::String(tag.tag().to_string()),
);
}
let html = self.templates.render(&req.template, &vars).await?;
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: None,
};
self.deliver(send_req, _request_id).await
}
/// 把事件写进 outbox，由 relay 投递到 RocketMQ；写库失败时退回直接投递。
async fn emit<T>(
&self,
event: RuntimeEvent,
labels: Labels,
payload: &T,
_request_id: Option<&str>,
) where
T: Serialize,
{
let Some(events) = self.events.as_ref() else {
return;
};
if let Some(database) = self.database.as_ref() {
let trace_id = _request_id
.map(str::to_owned)
.or_else(envelope::context::request_id)
.unwrap_or_default();
match ::events::outbox::enqueue_at(
database.write(),
event,
labels.clone(),
payload,
&trace_id,
chrono::Utc::now(),
_request_id,
)
.await
{
Ok(_) => return,
Err(error) => tracing::warn!(
event = %event,
%error,
"failed to enqueue the email event in the outbox; falling back to direct publish"
),
}
}
::events::retry_publish(events, event, labels, payload, "email", _request_id).await;
}
}
#[async_trait::async_trait]
impl JobHandler for EmailWorker {
async fn handle(&self, job: &Job) -> Result<(), EmailError> {
let request_id = job.request_id().to_string();
let request_id = Some(request_id.as_str());
match job {
Job::Send(req) => {
let sent = self.deliver(req.clone(), request_id).await?;
self.emit(
RuntimeEvent::Email(EmailEvent::Delivery {
operate: RestOperate::Create,
}),
Labels::new().with("request_id", sent.request_id),
&sent,
None,
)
.await;
Ok(())
}
Job::Batch(req) => {
self.deliver_batch(req.clone(), request_id).await?;
Ok(())
}
Job::Render(req) => {
let sent = self.deliver_render(req.clone(), request_id).await?;
self.emit(
RuntimeEvent::Email(EmailEvent::Delivery {
operate: RestOperate::Create,
}),
Labels::new().with("request_id", sent.request_id),
&sent,
None,
)
.await;
Ok(())
}
}
}
async fn give_up(&self, job: &Job, error: &EmailError) {
let (request_id, recipient_count, metadata) = match job {
Job::Send(req) => (
req.request_id,
req.to.len() + req.cc.len() + req.bcc.len(),
req.metadata.clone(),
),
Job::Batch(req) => (req.request_id, req.recipients.len(), req.metadata.clone()),
Job::Render(req) => (
req.request_id,
req.to.len() + req.cc.len() + req.bcc.len(),
req.metadata.clone(),
),
};
let failed = EmailFailed {
request_id,
error: error.to_string(),
recipient_count,
failed_at_ms: now_ms(),
metadata,
};
self.emit(
RuntimeEvent::Email(EmailEvent::Failure {
operate: RestOperate::Create,
}),
Labels::new().with("request_id", request_id),
&failed,
None,
)
.await;
}
}
impl Default for EmailService {
fn default() -> Self {
let worker = Arc::new(EmailWorker {
transport: None,
from: None,
templates: TemplateRegistry::new(),
max_attachment_bytes: 0,
events: None,
database: None,
});
let queue = EmailQueue::start(QueueSettings::default(), worker.clone());
Self { worker, queue }
}
}
pub struct EmailServiceBuilder {
pool_config: config::identity::email::EmailPoolConfig,
transport_config: Option<config::identity::smtp::SmtpConfig>,
events: Option<EventHandler>,
database: Option<DbConn>,
templates: TemplateRegistry,
}
impl EmailServiceBuilder {
pub fn new(pool_config: config::identity::email::EmailPoolConfig) -> Self {
Self {
pool_config,
transport_config: None,
events: None,
database: None,
templates: TemplateRegistry::new(),
}
}
pub fn with_transport(mut self, transport_config: config::identity::smtp::SmtpConfig) -> Self {
self.transport_config = Some(transport_config);
self
}
/// 注入事件处理器；投递成功/失败都会推事件到 RocketMQ。
pub fn with_events(mut self, events: EventHandler) -> Self {
self.events = Some(events);
self
}
/// 注入数据库连接；投递结果先写 `event_outbox`，由 relay 异步投递。
pub fn with_database(mut self, database: DbConn) -> Self {
self.database = Some(database);
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> {
self.pool_config
.is_valid()
.map_err(EmailError::InvalidConfig)?;
let mut transport = None;
let mut from = None;
let mut max_attachment_bytes = 0;
if let Some(smtp) = &self.transport_config {
smtp.is_valid().map_err(EmailError::InvalidConfig)?;
from = Some(smtp.from.trim().parse::<Mailbox>()?);
transport = Some(
EmailTransportBuilder::new(
smtp.host.clone(),
smtp.port,
smtp.security,
smtp.username.clone(),
smtp.password.clone(),
)
.connection_timeout(smtp.connect_timeout())
.build()?,
);
max_attachment_bytes = smtp.max_attachment_bytes;
info!(host = %smtp.host, port = smtp.port, security = ?smtp.security, "smtp transport ready");
} else {
info!("smtp transport not configured; email requests will be rejected");
}
let worker = Arc::new(EmailWorker {
transport,
from,
templates: self.templates,
max_attachment_bytes,
events: self.events,
database: self.database,
});
let queue = EmailQueue::start(
QueueSettings::from_config(&self.pool_config),
worker.clone(),
);
Ok(EmailService { worker, queue })
}
}