package sqluser

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"slices"
	"strconv"
	"strings"

	"braces.dev/errtrace"
	"github.com/ahmetb/go-linq/v3"
	"github.com/google/uuid"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/adapter"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/constant"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/repository"
	workflowcommon "github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

type Service interface {
	CreateSqlUsers(ctx context.Context, params *BatchCreateUserParams) error
	UpdateSqlUserPassword(ctx context.Context, req *globalserverlessv1.ChangeSqlUserPasswdRequest) error
	CreateSqlUser(ctx context.Context, params *globalserverlessv1.SqlUserServiceCreateSqlUserRequest) error
	GetSqlUser(
		ctx context.Context,
		req *globalserverlessv1.SqlUserServiceGetSqlUserRequest,
	) (*globalserverlessv1.SqlUser, error)
	UpdateSqlUser(ctx context.Context, req *globalserverlessv1.SqlUserServiceUpdateSqlUserRequest) error
	DeleteSqlUser(ctx context.Context, req *globalserverlessv1.SqlUserServiceDeleteSqlUserRequest) error
	IsSqlUserExisting(ctx context.Context, req *globalserverlessv1.SqlUserServiceIsSqlUserExistingRequest) (bool, error)

	CreateSqlUserRole(ctx context.Context, req *globalserverlessv1.SqlUserServiceCreateSqlRoleRequest) error
	UpdateSqlUserRole(ctx context.Context, req *globalserverlessv1.SqlUserServiceUpdateSqlRoleRequest) error

	List(
		ctx context.Context,
		req *globalserverlessv1.ListSqlUsersRequest,
	) (map[string]*globalserverlessv1.SqlValues, []*globalserverlessv1.SqlUser, error)
	RotateRootPassword(ctx context.Context, req *globalserverlessv1.RotateRootPasswordRequest) (*globalserverlessv1.RotateRootPasswordResponse, error)
	RedeemRootPassword(ctx context.Context, req *globalserverlessv1.RedeemRootPasswordRequest) (*globalserverlessv1.RedeemRootPasswordResponse, error)
}

type sqlUserSvcImpl struct {
	clusterProxyProvider adapter.ClusterProxyProvider
	encryptor            adapter.EncryptorService
	clusterRepo          repository.ClusterRepo
	regionalSvc          adapter.RegionalService
	sys9Lifecycle        adapter.Sys9Lifecycle
	flow                 workflow.Workflow

	eventCommitter adapter.WALCommitter
	roleGetter     RoleGetter
}

func NewSqlUserService(
	encryptor adapter.EncryptorService,
	clusterProxyProvider adapter.ClusterProxyProvider,
	clusterRepo repository.ClusterRepo,
	regionSvc adapter.RegionalService,
	sys9Lifecycle adapter.Sys9Lifecycle,
	flow workflow.Workflow,
	committer adapter.WALCommitter,
) Service {
	return &sqlUserSvcImpl{
		clusterRepo:          clusterRepo,
		encryptor:            encryptor,
		regionalSvc:          regionSvc,
		sys9Lifecycle:        sys9Lifecycle,
		flow:                 flow,
		roleGetter:           GetRoles,
		eventCommitter:       committer,
		clusterProxyProvider: clusterProxyProvider,
	}
}

