package authcenter

import (
	"context"
	"encoding/json"
	"fmt"
	"math/rand"
	"strings"
	"sync"
	"time"

	"github.com/duke-git/lancet/v2/slice"
	"github.com/google/uuid"
	"github.com/tidbcloud/account/central/utils"
	"github.com/tidbcloud/account/commons/config"
	"github.com/tidbcloud/account/commons/database"
	apikeydao "github.com/tidbcloud/account/internel/dao/apikey"
	"github.com/tidbcloud/account/internel/dao/dbuser"
	"github.com/tidbcloud/account/internel/dao/role"
	"github.com/tidbcloud/account/thirdparty/infra"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/tiinf/errs"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"

	"github.com/tidbcloud/account/internel/dao"
	"github.com/tidbcloud/account/internel/dao/entity"
)

type ClusterUserSvcIface interface {
	CreateClusterUser(ctx context.Context, param *CreateClusterUserParam) (*ClusterUserInfo, error)
	GetClusterUser(ctx context.Context, clusterID uint64, username string, tokenExpiredDuration time.Duration) (*ClusterUserInfo, error)
	GenJwtToken(ctx context.Context, clusterId uint64, clusterUsername, userEmail string, tokenExpiredDuration time.Duration) (string, error)
	GetClusterCloudAdmin(ctx context.Context, clusterID uint64, tokenExpiredDuration time.Duration) (*clusterproxy.ClusterOperator, error)
	GetExistDBUser(ctx context.Context, clusterId uint64) (*clusterproxy.ClusterOperator, error)
}

var (
	clusterUserSvcInitOnce sync.Once
	clusterUserSvcInstance ClusterUserSvcIface
)

func GetClusterUserSvc() (ClusterUserSvcIface, error) {
	clusterUserSvcInitOnce.Do(func() {
		clusterUserSvcInstance = NewClusterUserSvc()
	})
	return clusterUserSvcInstance, nil
}

type ClusterUserSvc struct {
}

func NewClusterUserSvc() ClusterUserSvcIface {
	return &ClusterUserSvc{}
}

func (s *ClusterUserSvc) CreateClusterUser(ctx context.Context, param *CreateClusterUserParam) (*ClusterUserInfo, error) {
	tilog.Info(ctx, "CreateClusterUser start", zap.Any("param", param))
	if err := checkCreateClusterUserParam(param); err != nil {
		return nil, err
	}
	// idempotent check
	dbClusterUser, err := dao.GetDaoHub().ClusterUser.GetClusterUserByClusterIDUsername(ctx, database.DBConn.DB, param.ClusterID, param.Username)
	if err != nil {
		tilog.Error(ctx, "CreateClusterUser GetClusterUserByClusterIDUsername error", zap.Any("param", param), zap.Any("err", err))
		return nil, err
	}
	if dbClusterUser != nil && (dbClusterUser.Status == entity.Active || dbClusterUser.Status == entity.Pending) {
		tilog.Info(ctx, "CreateClusterUser dbClusterUser already exist, ignore to create", zap.Any("dbClusterUser", dbClusterUser))
		clusterUserInfo, err := s.convClusterUserEntity2ClusterUserInfo(ctx, dbClusterUser, param.TokenExpiredDuration)
		tilog.Info(ctx, "CreateClusterUser create ClusterUserInfo", zap.Any("clusterUsername", clusterUserInfo.ClusterUsername), zap.Any("credentialType", clusterUserInfo.CredentialType), zap.Any("err", err))
		return clusterUserInfo, err
	}
	// 1. gen cluster user entity
	clusterUser, err := s.genClusterUserEntity(ctx, param)
	if err != nil {
		tilog.Error(ctx, "CreateClusterUser genClusterUserEntity error", zap.Any("param", param), zap.Any("err", err))
		return nil, err
	}
	// 2. persist to db
	if dbClusterUser == nil {
		clusterUser, err = dao.GetDaoHub().ClusterUser.CreateClusterUser(ctx, database.DBConn.DB, clusterUser)
	} else {
		clusterUser.ID = dbClusterUser.ID
		err = dao.GetDaoHub().ClusterUser.UpdateClusterUserByID(ctx, database.DBConn.DB, clusterUser)
	}
	if err != nil {
		tilog.Error(ctx, "CreateClusterUser DB create error", zap.Any("clusterUser", clusterUser), zap.Any("err", err))
		return nil, err
	}
	tilog.Info(ctx, "CreateClusterUser DB create success", zap.Any("clusterUser", clusterUser))
	// 3. create on kernel
	clusterUserPriv := genClusterUserPriv(clusterUser, param)
	err = s.pushClusterUserToKernel(ctx, clusterUser, clusterUserPriv)
	if err != nil {
		tilog.Error(ctx, "CreateClusterUser pushClusterUserToKernel error", zap.Any("clusterUser", clusterUser), zap.Any("param", param), zap.Any("err", err))
	}
	// 4. convert and return
	clusterUserInfo, err := s.convClusterUserEntity2ClusterUserInfo(ctx, clusterUser, param.TokenExpiredDuration)
	if err != nil {
		tilog.Error(ctx, "CreateClusterUser convClusterUserEntity2ClusterUserInfo error", zap.Any("clusterUser", clusterUser), zap.Any("err", err))
		return nil, err
	}
	tilog.Info(ctx, "CreateClusterUser create ClusterUserInfo", zap.Any("clusterUsername", clusterUserInfo.ClusterUsername), zap.Any("credentialType", clusterUserInfo.CredentialType))
	return clusterUserInfo, nil
}

