main
use crate::KenAuthError;
use crate::models::users::user_events::UserEventType;
use crate::models::users::{user_sessions, users};
use crate::protos::auth::v1::*;
use sea_orm::sea_query::{Expr, ExprTrait};
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect};
use uuid::Uuid;
use crate::Accounts;
use crate::errors::AuthError;
use crate::share::{auth_convert as convert, auth_error as error, auth_store as store};
pub const MAX_LIST_PAGE: u64 = 100;
pub const SESSION_LIST_MAX: usize = 200;
impl Accounts {
#[tracing::instrument(skip_all, fields(request_id = %envelope::context::resolve_request_id(request_id)))]
pub(crate) async fn list_user_sessions_inner(
&self,
request_id: Option<&str>,
request: ListUserSessionsRequest,
) -> Result<ListUserSessionsResponse, KenAuthError> {
let page = request.page.unwrap_or_default();
let offset = match store::parse_page_token(&page.page_token) {
Some(offset) => offset,
None => {
return Err(error::invalid_argument(
"page_token is malformed",
request_id,
));
}
};
let page_size = convert::page_size(page.page_size) as u64;
let mut limit = if page_size < MAX_LIST_PAGE {
page_size
} else {
MAX_LIST_PAGE
};
let remaining = (SESSION_LIST_MAX as u64).saturating_sub(offset);
if limit > remaining {
limit = remaining;
}
let mut query =
user_sessions::Entity::find().filter(user_sessions::Column::UserId.eq(request.user_id));
if !request.include_revoked {
query = query.filter(user_sessions::Column::RevokedAt.is_null());
}
if !request.include_expired {
query = query.filter(user_sessions::Column::ExpiresAt.gt(store::now()));
}
let rows = query
.order_by_desc(user_sessions::Column::LastSeenAt)
.limit(limit)
.offset(offset)
.all(self.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 sessions = rows.iter().map(convert::session_info).collect();
Ok(ListUserSessionsResponse {
sessions,
page: Some(crate::protos::auth::common::v1::PageResponse { next_page_token }),
})
}
#[tracing::instrument(skip_all, fields(request_id = %envelope::context::resolve_request_id(request_id)))]
pub(crate) async fn revoke_session_inner(
&self,
request_id: Option<&str>,
request: RevokeSessionRequest,
) -> Result<RevokeSessionResponse, KenAuthError> {
let session_id = Uuid::parse_str(request.session_id.trim())
.map_err(|_| error::invalid_argument("session_id is not a valid uuid", request_id))?;
let user = store::load_user(self.db(), request.user_id)
.await
.map_err(|err| error::status(&err, request_id))?;
let security_version = user.security_version;
let row = user_sessions::Entity::find_by_id(session_id)
.filter(user_sessions::Column::UserId.eq(request.user_id))
.one(self.db().write())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
let Some(row) = row else {
return Ok(RevokeSessionResponse {
revoked: false,
security_version,
});
};
if row.revoked_at.is_some() {
return Ok(RevokeSessionResponse {
revoked: true,
security_version,
});
}
let revoked_at = store::now();
// 条件更新 + 行数校验：并发双撤时只有一方生效。
let updated = user_sessions::Entity::update_many()
.col_expr(user_sessions::Column::RevokedAt, Expr::value(revoked_at))
.col_expr(
user_sessions::Column::Version,
Expr::col(user_sessions::Column::Version).add(1),
)
.col_expr(user_sessions::Column::UpdatedAt, Expr::value(revoked_at))
.filter(user_sessions::Column::Id.eq(session_id))
.filter(user_sessions::Column::RevokedAt.is_null())
.exec(self.db().write())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
if updated.rows_affected == 0 {
return Ok(RevokeSessionResponse {
revoked: true,
security_version,
});
}
if let Err(err) = store::drop_session_by_id(self.cache(), request.user_id, session_id).await
{
tracing::warn!(%err, %session_id, "failed to drop the revoked session state");
}
if let Err(err) = store::audit(
self.db(),
request.user_id,
None,
UserEventType::Logout,
Some(serde_json::json!({
"reason": request.reason,
"session_id": session_id,
})),
request_id,
None,
None,
)
.await
{
// 会话已失效，审计失败不回退结果。
tracing::warn!(%err, %session_id, "failed to record session revocation");
}
Ok(RevokeSessionResponse {
revoked: true,
security_version,
})
}
#[tracing::instrument(skip_all, fields(request_id = %envelope::context::resolve_request_id(request_id)))]
pub(crate) async fn revoke_user_sessions_inner(
&self,
request_id: Option<&str>,
request: RevokeUserSessionsRequest,
) -> Result<RevokeUserSessionsResponse, KenAuthError> {
let except = match request.except_session_id.trim() {
"" => None,
value => Some(Uuid::parse_str(value).map_err(|_| {
error::invalid_argument("except_session_id is not a valid uuid", request_id)
})?),
};
let now = store::now();
let mut update = user_sessions::Entity::update_many()
.col_expr(user_sessions::Column::RevokedAt, Expr::value(now))
.col_expr(
user_sessions::Column::Version,
Expr::col(user_sessions::Column::Version).add(1),
)
.col_expr(user_sessions::Column::UpdatedAt, Expr::value(now))
.filter(user_sessions::Column::UserId.eq(request.user_id))
.filter(user_sessions::Column::RevokedAt.is_null());
if let Some(except) = except {
update = update.filter(user_sessions::Column::Id.ne(except));
}
let affected = update
.exec(self.db().write())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
if let Err(err) =
store::drop_user_session_states(self.cache(), request.user_id, except).await
{
tracing::warn!(%err, user_id = request.user_id, "failed to drop revoked sessions");
}
let user = store::load_user(self.db(), request.user_id)
.await
.map_err(|err| error::status(&err, request_id))?;
let mut security_version = user.security_version;
if request.bump_security_version {
users::Entity::update_many()
.col_expr(
users::Column::SecurityVersion,
Expr::col(users::Column::SecurityVersion).add(1),
)
.col_expr(users::Column::UpdatedAt, Expr::value(now))
.filter(users::Column::Id.eq(request.user_id))
.exec(self.db().write())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
crate::services::me::invalidate_me_cache(self.cache(), request.user_id).await;
security_version = store::load_user(self.db(), request.user_id)
.await
.map_err(|err| error::status(&err, request_id))?
.security_version;
}
Ok(RevokeUserSessionsResponse {
revoked_count: affected.rows_affected as u32,
security_version,
})
}
}