func (s *sqlUserSvcImpl) RotateRootPassword(
	ctx context.Context,
	req *globalserverlessv1.RotateRootPasswordRequest,
) (*globalserverlessv1.RotateRootPasswordResponse, error) {
	if req == nil || req.GetClusterId() == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster id is required")
	}

	clusterEn, err := s.clusterRepo.Get(ctx, req.GetClusterId(), false)
	if err != nil {
		return nil, errtrace.Wrap(err)
	}
	if !isPGStarterClusterForPassword(clusterEn) {
		return nil, bizerr.New(bizerr.ErrCodeParams, "rotate root password currently only supports postgres starter")
	}
	if clusterEn.ExternalInstanceID == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "missing external instance id for pg cluster")
	}

	running, err := s.flow.ListRunningDagInstances(ctx, []string{workflowcommon.RotateRootPassword}, clusterEn.ClusterId)
	if err != nil {
		return nil, errtrace.Wrap(err)
	}
	if len(running) > 0 {
		return &globalserverlessv1.RotateRootPasswordResponse{Rotating: true}, nil
	}

	operationToken := req.GetOperationToken()
	if operationToken == "" {
		operationToken = uuid.NewString()
	}

	if err := s.flow.RunDagWithBizId(
		ctx,
		workflowcommon.RotateRootPassword,
		fmt.Sprintf("%d", clusterEn.TenantId),
		fmt.Sprintf("%d", clusterEn.ProjectId),
		clusterEn.ClusterId,
		map[string]string{
			workflow.KeyTraceId:              workflow.GetTraceId(ctx),
			workflowcommon.KeyClusterId:      clusterEn.ClusterId,
			workflowcommon.KeyOperationToken: operationToken,
		},
	); err != nil {
		return nil, errtrace.Wrap(err)
	}

	return &globalserverlessv1.RotateRootPasswordResponse{Rotating: true}, nil
}

func (s *sqlUserSvcImpl) RedeemRootPassword(
	ctx context.Context,
	req *globalserverlessv1.RedeemRootPasswordRequest,
) (*globalserverlessv1.RedeemRootPasswordResponse, error) {
	if req == nil || req.GetClusterId() == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster id is required")
	}

	clusterEn, err := s.clusterRepo.Get(ctx, req.GetClusterId(), false)
	if err != nil {
		return nil, errtrace.Wrap(err)
	}
	if !isPGStarterClusterForPassword(clusterEn) {
		return nil, bizerr.New(bizerr.ErrCodeParams, "redeem root password currently only supports postgres starter")
	}

	password := ""
	if clusterEn.EncryptedRootPassword != "" {
		password, err = s.encryptor.DecryptBase64(ctx, clusterEn.EncryptedRootPassword)
		if err != nil {
			return nil, errtrace.Wrap(err)
		}
	}

	running, err := s.flow.ListRunningDagInstances(ctx, []string{workflowcommon.RotateRootPassword}, clusterEn.ClusterId)
	if err != nil {
		return nil, errtrace.Wrap(err)
	}

	return &globalserverlessv1.RedeemRootPasswordResponse{
		Rotating: len(running) > 0,
		Password: password,
	}, nil
}

func IsRotationTerminal(op *adapter.Sys9Operation) bool {
	if op == nil {
		return false
	}
	return slices.Contains([]string{"OPERATION_PHASE_SUCCEEDED", "OPERATION_PHASE_FAILED", "OPERATION_PHASE_CANCELLED"}, op.Phase)
}

func isPGStarterClusterForPassword(cluster *entity.Cluster) bool {
	if cluster == nil || cluster.Region == nil {
		return false
	}
	if cluster.ServicePlan != globalcommonv1.ServicePlan_Starter {
		return false
	}
	return strings.EqualFold(cluster.EngineType, "POSTGRESQL_COMPATIBLE")
}