func (s *ClusterUserSvc) GetClusterUser(ctx context.Context, clusterID uint64, username string, tokenExpiredDuration time.Duration) (*ClusterUserInfo, error) {
	tilog.Info(ctx, "GetClusterUser start", zap.Any("clusterID", clusterID), zap.Any("username", username))
	dbClusterUser, err := dao.GetDaoHub().ClusterUser.GetClusterUserByClusterIDUsername(ctx, database.DBConn.DB, clusterID, username)
	if err != nil {
		tilog.Error(ctx, "GetClusterUser GetClusterUserByClusterIDUsername error", zap.Any("clusterID", clusterID), zap.Any("username", username), zap.Any("err", err))
		return nil, err
	}
	if dbClusterUser == nil || dbClusterUser.Status != entity.Active {
		dbClusterUser, err = s.toCompensateClusterUser(ctx, dbClusterUser, clusterID, username)
		if err != nil {
			tilog.Error(ctx, "GetClusterUser toCompensateClusterUser error", zap.Any("clusterID", clusterID), zap.Any("username", username), zap.Any("err", err))
			return nil, err
		}
	}
	if dbClusterUser == nil {
		tilog.Warn(ctx, "GetClusterUser dbClusterUser not found", zap.Any("clusterID", clusterID), zap.Any("username", username))
		return nil, errs.NewNotFound("dbClusterUser not found", nil)
	}
	// if it is necessary to check active status (means cluster user has already been created on kernel)
	if dbClusterUser.Status != entity.Active {
		tilog.Warn(ctx, "GetClusterUser dbClusterUser status not active", zap.Any("clusterID", clusterID), zap.Any("username", username), zap.Any("dbClusterUser", dbClusterUser), zap.Any("err", err))
		return nil, errs.NewNotFound("dbClusterUser status not active", nil)
	}
	clusterUserInfo, err := s.convClusterUserEntity2ClusterUserInfo(ctx, dbClusterUser, tokenExpiredDuration)
	if err != nil {
		tilog.Error(ctx, "GetClusterUser convClusterUserEntity2ClusterUserInfo error", zap.Any("dbClusterUser", dbClusterUser), zap.Any("err", err))
		return nil, err
	}
	tilog.Info(ctx, "GetClusterUser create ClusterUserInfo", zap.Any("clusterUsername", clusterUserInfo.ClusterUsername), zap.Any("credentialType", clusterUserInfo.CredentialType))
	return clusterUserInfo, nil
}

func (s *ClusterUserSvc) pushClusterUserToKernel(ctx context.Context, clusterUser *entity.ClusterUser, priv *ClusterUserPriv) error {
	err := s.createClusterUserOnKernel(ctx, clusterUser, priv)
	if err != nil {
		tilog.Warn(ctx, "pushClusterUserToKernel createClusterUserOnKernel error", zap.Any("clusterUser", clusterUser), zap.Any("priv", priv), zap.Any("err", err))
		// add offline task to create
		err = s.addCreateClusterUserTask(ctx, clusterUser, priv)
	} else {
		err = dao.GetDaoHub().ClusterUser.UpdateClusterUserByID(ctx, database.DBConn.DB, &entity.ClusterUser{
			Updatable: clusterUser.Updatable,
			Status:    entity.Active,
		})
	}
	return err
}

