use crate::models::users::user_events::UserEventType;
use crate::models::users::{user_sessions, users};
use crate::protos::auth::v1::session_service_server::{SessionService, SessionServiceServer};
use crate::protos::auth::v1::*;
use sea_orm::sea_query::{Expr, ExprTrait};
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect};
use tonic::{Request, Response, Status};
use uuid::Uuid;

use crate::AuthService;
use crate::api::{context, convert, error, store};
use crate::errors::AuthError;

pub const MAX_LIST_PAGE: u64 = 100;
pub const SESSION_LIST_MAX: usize = 200;

#[derive(Clone)]
pub struct SessionApi {
    svc: AuthService,
}

impl SessionApi {
    pub fn new(svc: AuthService) -> Self {
        Self { svc }
    }

    pub fn server(
        self,
    ) -> tonic::service::interceptor::InterceptedService<
        SessionServiceServer<Self>,
        context::Interceptor,
    > {
        let gate = context::Interceptor::new(self.svc.auth().rpc_key.clone());
        SessionServiceServer::with_interceptor(self, gate)
    }
}

#[tonic::async_trait]
impl SessionService for SessionApi {
    async fn list_user_sessions(
        &self,
        request: Request<ListUserSessionsRequest>,
    ) -> Result<Response<ListUserSessionsResponse>, Status> {
        let meta = context::meta(&request).await?;
        let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.SessionService");
        let result = self
            .list_user_sessions_inner(meta.request_id(), request.into_inner())
            .await;
        timer.finish("ListUserSessions", &result);
        result
    }

    async fn revoke_session(
        &self,
        request: Request<RevokeSessionRequest>,
    ) -> Result<Response<RevokeSessionResponse>, Status> {
        let meta = context::meta(&request).await?;
        let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.SessionService");
        let result = self
            .revoke_session_inner(meta.request_id(), request.into_inner())
            .await;
        timer.finish("RevokeSession", &result);
        result
    }

    async fn revoke_user_sessions(
        &self,
        request: Request<RevokeUserSessionsRequest>,
    ) -> Result<Response<RevokeUserSessionsResponse>, Status> {
        let meta = context::meta(&request).await?;
        let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.SessionService");
        let result = self
            .revoke_user_sessions_inner(meta.request_id(), request.into_inner())
            .await;
        timer.finish("RevokeUserSessions", &result);
        result
    }

    async fn bump_security_version(
        &self,
        request: Request<BumpSecurityVersionRequest>,
    ) -> Result<Response<BumpSecurityVersionResponse>, Status> {
        let meta = context::meta(&request).await?;
        let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.SessionService");
        let result = self
            .bump_security_version_inner(meta.request_id(), request.into_inner())
            .await;
        timer.finish("BumpSecurityVersion", &result);
        result
    }
}
impl SessionApi {
    #[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
    async fn list_user_sessions_inner(
        &self,
        request_id: Option<&str>,
        request: ListUserSessionsRequest,
    ) -> Result<Response<ListUserSessionsResponse>, Status> {
        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.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 sessions = rows.iter().map(convert::session_info).collect();
        Ok(Response::new(ListUserSessionsResponse {
            sessions,
            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 revoke_session_inner(
        &self,
        request_id: Option<&str>,
        request: RevokeSessionRequest,
    ) -> Result<Response<RevokeSessionResponse>, Status> {
        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.svc.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.svc.db().write())
            .await
            .map_err(|err| error::status(&AuthError::from(err), request_id))?;
        let Some(row) = row else {
            return Ok(Response::new(RevokeSessionResponse {
                revoked: false,
                security_version,
            }));
        };
        if row.revoked_at.is_some() {
            return Ok(Response::new(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.svc.db().write())
            .await
            .map_err(|err| error::status(&AuthError::from(err), request_id))?;
        if updated.rows_affected == 0 {
            return Ok(Response::new(RevokeSessionResponse {
                revoked: true,
                security_version,
            }));
        }

        if let Err(err) =
            store::drop_session_by_id(self.svc.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.svc.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(Response::new(RevokeSessionResponse {
            revoked: true,
            security_version,
        }))
    }
    #[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
    async fn revoke_user_sessions_inner(
        &self,
        request_id: Option<&str>,
        request: RevokeUserSessionsRequest,
    ) -> Result<Response<RevokeUserSessionsResponse>, Status> {
        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.svc.db().write())
            .await
            .map_err(|err| error::status(&AuthError::from(err), request_id))?;

        if let Err(err) =
            store::drop_user_session_states(self.svc.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.svc.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.svc.db().write())
                .await
                .map_err(|err| error::status(&AuthError::from(err), request_id))?;
            crate::services::me::invalidate_me_cache(self.svc.cache(), request.user_id).await;
            security_version = store::load_user(self.svc.db(), request.user_id)
                .await
                .map_err(|err| error::status(&err, request_id))?
                .security_version;
        }

        Ok(Response::new(RevokeUserSessionsResponse {
            revoked_count: affected.rows_affected as u32,
            security_version,
        }))
    }
    #[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
    async fn bump_security_version_inner(
        &self,
        request_id: Option<&str>,
        request: BumpSecurityVersionRequest,
    ) -> Result<Response<BumpSecurityVersionResponse>, Status> {
        store::load_user(self.svc.db(), request.user_id)
            .await
            .map_err(|err| error::status(&err, request_id))?;
        let now = store::now();

        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.svc.db().write())
            .await
            .map_err(|err| error::status(&AuthError::from(err), request_id))?;
        let security_version = store::load_user(self.svc.db(), request.user_id)
            .await
            .map_err(|err| error::status(&err, request_id))?
            .security_version;

        let mut revoked_sessions = 0u32;
        if request.revoke_sessions {
            let revoked = 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())
                .exec(self.svc.db().write())
                .await
                .map_err(|err| error::status(&AuthError::from(err), request_id))?;
            revoked_sessions = revoked.rows_affected as u32;
        }

        crate::services::me::invalidate_me_cache(self.svc.cache(), request.user_id).await;
        store::audit(
            self.svc.db(),
            request.user_id,
            None,
            UserEventType::Logout,
            Some(serde_json::json!({
                "reason": request.reason,
                "revoke_sessions": request.revoke_sessions,
            })),
            request_id,
            None,
            None,
        )
        .await
        .map_err(|err| error::status(&err, request_id))?;

        Ok(Response::new(BumpSecurityVersionResponse {
            security_version,
            revoked_sessions,
            bumped_at: Some(convert::now_timestamp()),
        }))
    }
}