d3d11f4
use std::net::SocketAddr;
use std::time::Duration;
use anyhow::Context;
use config::scoped::{MailsConfig, ScopedConfig};
use email::runtime::{WorkerOptions, run_until_shutdown};
use tokio::net::TcpListener;
use tracing::info;
#[tokio::main]
async fn main() -> anyhow::Result<()> {
export::telemetry::init_tracing();
let cfg = MailsConfig::load().context("loading configuration")?;
let redis_url = cfg.redis.url().to_string();
let redis_host = match redis_url.rsplit_once('@') {
Some((_, host)) => host.to_string(),
None => redis_url
.split_once("://")
.map(|x| x.1)
.unwrap_or("?")
.to_string(),
};
info!(
smtp_enabled = cfg.email.enabled(),
queue_enabled = cfg.email_queue.enabled(),
redis = %redis_host,
shutdown_timeout_secs = 5,
"starting mails worker"
);
if let Err(error) = export::telemetry::init_meter_provider() {
tracing::warn!(%error, "failed to init meter provider; metrics disabled");
}
let opts = WorkerOptions {
template_dir: None,
shutdown_timeout: Some(Duration::from_secs(5)),
};
let (health_tx, health_rx) = tokio::sync::oneshot::channel::<()>();
let mut health_handle = None;
if cfg.health.enabled() {
let redis = export::AppRedis::connect(cfg.redis.clone())
.await
.context("connecting to redis")?;
let addr: SocketAddr = format!("{}:{}", cfg.health.bind_addr(), cfg.health.mails_port())
.parse()
.context("parsing health bind address")?;
let listener = TcpListener::bind(addr)
.await
.with_context(|| format!("binding health {addr}"))?;
info!(%addr, "mails health listening");
let router = export::health::router_with(None, redis);
health_handle = Some(tokio::spawn(async move {
let _ = axum::serve(listener, router)
.with_graceful_shutdown(async {
let _ = health_rx.await;
})
.await;
}));
}
let result =
run_until_shutdown(cfg.email.clone(), cfg.email_queue.clone(), redis_url, opts).await;
drop(health_tx);
if let Some(handle) = health_handle {
let _ = handle.await;
}
result.map_err(|e| anyhow::anyhow!(e.to_string()))?;
export::telemetry::shutdown_telemetry();
info!("mails worker exited cleanly");
Ok(())
}