func genClusterUserPriv(clusterUser *entity.ClusterUser, param *CreateClusterUserParam) *ClusterUserPriv {
	var priv ClusterUserPriv
	if clusterUser.Username == CloudSystemSettingAdminUsername || clusterUser.Username == CloudRestrictUserAdminUsername {
		priv = ClusterUserPriv{
			Grants: genLimitedAdminGrants(clusterUser.Username),
		}
	} else {
		priv = ClusterUserPriv{
			BuiltinRole: param.BuiltinRole,
			CustomRoles: param.CustomRoles,
			Grants:      param.Grants,
		}
	}
	return &priv
}

func (s *ClusterUserSvc) toCompensateClusterUser(ctx context.Context, dbClusterUser *entity.ClusterUser, clusterID uint64, username string) (*entity.ClusterUser, error) {
	cluster, err := infra.GetHandler().GetClusterById(ctx, clusterID, true)
	if err != nil {
		tilog.Error(ctx, "toCompensateClusterUser GetClusterById error", zap.Any("clusterID", clusterID), zap.Any("err", err))
		return nil, err
	}
	if cluster == nil {
		tilog.Warn(ctx, "toCompensateClusterUser GetClusterById return nil", zap.Any("clusterID", clusterID))
		return nil, errs.NewNotFound("cluster not found", nil)
	}

	// cloud admin
	if username == CloudAdminUsername {
		return s.toSyncCmCloudAdmin(ctx, cluster)
	}
	// limited admin
	if username == CloudSystemSettingAdminUsername || username == CloudRestrictUserAdminUsername {
		return s.toCompensateLimitedAdmin(ctx, dbClusterUser, cluster, username)
	}

	return nil, nil
}

func (s *ClusterUserSvc) toCompensateLimitedAdmin(ctx context.Context, dbClusterUser *entity.ClusterUser, cluster *core.Cluster, username string) (*entity.ClusterUser, error) {
	// create cluster user in db if not exist
	if dbClusterUser == nil {
		var serverlessToken = ""
		if cluster.ServerlessToken != nil {
			serverlessToken = *cluster.ServerlessToken
		}
		clusterUser := &entity.ClusterUser{
			ClusterID:       entity.ID(cluster.Id),
			ServerlessToken: serverlessToken,
			Username:        username,
			Credential:      "",
			CredentialType:  entity.CredentialTypeToken,
			Description:     "",
			Status:          entity.Pending,
			TargetType:      entity.ClusterUserTargetLimitedUser,
			UserEmail:       "",
		}
		clusterUser, err := dao.GetDaoHub().ClusterUser.CreateClusterUser(ctx, database.DBConn.DB, clusterUser)
		if err != nil {
			return nil, err
		}
		dbClusterUser = clusterUser
	}

	priv := &ClusterUserPriv{
		Grants: genLimitedAdminGrants(username),
	}
	err := s.pushClusterUserToKernel(ctx, dbClusterUser, priv)
	return dbClusterUser, err
}

func genLimitedAdminGrants(username string) string {
	var grantArr []string
	if sqls, ok := KernelUserGrantSql[username]; ok {
		for _, sql := range sqls {
			oneSql := fmt.Sprintf(sql, username)
			grantArr = append(grantArr, oneSql)
		}
	}
	if len(grantArr) == 0 {
		return ""
	}
	return strings.Join(grantArr, " ")
}

func (s *ClusterUserSvc) toSyncCmCloudAdmin(ctx context.Context, cluster *core.Cluster) (*entity.ClusterUser, error) {
	clusterID := cluster.Id
	tilog.Info(ctx, "toSyncCmCloudAdmin start", zap.Any("clusterID", clusterID))
	infoRsp, err := infra.GetHandler().GetClusterConfidentialInfo(ctx, &core.GetClusterConfidentialInfoReq{
		ClusterId: clusterID,
	})
	tilog.Info(ctx, "toSyncCmCloudAdmin GetClusterConfidentialInfo result", zap.Any("clusterID", clusterID), zap.Any("infoRsp", infoRsp), zap.Any("err", err))

	if err != nil {
		return nil, err
	}
	if infoRsp.BaseResp.ErrCode != 0 || infoRsp.Info.EncryptedCloudAdminPasswd == "" {
		return nil, errs.NewInternalUnknown(fmt.Sprintf("get cluster confidential info fail: %d:%s", infoRsp.BaseResp.ErrCode, infoRsp.BaseResp.ErrMsg), nil)
	}

	serverlessToken := ""
	if cluster.ServerlessToken != nil {
		serverlessToken = *cluster.ServerlessToken
	}

	return createCmCloudAdminClusterUser(ctx, database.DBConn.DB, dao.GetDaoHub(), clusterID, serverlessToken, infoRsp.Info.EncryptedCloudAdminPasswd)
}

