use crate::models::users::users;
use crate::protos::auth::common::v1::{Code, PageResponse};
use crate::protos::auth::internal::v1::authority_service_server::{
    AuthorityService, AuthorityServiceServer,
};
use crate::protos::auth::internal::v1::{
    AuthorityAccess, AuthorityUser, AuthorityUserPatch, BatchGetUserRequest, BatchGetUserResponse,
    CancelAccountDeletionRequest, CancelAccountDeletionResponse, DeleteAccountRequest,
    DeleteAccountResponse, GetSessionRequest, GetSessionResponse, GetUserRequest, GetUserResponse,
    ListUsersRequest, ListUsersResponse, PurgeAccountRequest, PurgeAccountResponse,
    UpdateUserRequest, UpdateUserResponse, ValidateRequest, ValidateResponse,
};
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect};
use tonic::{Request, Response, Status};

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

pub struct AuthorityApi {
    svc: AuthService,
}

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

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

fn status(error_value: &AuthError, request_id: Option<&str>) -> Status {
    let code = match error_value {
        AuthError::Unauthenticated | AuthError::SecurityVersionStale { .. } => {
            Code::Unauthenticated
        }
        AuthError::Forbidden => Code::Forbidden,
        AuthError::UserNotFound => Code::UserNotFound,
        AuthError::SessionNotFound => Code::SessionNotFound,
        AuthError::MalformedRequest => Code::MalformedRequest,
        AuthError::VersionMismatch { .. } => Code::Conflict,
        _ => Code::Internal,
    };
    error::status_of(code, error_value.message(), request_id)
}

fn user_status(error_value: &AuthError, request_id: Option<&str>) -> Status {
    match error_value {
        AuthError::UserNotFound => {
            error::status_of(Code::UserNotFound, "user not found", request_id)
        }
        other => status(other, request_id),
    }
}

fn access(user: &users::Model, mfa_verified: bool) -> AuthorityAccess {
    AuthorityAccess {
        id: user.id,
        username: user.username.clone(),
        allow_use: user.allow_use,
        forbid_login: user.forbid_login,
        endtime_forbid_login_at: convert::datetime_opt(user.endtime_forbid_login_at),
        is_admin: user.is_admin,
        security_version: user.security_version,
        deleted_at: convert::datetime_opt(user.deleted_at),
        mfa_verified,
    }
}

fn user(record: &users::Model) -> AuthorityUser {
    AuthorityUser {
        id: record.id,
        username: record.username.clone(),
        lowername: record.lowername.clone(),
        displayname: record.displayname.clone(),
        birth_date: record.birth_date.and_then(convert::date_timestamp),
        email: record.email.clone(),
        public_show_email: record.public_show_email,
        email_verified_at: convert::datetime_opt(record.email_verified_at),
        allow_use: record.allow_use,
        forbid_login: record.forbid_login,
        forbid_login_reason: record.forbid_login_reason.clone(),
        endtime_forbid_login_at: convert::datetime_opt(record.endtime_forbid_login_at),
        is_admin: record.is_admin,
        bot: record.bot,
        system: record.system,
        created_at: Some(convert::timestamp_from_datetime(record.created_at)),
        updated_at: Some(convert::timestamp_from_datetime(record.updated_at)),
        security_version: record.security_version,
        version: record.version,
        deleted_at: convert::datetime_opt(record.deleted_at),
    }
}

fn patch_value(patch: &AuthorityUserPatch) -> Result<serde_json::Value, String> {
    if patch.displayname.is_some() && patch.clear_displayname {
        return Err("patch carries both displayname and clear_displayname".into());
    }
    if let Some(username) = patch.username.as_deref()
        && username.trim().is_empty()
    {
        return Err("patch carries an empty username".into());
    }
    if patch.username.is_none()
        && patch.displayname.is_none()
        && !patch.clear_displayname
        && patch.public_show_email.is_none()
    {
        return Err("patch carries no updatable field".into());
    }
    let mut value = serde_json::json!({ "version": i64::from(patch.version) });
    if let Some(username) = patch.username.as_deref() {
        value["username"] = serde_json::Value::String(username.to_owned());
    }
    if patch.clear_displayname {
        value["displayname"] = serde_json::Value::Null;
    } else if let Some(displayname) = patch.displayname.as_deref() {
        value["displayname"] = serde_json::Value::String(displayname.to_owned());
    }
    if let Some(show) = patch.public_show_email {
        value["public_show_email"] = serde_json::Value::Bool(show);
    }
    Ok(value)
}

