package service

import (
	"context"
	"encoding/json"

	"github.com/pingcap/errors"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/rpc"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	"go.uber.org/zap"
	"gorm.io/gorm"
)

type UserInfo struct {
	Username string
	// UserEmail is the full email address of this user
	// 1. optional
	// 2. is required if auth_method is token and empty string is acceptable
	UserEmail  string
	UserType   constant.SqlUserType
	AuthMethod constant.AuthMethod
	Credential string // optional, required if auth_method is password
	Roles      []string
}

type SqlOperator struct {
	Username   string
	AuthMethod constant.AuthMethod
	Credential string
}

func ConvertAuthMethod(m rpc.DBUser_AuthMethod) (constant.AuthMethod, error) {
	var authMethod constant.AuthMethod
	switch m {
	case rpc.DBUser_METHOD_PASSWORD:
		authMethod = constant.AuthMethodToken
	case rpc.DBUser_METHOD_TOKEN:
		authMethod = constant.AuthMethodToken
	default:
		return constant.AuthMethodUnknown, errors.New("unknown authmethod")
	}
	return authMethod, nil
}

func ConvertDBUserType(t rpc.DBUser_Type) (constant.SqlUserType, error) {
	var userType constant.SqlUserType
	switch t {
	case rpc.DBUser_TYPE_DELEGATE:
		userType = constant.SqlUserTypeDelegate
	case rpc.DBUser_TYPE_USER_DEFINED:
		userType = constant.SqlUserTypeUserDefined
	default:
		return constant.SqlUserTypeUnknown, errors.New("unknown user type")
	}
	return userType, nil
}

func ConvertRoles(b rpc.DBUser_BuiltinRole, cs []string) ([]string, error) {
	var roles []string
	switch b {
	case rpc.DBUser_ROLE_ADMIN:
		roles = append(roles, string(constant.BuiltinRoleAdmin))
	case rpc.DBUser_ROLE_MEMBER:
		roles = append(roles, string(constant.BuiltinRoleMember))
	default:
		return nil, errors.New("unknown builtin role")
	}
	roles = append(roles, cs...)
	return roles, nil
}

type UserRoleService struct {
	infraFacade        facade.InfraFacade
	clusterProxyFacade facade.ClusterProxyProviderFacade
	db                 *gorm.DB
}

func NewUserRoleService(
	infraFacade facade.InfraFacade,
	clusterProxyFacade facade.ClusterProxyProviderFacade,
	db *gorm.DB,
) *UserRoleService {
	return &UserRoleService{
		infraFacade:        infraFacade,
		clusterProxyFacade: clusterProxyFacade,
		db:                 db,
	}
}

func convertSqlOperatorToClusterOperator(operator *SqlOperator) *clusterproxy.ClusterOperator {
	return &clusterproxy.ClusterOperator{
		Username:   operator.Username,
		AuthMethod: clusterproxy.AuthMethod(operator.AuthMethod),
		Credential: operator.Credential,
	}
}

func (s *UserRoleService) CreateUser(
	ctx context.Context,
	operator *SqlOperator,
	branch *entity.Branch,
	userInfo *UserInfo,
) bizerr.BizError {
	logger := log.WithContextL(ctx).
		With(zap.String("branchId", branch.ID), zap.String("username", userInfo.Username), zap.String("operatorUsername", operator.Username))
	logger.Info("going to create user")

	if err := checkCreateUser(operator, branch, userInfo); err != nil {
		logger.Error("failed to check create user parameters", zap.Error(err))
		return bizerr.New(bizerr.ErrCodeParams, "failed to check create user parameters")
	}

	cr, err := s.infraFacade.GetBranch(ctx, branch.ProjectID, branch.ID)
	if err != nil {
		logger.Error("failed to get branch cr", zap.Error(err))
		return bizerr.New(bizerr.ErrCodeSvcInfraAPI, "failed to get branch cr")
	}
	clusterOperator := convertSqlOperatorToClusterOperator(operator)
	proxy, err := s.clusterProxyFacade.GetBranchProxy(ctx, cr, clusterOperator)
	if err != nil {
		logger.Error("failed to get proxy", zap.Error(err))
		return bizerr.New(errorcode.ErrCodeSvcClusterProxy, "failed to get proxy")
	}

	sqls := s.buildCreateAndGrantSqls(userInfo)
	if err := proxy.ExecuteMultiSQL(ctx, sqls); err != nil {
		logger.Error("failed to create user", zap.Error(err))
		return errorcode.MarkClusterSQLErrorUserActionable(err)
	}

	return nil
}

