use crate::models::users::users;
use crate::protos::auth::v1::identity_service_server::{IdentityService, IdentityServiceServer};
use crate::protos::auth::v1::*;
use chrono::Utc;
use rand::RngExt as _;
use sea_orm::sea_query::{Expr, ExprTrait};
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, Set};
use tonic::{Request, Response, Status};

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

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

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

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

fn random_bot_id() -> i64 {
    let mut rng = rand::rng();
    let digits = rng.random_range(5u32..=10);
    let upper = 10i64.pow(digits);
    rng.random_range((upper / 10)..upper)
}

fn bot_public(user: &users::Model) -> crate::protos::auth::common::v1::UserPublic {
    crate::api::store::user_public(user, String::new(), None)
}

impl IdentityApi {
    async fn change_password_impl(
        &self,
        request: ChangePasswordRequest,
        request_id: Option<&str>,
    ) -> AuthResult<ChangePasswordResponse> {
        let db = self.svc.db();
        let user = store::load_active_user(db, request.user_id).await?;

        let stored = user.password_hash.clone();
        if stored.is_empty() {
            // 没有存量密码就不能「凭旧密码」修改：否则持 rpc_key 的一方能直接设置密码。
            return Err(AuthError::InvalidPassword);
        }
        if !stored.is_empty() {
            let check = crate::share::password::verify(
                request.old_password_hash.clone(),
                stored,
                request_id,
            )
            .await?;
            if check == crate::share::password::PasswordCheck::Mismatch {
                return Err(AuthError::InvalidPassword);
            }
        }

        crate::share::password::validate_new_password(
            &request.new_password_hash,
            &self.svc.auth().password_policy,
        )?;
        let new_hash =
            crate::share::password::hash(request.new_password_hash.clone(), request_id).await?;

        let now = store::now();
        users::Entity::update_many()
            .col_expr(users::Column::PasswordHash, Expr::value(new_hash))
            .col_expr(users::Column::PasswordUpdatedAt, Expr::value(Some(now)))
            .col_expr(users::Column::PasswordMustChange, Expr::value(false))
            .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(user.id))
            .exec(db.write())
            .await?;

        let keep = match request.keep_session_id.trim() {
            "" => None,
            value => uuid::Uuid::parse_str(value).ok(),
        };
        let mut update = crate::models::users::user_sessions::Entity::update_many()
            .col_expr(
                crate::models::users::user_sessions::Column::RevokedAt,
                Expr::value(Some(now)),
            )
            .col_expr(
                crate::models::users::user_sessions::Column::Version,
                Expr::col(crate::models::users::user_sessions::Column::Version).add(1),
            )
            .col_expr(
                crate::models::users::user_sessions::Column::UpdatedAt,
                Expr::value(now),
            )
            .filter(crate::models::users::user_sessions::Column::UserId.eq(user.id))
            .filter(crate::models::users::user_sessions::Column::RevokedAt.is_null());
        if let Some(keep) = keep {
            update = update.filter(crate::models::users::user_sessions::Column::Id.ne(keep));
        }
        let revoked = update.exec(db.write()).await?;

        if let Err(err) = store::drop_user_session_states(self.svc.cache(), user.id, keep).await {
            tracing::warn!(%err, user_id = user.id, "failed to drop revoked sessions");
        }
        crate::services::me::invalidate_me_cache(self.svc.cache(), user.id).await;

        store::audit(
            db,
            user.id,
            Some(user.id),
            crate::models::users::user_events::UserEventType::PasswordChange,
            Some(serde_json::json!({ "action": "password_changed" })),
            request_id,
            match request.client_ip.trim() {
                "" => None,
                value => Some(value.to_owned()),
            },
            None,
        )
        .await?;

        let security_version = store::load_user(db, user.id).await?.security_version;
        if let Some(keep) = keep {
            if let Err(err) = store::refresh_session_version(
                self.svc.cache(),
                user.id,
                keep,
                security_version,
                self.svc.session_max_age_secs(),
            )
            .await
            {
                tracing::warn!(%err, user_id = user.id, "failed to refresh the kept session version");
            }
        }
        Ok(ChangePasswordResponse {
            changed: true,
            security_version,
            revoked_sessions: revoked.rows_affected as u32,
        })
    }
}