func (s *sqlUserSvcImpl) CreateSqlUsers(ctx context.Context, req *BatchCreateUserParams) error {
	ctx, span := Tracer.Start(ctx, "Domain.BatchCreateSqlUser")
	defer span.End()

	tilog.Info(
		ctx,
		"try to create user in batches",
		zap.String("operator", StringifyOperator(req.Operator)),
		zap.Int("len(users)", len(req.Users)),
	)
	if err := s.check4batch(ctx, req); err != nil {
		return errtrace.Wrap(err)
	}
	for idx, user := range req.Users {
		tilog.Info(
			ctx,
			"user information",
			zap.Int("user", idx),
			zap.String("userName", user.UserEmail),
			zap.Any("userType", user.UserType),
			zap.Any("authMethod", user.AuthMethod),
		)
	}

	regionalCluster := req.Cluster
	cid, err := strconv.ParseUint(regionalCluster.ClusterId, 10, 64)
	if err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "invalid cluster id: %s", regionalCluster.ClusterId)
	}
	proxy, err := s.clusterProxyProvider.GetClusterProxy(
		ctx,
		cid,
		regionalCluster.ProxyEndpoint,
		regionalCluster.Version,
		BuildClusterOperator(req.Operator),
	)
	if err != nil {
		return errtrace.Wrap(err)
	}

	var sqls []string
	for _, user := range req.Users {
		oneUserSqls, err := s.buildCreateAndGrantSqls(ctx, user)
		if err != nil {
			return errtrace.Wrap(err)
		}

		sqls = append(sqls, oneUserSqls...)
	}
	if err := proxy.ExecuteMultiSQL(ctx, sqls); err != nil {
		return errtrace.Wrap(err)
	}

	return nil
}

func (s *sqlUserSvcImpl) CreateSqlUser(
	ctx context.Context,
	req *globalserverlessv1.SqlUserServiceCreateSqlUserRequest,
) error {
	en, cluster, err := s.getCluster(ctx, req.ClusterId)
	if err != nil {
		return errtrace.Wrap(err)
	}

	if err := commonCheck(req.Operator); err != nil {
		return errtrace.Wrap(err)
	}

	param := &CreateUserParams{
		Operator: req.Operator,
		Cluster:  cluster,
		Settings: req.Settings,
		CreateUserInfo: &CreateUserInfo{
			UserType:    req.UserType,
			UserName:    req.Username,
			UserEmail:   req.GetEmail(),
			AuthMethod:  req.AuthMethod,
			Credential:  req.GetCredential(),
			BuiltinRole: req.BuiltinRole,
			CustomRoles: req.CustomRoles,
		},
	}

	return errtrace.Wrap(s.createSqlUser(ctx, en, param))
}

func (s *sqlUserSvcImpl) createSqlUser(
	ctx context.Context,
	en *entity.Cluster,
	params *CreateUserParams,
) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "Domain.CreateUser")
	defer span.End()

	if params.CreateUserInfo == nil {
		return bizerr.New(bizerr.ErrCodeParams, "CreateUserInfo is required")
	}

	tilog.Info(
		ctx,
		"try to create user",
		zap.String("name", params.UserName),
		zap.Any("type", params.UserType),
		zap.Any("authMethod", params.AuthMethod),
		zap.Any("operator", params.Operator),
	)
	if params.Operator == nil {
		return bizerr.New(bizerr.ErrCodeParams, "operator is required")
	}
	if err := checkUser(ctx, params.CreateUserInfo); err != nil {
		return err
	}
	if err := checkManageUserRole(en, params.Cluster); err != nil {
		return err
	}

	regionalCluster := params.Cluster
	cid, err := strconv.ParseUint(regionalCluster.ClusterId, 10, 64)
	if err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "invalid cluster id: %s", regionalCluster.ClusterId)
	}
	proxy, err := s.clusterProxyProvider.GetClusterProxy(
		ctx,
		cid,
		regionalCluster.ProxyEndpoint,
		regionalCluster.Version,
		BuildClusterOperator(params.Operator),
	)
	if err != nil {
		return bizerr.WrapError(err, "domain.GetClusterProxy")
	}

	if skip, err := s.skipCreationIfIgnoreExists(ctx, proxy, params); err != nil {
		return err
	} else if skip {
		tilog.Info(ctx, "this user is existing, ignore creation.")
		return nil
	}

	sqls, bErr := s.buildCreateAndGrantSqls(ctx, params.CreateUserInfo)
	if bErr != nil {
		return bizerr.WrapError(bErr, "domain.buildCreateAndGrantSqls")
	}
	if err := proxy.ExecuteMultiSQL(ctx, sqls); err != nil {
		return bizerr.WrapError(err, "domain.ExecuteMultiSQL")
	}
	return nil
}