func (s *ClusterUserSvc) GetClusterCloudAdmin(ctx context.Context, clusterID uint64, tokenExpiredPeriod time.Duration) (*clusterproxy.ClusterOperator, error) {
	cluster, err := infra.GetHandler().GetClusterById(ctx, clusterID, true)
	if err != nil {
		tilog.Error(ctx, "GetClusterCloudAdmin GetClusterById error", zap.Any("clusterID", clusterID), zap.Any("err", err))
		return nil, err
	}
	if cluster == nil {
		tilog.Warn(ctx, "GetClusterCloudAdmin GetClusterById return nil", zap.Any("clusterID", clusterID))
		return nil, errs.NewNotFound("cluster not found", nil)
	}
	infoRsp, err := infra.GetHandler().GetClusterConfidentialInfo(ctx, &core.GetClusterConfidentialInfoReq{
		ClusterId: clusterID,
	})
	tilog.Info(ctx, "GetClusterCloudAdmin GetClusterConfidentialInfo result", zap.Any("clusterID", clusterID), zap.Any("infoRsp", infoRsp), zap.Any("err", err))
	if err != nil {
		return nil, err
	}
	if infoRsp.BaseResp.ErrCode != 0 || infoRsp.Info.EncryptedCloudAdminPasswd == "" {
		return nil, errs.NewInternalUnknown(fmt.Sprintf("get cluster confidential info fail: %d:%s", infoRsp.BaseResp.ErrCode, infoRsp.BaseResp.ErrMsg), nil)
	}
	credential, err := DecryptCredential(ctx, infoRsp.Info.EncryptedCloudAdminPasswd)
	if err != nil {
		tilog.Error(ctx, "GetClusterCloudAdmin DecryptCredential error", zap.Any("clusterID", clusterID), zap.Any("infoRsp", infoRsp), zap.Any("err", err))
		return nil, err
	}
	serverlessToken := ""
	if cluster.ServerlessToken != nil {
		serverlessToken = *cluster.ServerlessToken
	}
	clusterUsername := genClusterUsername(serverlessToken, CloudAdminUsername)

	clusterOperator := &clusterproxy.ClusterOperator{
		Username:   clusterUsername,
		Credential: credential,
		AuthMethod: clusterproxy.AuthMethodPassword,
	}
	return clusterOperator, nil
}

func (s *ClusterUserSvc) genNativePassword() (string, error) {
	return uuid.New().String(), nil
}

func (s *ClusterUserSvc) GenJwtToken(ctx context.Context, clusterId uint64, clusterUsername, userEmail string, tokenExpiredDuration time.Duration) (string, error) {
	cluster, err := infra.GetHandler().GetClusterById(ctx, clusterId, true)
	if err != nil {
		return "", err
	}
	if cluster == nil {
		return "", errs.NewNotFound(ErrMsgClusterNotExist, nil)
	}

	var orgId uint64 = config.ShareKeyOrg
	if !cluster.IsDevTier {
		orgId = cluster.TenantId
	}
	orgKeys, err := dao.GetDaoHub().DBUser.ListOrgKeys(ctx, database.DBConn.DB, dbuser.OrgKeyFilter{
		OrgIDs: []entity.ID{entity.ID(orgId)},
		Status: []entity.OrgKeyStatus{entity.OrgKeyStatusActive},
	})
	if err != nil {
		return "", err
	}
	if len(orgKeys) == 0 {
		return "", errs.NewNotFound(ErrMsgOrgKeyNotExist, nil)
	}
	orgKey := orgKeys[0]

	kms := utils.GetKeyServiceSingleton()
	sub := clusterUsername
	iss := config.Iss

	if tokenExpiredDuration == 0 {
		tokenExpiredDuration = config.JwtTokenExpiredTime
	}

	return utils.GetJwtToken(ctx, kms, orgKey.KeyID, userEmail, sub, tokenExpiredDuration, iss)
}