#[tonic::async_trait]
impl AuthorityService for AuthorityApi {
    async fn redeem_socket_ticket(
        &self,
        request: Request<crate::protos::auth::internal::v1::RedeemSocketTicketRequest>,
    ) -> Result<Response<crate::protos::auth::internal::v1::RedeemSocketTicketResponse>, Status>
    {
        let meta = context::meta(&request).await?;
        let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.AuthorityService");
        let result = self
            .redeem_socket_ticket_inner(meta.request_id(), request.into_inner())
            .await;
        timer.finish("RedeemSocketTicket", &result);
        result
    }
    async fn validate(
        &self,
        request: Request<ValidateRequest>,
    ) -> Result<Response<ValidateResponse>, Status> {
        let meta = context::meta(&request).await?;
        let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.AuthorityService");
        let result = self
            .validate_inner(meta.request_id(), request.into_inner())
            .await;
        timer.finish("Validate", &result);
        result
    }

    async fn get_user(
        &self,
        request: Request<GetUserRequest>,
    ) -> Result<Response<GetUserResponse>, Status> {
        let meta = context::meta(&request).await?;
        let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.AuthorityService");
        let result = self
            .get_user_inner(meta.request_id(), request.into_inner())
            .await;
        timer.finish("GetUser", &result);
        result
    }

    async fn batch_get_user(
        &self,
        request: Request<BatchGetUserRequest>,
    ) -> Result<Response<BatchGetUserResponse>, Status> {
        let meta = context::meta(&request).await?;
        let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.AuthorityService");
        let result = self
            .batch_get_user_inner(meta.request_id(), request.into_inner())
            .await;
        timer.finish("BatchGetUser", &result);
        result
    }

    async fn get_session(
        &self,
        request: Request<GetSessionRequest>,
    ) -> Result<Response<GetSessionResponse>, Status> {
        let meta = context::meta(&request).await?;
        let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.AuthorityService");
        let result = self
            .get_session_inner(meta.request_id(), request.into_inner())
            .await;
        timer.finish("GetSession", &result);
        result
    }

    async fn update_user(
        &self,
        request: Request<UpdateUserRequest>,
    ) -> Result<Response<UpdateUserResponse>, Status> {
        let meta = context::meta(&request).await?;
        let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.AuthorityService");
        let result = self
            .update_user_inner(meta.request_id(), request.into_inner())
            .await;
        timer.finish("UpdateUser", &result);
        result
    }

    async fn list_users(
        &self,
        request: Request<ListUsersRequest>,
    ) -> Result<Response<ListUsersResponse>, Status> {
        let meta = context::meta(&request).await?;
        let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.AuthorityService");
        let result = self
            .list_users_inner(meta.request_id(), request.into_inner())
            .await;
        timer.finish("ListUsers", &result);
        result
    }

    async fn delete_account(
        &self,
        request: Request<DeleteAccountRequest>,
    ) -> Result<Response<DeleteAccountResponse>, Status> {
        let meta = context::meta(&request).await?;
        let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.AuthorityService");
        let result = self
            .delete_account_inner(meta.request_id(), request.into_inner())
            .await;
        timer.finish("DeleteAccount", &result);
        result
    }

    async fn cancel_account_deletion(
        &self,
        request: Request<CancelAccountDeletionRequest>,
    ) -> Result<Response<CancelAccountDeletionResponse>, Status> {
        let meta = context::meta(&request).await?;
        let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.AuthorityService");
        let result = self
            .cancel_account_deletion_inner(meta.request_id(), request.into_inner())
            .await;
        timer.finish("CancelAccountDeletion", &result);
        result
    }

