163952e
use crate::models::users::user_events::{self, UserEventType};
use crate::protos::auth::internal::v1::audit_service_server::{AuditService, AuditServiceServer};
use crate::protos::auth::internal::v1::*;
use chrono::{Duration, Utc};
use prost_types::Timestamp;
use sea_orm::sea_query::Condition;
use sea_orm::{
ColumnTrait, EntityTrait, ExprTrait, PaginatorTrait, QueryFilter, QueryOrder, QuerySelect,
};
use tonic::{Request, Response, Status};
use crate::AuthSvc;
use crate::errors::AuthError;
use crate::internal::{context, convert, error, store};
const DEFAULT_EXPORT_BATCH: u32 = 500;
const MAX_EXPORT_BATCH: u32 = 5_000;
const MAX_EXPORT_RECORDS: u64 = 1_000_000;
const EXPORT_CHANNEL_CAPACITY: usize = 256;
const PURGE_BATCH: u32 = 5_000;
const MAX_RETENTION_DAYS: u64 = 3_650;
const MAX_PURGE_RECORDS: u64 = 1_000_000;
pub struct EventStream {
rx: tokio::sync::mpsc::Receiver<Result<AuthEventRecord, Status>>,
}
impl tonic::codegen::tokio_stream::Stream for EventStream {
type Item = Result<AuthEventRecord, Status>;
fn poll_next(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<Option<Self::Item>> {
self.rx.poll_recv(cx)
}
}
#[derive(Clone)]
pub struct AuditApi {
svc: AuthSvc,
}
fn kind_name(kind: &UserEventType) -> &'static str {
match kind {
UserEventType::Register => "register",
UserEventType::Login => "login",
UserEventType::Logout => "logout",
UserEventType::PasswordChange => "password_change",
UserEventType::ProfileUpdate => "profile_update",
UserEventType::TwoFactorEnabled => "two_factor_enabled",
UserEventType::TwoFactorDisabled => "two_factor_disabled",
UserEventType::FailedLoginAttempt => "failed_login_attempt",
UserEventType::AccountLocked => "account_locked",
UserEventType::AccountUnlocked => "account_unlocked",
UserEventType::EmailChange => "email_change",
UserEventType::EmailVerified => "email_verified",
UserEventType::RoleChanged => "role_changed",
UserEventType::PasswordResetRequested => "password_reset_requested",
UserEventType::PublicKeyAdded => "public_key_added",
UserEventType::PublicKeyRemoved => "public_key_removed",
UserEventType::AccessKeyAdded => "access_key_added",
UserEventType::AccessKeyRemoved => "access_key_removed",
UserEventType::Follow => "follow",
UserEventType::Unfollow => "unfollow",
UserEventType::Block => "block",
UserEventType::Unblock => "unblock",
}
}
fn parse_kind(name: &str) -> Option<UserEventType> {
let value = name.trim().to_ascii_lowercase();
Some(match value.as_str() {
"register" => UserEventType::Register,
"login" => UserEventType::Login,
"logout" => UserEventType::Logout,
"password_change" => UserEventType::PasswordChange,
"profile_update" => UserEventType::ProfileUpdate,
"two_factor_enabled" => UserEventType::TwoFactorEnabled,
"two_factor_disabled" => UserEventType::TwoFactorDisabled,
"failed_login_attempt" => UserEventType::FailedLoginAttempt,
"account_locked" => UserEventType::AccountLocked,
"account_unlocked" => UserEventType::AccountUnlocked,
"email_change" => UserEventType::EmailChange,
"email_verified" => UserEventType::EmailVerified,
"role_changed" => UserEventType::RoleChanged,
"password_reset_requested" => UserEventType::PasswordResetRequested,
"public_key_added" => UserEventType::PublicKeyAdded,
"public_key_removed" => UserEventType::PublicKeyRemoved,
"access_key_added" => UserEventType::AccessKeyAdded,
"access_key_removed" => UserEventType::AccessKeyRemoved,
"follow" => UserEventType::Follow,
"unfollow" => UserEventType::Unfollow,
"block" => UserEventType::Block,
"unblock" => UserEventType::Unblock,
_ => return None,
})
}
fn kinds_of(raw: &[String], request_id: Option<&str>) -> Result<Vec<UserEventType>, Status> {
let mut kinds = Vec::with_capacity(raw.len());
for name in raw {
match parse_kind(name) {
Some(kind) => kinds.push(kind),
None => {
return Err(error::invalid_argument(
format!("unknown event kind: {}", name.trim()),
request_id,
));
}
}
}
Ok(kinds)
}
fn record(row: &user_events::Model) -> AuthEventRecord {
AuthEventRecord {
event_id: row.id.to_string(),
kind: kind_name(&row.event_type).to_owned(),
subject_user_id: row.user_id,
actor_user_id: row.actor_user_id.unwrap_or(0),
occurred_at: convert::datetime_opt(Some(row.occurred_at)),
client_ip: row.ip_address.clone().unwrap_or_default(),
user_agent: row.user_agent.clone().unwrap_or_default(),
trace_id: row.trace_id.clone().unwrap_or_default(),
details_json: row
.details
.as_ref()
.map(|details| details.to_string())
.unwrap_or_default(),
credential_fingerprint: String::new(),
}
}
#[allow(clippy::too_many_arguments)]
fn base_query<Q: QueryFilter>(
query: Q,
user_id: i64,
actor_user_id: i64,
kinds: &[UserEventType],
from: Option<&Timestamp>,
to: Option<&Timestamp>,
request_id: Option<&str>,
) -> Result<Q, Status> {
let mut query = query;
if user_id > 0 {
query = query.filter(user_events::Column::UserId.eq(user_id));
}
if actor_user_id > 0 {
query = query.filter(user_events::Column::ActorUserId.eq(actor_user_id));
}
if !kinds.is_empty() {
query = query.filter(user_events::Column::EventType.is_in(kinds.to_vec()));
}
if let Some(from) = from {
let value = convert::datetime(from)
.ok_or_else(|| error::invalid_argument("from is malformed", request_id))?;
query = query.filter(user_events::Column::OccurredAt.gte(value));
}
if let Some(to) = to {
let value = convert::datetime(to)
.ok_or_else(|| error::invalid_argument("to is malformed", request_id))?;
query = query.filter(user_events::Column::OccurredAt.lte(value));
}
Ok(query)
}
impl AuditApi {
pub fn new(svc: AuthSvc) -> Self {
Self { svc }
}
pub fn server(
self,
) -> tonic::service::interceptor::InterceptedService<
AuditServiceServer<Self>,
context::Interceptor,
> {
let gate = context::Interceptor::new(self.svc.auth().rpc_key.clone());
AuditServiceServer::with_interceptor(self, gate)
}
}
#[tonic::async_trait]
impl AuditService for AuditApi {
type ExportAuthEventsStream = EventStream;
async fn list_auth_events(
&self,
request: Request<ListAuthEventsRequest>,
) -> Result<Response<ListAuthEventsResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.AuditService");
let result = self
.list_auth_events_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("ListAuthEvents", &result);
result
}
async fn export_auth_events(
&self,
request: Request<ExportAuthEventsRequest>,
) -> Result<Response<<AuditApi as crate::protos::auth::internal::v1::audit_service_server::AuditService>::ExportAuthEventsStream>, Status>{
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.AuditService");
let result = self
.export_auth_events_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("ExportAuthEvents", &result);
result
}
async fn purge_auth_events(
&self,
request: Request<PurgeAuthEventsRequest>,
) -> Result<Response<PurgeAuthEventsResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.AuditService");
let result = self
.purge_auth_events_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("PurgeAuthEvents", &result);
result
}
}
impl AuditApi {
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn list_auth_events_inner(
&self,
request_id: Option<&str>,
request: ListAuthEventsRequest,
) -> Result<Response<ListAuthEventsResponse>, Status> {
let kinds = kinds_of(&request.kinds, request_id)?;
let page = request.page.unwrap_or_default();
let offset = store::parse_page_token(&page.page_token)
.ok_or_else(|| error::invalid_argument("page_token is malformed", request_id))?;
let limit = convert::page_size(page.page_size) as u64;
let query = base_query(
user_events::Entity::find(),
request.user_id,
request.actor_user_id,
&kinds,
request.from.as_ref(),
request.to.as_ref(),
request_id,
)?;
let rows = query
.order_by_desc(user_events::Column::OccurredAt)
.limit(limit)
.offset(offset)
.all(self.svc.db().read())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
let next_page_token = if limit > 0 && rows.len() as u64 == limit {
store::page_token(offset + rows.len() as u64)
} else {
String::new()
};
let events = rows.iter().map(record).collect();
Ok(Response::new(ListAuthEventsResponse {
events,
page: Some(crate::protos::auth::common::v1::PageResponse { next_page_token }),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn export_auth_events_inner(
&self,
request_id: Option<&str>,
request: ExportAuthEventsRequest,
) -> Result<Response<<AuditApi as crate::protos::auth::internal::v1::audit_service_server::AuditService>::ExportAuthEventsStream>, Status>{
let kinds = kinds_of(&request.kinds, request_id)?;
let batch_size = if request.batch_size == 0 {
DEFAULT_EXPORT_BATCH
} else {
Ord::min(request.batch_size, MAX_EXPORT_BATCH)
} as u64;
let base = base_query(
user_events::Entity::find(),
0,
0,
&kinds,
request.from.as_ref(),
request.to.as_ref(),
request_id,
)?;
let db = self.svc.db().clone();
let request_id = request_id.map(str::to_owned);
let (tx, rx) = tokio::sync::mpsc::channel(EXPORT_CHANNEL_CAPACITY);
// keyset 分页：以上一页最后一行的 (occurred_at, id) 为游标，避免深 offset O(n²)。
tokio::spawn(async move {
let mut cursor: Option<(chrono::DateTime<chrono::Utc>, uuid::Uuid)> = None;
let mut sent = 0u64;
while sent < MAX_EXPORT_RECORDS {
let limit = Ord::min(batch_size, MAX_EXPORT_RECORDS - sent);
let mut query = base
.clone()
.order_by_asc(user_events::Column::OccurredAt)
.order_by_asc(user_events::Column::Id)
.limit(limit);
if let Some((occurred_at, id)) = cursor {
query = query.filter(
user_events::Column::OccurredAt.gt(occurred_at).or(
user_events::Column::OccurredAt
.eq(occurred_at)
.and(user_events::Column::Id.gt(id)),
),
);
}
let rows = query.all(db.read()).await;
match rows {
Ok(rows) => {
let fetched = rows.len() as u64;
if let Some(last) = rows.last() {
cursor = Some((last.occurred_at, last.id));
}
for row in &rows {
sent += 1;
if tx.send(Ok(record(row))).await.is_err() {
return;
}
}
if fetched < limit {
return;
}
}
Err(err) => {
let _ = tx
.send(Err(error::status(
&AuthError::from(err),
request_id.as_deref(),
)))
.await;
return;
}
}
}
});
Ok(Response::new(EventStream { rx }))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn purge_auth_events_inner(
&self,
request_id: Option<&str>,
request: PurgeAuthEventsRequest,
) -> Result<Response<PurgeAuthEventsResponse>, Status> {
let older_than =
match request.older_than.as_ref() {
Some(value) => Some(convert::datetime(value).ok_or_else(|| {
error::invalid_argument("older_than is malformed", request_id)
})?),
None => None,
};
let retention_cutoff = if request.retention_days > 0 {
if request.retention_days > MAX_RETENTION_DAYS {
return Err(error::invalid_argument(
format!("retention_days exceeds the maximum of {MAX_RETENTION_DAYS}"),
request_id,
));
}
let days = Duration::try_days(request.retention_days as i64).ok_or_else(|| {
error::invalid_argument("retention_days is out of range", request_id)
})?;
Some(Utc::now().checked_sub_signed(days).ok_or_else(|| {
error::invalid_argument("retention_days is out of range", request_id)
})?)
} else {
None
};
let cutoff = match (older_than, retention_cutoff) {
(Some(older_than), Some(retention)) => Ord::min(older_than, retention),
(Some(older_than), None) => older_than,
(None, Some(retention)) => retention,
(None, None) => {
return Err(error::invalid_argument(
"older_than or retention_days is required",
request_id,
));
}
};
let kinds = kinds_of(&request.kinds, request_id)?;
let mut condition = Condition::all().add(user_events::Column::OccurredAt.lt(cutoff));
if !kinds.is_empty() {
condition = condition.add(user_events::Column::EventType.is_in(kinds));
}
if request.dry_run {
let matched = user_events::Entity::find()
.filter(condition)
.count(self.svc.db().read())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
return Ok(Response::new(PurgeAuthEventsResponse {
matched_records: matched,
purged_records: 0,
dry_run: true,
}));
}
// 删除前先数一次：单次调用不得清空超过上限的事件，避免错误 cutoff 造成不可逆损失。
let matched = user_events::Entity::find()
.filter(condition.clone())
.count(self.svc.db().read())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
if matched > MAX_PURGE_RECORDS {
return Err(error::invalid_argument(
format!(
"matched {matched} records, which exceeds the single purge limit of {MAX_PURGE_RECORDS}; narrow the range or run with dry_run"
),
request_id,
));
}
// 分批删除：单条大 DELETE 会长时间锁表并撑大 WAL，按批次循环清。
let mut purged = 0u64;
let mut matched_records = 0u64;
loop {
let ids: Vec<uuid::Uuid> = user_events::Entity::find()
.filter(condition.clone())
.limit(PURGE_BATCH as u64)
.all(self.svc.db().read())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?
.into_iter()
.map(|row| row.id)
.collect();
if ids.is_empty() {
break;
}
matched_records += ids.len() as u64;
let batch = user_events::Entity::delete_many()
.filter(user_events::Column::Id.is_in(ids))
.exec(self.svc.db().write())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
purged += batch.rows_affected;
if batch.rows_affected < PURGE_BATCH as u64 {
break;
}
}
Ok(Response::new(PurgeAuthEventsResponse {
matched_records,
purged_records: purged,
dry_run: false,
}))
}
}