func (s *ClusterUserSvc) addCreateClusterUserTask(ctx context.Context, clusterUser *entity.ClusterUser, priv *ClusterUserPriv) error {
	var userId entity.ID = 0
	if clusterUser.TargetType == entity.ClusterUserTargetCloudUser {
		cloudUser, err := dao.GetDaoHub().User.GetUserByEmail(ctx, database.DBConn.DB, clusterUser.UserEmail)
		if err != nil {
			return err
		}
		userId = cloudUser.ID
	}

	marshal, _ := json.Marshal(priv)

	commandDBUserCluster := entity.CommandDBUserCluster{
		OrgID:        0,
		ClusterID:    clusterUser.ClusterID,
		Operator:     0,
		UserID:       userId,
		DbUser:       genClusterUsername(clusterUser.ServerlessToken, clusterUser.Username),
		Status:       entity.CommandStatusWaiting,
		Action:       entity.CommandActionCreate,
		TargetType:   clusterUser.TargetType,
		DbRole:       "",
		OperatorType: "",
		Priv:         string(marshal),
	}
	err := dao.GetDaoHub().DBUser.BatchCreateCommandDBUserCluster(ctx, database.DBConn.DB, []entity.CommandDBUserCluster{commandDBUserCluster})
	tilog.Info(ctx, "addCreateClusterUserTask result", zap.Any("commandDBUserCluster", commandDBUserCluster), zap.Any("err", err))
	return err
}

func checkCreateClusterUserParam(param *CreateClusterUserParam) error {
	if param.ClusterID == 0 {
		return errs.NewBadRequest("clusterId is required", nil)
	}
	if param.Username == "" {
		return errs.NewBadRequest("username is required", nil)
	}
	if _, ok := SupportClusterUserTargetType[param.TargetType]; !ok {
		return errs.NewBadRequest(fmt.Sprintf("targetType %s not support", param.TargetType), nil)
	}
	if param.TargetType == entity.ClusterUserTargetCloudUser && param.UserEmail == "" {
		return errs.NewBadRequest("userEmail is required when targetType is CloudUser", nil)
	}
	return nil
}

func (s *ClusterUserSvc) genClusterUserEntity(ctx context.Context, param *CreateClusterUserParam) (*entity.ClusterUser, error) {
	clusterUser := entity.ClusterUser{
		ClusterID:       entity.ID(param.ClusterID),
		ServerlessToken: param.ServerlessToken,
		Username:        param.Username,
		Credential:      "",
		CredentialType:  param.CredentialType,
		Description:     param.Description,
		Status:          entity.Pending,
		TargetType:      param.TargetType,
		UserEmail:       param.UserEmail, // for cloud user get jwt need email
	}
	if clusterUser.CredentialType == entity.CredentialTypePassword {
		// use param credential first
		var credential = param.Credential
		if param.Credential == "" {
			credential, _ = s.genNativePassword()
		}
		encryptCredential, err := EncryptCredential(ctx, credential)
		if err != nil {
			return nil, err
		}
		clusterUser.Credential = encryptCredential
	}

	return &clusterUser, nil
}

func (s *ClusterUserSvc) convClusterUserEntity2ClusterUserInfo(ctx context.Context, clusterUser *entity.ClusterUser, tokenExpiredDuration time.Duration) (*ClusterUserInfo, error) {
	var credential string
	var err error
	if clusterUser.CredentialType == entity.CredentialTypePassword {
		credential, err = DecryptCredential(ctx, clusterUser.Credential)
	} else {
		credential, err = s.GenJwtToken(ctx, clusterUser.ClusterID.Uint64(), clusterUser.Username, clusterUser.UserEmail, tokenExpiredDuration)
	}
	if err != nil {
		return nil, err
	}

	clusterUserInfo := &ClusterUserInfo{
		ClusterUsername: genClusterUsername(clusterUser.ServerlessToken, clusterUser.Username),
		Credential:      credential,
		CredentialType:  clusterUser.CredentialType,
	}
	return clusterUserInfo, nil
}

func genClusterUsername(serverlessToken, username string) string {
	if serverlessToken == "" {
		return username
	}
	return fmt.Sprintf("%s.%s", serverlessToken, username)
}