    async fn purge_account(
        &self,
        request: Request<PurgeAccountRequest>,
    ) -> Result<Response<PurgeAccountResponse>, Status> {
        let meta = context::meta(&request).await?;
        let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.AuthorityService");
        let result = self
            .purge_account_inner(meta.request_id(), request.into_inner())
            .await;
        timer.finish("PurgeAccount", &result);
        result
    }
}

impl AuthorityApi {
    #[tracing::instrument(skip_all, fields(request_id = %_request_id.unwrap_or_default()))]
    async fn redeem_socket_ticket_inner(
        &self,
        _request_id: Option<&str>,
        _payload: crate::protos::auth::internal::v1::RedeemSocketTicketRequest,
    ) -> Result<Response<crate::protos::auth::internal::v1::RedeemSocketTicketResponse>, Status>
    {
        Err(Status::unimplemented("socket tickets are not available"))
    }

    #[tracing::instrument(skip_all, fields(request_id = %_request_id.unwrap_or_default()))]
    async fn validate_inner(
        &self,
        _request_id: Option<&str>,
        _payload: ValidateRequest,
    ) -> Result<Response<ValidateResponse>, Status> {
        Err(Status::unimplemented(
            "access token validation is not available",
        ))
    }

    #[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
    async fn get_user_inner(
        &self,
        request_id: Option<&str>,
        payload: GetUserRequest,
    ) -> Result<Response<GetUserResponse>, Status> {
        let user_id = payload.user_id;
        if user_id <= 0 {
            return Err(error::invalid_argument(
                "user_id must be positive",
                request_id,
            ));
        }
        let record = self
            .svc
            .user(user_id)
            .await
            .map_err(|error_value| user_status(&error_value, request_id))?;
        Ok(Response::new(GetUserResponse {
            user: Some(user(&record)),
        }))
    }

    #[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
    async fn batch_get_user_inner(
        &self,
        request_id: Option<&str>,
        payload: BatchGetUserRequest,
    ) -> Result<Response<BatchGetUserResponse>, Status> {
        let user_ids = payload.user_ids;
        if user_ids.is_empty() {
            return Ok(Response::new(BatchGetUserResponse { users: Vec::new() }));
        }
        if user_ids.len() > crate::api::store::MAX_BATCH_USER_IDS {
            return Err(error::invalid_argument(
                "too many user ids in one batch",
                request_id,
            ));
        }
        let records = self
            .svc
            .users(&user_ids)
            .await
            .map_err(|error_value| user_status(&error_value, request_id))?;
        Ok(Response::new(BatchGetUserResponse {
            users: records.iter().map(user).collect(),
        }))
    }

    #[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
    async fn get_session_inner(
        &self,
        request_id: Option<&str>,
        payload: GetSessionRequest,
    ) -> Result<Response<GetSessionResponse>, Status> {
        if payload.user_id <= 0 {
            return Err(error::invalid_argument(
                "user_id must be positive",
                request_id,
            ));
        }
        if payload.version <= 0 {
            return Err(error::invalid_argument(
                "version must be positive",
                request_id,
            ));
        }
        if payload.session_id.is_empty() {
            return Err(error::invalid_argument(
                "session_id is required",
                request_id,
            ));
        }
        let (user, mfa_verified) = self
            .svc
            .check_session(
                payload.user_id,
                payload.version,
                &payload.session_id,
                request_id,
            )
            .await
            .map_err(|error_value| status(&error_value, request_id))?;
        tracing::debug!(
            request_id,
            user_id = payload.user_id,
            "authority revalidated a session"
        );
        Ok(Response::new(GetSessionResponse {
            access: Some(access(&user, mfa_verified)),
        }))
    }