func (s *sqlUserSvcImpl) skipCreationIfIgnoreExists(
	ctx context.Context,
	proxy adapter.ClusterProxy,
	params *CreateUserParams,
) (skip bool, _ bizerr.BizError) {
	if params.Settings == nil || !params.Settings.IgnoreExists {
		return false, nil
	}

	existing, err := s.IsUserExisting(ctx, proxy, params.UserName)
	if err != nil {
		return false, err
	}

	tilog.Info(
		ctx,
		"user existence",
		zap.Bool("", existing),
		zap.String("username", params.UserName),
		zap.String("cluster", params.Cluster.ClusterId),
	)
	return existing, nil
}

func (s *sqlUserSvcImpl) IsUserExisting(
	ctx context.Context,
	proxy adapter.ClusterProxy,
	username string,
) (bool, bizerr.BizError) {
	_, err := GetRoles(ctx, proxy, username)
	if err != nil {
		tilog.Info(ctx, "check user existence", zap.Error(err))
		if err.ErrCode() == bizerr.ErrCodeClusterProxy && strings.Contains(err.ErrMsg(), SqlErNonExistingGrant) {
			tilog.Info(ctx, "sqluser is not existing", zap.String("username", username))
			return false, nil
		}
		return false, err
	}
	return true, nil
}

func (s *sqlUserSvcImpl) buildCreateAndGrantSqls(ctx context.Context, user *CreateUserInfo) ([]string, error) {
	var sqls []string

	createUserSQL, err := s.buildCreateUserSql(ctx, user)
	if err != nil {
		return nil, errtrace.Wrap(err)
	}
	sqls = append(sqls, createUserSQL)
	desiredRoles := s.getCreateUserDesiredRoles(ctx, user)
	sqls = append(sqls, BuildGrantUserRolesSql(user.UserName, desiredRoles)...)
	sqls = append(sqls, BuildEnableAllRoleSql(user.UserName))

	return sqls, nil
}

func (s *sqlUserSvcImpl) check4batch(ctx context.Context, params *BatchCreateUserParams) error {
	if params.Operator == nil {
		return errtrace.Wrap(bizerr.New(bizerr.ErrCodeParams, "operator is required"))
	}
	for _, each := range params.Users {
		if err := checkUser(ctx, each); err != nil {
			return errtrace.Wrap(err)
		}
	}
	if !params.ClusterAttributes.TokenBasedAuthEnabled {
		return errtrace.Wrap(errors.New("token based auth is not enabled"))
	}
	return nil
}

func checkUser(ctx context.Context, user *CreateUserInfo) bizerr.BizError {
	if user.AuthMethod == globalserverlessv1.SqlUser_AUTH_METHOD_UNSPECIFIED {
		return bizerr.New(bizerr.ErrCodeParams, "invalid authMethod: %v", user.AuthMethod)
	}
	if user.UserType == globalserverlessv1.SqlUser_TYPE_UNSPECIFIED {
		return bizerr.New(bizerr.ErrCodeParams, "invalid userType: %v", user.UserType)
	}

	if user.UserType == globalserverlessv1.SqlUser_TYPE_DELEGATE {
		if user.AuthMethod != globalserverlessv1.SqlUser_AUTH_METHOD_TOKEN {
			return bizerr.New(bizerr.ErrCodeParams, "only support token as auth_method if the user is delegate user")
		}

		if user.UserEmail == "" {
			tilog.Info(ctx, "userEmail is empty")
		}
	}

	if user.AuthMethod == globalserverlessv1.SqlUser_AUTH_METHOD_PASSWORD && user.Credential == "" {
		return bizerr.New(bizerr.ErrCodeParams, "credential is required if auth_method is password")
	}

	if err := checkCustomRoles(user.CustomRoles); err != nil {
		return err
	}
	return nil
}