func (s *ClusterUserSvc) createClusterUserOnKernel(ctx context.Context, clusterUser *entity.ClusterUser, priv *ClusterUserPriv) error {
	if _, ok := noCreateKernelUsers[clusterUser.Username]; ok {
		return nil
	}
	cluster, err := infra.GetHandler().GetClusterById(ctx, clusterUser.ClusterID.Uint64(), true)
	if err != nil {
		return err
	}
	if cluster == nil {
		return errs.NewNotFound(fmt.Sprintf("cluster %d is not exist", clusterUser.ClusterID), nil)
	}

	operator, err := s.GetExistDBUser(ctx, cluster.Id)
	if err != nil {
		return err
	}

	createUserInfo := &infra.CreateUserInfo{
		UserName: genClusterUsername(clusterUser.ServerlessToken, clusterUser.Username),
		UserAttribute: map[string]string{
			"email": clusterUser.UserEmail,
		},
		AuthMethod:  clusterproxy.AuthMethodToken,
		BuiltinRole: &priv.BuiltinRole,
		CustomRoles: priv.CustomRoles,
		Grants:      priv.Grants,
	}
	if clusterUser.CredentialType == entity.CredentialTypePassword {
		createUserInfo.AuthMethod = clusterproxy.AuthMethodPassword
		createUserInfo.EncryptCredential = clusterUser.Credential
	}

	params := infra.CreateUserParams{
		Cluster:  cluster,
		Operator: operator,
		Settings: &core.CreateDBUserReq_Settings{
			IgnoreExists: true,
		},
		CreateUserInfo: createUserInfo,
	}
	err = infra.CreateUser(ctx, &params)
	tilog.Info(ctx, "createClusterUserOnKernel result", zap.Any("createUserInfo", createUserInfo), zap.Any("err", err))
	return err
}

func (s *ClusterUserSvc) GetExistDBUser(ctx context.Context, clusterId uint64) (*clusterproxy.ClusterOperator, error) {
	bm := utils.Begin(ctx, "GetExistDBUser", zap.Any("clusterId", clusterId))
	defer bm.OnExit(ctx)

	cluster, err := infra.GetHandler().GetClusterById(ctx, clusterId, true)
	if err != nil {
		tilog.Warn(ctx, "GetExistDbUser GetClusterById error", zap.Any("clusterID", clusterId), zap.Error(err))
		return nil, err
	}
	if cluster == nil {
		tilog.Info(ctx, "GetExistDbUser cluster is nil", zap.Any("clusterID", clusterId))
		return nil, errs.NewNotFound(ErrMsgClusterNotExist, nil)
	}

	candidates, err := s.getOperatorCandidates(ctx, cluster)
	if err != nil {
		tilog.Warn(ctx, "GetExistDbUser getOperatorCandidates error", zap.Any("clusterID", cluster.Id), zap.Error(err))
		return nil, err
	}
	tilog.Info(ctx, fmt.Sprintf("GetExistDbUser get operator candidates count: %d", len(candidates)))
	if len(candidates) == 0 {
		return nil, errs.NewNotFound(ErrMsgOperatorCandidatesEmpty, nil)
	}

	cloudAdmin, err := s.GetClusterCloudAdmin(ctx, cluster.Id, 0)
	if err != nil {
		tilog.Warn(ctx, "GetExistDbUser GetClusterCloudAdmin error", zap.Any("clusterID", cluster.Id), zap.Error(err))
		return nil, err
	}

	isDedicated := !cluster.IsDevTier

	var dbUsername2CandidateMap = map[string]OperatorCandidate{}
	var dbUsernames []string
	for i := range candidates {
		u := candidates[i]
		req := utils.GetDBUserReq{
			Email:           u.TargetVal,
			ClusterId:       &cluster.Id,
			IsDedicated:     &isDedicated,
			ServerlessToken: cluster.ServerlessToken,
		}
		if u.TargetType == entity.RoleBindingTargetAPIKey {
			req.TargetType = entity.RoleBindingTargetAPIKey.String()
		}
		operatorDbUser, err := utils.GetDBUser(ctx, req)
		if err != nil {
			tilog.Warn(ctx, "GetExistDBUser GetDBUser error", zap.Any("candidate", u), zap.Any("err", err))
			continue
		}
		dbUsernames = append(dbUsernames, *operatorDbUser)
		dbUsername2CandidateMap[*operatorDbUser] = u
	}
	if len(dbUsernames) == 0 {
		tilog.Warn(ctx, "GetExistDbUser get dbUsernames empty", zap.Any("clusterID", cluster.Id), zap.Error(err))
		return nil, errs.NewNotFound(ErrMsgNotFoundValidOperatorDbUsername, nil)
	}
	tilog.Info(ctx, fmt.Sprintf("GetExistDbUser get dbUsernames count: %d", len(dbUsernames)))

	chunkList := slice.Chunk(dbUsernames, 30)
	var rerr error
	for _, chunk := range chunkList {
		operator, err := s.selectOperator(ctx, cluster, cloudAdmin, dbUsername2CandidateMap, chunk)
		if err == nil {
			return operator, nil
		}
		rerr = err
	}
	return nil, rerr
}