    #[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
    async fn update_user_inner(
        &self,
        request_id: Option<&str>,
        payload: UpdateUserRequest,
    ) -> Result<Response<UpdateUserResponse>, Status> {
        if payload.user_id <= 0 {
            return Err(error::invalid_argument(
                "user_id must be positive",
                request_id,
            ));
        }
        let patch = payload
            .patch
            .ok_or_else(|| error::invalid_argument("patch is required", request_id))?;
        if patch.version <= 0 {
            return Err(error::invalid_argument(
                "version must be positive",
                request_id,
            ));
        }
        let value = patch_value(&patch).map_err(|message| {
            error::invalid_argument(format!("invalid patch: {message}"), request_id)
        })?;
        let record =
            self.svc.update(payload.user_id, value).await.map_err(
                |error_value| match error_value {
                    AuthError::VersionMismatch { .. } => {
                        error::status_of(Code::Conflict, "account version conflict", request_id)
                    }
                    other => status(&other, request_id),
                },
            )?;
        tracing::info!(
            request_id,
            user_id = record.id,
            version = record.version,
            "authority updated an account"
        );
        Ok(Response::new(UpdateUserResponse {
            user: Some(user(&record)),
        }))
    }

    #[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
    async fn list_users_inner(
        &self,
        request_id: Option<&str>,
        payload: ListUsersRequest,
    ) -> Result<Response<ListUsersResponse>, Status> {
        let filter = payload.filter.unwrap_or_default();
        if filter.user_ids.len() > crate::api::store::MAX_BATCH_USER_IDS {
            return Err(error::invalid_argument(
                "too many user ids in one filter",
                request_id,
            ));
        }
        let page = payload.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 page_size = convert::page_size(page.page_size);

        let query = order_by(
            users::Entity::find().filter(filter_condition(&filter)),
            payload.order,
        );

        let rows = query
            .limit(page_size as u64)
            .offset(offset)
            .all(self.svc.db().read())
            .await
            .map_err(|value| error::status(&crate::errors::AuthError::from(value), request_id))?;
        let next_page_token = if convert::has_more(offset, page_size, rows.len()) {
            store::page_token(offset + rows.len() as u64)
        } else {
            String::new()
        };
        Ok(Response::new(ListUsersResponse {
            users: rows.iter().map(user).collect(),
            page: Some(PageResponse { next_page_token }),
        }))
    }

    async fn delete_account_inner(
        &self,
        request_id: Option<&str>,
        payload: DeleteAccountRequest,
    ) -> Result<Response<DeleteAccountResponse>, Status> {
        if payload.user_id <= 0 {
            return Err(error::invalid_argument(
                "user_id must be positive",
                request_id,
            ));
        }
        self.svc
            .delete_account(
                payload.user_id,
                &payload.reason,
                request_id.unwrap_or_default(),
            )
            .await
            .map_err(|failure| error::status(&failure, request_id))?;
        Ok(Response::new(DeleteAccountResponse {}))
    }

    async fn cancel_account_deletion_inner(
        &self,
        request_id: Option<&str>,
        payload: CancelAccountDeletionRequest,
    ) -> Result<Response<CancelAccountDeletionResponse>, Status> {
        if payload.user_id <= 0 {
            return Err(error::invalid_argument(
                "user_id must be positive",
                request_id,
            ));
        }
        self.svc
            .cancel_account_deletion(payload.user_id, request_id.unwrap_or_default())
            .await
            .map_err(|failure| error::status(&failure, request_id))?;
        Ok(Response::new(CancelAccountDeletionResponse {}))
    }

    async fn purge_account_inner(
        &self,
        request_id: Option<&str>,
        payload: PurgeAccountRequest,
    ) -> Result<Response<PurgeAccountResponse>, Status> {
        if payload.user_id <= 0 {
            return Err(error::invalid_argument(
                "user_id must be positive",
                request_id,
            ));
        }
        self.svc
            .purge_account(payload.user_id, request_id.unwrap_or_default())
            .await
            .map_err(|failure| error::status(&failure, request_id))?;
        Ok(Response::new(PurgeAccountResponse {}))
    }
}

