02d8172
use std::time::Duration;
use config::AppConfig;
use config::auth::AuthConfig;
use config::pow::PowConfig;
use export::{AppRedis, AppState};
use serde::Serialize;
use crate::cache::Cache;
use crate::db::DbConn;
use crate::email::MailPublisher;
use crate::errors::{AuthError, AuthResult};
use crate::events::{EventHandler, Labels, label};
async fn actor_details(
conn: &impl sea_orm::ConnectionTrait,
labels: &Labels,
) -> Option<serde_json::Value> {
use sea_orm::EntityTrait;
let uid = labels.get(label::USER_ID)?.parse::<i64>().ok()?;
let user = crate::models::users::users::Entity::find_by_id(uid)
.one(conn)
.await
.ok()??;
Some(serde_json::json!({
"id": user.id,
"username": user.username,
"display_name": user.displayname,
"email": user.email,
"is_admin": user.is_admin,
"bot": user.bot,
}))
}
#[derive(Clone)]
pub struct AuthService {
pub state: AppState,
db: DbConn,
pub(crate) read_timeout: Duration,
pub(crate) write_timeout: Duration,
}
impl AuthService {
pub fn new(state: AppState) -> Self {
let db = DbConn::new(
state.database.primary().clone(),
state.database.replicas().to_vec(),
);
Self {
state,
db,
read_timeout: Duration::from_millis(300),
write_timeout: Duration::from_millis(2_000),
}
}
pub fn config(&self) -> &AppConfig {
&self.state.config
}
pub fn db(&self) -> &DbConn {
&self.db
}
pub fn redis(&self) -> &AppRedis {
&self.state.redis
}
pub fn cache(&self) -> &Cache {
&self.state.cache
}
pub fn mail(&self) -> Option<&MailPublisher> {
self.state.mail.as_ref()
}
pub fn events(&self) -> Option<&EventHandler> {
self.state.events.as_ref()
}
pub fn auth(&self) -> &AuthConfig {
&self.state.config.auth
}
pub fn pow(&self) -> &PowConfig {
&self.state.config.pow
}
pub fn session_max_age_secs(&self) -> u64 {
self.state.config.session.max_age.max(1) as u64
}
pub async fn mail_locale(
&self,
uid: Option<i64>,
accept_language: Option<&str>,
) -> i18n::UiLocale {
let fallback = i18n::UiLocale::parse(self.config().email.default_locale())
.unwrap_or(i18n::UiLocale::En);
match uid {
Some(uid) => i18n::resolve_user_locale(self.db().read(), uid, fallback).await,
None => i18n::negotiate_accept_language(accept_language, fallback),
}
}
pub async fn emit<T>(&self, kind: &str, payload: T, request_id: Option<&str>)
where
T: Serialize + Clone,
{
self.emit_labeled(kind, Labels::default(), payload, request_id)
.await;
}
pub async fn emit_labeled<T>(
&self,
kind: &str,
labels: Labels,
payload: T,
request_id: Option<&str>,
) where
T: Serialize + Clone,
{
let Some(events) = self.events() else {
tracing::error!(
kind,
"event handler is not configured; event was not forwarded to rocketmq"
);
return;
};
let trace_id = request_id
.map(str::to_owned)
.or_else(crate::envelope::context::request_id)
.unwrap_or_default();
let mut payload = serde_json::to_value(&payload).unwrap_or(serde_json::Value::Null);
if let Some(actor) = actor_details(self.db.write(), &labels).await {
payload["actor"] = actor;
}
let enqueued = crate::events::outbox::enqueue_at(
self.db.write(),
kind,
labels.clone(),
&payload,
&trace_id,
chrono::Utc::now(),
request_id,
)
.await;
if let Err(error) = enqueued {
tracing::error!(
kind,
%error,
"failed to enqueue the event in the outbox; falling back to direct publish"
);
crate::events::retry_publish(events, kind, labels, payload, "auth", request_id).await;
}
}
pub async fn emit_tx<T>(
&self,
conn: &impl sea_orm::ConnectionTrait,
kind: &str,
labels: Labels,
payload: T,
request_id: Option<&str>,
) -> AuthResult<()>
where
T: Serialize,
{
let trace_id = request_id
.map(str::to_owned)
.or_else(crate::envelope::context::request_id)
.unwrap_or_default();
let mut payload = serde_json::to_value(&payload).unwrap_or(serde_json::Value::Null);
if let Some(actor) = actor_details(conn, &labels).await {
payload["actor"] = actor;
}
crate::events::outbox::enqueue_at(
conn,
kind,
labels,
&payload,
&trace_id,
chrono::Utc::now(),
request_id,
)
.await?;
Ok(())
}
pub async fn publish_mail(
&self,
to: &str,
subject: &str,
text: &str,
html: &str,
request_id: Option<&str>,
) -> AuthResult<()> {
let Some(mail) = self.mail() else {
return Err(AuthError::Unavailable(
"mail publisher is not configured".to_owned(),
));
};
mail.send_mail(to, subject, text, html, request_id)
.await
.map_err(|error| AuthError::Internal(error.to_string()))
}
}