func checkCustomRoles(roles []string) bizerr.BizError {
	for _, each := range roles {
		if _, ok := BuiltinRoleNameSet[each]; ok {
			return bizerr.New(bizerr.ErrCodeParams, "use builtin role in custom roles is prohibited, role=%s", each)
		}
	}

	return nil
}

func (s *sqlUserSvcImpl) buildCreateUserSql(ctx context.Context, user *CreateUserInfo) (string, error) {
	var createUserSql string
	switch user.AuthMethod {
	case globalserverlessv1.SqlUser_AUTH_METHOD_TOKEN:
		email, err := buildDelegateUserAttribute(user.UserEmail)
		if err != nil {
			return "", err
		}
		createUserSql = BuildCreateTokenBasedUserSql(user.UserName, email, constant.TiDBAuthTokenIssuer)
	case globalserverlessv1.SqlUser_AUTH_METHOD_PASSWORD:
		plaintextPassword, err := s.encryptor.DecryptBase64(ctx, user.Credential)
		if err != nil {
			return "", errtrace.Wrap(err)
		}
		createUserSql = BuildCreatePasswordBasedUserSql(user.UserName, plaintextPassword)
	default:
		return "", fmt.Errorf("invalid authMethod: %v", user.AuthMethod)
	}
	return createUserSql, nil
}

func (s *sqlUserSvcImpl) getCreateUserDesiredRoles(_ context.Context, user *CreateUserInfo) []string {
	var desiredRoles []string

	if user.BuiltinRole != globalserverlessv1.SqlUser_BUILTIN_ROLE_UNSPECIFIED {
		desiredRoles = append(desiredRoles, BuiltinRoleNameMap[user.BuiltinRole])
	}
	desiredRoles = append(desiredRoles, user.CustomRoles...)

	return desiredRoles
}

func (s *sqlUserSvcImpl) getCluster(
	ctx context.Context,
	clusterId string,
) (*entity.Cluster, *regionalserverlessv1.ServerlessRegionalServiceClusterInfo, error) {
	en, err := s.clusterRepo.Get(ctx, clusterId, false)
	if err != nil {
		return nil, nil, errtrace.Wrap(err)
	}
	if en == nil {
		return nil, nil, errtrace.Wrap(
			bizerr.New(bizerr.GrayscaleUpgradeDataNotInTargetSystem, "not found the cluster %s", clusterId),
		)
	}

	cluster, err := s.regionalSvc.ListCluster(ctx, en.Region, &globalserverlessv1.ServerlessServiceListRequest{
		Filter: &globalserverlessv1.ClusterFilter{
			ClusterIds: []string{clusterId},
			ProjectIds: []uint64{en.ProjectId},
		},
	})
	if err != nil {
		return nil, nil, errtrace.Wrap(err)
	}
	if len(cluster) != 1 {
		return nil, nil, errtrace.Wrap(
			bizerr.New(
				bizerr.GrayscaleUpgradeDataNotInTargetSystem,
				"not found the cluster %s in regional server. the length is %d",
				clusterId,
				len(cluster),
			),
		)
	}

	return en, cluster[0], nil
}

func buildDelegateUserAttribute(userEmail string) (string, error) {
	attribute := map[string]string{
		"email": userEmail,
	}
	out, err := json.Marshal(attribute)
	if err != nil {
		return "", err
	}
	return string(out), nil
}

