8cc1265
use bytes::Bytes;
use uuid::Uuid;
use crate::error::EmailError;
use crate::events::{EmailAddress, SendEmailRequest, now_ms};
use queue::Publisher;
#[derive(Clone)]
pub struct MailPublisher {
publisher: Publisher,
topic: String,
}
impl MailPublisher {
pub fn new(publisher: Publisher, topic: impl Into<String>) -> Self {
Self {
publisher,
topic: topic.into(),
}
}
pub fn topic(&self) -> &str {
&self.topic
}
pub async fn send(
&self,
mut request: SendEmailRequest,
request_id: Option<&str>,
) -> Result<(), EmailError> {
if request.request_id.is_nil() {
request.request_id = Uuid::now_v7();
}
if request.queued_at_ms == 0 {
request.queued_at_ms = now_ms();
}
let key = request.request_id.to_string();
let trace_id = request_id.filter(|request_id| !request_id.is_empty());
let mut properties = Vec::with_capacity(2);
properties.push(("request_id".to_owned(), key.clone()));
if let Some(request_id) = trace_id {
properties.push(("trace_id".to_owned(), request_id.to_owned()));
}
let body = Bytes::from(serde_json::to_vec(&request)?);
self.publisher
.publish(&self.topic, "send", vec![key], properties, body)
.await
.map_err(EmailError::from)
}
pub async fn send_mail(
&self,
to: &str,
subject: &str,
text: &str,
html: &str,
request_id: Option<&str>,
) -> Result<(), EmailError> {
let request = SendEmailRequest {
request_id: Uuid::now_v7(),
to: vec![EmailAddress::new(to)],
cc: Vec::new(),
bcc: Vec::new(),
subject: subject.to_owned(),
text: (!text.is_empty()).then(|| text.to_owned()),
html: (!html.is_empty()).then(|| html.to_owned()),
reply_to: None,
attachments: Vec::new(),
metadata: Default::default(),
queued_at_ms: now_ms(),
trace: request_id
.filter(|request_id| !request_id.is_empty())
.map(str::to_owned),
};
self.send(request, request_id).await
}
}