/// 把 proto 里的筛选条件翻成 SeaORM 条件树。未设置的字段一律跳过，
/// 因此空 filter 等价于「全量列表」。
fn filter_condition(
    filter: &crate::protos::auth::internal::v1::ListUsersFilter,
) -> sea_orm::Condition {
    use crate::protos::auth::internal::v1::{UserKind, UserState};
    use sea_orm::Condition;

    let mut any = Condition::all();

    // 子串匹配统一走 LIKE：username / lowername / displayname / email 四列任一命中即可。
    // 大小写不敏感靠 lowername 与 email 落库即小写来保证，displayname 则补一趟小写比较。
    let needle = filter.query.trim();
    if !needle.is_empty() {
        let lowered = needle.to_lowercase();
        any = any.add(
            Condition::any()
                .add(users::Column::Username.contains(needle))
                .add(users::Column::Lowername.contains(&lowered))
                .add(users::Column::Displayname.contains(needle))
                .add(users::Column::Email.contains(&lowered)),
        );
    }

    if !filter.user_ids.is_empty() {
        any = any.add(users::Column::Id.is_in(filter.user_ids.iter().copied()));
    }

    // 形态筛选：system 优先于 bot，与 UserKind 的互斥语义一致。
    match UserKind::try_from(filter.kind).ok() {
        Some(UserKind::Human) => {
            any = any
                .add(users::Column::Bot.eq(false))
                .add(users::Column::System.eq(false));
        }
        Some(UserKind::Bot) => {
            any = any
                .add(users::Column::Bot.eq(true))
                .add(users::Column::System.eq(false));
        }
        Some(UserKind::System) => {
            any = any.add(users::Column::System.eq(true));
        }
        Some(UserKind::Unspecified) | None => {}
    }

    // 状态筛选：优先级 DELETED > LOGIN_FORBIDDEN > UNAVAILABLE > ACTIVE，
    // 与 UserState 的注释一致；DB 侧用同一顺序表达，避免一条记录落两档。
    match UserState::try_from(filter.state).ok() {
        Some(UserState::Active) => {
            any = any
                .add(users::Column::DeletedAt.is_null())
                .add(users::Column::AllowUse.eq(true))
                .add(users::Column::ForbidLogin.eq(false));
        }
        Some(UserState::Unavailable) => {
            any = any
                .add(users::Column::DeletedAt.is_null())
                .add(users::Column::AllowUse.eq(false));
        }
        Some(UserState::LoginForbidden) => {
            any = any
                .add(users::Column::DeletedAt.is_null())
                .add(users::Column::ForbidLogin.eq(true));
        }
        Some(UserState::Deleted) => {
            any = any.add(users::Column::DeletedAt.is_not_null());
        }
        Some(UserState::Unspecified) | None => {}
    }

    if let Some(is_admin) = filter.is_admin {
        any = any.add(users::Column::IsAdmin.eq(is_admin));
    }
    if let Some(verified) = filter.email_verified {
        any = if verified {
            any.add(users::Column::EmailVerifiedAt.is_not_null())
        } else {
            any.add(users::Column::EmailVerifiedAt.is_null())
        };
    }

    any
}

fn order_by(
    mut query: sea_orm::Select<users::Entity>,
    order: i32,
) -> sea_orm::Select<users::Entity> {
    use crate::protos::auth::internal::v1::UserOrder;
    match UserOrder::try_from(order).ok() {
        Some(UserOrder::CreatedAtAsc) => query = query.order_by_asc(users::Column::CreatedAt),
        Some(UserOrder::LastLoginDesc) => {
            query = query
                .order_by_desc(users::Column::LastLoginAt)
                .order_by_desc(users::Column::Id);
        }
        Some(UserOrder::IdAsc) => query = query.order_by_asc(users::Column::Id),
        // 默认与 UNSPECIFIED 都走创建时间倒序；补 id 做稳定排序，翻页不跳行。
        Some(UserOrder::CreatedAtDesc) | Some(UserOrder::Unspecified) | None => {
            query = query
                .order_by_desc(users::Column::CreatedAt)
                .order_by_desc(users::Column::Id);
        }
    }
    query
}