main
use std::sync::Arc;
use config::email::EmailPoolConfig;
use config::rocketmq::RocketMqConfig;
use config::smtp::SmtpConfig;
use events::{EventHandler, Labels, label};
use lettre::message::Mailbox;
use queue::{Consumer, ConsumerConfig, MessageHandler, QueueRuntime};
use serde::Serialize;
use tokio_util::sync::CancellationToken;
use tracing::{Instrument, info, warn};
use uuid::Uuid;
use crate::error::EmailError;
use crate::events::{EmailFailed, EmailSent, SendEmailRequest, now_ms};
use crate::transport::{EmailTransport, EmailTransportBuilder, build_message};
const EVENT_PUBLISH_ATTEMPTS: usize = 3;
async fn publish_event<T: Serialize>(
events: &EventHandler,
kind: events::EventKind,
labels: Labels,
payload: &T,
request_id: &str,
) -> bool {
let mut attempt = 1_usize;
loop {
match events
.send_labeled(kind, labels.clone(), payload, Some(request_id))
.await
{
Ok(()) => return true,
Err(error) if attempt >= EVENT_PUBLISH_ATTEMPTS => {
tracing::error!(kind = %kind, attempt, %error, "failed to publish the email event");
return false;
}
Err(error) => {
tracing::warn!(kind = %kind, attempt, %error, "retrying the email event publish");
tokio::time::sleep(std::time::Duration::from_millis(50) * attempt as u32).await;
attempt += 1;
}
}
}
}
pub struct EmailConsumer {
transport: EmailTransport,
from: Mailbox,
max_attachment_bytes: usize,
pool: EmailPoolConfig,
events: Option<EventHandler>,
}
impl EmailConsumer {
pub fn new(
smtp: SmtpConfig,
pool: EmailPoolConfig,
events: Option<EventHandler>,
) -> Result<Self, EmailError> {
smtp.is_valid().map_err(EmailError::InvalidConfig)?;
pool.is_valid().map_err(EmailError::InvalidConfig)?;
let from = smtp.from.trim().parse::<Mailbox>()?;
let transport = EmailTransportBuilder::new(
smtp.host.clone(),
smtp.port,
smtp.security,
smtp.username.clone(),
smtp.password.clone(),
)
.connection_timeout(smtp.connect_timeout())
.build()?;
Ok(Self {
transport,
from,
max_attachment_bytes: smtp.max_attachment_bytes,
pool,
events,
})
}
pub async fn run(
self,
rocketmq: RocketMqConfig,
runtime: &QueueRuntime,
cancel: CancellationToken,
) -> Result<(), EmailError> {
let topic = self.pool.mail_topic.clone();
let group = self.pool.consumer_group.clone();
let workers = self.pool.workers.max(1) as u32;
let max_reconsume = self.pool.max_reconsume.min(16) as i32;
let consumer = Arc::new(self);
let handler_consumer = consumer.clone();
let handle = tokio::runtime::Handle::current();
let handler: MessageHandler = Arc::new(move |messages| {
let mut success = true;
for message in messages {
let request: SendEmailRequest = match message
.body()
.ok_or_else(|| EmailError::Missing("message body is empty".into()))
.and_then(|body| Ok(serde_json::from_slice(&body)?))
{
Ok(request) => request,
Err(error) => {
warn!(error = %error, "mail message rejected");
success = false;
continue;
}
};
let request_id = request
.trace
.as_deref()
.filter(|value| !value.is_empty())
.map(str::to_owned)
.unwrap_or_else(|| request.request_id.to_string());
let span = tracing::info_span!("email.delivery", request_id = %request_id);
match handle.block_on(handler_consumer.deliver(request.clone()).instrument(span)) {
Ok(sent) => handle.block_on(handler_consumer.emit_sent(&sent, &request_id)),
Err(error) => {
success = false;
handle.block_on(handler_consumer.emit_failed(
&request,
&error,
&request_id,
));
}
}
}
success
});
let mut consumer = Consumer::start(
ConsumerConfig {
name_server: rocketmq.name_server,
consumer_group: group.clone(),
topic: topic.clone(),
tag: "send".to_owned(),
workers,
max_reconsume,
},
runtime,
handler,
)
.await
.map_err(EmailError::from)?;
info!(
topic,
group, workers, max_reconsume, "mail consumer started"
);
cancel.cancelled().await;
consumer.shutdown().await;
info!("mail consumer stopped");
Ok(())
}
async fn deliver(&self, mut request: SendEmailRequest) -> Result<EmailSent, EmailError> {
if request.request_id.is_nil() {
request.request_id = Uuid::now_v7();
}
let recipient_count = request.to.len() + request.cc.len() + request.bcc.len();
if recipient_count == 0 {
return Err(EmailError::Missing("email has no recipients".into()));
}
let message = build_message(&request, self.from.clone(), self.max_attachment_bytes)?;
let request_id = request.request_id.to_string();
let smtp_message_id = self.transport.send(message, Some(&request_id)).await?;
Ok(EmailSent {
request_id: request.request_id,
recipient_count,
smtp_message_id,
sent_at_ms: now_ms(),
metadata: request.metadata,
})
}
async fn emit_sent(&self, sent: &EmailSent, request_id: &str) {
let Some(events) = self.events.as_ref() else {
return;
};
let labels = Labels::new().with(label::REQUEST_ID, request_id);
let _ = publish_event(
events,
label::EMAIL_DELIVERY_CREATE,
labels,
sent,
request_id,
)
.await;
}
async fn emit_failed(&self, request: &SendEmailRequest, error: &EmailError, request_id: &str) {
let Some(events) = self.events.as_ref() else {
return;
};
let failed = EmailFailed {
request_id: request.request_id,
error: error.to_string(),
recipient_count: request.to.len() + request.cc.len() + request.bcc.len(),
failed_at_ms: now_ms(),
metadata: request.metadata.clone(),
};
let labels = Labels::new().with(label::REQUEST_ID, request_id);
let _ = publish_event(
events,
label::EMAIL_FAILURE_CREATE,
labels,
&failed,
request_id,
)
.await;
}
}