func (s *sqlUserSvcImpl) UpdateSqlUser(
	ctx context.Context,
	req *globalserverlessv1.SqlUserServiceUpdateSqlUserRequest,
) error {
	ctx, span := Tracer.Start(ctx, "Domain.UpdateUser")
	defer span.End()

	if err := commonCheck(req.Operator); err != nil {
		return errtrace.Wrap(err)
	}

	en, cluster, err := s.getCluster(ctx, req.ClusterId)
	if err != nil {
		return errtrace.Wrap(err)
	}

	param := &UpdateUserParams{
		Operator:         req.Operator,
		Cluster:          cluster,
		Username:         req.Username,
		BuiltinRole:      req.BuiltinRole,
		CustomRoles:      req.CustomRoles,
		Password:         req.Password,
		ClearCustomRoles: req.ClearCustomRoles,
	}

	if err := updateSqlUserCheck(param, en); err != nil {
		return errtrace.Wrap(err)
	}

	return errtrace.Wrap(s.updateSqlUser(ctx, param))
}

func (s *sqlUserSvcImpl) updateSqlUser(ctx context.Context, params *UpdateUserParams) error {
	tilog.Info(
		ctx,
		"try to update user",
		zap.String("username", params.Username),
		zap.String("operator", StringifyOperator(params.Operator)),
	)
	regionalCluster := params.Cluster
	cid, err := strconv.ParseUint(regionalCluster.ClusterId, 10, 64)
	if err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "invalid cluster id: %s", regionalCluster.ClusterId)
	}
	proxy, err := s.clusterProxyProvider.GetClusterProxy(
		ctx,
		cid,
		regionalCluster.ProxyEndpoint,
		regionalCluster.Version,
		BuildClusterOperator(params.Operator),
	)
	if err != nil {
		return errtrace.Wrap(err)
	}

	if err := s.updateRolesAndPassword(ctx, proxy, params); err != nil {
		return errtrace.Wrap(err)
	}
	return nil
}

func (s *sqlUserSvcImpl) updateRolesAndPassword(
	ctx context.Context,
	proxy adapter.ClusterProxy,
	params *UpdateUserParams,
) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "updateRolesAndPassword")
	defer span.End()

	currentRoles, err := s.roleGetter(ctx, proxy, params.Username)
	if err != nil {
		return err
	}
	desiredRoles := s.getUpdateUserDesiredRoles(ctx, params, currentRoles)

	tilog.Info(ctx, "updateRolesAndPassword, currentRoles%s", zap.Strings("currentRoles", currentRoles))

	tilog.Info(ctx, "updateRolesAndPassword desiredRoles=%s", zap.Strings("desiredRoles", desiredRoles))

	var toDeleted, toAdded []string
	linq.From(currentRoles).Except(linq.From(desiredRoles)).ToSlice(&toDeleted)
	linq.From(desiredRoles).Except(linq.From(currentRoles)).ToSlice(&toAdded)

	tilog.Info(ctx, "update roles and password", zap.Strings("toAdded", toAdded), zap.Strings("toDeleted", toDeleted))
	updateFlag := false
	updateRoleFlag := false
	var sqls []string
	if len(toAdded) != 0 {
		sqls = append(sqls, BuildGrantUserRolesSql(params.Username, toAdded)...)
		updateFlag = true
		updateRoleFlag = true
	}
	if len(toDeleted) != 0 {
		sqls = append(sqls, BuildRevokeUserRolesSql(params.Username, toDeleted)...)
		updateFlag = true
		updateRoleFlag = true
	}

	if updateRoleFlag {
		sqls = append(sqls, BuildEnableAllRoleSql(params.Username))
	}

	if params.Password != nil {
		sqls = append(sqls, buildChangePasswordUserSql(params.Username, *params.Password))
		updateFlag = true
	}
	if updateFlag {
		tilog.Info(ctx, "updateRolesAndPassword", zap.Strings("MultiSQL", sqls))
		return proxy.ExecuteMultiSQL(ctx, sqls)
	}

	tilog.Info(ctx, "updateRolesAndPassword", zap.Strings("MultiSQL", sqls))
	return nil
}

