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,
        })
    }
}