#[tonic::async_trait]
impl IdentityService for IdentityApi {
    async fn authenticate(
        &self,
        _request: Request<AuthenticateRequest>,
    ) -> Result<Response<AuthenticateResponse>, Status> {
        Err(Status::unimplemented("authenticate is not available"))
    }

    async fn verify_credential(
        &self,
        _request: Request<VerifyCredentialRequest>,
    ) -> Result<Response<VerifyCredentialResponse>, Status> {
        Err(Status::unimplemented("verify_credential is not available"))
    }

    async fn get_user_public(
        &self,
        request: Request<GetUserPublicRequest>,
    ) -> Result<Response<GetUserPublicResponse>, Status> {
        let meta = context::meta(&request).await?;
        let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.IdentityService");
        let result = self
            .get_user_public_inner(meta.request_id(), request.into_inner())
            .await;
        timer.finish("GetUserPublic", &result);
        result
    }

    async fn batch_get_user_public(
        &self,
        request: Request<BatchGetUserPublicRequest>,
    ) -> Result<Response<BatchGetUserPublicResponse>, Status> {
        let meta = context::meta(&request).await?;
        let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.IdentityService");
        let result = self
            .batch_get_user_public_inner(meta.request_id(), request.into_inner())
            .await;
        timer.finish("BatchGetUserPublic", &result);
        result
    }

    async fn get_user_status(
        &self,
        request: Request<GetUserStatusRequest>,
    ) -> Result<Response<GetUserStatusResponse>, Status> {
        let meta = context::meta(&request).await?;
        let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.IdentityService");
        let result = self
            .get_user_status_inner(meta.request_id(), request.into_inner())
            .await;
        timer.finish("GetUserStatus", &result);
        result
    }

    async fn ensure_bot_user(
        &self,
        request: Request<EnsureBotUserRequest>,
    ) -> Result<Response<EnsureBotUserResponse>, Status> {
        let meta = context::meta(&request).await?;
        let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.IdentityService");
        let result = self
            .ensure_bot_user_inner(meta.request_id(), request.into_inner())
            .await;
        timer.finish("EnsureBotUser", &result);
        result
    }

    async fn change_password(
        &self,
        request: Request<ChangePasswordRequest>,
    ) -> Result<Response<ChangePasswordResponse>, Status> {
        let meta = context::meta(&request).await?;
        let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.IdentityService");
        let result = self
            .change_password_inner(meta.request_id(), request.into_inner())
            .await;
        timer.finish("ChangePassword", &result);
        result
    }
}

impl IdentityApi {
    #[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
    async fn get_user_public_inner(
        &self,
        request_id: Option<&str>,
        payload: GetUserPublicRequest,
    ) -> Result<Response<GetUserPublicResponse>, Status> {
        let reference = payload
            .user
            .as_ref()
            .ok_or_else(|| error::status(&AuthError::MalformedRequest, request_id))?;
        let user = store::find_user(self.svc.db(), reference)
            .await
            .map_err(|error_value| error::status(&error_value, request_id))?
            .ok_or_else(|| error::status(&AuthError::UserNotFound, request_id))?;
        let email = user.email.clone();
        Ok(Response::new(GetUserPublicResponse {
            user: Some(store::user_public(&user, email, None)),
        }))
    }

    #[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
    async fn batch_get_user_public_inner(
        &self,
        request_id: Option<&str>,
        payload: BatchGetUserPublicRequest,
    ) -> Result<Response<BatchGetUserPublicResponse>, Status> {
        let ids = payload.user_ids;
        if ids.is_empty() {
            return Ok(Response::new(BatchGetUserPublicResponse {
                users: Vec::new(),
            }));
        }
        if ids.len() > store::MAX_BATCH_USER_IDS {
            return Err(error::invalid_argument(
                format!(
                    "at most {} user ids can be fetched at once",
                    store::MAX_BATCH_USER_IDS
                ),
                request_id,
            ));
        }
        let mut unique: Vec<i64> = Vec::with_capacity(ids.len());
        for id in ids {
            if !unique.contains(&id) {
                unique.push(id);
            }
        }
        let rows = users::Entity::find()
            .filter(users::Column::Id.is_in(unique.iter().copied()))
            .all(self.svc.db().read())
            .await
            .map_err(|error_value| error::status(&AuthError::from(error_value), request_id))?;
        let mut users = Vec::with_capacity(unique.len());
        for user in &rows {
            users.push(store::user_public(user, user.email.clone(), None));
        }
        Ok(Response::new(BatchGetUserPublicResponse { users }))
    }

