9301161
use std::time::Duration;
use tracing::{Instrument as _, debug, info};
use crate::error::EmailError;
use crate::events::{
CHANNEL_BATCH, CHANNEL_RENDER, CHANNEL_SEND, EmailFailed, RenderAndSendEmailRequest,
SendBatchEmailRequest, SendEmailRequest, now_ms,
};
use crate::queue::{SubscribeHandle, TypedMessageMeta};
use crate::service::EmailService;
pub struct EmailSubscribers {
single: Option<SubscribeHandle<SendEmailRequest>>,
batch: Option<SubscribeHandle<SendBatchEmailRequest>>,
render: Option<SubscribeHandle<RenderAndSendEmailRequest>>,
}
impl EmailSubscribers {
pub fn empty() -> Self {
Self {
single: None,
batch: None,
render: None,
}
}
pub async fn shutdown(self, timeout: Duration) -> Result<(), EmailError> {
let s = self.single;
let b = self.batch;
let r = self.render;
let mut last_err: Option<EmailError> = None;
if let Some(h) = s
&& let Err(e) = h.shutdown(timeout).await
{
last_err = Some(e);
}
if let Some(h) = b
&& let Err(e) = h.shutdown(timeout).await
{
last_err = Some(e);
}
if let Some(h) = r
&& let Err(e) = h.shutdown(timeout).await
{
last_err = Some(e);
}
match last_err {
Some(e) => Err(e),
None => Ok(()),
}
}
}
pub async fn start_subscribers(service: EmailService) -> Result<EmailSubscribers, EmailError> {
let queue = match service.queue() {
Some(q) if q.enabled() => q,
_ => {
info!("email queue disabled; skipping pubsub subscribers");
return Ok(EmailSubscribers::empty());
}
};
let group = service.config().smtp_host().to_string();
info!(
group = %group,
prefix = %queue.config().channel_prefix,
"subscribing to email channels"
);
let single = queue
.subscribe(CHANNEL_SEND, &group, {
let svc = service.clone();
move |req: SendEmailRequest, _meta: TypedMessageMeta| {
let svc = svc.clone();
let span = tracing::info_span!("email.send", request_id = %req.request_id);
crate::trace::adopt(&span, req.trace.as_deref());
async move {
let recipient_count = req.to.len() + req.cc.len() + req.bcc.len();
let request_id = req.request_id;
let metadata = req.metadata.clone();
debug!(
request_id = %request_id,
recipients = recipient_count,
channel = CHANNEL_SEND,
"handler invoked"
);
match svc.send(req).await {
Ok(sent) => {
debug!(
request_id = %sent.request_id,
smtp_message_id = %sent.smtp_message_id,
"smtp send ok"
);
if let Err(err) = svc.publish_sent(&sent).await {
tracing::warn!(error = %err, "publish EmailSent failed");
}
}
Err(err) => {
let failed = EmailFailed {
request_id,
error: err.to_string(),
recipient_count,
failed_at_ms: now_ms(),
metadata,
};
tracing::error!(
request_id = %request_id,
error = %err,
"email send failed"
);
if let Err(pub_err) = svc.publish_failed(&failed).await {
tracing::warn!(error = %pub_err, "publish EmailFailed failed");
}
}
}
Ok(())
}
.instrument(span)
}
})
.await?;
info!(channel = CHANNEL_SEND, "subscribed channel");
let batch = queue
.subscribe(CHANNEL_BATCH, &group, {
let svc = service.clone();
move |req: SendBatchEmailRequest, _meta: TypedMessageMeta| {
let svc = svc.clone();
let span = tracing::info_span!("email.batch", request_id = %req.request_id);
crate::trace::adopt(&span, req.trace.as_deref());
async move {
let total = req.recipients.len();
let request_id = req.request_id;
debug!(
request_id = %request_id,
total,
channel = CHANNEL_BATCH,
"batch handler invoked"
);
match svc.send_batch(req).await {
Ok(_) => {
tracing::debug!(request_id = %request_id, total, "batch completed");
}
Err(err) => {
let failed = EmailFailed {
request_id,
error: err.to_string(),
recipient_count: total,
failed_at_ms: now_ms(),
metadata: Default::default(),
};
tracing::error!(
request_id = %request_id,
error = %err,
"batch failed"
);
if let Err(pub_err) = svc.publish_failed(&failed).await {
tracing::warn!(error = %pub_err, "publish EmailFailed failed");
}
}
}
Ok(())
}
.instrument(span)
}
})
.await?;
info!(channel = CHANNEL_BATCH, "subscribed channel");
let render = queue
.subscribe(CHANNEL_RENDER, &group, {
let svc = service.clone();
move |req: RenderAndSendEmailRequest, _meta: TypedMessageMeta| {
let svc = svc.clone();
let span = tracing::info_span!("email.render", request_id = %req.request_id);
crate::trace::adopt(&span, req.trace.as_deref());
async move {
let request_id = req.request_id;
let recipient_count = req.to.len() + req.cc.len() + req.bcc.len();
let metadata = req.metadata.clone();
debug!(
request_id = %request_id,
template = %req.template,
recipients = recipient_count,
channel = CHANNEL_RENDER,
"render handler invoked"
);
match svc.render_and_send(req).await {
Ok(sent) => {
debug!(
request_id = %sent.request_id,
smtp_message_id = %sent.smtp_message_id,
"render+send ok"
);
if let Err(err) = svc.publish_sent(&sent).await {
tracing::warn!(error = %err, "publish EmailSent failed");
}
}
Err(err) => {
let failed = EmailFailed {
request_id,
error: err.to_string(),
recipient_count,
failed_at_ms: now_ms(),
metadata,
};
tracing::error!(
request_id = %request_id,
error = %err,
"render+send failed"
);
if let Err(pub_err) = svc.publish_failed(&failed).await {
tracing::warn!(error = %pub_err, "publish EmailFailed failed");
}
}
}
Ok(())
}
.instrument(span)
}
})
.await?;
info!(channel = CHANNEL_RENDER, "subscribed channel");
Ok(EmailSubscribers {
single: Some(single),
batch: Some(batch),
render: Some(render),
})
}