func (s *ClusterUserSvc) selectOperator(ctx context.Context, cluster *core.Cluster, operator *clusterproxy.ClusterOperator, dbUsername2CandidateMap map[string]OperatorCandidate, dbUsernames []string) (*clusterproxy.ClusterOperator, error) {
	_, existDbUsers, err := infra.ListUsers(ctx, &infra.ListUsersParams{
		Cluster:  cluster,
		Operator: operator,
		Filter: &core.ListDBUsersFilter{
			Users: dbUsernames,
		},
	})
	if err != nil {
		tilog.Warn(ctx, "GetExistDbUser infra ListUsers error", zap.Any("clusterID", cluster.Id), zap.Any("dbUsernames", dbUsernames), zap.Error(err))
		return nil, err
	}

	var roleAdminDbUsers []infra.DBUser
	for i := range existDbUsers {
		eu := existDbUsers[i]
		if eu.BuiltinRole == entity.DbRoleAdmin {
			roleAdminDbUsers = append(roleAdminDbUsers, eu)
		}
	}
	if len(roleAdminDbUsers) == 0 {
		tilog.Info(ctx, "GetExistDbUser not found role admin db user", zap.Any("clusterID", cluster.Id))
		return nil, errs.NewNotFound(ErrMsgNotFoundRoleAdminDbUser, nil)
	}

	var useOperator *clusterproxy.ClusterOperator
	cnt := len(roleAdminDbUsers)
	rand.Seed(time.Now().UnixNano())
	retryCnt := 0
	var terr error
	for {
		if retryCnt > 3 {
			break
		}
		index := rand.Intn(cnt)
		findRoleAdminDbUser := roleAdminDbUsers[index]
		clusterUsername := findRoleAdminDbUser.Username
		candidate := dbUsername2CandidateMap[clusterUsername]
		email := ""
		if candidate.TargetType == entity.RoleBindingTargetUser {
			email = candidate.TargetVal
		}
		token, err := s.GenJwtToken(ctx, cluster.Id, clusterUsername, email, 0)
		if err != nil {
			tilog.Warn(ctx, "GetExistDbUser GetJwtToken error", zap.Any("email", email), zap.Any("dbUser", findRoleAdminDbUser.Username), zap.Any("err", err))
			terr = err
		} else {
			useOperator = &clusterproxy.ClusterOperator{
				Username:   findRoleAdminDbUser.Username,
				AuthMethod: clusterproxy.AuthMethodToken,
				Credential: token,
			}
			terr = nil
			tilog.Info(ctx, fmt.Sprintf("GetExistDBUser find role admin operator %s", useOperator.Username))
			break
		}
		retryCnt++
	}

	return useOperator, terr
}