    #[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
    async fn get_user_status_inner(
        &self,
        request_id: Option<&str>,
        payload: GetUserStatusRequest,
    ) -> Result<Response<GetUserStatusResponse>, Status> {
        let reference = payload
            .user
            .as_ref()
            .ok_or_else(|| error::status(&AuthError::MalformedRequest, request_id))?;
        let user = store::find_user(self.svc.db(), reference)
            .await
            .map_err(|error_value| error::status(&error_value, request_id))?;
        let Some(user) = user else {
            return Ok(Response::new(GetUserStatusResponse {
                status: Some(store::empty_status()),
            }));
        };
        let mfa = store::active_mfa_count(self.svc.db(), user.id)
            .await
            .map_err(|error_value| error::status(&error_value, request_id))?;
        Ok(Response::new(GetUserStatusResponse {
            status: Some(store::user_status_of(&user, mfa)),
        }))
    }

    #[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
    async fn ensure_bot_user_inner(
        &self,
        request_id: Option<&str>,
        payload: EnsureBotUserRequest,
    ) -> Result<Response<EnsureBotUserResponse>, Status> {
        let username = payload.username.trim();
        let lower = username.to_lowercase();
        if lower.is_empty()
            || lower.len() > 32
            || !lower
                .chars()
                .all(|value| value.is_ascii_alphanumeric() || value == '-' || value == '_')
        {
            return Err(error::status(&AuthError::InvalidUsername, request_id));
        }
        let db = self.svc.db().write();
        if let Some(existing) = users::Entity::find()
            .filter(users::Column::Lowername.eq(&lower))
            .one(db)
            .await
            .map_err(|value| error::status(&AuthError::from(value), request_id))?
        {
            if !existing.bot {
                return Err(error::status(&AuthError::UsernameTaken, request_id));
            }
            return Ok(Response::new(EnsureBotUserResponse {
                user: Some(bot_public(&existing)),
                created: false,
            }));
        }

        let displayname = if payload.displayname.trim().is_empty() {
            username.to_string()
        } else {
            payload.displayname.trim().to_string()
        };
        let now = Utc::now();
        let password_hash = crate::share::password::hash(format!("bot-{lower}-{now}"), request_id)
            .await
            .map_err(|value| error::status(&value, request_id))?;
        let inserted = users::ActiveModel {
            id: Set(random_bot_id()),
            username: Set(username.to_string()),
            lowername: Set(lower.clone()),
            displayname: Set(Some(displayname)),
            email: Set(format!("{lower}@bots.invalid")),
            email_verified_at: Set(Some(now)),
            password_hash: Set(password_hash),
            password_updated_at: Set(Some(now)),
            allow_use: Set(true),
            forbid_login: Set(true),
            forbid_login_reason: Set(Some("bot account".to_string())),
            bot: Set(true),
            system: Set(payload.system),
            created_at: Set(now),
            ..Default::default()
        }
        .insert(db)
        .await;
        match inserted {
            Ok(user) => Ok(Response::new(EnsureBotUserResponse {
                user: Some(bot_public(&user)),
                created: true,
            })),
            Err(failure) => {
                // 只有唯一约束冲突才能走「已存在」分支；其余数据库错误必须原样上抛。
                if !matches!(
                    failure.sql_err(),
                    Some(sea_orm::SqlErr::UniqueConstraintViolation(_))
                ) {
                    return Err(error::status(&AuthError::from(failure), request_id));
                }
                let existing = users::Entity::find()
                    .filter(users::Column::Lowername.eq(&lower))
                    .one(db)
                    .await
                    .map_err(|value| error::status(&AuthError::from(value), request_id))?
                    .ok_or_else(|| error::status(&AuthError::UsernameTaken, request_id))?;
                if !existing.bot {
                    return Err(error::status(&AuthError::UsernameTaken, request_id));
                }
                Ok(Response::new(EnsureBotUserResponse {
                    user: Some(bot_public(&existing)),
                    created: false,
                }))
            }
        }
    }

    #[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
    async fn change_password_inner(
        &self,
        request_id: Option<&str>,
        payload: ChangePasswordRequest,
    ) -> Result<Response<ChangePasswordResponse>, Status> {
        self.change_password_impl(payload, request_id)
            .await
            .map(Response::new)
            .map_err(|failure| error::status(&failure, request_id))
    }
}