func (s *sqlUserSvcImpl) getUpdateUserDesiredRoles(
	_ context.Context,
	params *UpdateUserParams,
	currentRoles []string,
) []string {
	currentBuiltinRoles := GetBuiltinRoles(currentRoles)
	currentCustomRoles := GetCustomRoles(currentRoles)

	var desiredRoles []string

	// For params.BuiltinRole, nil and unspecified value has different meanings.
	if params.BuiltinRole != nil {
		if *params.BuiltinRole != globalserverlessv1.SqlUser_BUILTIN_ROLE_UNSPECIFIED {
			desiredRoles = append(desiredRoles, BuiltinRoleNameMap[*params.BuiltinRole])
		}
	} else {
		desiredRoles = append(desiredRoles, currentBuiltinRoles...)
	}

	if params.ClearCustomRoles {
		return desiredRoles
	}

	if params.CustomRoles != nil {
		desiredRoles = append(desiredRoles, params.CustomRoles...)
	} else {
		desiredRoles = append(desiredRoles, currentCustomRoles...)
	}

	return desiredRoles
}

func updateSqlUserCheck(params *UpdateUserParams, en *entity.Cluster) bizerr.BizError {
	if params.Operator == nil {
		return bizerr.New(bizerr.ErrCodeParams, "operator is required")
	}
	if err := checkCustomRoles(params.CustomRoles); err != nil {
		return err
	}

	return checkManageUserRole(en, params.Cluster)
}

func (s *sqlUserSvcImpl) DeleteSqlUser(
	ctx context.Context,
	req *globalserverlessv1.SqlUserServiceDeleteSqlUserRequest,
) error {
	ctx, span := Tracer.Start(ctx, "Domain.DeleteUser")
	defer span.End()

	if err := commonCheck(req.Operator); err != nil {
		return errtrace.Wrap(err)
	}

	en, cluster, err := s.getCluster(ctx, req.ClusterId)
	if err != nil {
		return errtrace.Wrap(err)
	}

	param := &DeleteUserParams{
		Operator: req.Operator,
		Cluster:  cluster,
		Username: req.Username,
	}

	if err := deleteSqlUserCheck(param, en); err != nil {
		return errtrace.Wrap(err)
	}

	return errtrace.Wrap(s.deleteSqlUser(ctx, param))
}

func deleteSqlUserCheck(params *DeleteUserParams, en *entity.Cluster) bizerr.BizError {
	if params.Operator == nil {
		return bizerr.New(bizerr.ErrCodeParams, "operator is required")
	}

	return checkManageUserRole(en, params.Cluster)
}

func (s *sqlUserSvcImpl) deleteSqlUser(ctx context.Context, params *DeleteUserParams) error {
	tilog.Info(
		ctx,
		"try to delete user",
		zap.String("name", params.Username),
		zap.String("operator", StringifyOperator(params.Operator)),
	)

	regionalCluster := params.Cluster
	cid, err := strconv.ParseUint(regionalCluster.ClusterId, 10, 64)
	if err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "invalid cluster id: %s", regionalCluster.ClusterId)
	}
	proxy, err := s.clusterProxyProvider.GetClusterProxy(
		ctx,
		cid,
		regionalCluster.ProxyEndpoint,
		regionalCluster.Version,
		BuildClusterOperator(params.Operator),
	)
	if err != nil {
		return errtrace.Wrap(err)
	}

	if err := proxy.ExecuteSQL(ctx, BuildDropUserSql(params.Username)); err != nil {
		return errtrace.Wrap(err)
	}
	return nil
}

