f0a1552
ZhenYi
feat(chat): add chat module with room management, message handling, and session revocation
f0a1552
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<()> {
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.splitn(2, "://").nth(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"
);
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()))?;
info!("mails worker exited cleanly");
Ok(())
}
fn init_tracing() {
use tracing_subscriber::EnvFilter;
let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
tracing_subscriber::fmt().with_env_filter(filter).init();
}