func (s *UserRoleService) CreateUsers(
	ctx context.Context,
	operator *SqlOperator,
	branch *entity.Branch,
	userInfos []*UserInfo,
) bizerr.BizError {
	logger := log.WithContextL(ctx).
		With(zap.String("branchId", branch.ID), zap.String("operatorUsername", operator.Username))
	logger.Info("going to create users")

	if err := checkCreateUsers(operator, branch, userInfos); err != nil {
		logger.Error("failed to check create users parameters", zap.Error(err))
		return bizerr.New(bizerr.ErrCodeParams, "failed to check create users parameters")
	}

	for idx, userInfo := range userInfos {
		logger.Sugar().
			Infof("users[%d], name=%s, tp=%s, authMethod=%s", idx, userInfo.Username, userInfo.UserType, userInfo.AuthMethod)
	}

	cr, err := s.infraFacade.GetBranch(ctx, branch.ProjectID, branch.ID)
	if err != nil {
		logger.Error("failed to get branch cr", zap.Error(err))
		return bizerr.New(bizerr.ErrCodeSvcInfraAPI, "failed to get branch cr")
	}
	clusterOperator := convertSqlOperatorToClusterOperator(operator)
	proxy, err := s.clusterProxyFacade.GetBranchProxy(ctx, cr, clusterOperator)
	if err != nil {
		logger.Error("failed to get proxy", zap.Error(err))
		return bizerr.New(errorcode.ErrCodeSvcClusterProxy, "failed to get proxy")
	}

	var sqls []string
	for _, userInfo := range userInfos {
		oneUserSqls := s.buildCreateAndGrantSqls(userInfo)
		sqls = append(sqls, oneUserSqls...)
	}
	if err := proxy.ExecuteMultiSQL(ctx, sqls); err != nil {
		logger.Error("failed to create users", zap.Error(err))
		return errorcode.MarkClusterSQLErrorUserActionable(err)
	}

	return nil
}

func checkCreateUser(operator *SqlOperator, branch *entity.Branch, userInfo *UserInfo) error {
	if err := checkOperator(operator); err != nil {
		return err
	}
	if err := checkUser(userInfo); err != nil {
		return err
	}
	return checkBranchStatus(branch)
}

func checkCreateUsers(operator *SqlOperator, branch *entity.Branch, userInfos []*UserInfo) error {
	if err := checkOperator(operator); err != nil {
		return err
	}
	for _, userInfo := range userInfos {
		if err := checkUser(userInfo); err != nil {
			return err
		}
	}
	return checkBranchStatus(branch)
}

func checkOperator(operator *SqlOperator) error {
	if operator == nil {
		return errors.New("operator is required")
	}
	return nil
}

func checkBranchStatus(branch *entity.Branch) error {
	if branch.Status != constant.BranchStatusRunning {
		return errors.New("branch is not running")
	}
	return nil
}

func checkUser(userInfo *UserInfo) error {
	if userInfo.UserType == constant.SqlUserTypeDelegate && userInfo.AuthMethod != constant.AuthMethodToken {
		return errors.New("only support token as auth_method if the user is delegate user")
	}

	if userInfo.AuthMethod == constant.AuthMethodPassword && userInfo.Credential == "" {
		return errors.New("credential is required if auth_method is password")
	}

	return nil
}

func (s *UserRoleService) buildCreateAndGrantSqls(userInfo *UserInfo) []string {
	var sqls []string

	createUserSql := s.buildCreateUserSql(userInfo)
	sqls = append(sqls, createUserSql)
	sqls = append(sqls, s.buildGrantUserRolesSql(userInfo.Username, userInfo.Roles)...)
	sqls = append(sqls, s.buildEnableAllRoleSql(userInfo.Username))

	return sqls
}

func (s *UserRoleService) buildCreateUserSql(userInfo *UserInfo) string {
	var createUserSql string
	switch userInfo.AuthMethod {
	case constant.AuthMethodToken:
		createUserSql = s.buildCreateTokenBasedUserSql(
			userInfo.Username,
			buildDelegateUserAttribute(userInfo.UserEmail),
			constant.TiDBAuthTokenIssuer,
		)
	case constant.AuthMethodPassword:
		createUserSql = s.buildCreatePasswordBasedUserSql(userInfo.Username, userInfo.Credential)
	}
	return createUserSql
}

func buildDelegateUserAttribute(userEmail string) string {
	attribute := map[string]string{
		"email": userEmail,
	}
	out, _ := json.Marshal(attribute)
	return string(out)
}

func (s *UserRoleService) DeleteUser(
	ctx context.Context,
	operator *SqlOperator,
	branch *entity.Branch,
	userInfo *UserInfo,
) bizerr.BizError {
	logger := log.WithContextL(ctx).
		With(zap.String("branchId", branch.ID), zap.String("username", userInfo.Username), zap.String("operatorUsername", operator.Username))
	logger.Info("going to delete user")

	if err := checkDeleteUser(operator, branch); err != nil {
		logger.Error("failed to check delete user parameters", zap.Error(err))
		return bizerr.New(bizerr.ErrCodeParams, "failed to check delete user parameters")
	}

	cr, err := s.infraFacade.GetBranch(ctx, branch.ProjectID, branch.ID)
	if err != nil {
		logger.Error("failed to get branch cr", zap.Error(err))
		return bizerr.New(bizerr.ErrCodeSvcInfraAPI, "failed to get branch cr")
	}
	clusterOperator := convertSqlOperatorToClusterOperator(operator)
	proxy, err := s.clusterProxyFacade.GetBranchProxy(ctx, cr, clusterOperator)
	if err != nil {
		logger.Error("failed to get proxy", zap.Error(err))
		return bizerr.New(errorcode.ErrCodeSvcClusterProxy, "failed to get proxy")
	}
	if err := proxy.ExecuteSQL(ctx, s.buildDropUserSql(userInfo.Username)); err != nil {
		logger.Error("failed to delete user", zap.Error(err))
		return errorcode.MarkClusterSQLErrorUserActionable(err)
	}

	return nil
}

func checkDeleteUser(operator *SqlOperator, branch *entity.Branch) error {
	if operator == nil {
		return errors.New("operator is required")
	}

	return checkBranchStatus(branch)
}