9362a10
ZhenYi
refactor: remove OAuth-related modules and update references for identity service integration
9362a10
use std::sync::Arc;
use crate::cache::Cache;
use crate::config::app::AppRegistry;
use crate::config::auth::AuthConfig;
use crate::config::jwt::JwtConfig;
use crate::config::pow::PowConfig;
use crate::db::DbConn;
use crate::email::EmailService;
use crate::events::{EventHandler, Labels, label};
use serde::Serialize;
pub use super::errors::{AuthError, AuthResult};
const EVENT_PUBLISH_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
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 AuthSvc {
db: DbConn,
cache: Cache,
email: Option<EmailService>,
events: Option<EventHandler>,
auth: Arc<AuthConfig>,
pow: Arc<PowConfig>,
apps: Arc<AppRegistry>,
signing: Arc<crate::server::keys::SigningState>,
jwt: crate::jwt::JwtState,
session_max_age_secs: u64,
}
impl AuthSvc {
#[allow(clippy::too_many_arguments)]
pub async fn new(
db: DbConn,
cache: Cache,
email: Option<EmailService>,
events: Option<EventHandler>,
auth: AuthConfig,
pow: PowConfig,
apps: AppRegistry,
jwt_config: JwtConfig,
session_max_age_secs: u64,
) -> AuthResult<Self> {
let rpc_key = auth.rpc_key.trim();
if !rpc_key.is_empty() && rpc_key.len() < 32 {
return Err(AuthError::Unavailable(
"auth.rpc_key must be at least 32 characters".to_owned(),
));
}
if rpc_key.starts_with("REPLACE_WITH_") {
return Err(AuthError::Unavailable(
"auth.rpc_key is still set to the placeholder value".to_owned(),
));
}
let signing = crate::server::keys::SigningState::from_config(&jwt_config)?;
let jwt_state = crate::jwt::JwtState::load(&jwt_config)
.await
.map_err(|error| AuthError::Unavailable(error.to_string()))?;
let svc = Self {
db,
cache,
email,
events,
auth: Arc::new(auth),
pow: Arc::new(pow),
apps: Arc::new(apps),
signing: Arc::new(signing),
jwt: jwt_state,
session_max_age_secs: session_max_age_secs.max(1),
};
crate::jwt::authority::install(Arc::new(svc.clone()))
.map_err(|error| AuthError::Unavailable(error.to_string()))?;
Ok(svc)
}
pub fn db(&self) -> &DbConn {
&self.db
}
pub fn cache(&self) -> &Cache {
&self.cache
}
pub fn email(&self) -> Option<&EmailService> {
self.email.as_ref()
}
pub fn auth(&self) -> &AuthConfig {
&self.auth
}
pub fn pow(&self) -> &PowConfig {
&self.pow
}
pub fn apps(&self) -> &AppRegistry {
&self.apps
}
pub fn signing(&self) -> &crate::server::keys::SigningState {
&self.signing
}
pub fn jwt(&self) -> &crate::jwt::JwtState {
&self.jwt
}
pub fn events(&self) -> Option<&EventHandler> {
self.events.as_ref()
}
/// 逻辑会话有效期（秒）：启动时从配置读取一次，热路径避免重读配置文件。
pub fn session_max_age_secs(&self) -> u64 {
self.session_max_age_secs
}
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;
}
/// 事件先写进 `event_outbox`，由 relay 投递到 RocketMQ；进程在投递前退出也不会丢。
///
/// 写库失败时退回直接投递：outbox 表缺失或数据库抖动不应该让事件链静默断掉，
/// 只是这时又回到「重试 3 次后丢弃」的旧语义。
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.as_ref() 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;
}
}
/// 事务内入队：事件行与业务写入同一个事务，提交时一起可见、回滚时一起消失。
///
/// 与 `emit_labeled` 的差别在于**错误会向上传播**：写不进 outbox 就让业务事务回滚，
/// 而不是退回直接投递。这是 KenShell 的语义——宁可这次写操作失败重试，
/// 也不要出现「业务成功、事件永久缺失」。
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(())
}
/// 启动时验证事件链路：查询一次 RocketMQ 路由，让 broker 不可达如实暴露。
pub async fn verify_event_forwarding(&self) -> bool {
let Some(events) = self.events.as_ref() else {
return false;
};
match tokio::time::timeout(EVENT_PUBLISH_TIMEOUT, events.health()).await {
Ok(Ok(())) => true,
Ok(Err(error)) => {
tracing::error!(%error, "rocketmq event forwarding is unavailable");
false
}
Err(_) => {
tracing::error!(
timeout_ms = EVENT_PUBLISH_TIMEOUT.as_millis() as u64,
"rocketmq event forwarding timed out"
);
false
}
}
}
pub async fn publish_mail(
&self,
to: &str,
subject: &str,
text: &str,
html: &str,
request_id: Option<&str>,
) -> AuthResult<()> {
match self.email.as_ref() {
Some(service) => {
service
.enqueue_mail(to, subject, text, html, request_id)
.await?;
Ok(())
}
None => {
tracing::warn!(to, subject, "auth email service is not configured");
Ok(())
}
}
}
}