func (s *ClusterUserSvc) getOperatorCandidates(ctx context.Context, cluster *core.Cluster) ([]OperatorCandidate, error) {
	orgId := cluster.TenantId
	projectId := cluster.ProjectId

	roleBindings, err := dao.GetDaoHub().Role.ListRoleBindings(ctx, database.DBConn.DB, role.RoleBindingFilter{
		TargetType: []entity.TargetType{entity.RoleBindingTargetUser},
		ScopeTypes: []entity.ScopeType{entity.RoleBindingScopeOrg},
		ScopeIDs:   []entity.ID{entity.ID(orgId)},
		Status:     []string{entity.RoleBindingStatusActive},
		Roles:      []string{entity.RoleOrgOwner},
	})
	if err != nil {
		tilog.Warn(ctx, "getOperatorCandidates list org role bindings error", zap.Any("clusterID", cluster.Id), zap.Any("orgID", orgId), zap.Error(err))
	}
	projectRoleBindings, err := dao.GetDaoHub().Role.ListRoleBindings(ctx, database.DBConn.DB, role.RoleBindingFilter{
		TargetType: []entity.TargetType{entity.RoleBindingTargetUser},
		ScopeTypes: []entity.ScopeType{entity.RoleBindingScopeProject},
		ScopeIDs:   []entity.ID{entity.ID(projectId)},
		Status:     []string{entity.RoleBindingStatusActive},
		Roles:      []string{entity.RoleProjectOwner},
	})
	if err != nil {
		tilog.Warn(ctx, "getOperatorCandidates list project role bindings error", zap.Any("clusterID", cluster.Id), zap.Any("projectID", projectId), zap.Error(err))
	}
	// find open api key
	apiKeyRoleBindings, err := dao.GetDaoHub().Role.ListRoleBindings(ctx, database.DBConn.DB, role.RoleBindingFilter{
		TargetType: []entity.TargetType{entity.RoleBindingTargetAPIKey},
		ScopeTypes: []entity.ScopeType{entity.RoleBindingScopeOrg},
		ScopeIDs:   []entity.ID{entity.ID(orgId)},
		Status:     []string{entity.RoleBindingStatusActive},
		Roles:      []string{entity.RoleApiKeyOrgOwner, entity.RoleOrgOwner},
	})
	if err != nil {
		tilog.Warn(ctx, "getOperatorCandidates list api key relate org role bindings error", zap.Any("clusterID", cluster.Id), zap.Any("orgID", orgId), zap.Error(err))
	}

	if len(roleBindings) == 0 && len(projectRoleBindings) == 0 && len(apiKeyRoleBindings) == 0 {
		return nil, errs.NewNotFound("not found org/project/apikey owner", nil)
	}

	// find active user
	var orgOwnerIdMap = map[uint64]struct{}{}
	var ownerIds []uint64
	for _, r := range roleBindings {
		ownerIds = append(ownerIds, r.TargetID.Uint64())
		orgOwnerIdMap[r.TargetID.Uint64()] = struct{}{}
	}
	for _, r := range projectRoleBindings {
		if _, ok := orgOwnerIdMap[r.TargetID.Uint64()]; !ok {
			ownerIds = append(ownerIds, r.TargetID.Uint64())
		}
	}
	ownerUsers, err := dao.GetDaoHub().User.ListUsers(ctx, database.DBConn.DB, ownerIds, []entity.UserStatus{entity.UserStatusActive})
	if err != nil {
		tilog.Warn(ctx, "getOperatorCandidates ListUsers error", zap.Any("clusterID", cluster.Id), zap.Any("userIds", ownerIds), zap.Error(err))
	}

	// find active api key
	var apikeyIds []uint64
	for _, r := range apiKeyRoleBindings {
		apikeyIds = append(apikeyIds, r.TargetID.Uint64())
	}
	var ownerApiKeys []entity.ApiKey
	if len(apikeyIds) > 0 {
		param := apikeydao.ListApiKeysParam{
			Filter: apikeydao.ApiKeyFilter{
				ID: apikeyIds,
			},
		}
		keys, err := dao.GetDaoHub().ApiKey.ListApiKeys(ctx, database.DBConn.DB, &param)
		if err != nil {
			tilog.Warn(ctx, "getOperatorCandidates ListApiKeys error", zap.Any("clusterID", cluster.Id), zap.Any("apikeyIds", apikeyIds), zap.Error(err))
		}
		if keys != nil && len(keys.List) > 0 {
			ownerApiKeys = keys.List
		}
	}

	var candidates []OperatorCandidate
	if len(ownerUsers) > 0 {
		for _, ou := range ownerUsers {
			candidate := OperatorCandidate{
				TargetType: entity.RoleBindingTargetUser,
				TargetID:   ou.ID.Uint64(),
				TargetVal:  ou.Email,
			}
			candidates = append(candidates, candidate)
		}
	}
	if len(ownerApiKeys) > 0 {
		for _, ok := range ownerApiKeys {
			candidate := OperatorCandidate{
				TargetType: entity.RoleBindingTargetAPIKey,
				TargetID:   ok.ID.Uint64(),
				TargetVal:  ok.AccessKey,
			}
			candidates = append(candidates, candidate)
		}
	}

	return candidates, nil
}