func (s *sqlUserSvcImpl) GetSqlUser(
	ctx context.Context,
	req *globalserverlessv1.SqlUserServiceGetSqlUserRequest,
) (*globalserverlessv1.SqlUser, error) {
	ctx, span := Tracer.Start(ctx, "GetUserSvc")
	defer span.End()

	en, regionalCluster, err := s.getCluster(ctx, req.ClusterId)
	if err != nil {
		return nil, errtrace.Wrap(err)
	}

	if req.Operator == nil {
		return nil, errtrace.Wrap(bizerr.New(bizerr.ErrCodeParams, "operator is required"))
	}

	// NOTE: we have to decrypt the credential if the auth_method is PASSWORD
	if req.Operator.AuthMethod != globalserverlessv1.SqlUser_AUTH_METHOD_TOKEN {
		return nil, errtrace.Wrap(bizerr.New(bizerr.ErrCodeParams, "invalid auth_method: %s", req.Operator.AuthMethod))
	}

	if err := checkManageUserRole(en, regionalCluster); err != nil {
		return nil, errtrace.Wrap(err)
	}

	cid, err := strconv.ParseUint(regionalCluster.ClusterId, 10, 64)
	if err != nil {
		return nil, bizerr.WrapError(err, "invalid cluster id: %s", regionalCluster.ClusterId)
	}
	proxy, err := s.clusterProxyProvider.GetClusterProxy(
		ctx,
		cid,
		regionalCluster.ProxyEndpoint,
		regionalCluster.Version,
		BuildClusterOperator(req.Operator),
	)
	if err != nil {
		return nil, errtrace.Wrap(err)
	}

	roles, err := GetRoles(ctx, proxy, req.Username)
	if err != nil {
		return nil, errtrace.Wrap(err)
	}
	builtinRole, customRoles := ParseRoles(roles)

	out := &globalserverlessv1.SqlUser{
		Username:    req.Username,
		BuiltinRole: builtinRole,
		CustomRoles: customRoles,
	}
	return out, nil
}

func checkManageUserRole(
	en *entity.Cluster,
	cluster *regionalserverlessv1.ServerlessRegionalServiceClusterInfo,
) bizerr.BizError {
	if !en.Attributes.TokenBasedAuthEnabled {
		return bizerr.New(bizerr.ErrCodeTokenBasedAuthRequired, "token based auth is not enabled")
	}

	if cluster.State != regionalserverlessv1.ClusterState_CLUSTER_STATE_READY {
		return bizerr.New(
			bizerr.ErrCodeUnexpectedClusterStatus,
			"can't manage user/role when the cluster is in %v",
			cluster.State,
		)
	}

	return nil
}

func (s *sqlUserSvcImpl) IsSqlUserExisting(
	ctx context.Context,
	req *globalserverlessv1.SqlUserServiceIsSqlUserExistingRequest,
) (bool, error) {
	ctx, span := Tracer.Start(ctx, "IsUserExistingSvc")
	defer span.End()

	_, regionalCluster, err := s.getCluster(ctx, req.ClusterId)
	if err != nil {
		return false, errtrace.Wrap(err)
	}

	if err := commonCheck(req.Operator); err != nil {
		return false, errtrace.Wrap(err)
	}
	cid, err := strconv.ParseUint(regionalCluster.ClusterId, 10, 64)
	if err != nil {
		return false, bizerr.WrapError(err, "invalid cluster id: %s", regionalCluster.ClusterId)
	}
	proxy, err := s.clusterProxyProvider.GetClusterProxy(
		ctx,
		cid,
		regionalCluster.ProxyEndpoint,
		regionalCluster.Version,
		BuildClusterOperator(req.Operator),
	)
	if err != nil {
		return false, errtrace.Wrap(err)
	}

	return errtrace.Wrap2(s.IsUserExisting(ctx, proxy, req.Username))
}

func commonCheck(operator *globalserverlessv1.SqlUserOperator) error {
	if operator == nil {
		return errtrace.Wrap(bizerr.New(bizerr.ErrCodeParams, "operator is required"))
	}

	// NOTE: we have to decrypt the credential if the auth_method is PASSWORD
	if operator.AuthMethod != globalserverlessv1.SqlUser_AUTH_METHOD_TOKEN {
		return errtrace.Wrap(bizerr.New(bizerr.ErrCodeParams, "invalid auth_method: %s", operator.AuthMethod))
	}

	return nil
}