package cluster

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"net/mail"
	"net/url"
	"path"
	"regexp"
	"strings"
	"time"

	"github.com/gogo/protobuf/types"
	"github.com/google/uuid"
	"github.com/pingcap/dbaas/central/config"
	"github.com/pingcap/dbaas/central/db"
	"github.com/pingcap/dbaas/central/db/dbutil"
	"github.com/pingcap/dbaas/central/db/entity"
	"github.com/pingcap/dbaas/central/grpc/grpcli"
	infraapi "github.com/pingcap/dbaas/central/infra-api"
	"github.com/pingcap/dbaas/central/model/billing"
	"github.com/pingcap/dbaas/central/model/cluster"
	"github.com/pingcap/dbaas/central/model/cluster/clusterproxy/tidbcloud"
	"github.com/pingcap/dbaas/central/model/project"
	"github.com/pingcap/dbaas/central/model/util"
	"github.com/pingcap/dbaas/central/service"
	serviceUtils "github.com/pingcap/dbaas/central/service/utils"
	"github.com/pingcap/dbaas/central/singleton"
	"github.com/pingcap/dbaas/pkg/tidb/component"
	"github.com/pingcap/errcode"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/idl/pbgen/proto/central"
	"github.com/tidbcloud/idl/pbgen/proto/core"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/log"
	"go.opentelemetry.io/otel"
	"go.uber.org/zap"
	"gorm.io/gorm"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
)

type clusterService struct {
	central.UnimplementedClusterServiceServer
	service.OperationsService
	clustersModel cluster.Model
	billingModel  *billing.Model
}

var _ central.ClusterServiceServer = (*clusterService)(nil)
var Tracer = otel.Tracer("central/service/cluster")

// NewService constructs a ClusterService.
func NewService(s service.OperationsService, model cluster.Model, billingMode *billing.Model) central.ClusterServiceServer {
	return &clusterService{
		OperationsService: s,
		clustersModel:     model,
		billingModel:      billingMode,
	}
}

func (s *clusterService) userCluster(ctx context.Context) cluster.UserContext {
	user := s.CurrentUser(ctx)
	return s.clustersModel.User(ctx, user)
}

func (s *clusterService) Log(ctx context.Context) *zap.Logger {
	return s.userCluster(ctx).Log().With(log.FieldsFromContext(ctx)...)
}

func (s *clusterService) Get(ctx context.Context, req *central.ClusterGetReq) (*central.GetClusterResp, error) {
	logger := s.Log(ctx)
	context := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}
	cluster, err := context.Get(ctx, orgID, projectID, entity.ID(req.Id))
	if err != nil {
		logger.Error("failed to get cluster", zap.Uint64("clusterID", req.Id), zap.Error(err))
		return nil, err
	}

	return &central.GetClusterResp{
		Cluster:  s.clusterOutput(logger, cluster),
		BaseResp: &central.BaseResp{},
	}, nil
}

func (s *clusterService) convertDeletedCluster(logger *zap.Logger, dc *cluster.DeletedCluster) *central.DeletedCluster {
	deleteCluster := &central.DeletedCluster{
		Id:                       dc.ID,
		Name:                     dc.Name,
		Provider:                 dc.Provider,
		Region:                   dc.Region,
		DeleteTime:               nil, // fill later
		LatestBackupSize:         dc.LatestBackupSize,
		CrossAccountDeploymentId: dc.XAccountDeployId,
	}
	deleteTime, err := types.TimestampProto(dc.DeleteTime)
	if err == nil {
		deleteCluster.DeleteTime = deleteTime
	} else {
		logger.Error("failed to parse deletedCluster's DeleteTime field", zap.Uint64("clusterID", dc.ID), zap.Error(err))
	}

	return deleteCluster
}

func (s *clusterService) clusterOutput(logger *zap.Logger, cluster *cluster.Cluster) *central.Cluster {
	clusterItem := central.Cluster{
		Id:              int64(cluster.ID),
		Name:            cluster.Name,
		Status:          cluster.Status,
		SecondaryStatus: cluster.SecondaryStatus,
		Version:         cluster.Version,
		Provider:        string(cluster.Provider),
		Region:          cluster.Region,
		CreatedAt:       nil,
		CreatedBy:       nil,
		Endpoint: &central.Endpoint{
			Ready:   cluster.Endpoints.MySQL.Ready,
			Address: cluster.Endpoints.MySQL.Host,
			Port:    cluster.Endpoints.MySQL.Port,
		},
		VpcCidrBlock:      cluster.VPCCidrBlock,
		Capacity:          0,
		ClusterComponents: nil,
		CreateProgress:    cluster.Progress,
		PrometheusReady:   cluster.Endpoints.PrometheusReady,
		PrivateEndpoint: &central.Endpoint{
			Ready:   cluster.Endpoints.PrivateMySQL.Ready,
			Address: cluster.Endpoints.PrivateMySQL.Host,
			Port:    cluster.Endpoints.PrivateMySQL.Port,
		},
		Restored:                 cluster.Restored,
		RoleArn:                  cluster.RoleARN,
		ImportId:                 0,
		LastErrors:               cluster.LastErrors,
		Replications:             cluster.Replications,
		ClusterType:              cluster.ClusterType,
		WriteStatus:              cluster.WriteStatus,
		IsDevTier:                cluster.IsDevTier,
		CrossAccountDeploymentId: cluster.XAccountDeployId,
	}

	if cluster.ImportID != nil {
		clusterItem.ImportId = *cluster.ImportID
	}

	createdAt, err := types.TimestampProto(cluster.CreatedAt)
	if err != nil {
		logger.Error("failed to parse cluster's CreatedAt field", zap.Int64("clusterID", int64(cluster.ID)), zap.Error(err))
	} else {
		clusterItem.CreatedAt = createdAt
	}

	if cluster.CreatedBy != nil {
		clusterItem.CreatedBy = &central.User{
			Email:  cluster.CreatedBy.Email,
			Id:     cluster.CreatedBy.ID.Uint64(),
			Name:   fmt.Sprintf("%s %s", cluster.CreatedBy.FirstName, cluster.CreatedBy.LastName),
			Status: string(cluster.CreatedBy.Status),
			Role:   string(cluster.CreatedBy.Role),
		}

		if cluster.CreatedBy.InvitedAt != nil {
			inviteTime, err := types.TimestampProto(*cluster.CreatedBy.InvitedAt)
			if err != nil {
				logger.Error("failed to parse cluster's InviteTime field", zap.Int64("clusterID", int64(cluster.ID)), zap.Error(err))
			} else {
				clusterItem.CreatedBy.InviteTime = inviteTime
			}
		}
	}

	resources := cluster.Resources
	clusterComponents := &central.ClusterComponents{}
	for _, resource := range resources {
		capy := resource.Capacity
		if resource.ComponentName == component.TiDB {
			tidb := central.Component{
				Replicas:    capy.Replicas,
				NodeProfile: &central.NodeProfile{},
			}
			if capy.Node != nil {
				tidb.NodeProfile.NodeProfileId = capy.Node.NodeProfileID
				tidb.NodeProfile.Name = capy.Node.Name
			}
			if capy.Compute != nil {
				tidb.NodeProfile.Compute = &central.Compute{
					Cpu:      capy.Compute.CPU,
					MemoryMi: capy.Compute.MemoryMi,
				}
			}
			clusterComponents.Tidb = &tidb
		}

		if resource.ComponentName == component.TiKV {
			tikv := central.Component{
				Replicas:    capy.Replicas,
				NodeProfile: &central.NodeProfile{},
			}
			if capy.Node != nil {
				tikv.NodeProfile.NodeProfileId = capy.Node.NodeProfileID
				tikv.NodeProfile.Name = capy.Node.Name
			}
			if capy.Compute != nil {
				tikv.NodeProfile.Compute = &central.Compute{
					Cpu:      capy.Compute.CPU,
					MemoryMi: capy.Compute.MemoryMi,
				}
			}
			if capy.Disk != nil {
				tikv.NodeProfile.StorageGi = capy.Disk.SizeGi
				clusterItem.Capacity = tikv.NodeProfile.StorageGi * tikv.Replicas
			}
			clusterComponents.Tikv = &tikv
		}
		if resource.ComponentName == component.TiFlash {
			tiflash := central.Component{
				Replicas:    capy.Replicas,
				NodeProfile: &central.NodeProfile{},
			}
			if capy.Node != nil {
				tiflash.NodeProfile.NodeProfileId = capy.Node.NodeProfileID
				tiflash.NodeProfile.Name = capy.Node.Name
			}
			if capy.Compute != nil {
				tiflash.NodeProfile.Compute = &central.Compute{
					Cpu:      capy.Compute.CPU,
					MemoryMi: capy.Compute.MemoryMi,
				}
			}
			if capy.Disk != nil {
				tiflash.NodeProfile.StorageGi = capy.Disk.SizeGi
			}
			clusterComponents.Tiflash = &tiflash
		}
	}

	clusterItem.ClusterComponents = clusterComponents
	return &clusterItem
}

func (s *clusterService) List(ctx context.Context, req *central.ClusterListReq) (*central.Clusters, error) {
	context := s.userCluster(ctx)
	projectID := entity.ID(req.ProjectId)
	orgID := entity.ID(req.OrgId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	logger := s.Log(ctx)
	clusters, total, err := context.ListV2(ctx, projectID, entity.Cloud(req.Provider), req.Page, req.PerPage)
	if err != nil {
		if errors.Cause(err) == gorm.ErrRecordNotFound {
			logger.Error("failed to get provider", zap.String("provider", req.Provider))
			return nil, errcode.NewNotFoundErr(err)
		}

		logger.Error("failed to list clusters", zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New("failed to list clusters"))
	}
	items := make([]*central.Cluster, len(clusters))
	for i, cluster := range clusters {
		items[i] = s.clusterOutput(logger, &cluster)
	}
	return &central.Clusters{
		Total:    total,
		Items:    items,
		BaseResp: &central.BaseResp{},
	}, nil
}

func (s *clusterService) ListDeletedClusters(ctx context.Context, req *central.ClusterListReq) (*central.DeletedClusters, error) {
	context := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	logger := s.Log(ctx)
	dcs, err := context.ListDeletedClusters(ctx, entity.Cloud(req.Provider), projectID)
	if err != nil {
		logger.Error("failed to list deleted clusters", zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New("failed to list deleted clusters"))
	}
	items := make([]*central.DeletedCluster, len(dcs))
	for i, item := range dcs {
		items[i] = s.convertDeletedCluster(logger, item)
	}
	return &central.DeletedClusters{
		Total: int32(len(items)),
		Items: service.Pagination(req.Page, req.PerPage, items).([]*central.DeletedCluster),
	}, nil
}

func (s *clusterService) SendPasswordVerifyCode(ctx context.Context, req *central.SendPasswordVerifyCodeReq) (*types.Empty, error) {
	context := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	if err := s.rateControl4SendVerifyCodeEmail(ctx, s.CurrentUser(ctx).ID.Uint64(), clusterID.Uint64()); err != nil {
		return &types.Empty{}, err
	}

	err = context.SendPasswordVerifyCode(ctx, orgID, projectID, clusterID)
	if err != nil {
		logger := s.Log(ctx)
		logger.Error("failed to send password verification code", zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New("failed to send password verification code"))
	}
	return &types.Empty{}, nil
}

func (s *clusterService) rateControl4SendVerifyCodeEmail(ctx context.Context, userID, clusterID uint64) error {
	// limitation of send send password verify code email
	accessor := fmt.Sprint(userID)
	target := fmt.Sprint(clusterID)
	rateConf := config.Global().RateControlConfig.SendPasswordVerifyCode
	now := time.Now()
	rateController := singleton.RateControllerManager.WithScene(rateConf.Scene)
	if !rateController.AllowAll(ctx, accessor, target, now, rateConf.Rates...) {
		return errcode.NewInvalidInputErr(errors.New("request send verify code email too often"))
	}
	rateController.Access(ctx, accessor, target, now)
	return nil
}

func (s *clusterService) ChangePassword(ctx context.Context, req *central.ChangePasswordReq) (*types.Empty, error) {
	if err := service.IsValidRootPassword(req.GetPassword()); err != nil {
		return &types.Empty{}, errcode.NewInvalidInputErr(err)
	}
	context := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	err = context.ChangePassword(ctx, orgID, projectID, clusterID, req.Password, req.VerifyCode)
	if err != nil {
		return nil, err
	}
	return &types.Empty{}, nil
}

// TODO(Louis)：migrate it to core service
func (s *clusterService) ChangeClusterName(ctx context.Context, req *central.ChangeClusterNameReq) (*types.Empty, error) {
	if err := service.IsValidClusterName(req.GetClusterName()); err != nil {
		return &types.Empty{}, errcode.NewInvalidInputErr(err)
	}
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)
	clusterName := req.ClusterName
	context := s.userCluster(ctx)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	err = context.ChangeClusterName(orgID, projectID, clusterID, clusterName)
	if err != nil {
		return nil, err
	}
	return &types.Empty{}, nil
}

// Delete marks a cluster as deleted in the DB.
func (s *clusterService) Delete(ctx context.Context, req *central.ClusterIdReq) (*types.Empty, error) {
	context := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	err = context.Delete(ctx, orgID, projectID, clusterID)
	if err != nil {
		logger := s.Log(ctx)
		logger.Error("failed to delete cluster", zap.Uint64("clusterID", clusterID.Uint64()), zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New("failed to delete cluster"))
	}
	return &types.Empty{}, nil
}

func (s *clusterService) GetTiDBNodes(ctx context.Context, req *central.TiDBNodesReq) (*central.TiDBNodes, error) {
	context := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	nodes, err := context.ListTiDBNodesV1(ctx, orgID, projectID, clusterID)
	if err != nil {
		logger := s.Log(ctx)
		logger.Error("failed to get tidb nodes", zap.Uint64("clusterID", clusterID.Uint64()), zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New("failed to get tidb nodes"))
	}

	return &central.TiDBNodes{
		Total: int32(len(nodes)),
		Items: service.Pagination(req.Page, req.PerPage, nodes).([]*central.TiDBNode),
	}, nil
}

func (s *clusterService) GetTiKVNodes(ctx context.Context, req *central.TiKVNodesReq) (*central.TiKVNodes, error) {
	context := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	nodes, err := context.ListTiKVNodesV1(ctx, orgID, projectID, clusterID)
	if err != nil {
		logger := s.Log(ctx)
		logger.Error("failed to get tikv nodes", zap.Uint64("clusterID", clusterID.Uint64()), zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New("failed to get tikv nodes"))
	}

	return &central.TiKVNodes{
		Total: int32(len(nodes)),
		Items: service.Pagination(req.Page, req.PerPage, nodes).([]*central.TiKVNode),
	}, nil
}

func (s *clusterService) GetTiFlashNodes(ctx context.Context, req *central.TiFlashNodesReq) (*central.TiFlashNodes, error) {
	context := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	nodes, err := context.ListTiFlashNodesV1(ctx, orgID, projectID, clusterID)
	if err != nil {
		logger := s.Log(ctx)
		logger.Error("failed to get tiflash nodes", zap.Uint64("clusterID", clusterID.Uint64()), zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New("failed to get tiflash nodes"))
	}

	return &central.TiFlashNodes{
		Total: int32(len(nodes)),
		Items: service.Pagination(req.Page, req.PerPage, nodes).([]*central.TiFlashNode),
	}, nil
}

func (s *clusterService) CreatePrivateEndpoint(ctx context.Context, req *central.ClusterIdReq) (*types.Empty, error) {
	context := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	if _, err := grpcli.GetCluster(ctx, orgID.Uint64(), projectID.Uint64(), clusterID.Uint64()); err != nil {
		return nil, err
	}

	err = grpcli.CreateClusterConn(ctx, clusterID.Uint64(), core.ClusterConnectionType_CLUSTER_CONNECTION_TYPE_PRIVATE)
	if err != nil {
		logger := s.Log(ctx)
		logger.Error("failed to create private endpoint", zap.Uint64("clusterID", clusterID.Uint64()), zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New("failed to create private endpoint"))
	}
	return &types.Empty{}, nil
}

func (s *clusterService) GetTrafficFilter(ctx context.Context, req *central.ClusterIdReq) (*central.TrafficFilterResp, error) {
	context := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	trafficFilters, err := context.GetTrafficFilter(orgID, projectID, clusterID)
	if err != nil {
		logger := s.Log(ctx)
		logger.Error("failed to get traffic filter", zap.Uint64("clusterID", clusterID.Uint64()), zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New("failed to get traffic filter"))
	}
	return &central.TrafficFilterResp{
		Total:          int32(len(trafficFilters.GetCidrs())),
		TrafficFilters: trafficFilters,
	}, nil
}

func (s *clusterService) UpdateTrafficFilter(ctx context.Context, req *central.UpdateTrafficFilterReq) (*types.Empty, error) {
	context := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}
	if err := util.ValidateCidr(req.GetTrafficFilters()); err != nil {
		return &types.Empty{}, err
	}

	err = context.UpdateTrafficFilter(ctx, orgID, projectID, clusterID, req.GetTrafficFilters())
	if err != nil {
		logger := s.Log(ctx)
		logger.Error("failed to update traffic filter", zap.Int64("clusterID", int64(clusterID.Uint64())), zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New("failed to update traffic filter"))
	}
	return &types.Empty{}, nil
}

func (s *clusterService) GetPrivateEndpoint(ctx context.Context, req *central.ClusterIdReq) (*central.GetPrivateEndpointResp, error) {
	context := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	ret, err := context.GetPrivateEndpoint(ctx, orgID, projectID, clusterID)
	if err != nil {
		logger := s.Log(ctx)
		logger.Error("failed to get private endpoint", zap.Uint64("clusterID", clusterID.Uint64()), zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New("failed to get private endpoint"))
	}
	return ret, nil
}

func (s *clusterService) DetachReplication(ctx context.Context, req *central.DetachReplicationReq) (*central.DetachReplicationResp, error) {
	context := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	err := context.CheckOrgID(orgID)
	if err != nil {
		return nil, err
	}
	ret, err := context.DetachReplication(orgID, entity.ID(req.GetClusterId()), entity.ID(req.GetRepId()), req.GetDetachType())
	if err != nil {
		logger := s.Log(ctx)
		logger.Error("failed to detach replications", zap.Uint64("clusterID", req.ClusterId), zap.Uint64("replicationID", req.RepId), zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New("failed to detach replication"))
	}
	return ret, nil
}

func (s *clusterService) GetProjectCIDR(ctx context.Context, req *central.GetCIDRReq) (*central.ProjectCIDR, error) {
	logger := s.Log(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	providerName := entity.Cloud(req.Provider)
	userContext := s.userCluster(ctx)
	err := userContext.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}
	if !entity.IsValidProvider(providerName) {
		return nil, errcode.NewInvalidInputErr(errors.New("bad provider"))
	}

	var provider *entity.Provider
	err = db.Transaction(func(tx *gorm.DB) error {
		provider, err = dbutil.GetProviderByCloudName(ctx, tx, providerName)
		return err
	})
	if err != nil {
		return nil, errcode.NewInvalidInputErr(errors.New("bad provider"))
	}

	// NOTE for the region here, it may be empty string, which means we don't care the region CIDR.
	// Frontend query this API with region while create cluster, but not in the proejct CIDR page.
	regionCIDR, err := project.GetRegionCIDRFromInfraAPI(ctx, s.clustersModel.InfraAPI(), *provider, projectID, req.Region)
	if err != nil && !apierrors.IsNotFound(errors.Cause(err)) {
		logger.Error("failed to get region cidr", zap.Error(err))
	}

	var projectCIDR central.ProjectCIDR

	projectNetwork, err := project.GetProjectNetwork(ctx, projectID, provider.ID)
	switch err {
	case nil:
		projectCIDR.Cidr = projectNetwork.CIDR
		projectCIDR.IsDefault = false
	case gorm.ErrRecordNotFound:
		projectCIDR.Cidr = project.GetDefaultCIDR()
		projectCIDR.IsDefault = true
	default:
		return nil, err
	}
	projectCIDR.RegionCidr = regionCIDR

	return &projectCIDR, nil
}

// validateCreateProjectCIDRReq returns providerID and error
func validateCreateProjectCIDRReq(ctx context.Context, req *central.CreateProjectCIDRReq) (entity.ID, error) {
	providerName := entity.Cloud(req.Provider)
	projectID := entity.ID(req.ProjectId)
	var providerID entity.ID
	if !entity.IsValidProvider(providerName) {
		return providerID, errcode.NewInvalidInputErr(errors.New("bad provider"))
	}

	if err := isValidProjectCIDR(ctx, req.Cidr, providerName); err != nil {
		return providerID, err
	}

	var (
		provider *entity.Provider
		err      error
	)
	err = db.Transaction(func(tx *gorm.DB) error {
		provider, err = dbutil.GetProviderByCloudName(ctx, tx, providerName)
		return err
	})
	if err != nil {
		return providerID, errcode.NewInvalidInputErr(errors.New("bad provider"))
	}
	providerID = provider.ID

	projectNetwork, err := project.GetProjectNetwork(ctx, projectID, provider.ID)
	if err != nil && err != gorm.ErrRecordNotFound {
		return providerID, err
	}
	if projectNetwork.CIDR != "" {
		return providerID, errcode.NewInvalidInputErr(errors.New(config.ErrProjectCIDRExist))
	}
	return providerID, nil
}

func (s *clusterService) CreateProjectCIDR(ctx context.Context, req *central.CreateProjectCIDRReq) (*central.ProjectCIDR, error) {
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	userContext := s.userCluster(ctx)
	err := userContext.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	providerID, err := validateCreateProjectCIDRReq(ctx, req)
	if err != nil {
		return nil, err
	}

	projectNetwork, err := project.CreateProjectNetwork(ctx, projectID, providerID, req.Cidr)
	if err != nil {
		return nil, err
	}

	projectCIDR := central.ProjectCIDR{
		Cidr:      projectNetwork.CIDR,
		IsDefault: false,
	}
	return &projectCIDR, nil
}

func buildRegionCIDR(regionCIDR *core.RegionCIDR) *central.RegionCIDR {
	return &central.RegionCIDR{
		Region:     regionCIDR.Region,
		RegionName: regionCIDR.RegionDisplayName,
		Cidr:       regionCIDR.Cidr,
		Active:     regionCIDR.Active,
	}
}

func (s *clusterService) getProjectRegionCIDR(ctx context.Context, projectID entity.ID, provider entity.Cloud) (*central.ProjectRegionCIDR, error) {
	ret := central.ProjectRegionCIDR{}
	ret.Provider = provider.String()

	resp, err := grpcli.ListRegionCIDROfProject(ctx, &core.ListRegionCIDROfProjectReq{
		ProjectId:     projectID.Uint64(),
		CloudProvider: provider.String(),
		BaseReq:       nil,
	})
	if err != nil {
		return nil, err
	}

	ret.Cidr = resp.ProjectCidr
	for _, regionCIDR := range resp.RegionCidrs {
		ret.Regions = append(ret.Regions, buildRegionCIDR(regionCIDR))
	}

	return &ret, nil
}

func (s *clusterService) GetProjectRegionCIDR(ctx context.Context, req *central.GetProjectRegionCIDRReq) (*central.GetProjectRegionCIDRResp, error) {
	logger := s.Log(ctx)

	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	userContext := s.userCluster(ctx)
	err := userContext.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	awsResp, err := s.getProjectRegionCIDR(ctx, projectID, entity.Aws)
	if err != nil {
		logger.Error("failed to get aws region cidr", zap.Error(err))
		return nil, err
	}
	gcpResp, err := s.getProjectRegionCIDR(ctx, projectID, entity.Gcp)
	if err != nil {
		logger.Error("failed to get gcp region cidr", zap.Error(err))
		return nil, err
	}

	ret := central.GetProjectRegionCIDRResp{
		ProjectRegionCidrs: []*central.ProjectRegionCIDR{awsResp, gcpResp},
		BaseResp:           &central.BaseResp{},
	}
	return &ret, nil
}

func (s *clusterService) MakeClusterWritable(ctx context.Context, req *central.MakeClusterWritableReq) (*types.Empty, error) {
	context := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	err = context.SetReadOnly(ctx, clusterID, projectID, "OFF")
	if err != nil {
		return nil, err
	}
	return &types.Empty{}, nil
}

func (s *clusterService) SubscribeUserAlert(ctx context.Context, req *central.SubscribeUserAlertReq) (*central.SubscribeUserAlertResp, error) {
	logger := s.Log(ctx)
	context := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	if _, err := mail.ParseAddress(req.Email); err != nil {
		return nil, errcode.NewInvalidInputErr(fmt.Errorf("No active user of current project has the email %s", req.Email))
	}

	err = db.Transaction(func(tx *gorm.DB) error {
		var user entity.User
		// query existing users by emails
		user, err = dbutil.GetUserByEmail(tx, req.Email)
		if err != nil {
			return errcode.NewInvalidInputErr(fmt.Errorf("No active user of current project has the email %s", req.Email))
		}

		if user.TenantID.Uint64() != orgID.Uint64() {
			return errcode.NewInvalidInputErr(fmt.Errorf("No active user of current project has the email %s", req.Email))
		}

		if user.Status != entity.UserStatusActive {
			return errcode.NewInvalidInputErr(fmt.Errorf("No active user of current project has the email %s", req.Email))
		}

		userAlert, _ := s.clustersModel.UserAlertService.GetUserAlertByUserAndProjectID(ctx, tx, user.ID, projectID)
		if userAlert.ID.Uint64() > 0 {
			return errcode.NewInvalidInputErr(fmt.Errorf("A user with email %s is already subscribed", req.Email))
		}

		return s.clustersModel.UserAlertService.CreateUserAlert(ctx, tx, projectID, user.ID)
	})
	if err != nil {
		logger.Error("failed to subscribe user alerts", zap.Error(err), zap.Uint64("projectID", projectID.Uint64()), zap.String("email", req.Email))
		return nil, err
	}

	return &central.SubscribeUserAlertResp{}, nil
}

func (s *clusterService) UnsubscribeUserAlert(ctx context.Context, req *central.UnsubscribeUserAlertReq) (*central.UnsubscribeUserAlertResp, error) {
	logger := s.Log(ctx)
	context := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	err = db.Transaction(func(tx *gorm.DB) error {
		// query existing users by emails
		userAlert, err := s.clustersModel.UserAlertService.GetUserAlertByID(ctx, tx, entity.ID(req.UserAlertId))
		if err != nil {
			return err
		}

		if userAlert.ProjectID.Uint64() != projectID.Uint64() {
			return errcode.NewInvalidInputErr(fmt.Errorf("No active user of current project has the email"))
		}

		return s.clustersModel.UserAlertService.DeleteUserAlerts(ctx, tx, projectID, userAlert.ID)
	})
	if err != nil {
		logger.Error("failed to unsubscribe user alerts", zap.Error(err), zap.Uint64("projectID", projectID.Uint64()), zap.Uint64("user_alert_id", req.UserAlertId))

		return nil, err
	}

	return &central.UnsubscribeUserAlertResp{}, nil
}

func (s *clusterService) GetUserAlertList(ctx context.Context, req *central.GetUserAlertListReq) (*central.GetUserAlertListResp, error) {
	logger := s.Log(ctx)
	context := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	var items []*central.UserAlertItem
	err = db.Transaction(func(tx *gorm.DB) error {
		userAlerts, err := s.clustersModel.UserAlertService.GetUserAlertsByProjectID(ctx, tx, projectID)
		if err != nil {
			return err
		}

		for _, i := range userAlerts {
			if i.User != nil {
				items = append(items, &central.UserAlertItem{
					Email:       i.User.Email,
					UserAlertId: i.ID.Uint64(),
				})
			}
		}
		return nil
	})
	if err != nil {
		logger.Error("failed to get latest subscribed users", zap.Error(err), zap.Uint64("projectID", projectID.Uint64()))
		return nil, errcode.NewInternalErr(err)
	}

	return &central.GetUserAlertListResp{Items: items}, nil
}

func (s *clusterService) CreateMetricsKey(ctx context.Context, req *central.CreateMetricsKeyReq) (*central.CreateMetricsKeyResp, error) {
	userContext := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	err := userContext.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	user := s.CurrentUser(ctx)

	key := strings.ReplaceAll(uuid.New().String(), "-", "")
	var metricsKey *entity.MetricsKey
	err = db.Transaction(func(tx *gorm.DB) error {
		allExistsMetricsKey, err := dbutil.GetAllMetricsKeyByProject(tx, projectID)
		if err != nil && err != gorm.ErrRecordNotFound {
			return err
		}

		if len(allExistsMetricsKey) >= config.MaxMetricsKeyCount {
			return errcode.NewInvalidInputErr(fmt.Errorf(
				"more than %d active scrape_config files is not allowed. You must first delete a file to continue",
				config.MaxMetricsKeyCount,
			))
		}

		metricsKey, err = dbutil.CreateMetricsKey(tx, projectID, user.ID, key)
		if err != nil {
			return err
		}

		return dbutil.AddAccessLogNow(tx, user.TenantID, user.Email, fmt.Sprintf("add metrics key %d on project %d", metricsKey.ID, projectID))
	})
	if err != nil {
		return nil, err
	}
	t, err := types.TimestampProto(metricsKey.CreatedAt)
	if err != nil {
		return nil, err
	}

	return &central.CreateMetricsKeyResp{Key: &central.MetricsKey{KeyId: metricsKey.ID.Uint64(), Key: key, CreatedAt: t}}, nil
}

func (s *clusterService) GetAllMetricsKey(ctx context.Context, req *central.GetAllMetricsKeyReq) (*central.GetAllMetricsKeyResp, error) {
	userContext := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	err := userContext.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	var metricsKeys []entity.MetricsKey

	if err = db.Transaction(func(tx *gorm.DB) error {
		metricsKeys, err = dbutil.GetAllMetricsKeyByProject(tx, projectID)
		return err
	}); err != nil {
		return nil, err
	}

	resp := central.GetAllMetricsKeyResp{}

	for _, metricsKey := range metricsKeys {
		t, err := types.TimestampProto(metricsKey.CreatedAt)
		if err != nil {
			return nil, err
		}

		resp.MetricsKeys = append(resp.MetricsKeys, &central.MetricsKey{KeyId: metricsKey.ID.Uint64(), Key: "", CreatedAt: t})
	}

	return &resp, nil
}

func (s *clusterService) DeleteMetricsKey(ctx context.Context, req *central.DeleteMetricsKeyReq) (*types.Empty, error) {
	logger := s.Log(ctx)
	user := s.CurrentUser(ctx)
	userContext := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	err := userContext.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	keyID := entity.ID(req.KeyId)

	var metricsKey *entity.MetricsKey
	err = db.Transaction(func(tx *gorm.DB) error {
		metricsKey, err = dbutil.GetMetricsKeyByID(tx, keyID)
		return err
	})
	if err != nil {
		return nil, errcode.NewInvalidInputErr(errors.New("bad key id"))
	}

	if metricsKey.ProjectID.Uint64() != projectID.Uint64() {
		logger.Error("metrics projectID != projectID", zap.Uint64("metricsProjectID", metricsKey.ProjectID.Uint64()), zap.Uint64("projectID", projectID.Uint64()))
		return nil, errcode.NewForbiddenErr(errors.New("only owner can perform this operation"))
	}

	if err := db.Transaction(func(tx *gorm.DB) error {
		if err := dbutil.DeleteMetricsKey(tx, keyID); err != nil {
			return err
		}

		return dbutil.AddAccessLogNow(tx, user.TenantID, user.Email, fmt.Sprintf("delete metrics key %d on project %d", keyID, projectID))
	}); err != nil {
		return nil, err
	}

	return &types.Empty{}, nil
}

func (s *clusterService) validateAuditLogReq(ctx context.Context, orgID, projectID, clusterID entity.ID, auditLogConfig *central.AuditLogConfig) error {
	userContext := s.userCluster(ctx)
	logger := s.Log(ctx)

	// check if the user has access to the project
	err := userContext.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return err
	}

	// check if user's role is audit admin
	if err = userContext.CheckIfUserIsAuditAdmin(); err != nil {
		return err
	}

	// check audit log config
	if auditLogConfig != nil {
		if err = serviceUtils.ValidateBucketURL(auditLogConfig.DataSourceType, auditLogConfig.BucketUrl); err != nil {
			return err
		}

		if entity.DataSource(auditLogConfig.DataSourceType) == entity.DataSourceTypeS3 && auditLogConfig.RoleArn == "" {
			return errors.Errorf("Role ARN is required")
		}
	}

	// check if the cluster is in the project
	var c entity.Cluster
	err = db.Transaction(func(tx *gorm.DB) error {
		c, err = dbutil.GetCluster(tx, clusterID)
		return err
	})
	if err != nil {
		return err
	}

	if c.ProjectID.Uint64() != projectID.Uint64() {
		return errcode.NewInvalidInputErr(errors.New(config.TEXT_INTERNAL_ERR))
	}

	// check if the cluster is dev tier
	if c.IsDevTier() {
		return errcode.NewInvalidInputErr(errors.New(config.TEXT_INTERNAL_ERR))
	}

	api := s.clustersModel.InfraAPI()
	// check db version
	cr, err := api.GetCluster(ctx, projectID.Uint64(), clusterID.Uint64())
	if err != nil {
		logger.Error("failed to get cluster", zap.Error(err))
		return errcode.NewInternalErr(errors.New(config.TEXT_INTERNAL_ERR))
	}

	versionIsSupport, err := cluster.CheckIfAuditLogSupported(ctx, cr)
	if err != nil {
		logger.Error("failed to check if audit log supported", zap.Error(err))
		return errcode.NewInternalErr(errors.New(config.TEXT_INTERNAL_ERR))
	}

	if !versionIsSupport {
		return errcode.NewInvalidInputErr(errors.New(config.TEXT_PLEASE_UPGRADE))
	}

	return nil
}

func (s *clusterService) precheckAuditLogConfig(ctx context.Context, req *central.AuditLogConfig, projectID, clusterID uint64) (*clusterproxy.AuditCheckResp, error) {
	infraapi := s.clustersModel.InfraAPI()
	logger := s.Log(ctx).Sugar()

	cr, err := infraapi.GetCluster(ctx, projectID, clusterID)
	if err != nil {
		return nil, err
	}

	proxy, err := tidbcloud.BuildClusterProxy(clusterID, cr.Status, config.AuditLogClusterproxyTimeout)
	if err != nil {
		return nil, err
	}

	profile, err := s.userCluster(ctx).GetDataIngestionProfile(ctx, entity.ID(clusterID))
	if err != nil {
		logger.Errorf("get data ingestion profile failed, err: %s", err)
		return nil, err
	}

	var args clusterproxy.AuditCheckModel
	switch entity.DataSource(req.DataSourceType) {
	case entity.DataSourceTypeS3:
		args.ExternalID = profile.AwsExternalId
		args.Audit.BucketURL = req.BucketUrl
		args.Audit.Region = req.Region
		args.Audit.RoleARN = req.RoleArn
	case entity.DataSourceTypeGCS:
		args.ServiceAccount = profile.GcpAccountId
		args.Audit.BucketURL = req.BucketUrl
		args.Audit.Region = req.Region
	default:
		logger.Infof("unsupported data source type: %s", req.DataSourceType)
		return nil, errors.Errorf(config.TEXT_INTERNAL_ERR)
	}

	resp, err := proxy.TestAuditConnection(ctx, args)
	if err != nil {
		return nil, err
	}
	logger.Infof("audit log precheck result: %v", resp)

	return resp, nil
}

func (s *clusterService) GetAuditLogStatus(ctx context.Context, req *central.GetAuditLogStatusReq) (*central.GetAuditLogStatusResp, error) {
	logger := s.Log(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)

	err := s.validateAuditLogReq(ctx, orgID, projectID, clusterID, nil)
	if err != nil {
		logger.Info("validate audit log req failed", zap.Error(err))
		return nil, errcode.NewInvalidInputErr(err)
	}

	var clusterConfig *entity.ClusterConfig
	err = db.Transaction(func(tx *gorm.DB) error {
		clusterConfig, err = dbutil.GetClusterConfig(ctx, tx, clusterID)
		return err
	})
	if err != nil {
		logger.Info("get cluster extra config failed", zap.Error(err))
		if err == gorm.ErrRecordNotFound {
			return &central.GetAuditLogStatusResp{Enabled: false}, nil
		}

		return nil, errcode.NewInternalErr(err)
	}

	var auditLogconfig central.AuditLogConfig
	if clusterConfig.AuditLogConfig != "" {
		if err := json.Unmarshal([]byte(clusterConfig.AuditLogConfig), &auditLogconfig); err != nil {
			logger.Info("failed to unmarshal audit config", zap.Error(err))
			return nil, errcode.NewInternalErr(errors.New(config.TEXT_PLEASE_RETRY))
		}
	}

	return &central.GetAuditLogStatusResp{Enabled: clusterConfig.IsAuditLogOn, AuditLogConfig: &auditLogconfig}, nil
}

func (s *clusterService) checkInfraAuditReady(ctx context.Context, api *infraapi.API, projectID, clusterID uint64) bool {
	logger := s.Log(ctx)

	cr, err := api.GetCluster(ctx, projectID, clusterID)
	if err != nil {
		logger.Error("get cluster failed", zap.Error(err))
		return false
	}

	if cr.Status.DB != nil && cr.Status.DB.TiDB != nil && cr.Status.DB.TiDB.AuditReady {
		logger.Info("infra audit ready", zap.Uint64("projectID", projectID), zap.Uint64("clusterID", clusterID))
		return true
	}

	logger.Info("infra audit not ready", zap.Uint64("projectID", projectID), zap.Uint64("clusterID", clusterID))
	return false
}

func (s *clusterService) UpdateAuditLogStatus(ctx context.Context, req *central.UpdateAuditLogStatusReq) (*central.UpdateAuditLogStatusResp, error) {
	logger := s.Log(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)

	err := s.validateAuditLogReq(ctx, orgID, projectID, clusterID, req.AuditLogConfig)
	if err != nil {
		logger.Info("validate audit log req failed", zap.Error(err))
		return nil, errcode.NewInvalidInputErr(err)
	}

	// precheck
	precheckResp, err := s.precheckAuditLogConfig(ctx, req.AuditLogConfig, req.ProjectId, req.ClusterId)
	if err != nil {
		logger.Sugar().Errorf("precheck audit log config failed, err: %s", err)
		return nil, errcode.NewInternalErr(errors.New(config.TEXT_PLEASE_RETRY))
	}

	if !precheckResp.Pass {
		logger.Sugar().Errorf("precheck audit log config failed, err: %s", precheckResp.Message)
		errText := config.TEXT_PLEASE_CHECK_BUCKET
		if entity.DataSource(req.AuditLogConfig.DataSourceType) == entity.DataSourceTypeGCS {
			errText = config.TEXT_PLEASE_CHECK_BUCKET_GCS
		}
		return nil, errcode.NewInvalidInputErr(errors.Errorf(errText))
	}

	// if it's restart
	var auditEnabled = req.Enabled
	if req.Restart {
		auditEnabled = true
	}

	api := s.clustersModel.InfraAPI()
	auditLogconfig, err := json.Marshal(req.AuditLogConfig)
	if err != nil {
		logger.Info("marshal audit log config failed", zap.Error(err))
		errText := config.TEXT_PLEASE_CHECK_BUCKET
		if entity.DataSource(req.AuditLogConfig.DataSourceType) == entity.DataSourceTypeGCS {
			errText = config.TEXT_PLEASE_CHECK_BUCKET_GCS
		}
		return nil, errcode.NewInvalidInputErr(errors.Errorf(errText))
	}

	start := time.Now()
	bucketURL, err := url.Parse(req.AuditLogConfig.BucketUrl)
	if err != nil {
		logger.Error("parse bucket url failed", zap.Error(err))
		return nil, errcode.NewInvalidInputErr(err)
	}
	bucketURL.Path = path.Join(bucketURL.Path, fmt.Sprint(req.ClusterId))

	err = cluster.SyncAuditLogStatusToInfra(
		ctx, api, projectID.Uint64(), clusterID.Uint64(),
		bucketURL.String(), req.AuditLogConfig.Region, req.AuditLogConfig.RoleArn,
	)
	if err != nil {
		logger.Info("sync audit log status to infra failed", zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New(config.TEXT_PLEASE_RETRY))
	}
	logger.Info("sync audit log status to infra success", zap.Duration("time elapsed", time.Since(start)))

	err = cluster.UpdateAuditLogStatus(ctx, api, s.clustersModel.Encryptor(), clusterID, projectID, auditEnabled)
	if err != nil {
		logger.Info("update audit log status to cluster failed", zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New(config.TEXT_PLEASE_RETRY))
	}

	logger.Info("update audit log status to cluster success", zap.Duration("time elapsed", time.Since(start)))
	err = db.Transaction(func(tx *gorm.DB) error {
		return dbutil.UpsertClusterConfig(ctx, tx, clusterID, auditEnabled, string(auditLogconfig))
	})
	if err != nil {
		logger.Error("failed to update audit log config", zap.Error(err), zap.Uint64("projectID", projectID.Uint64()), zap.Uint64("clusterID", clusterID.Uint64()))
		return nil, errcode.NewInternalErr(errors.New(config.TEXT_PLEASE_RETRY))
	}

	return &central.UpdateAuditLogStatusResp{Enabled: auditEnabled}, nil
}

func (s *clusterService) PrecheckAuditLogConfig(ctx context.Context, req *central.PrecheckAuditLogConfigReq) (*central.PrecheckAuditLogConfigResp, error) {
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)
	logger := s.Log(ctx)

	err := s.validateAuditLogReq(ctx, orgID, projectID, clusterID, req.AuditLogConfig)
	if err != nil {
		logger.Info("validate audit log req failed", zap.Error(err))
		return nil, errcode.NewInvalidInputErr(err)
	}

	// precheck
	precheckResp, err := s.precheckAuditLogConfig(ctx, req.AuditLogConfig, req.ProjectId, req.ClusterId)
	if err != nil {
		logger.Sugar().Errorf("precheck audit log config failed, err: %s", err)
		return nil, errcode.NewInternalErr(errors.Errorf(config.TEXT_PLEASE_RETRY))
	}

	if !precheckResp.Pass {
		logger.Sugar().Errorf("precheck audit log config failed, err: %s", precheckResp.Message)
		errText := config.TEXT_PLEASE_CHECK_BUCKET
		if entity.DataSource(req.AuditLogConfig.DataSourceType) == entity.DataSourceTypeGCS {
			errText = config.TEXT_PLEASE_CHECK_BUCKET_GCS
		}

		return nil, errcode.NewInvalidInputErr(errors.Errorf(errText))
	}

	return &central.PrecheckAuditLogConfigResp{Valid: precheckResp.Pass, Message: ""}, nil
}

func (s *clusterService) CheckAuditLogPluginStatus(ctx context.Context, req *central.CheckAuditLogPluginStatusReq) (*central.CheckAuditLogPluginStatusResp, error) {
	logger := s.Log(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)

	err := s.validateAuditLogReq(ctx, orgID, projectID, clusterID, nil)
	if err != nil {
		logger.Info("validate audit log req failed", zap.Error(err))
		return nil, errcode.NewInvalidInputErr(err)
	}

	api := s.clustersModel.InfraAPI()
	cr, err := api.GetCluster(ctx, projectID.Uint64(), clusterID.Uint64())
	if err != nil {
		logger.Error("get cluster failed", zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New(config.TEXT_PLEASE_RETRY))
	}

	auditEnabled := cr.Spec.Cluster.TiDB.Audit != nil && cr.Spec.Cluster.TiDB.Audit.Enable

	return &central.CheckAuditLogPluginStatusResp{Enabled: auditEnabled}, nil
}

func (s *clusterService) GetAuditLogAccessRecords(ctx context.Context, req *central.GetAuditLogAccessRecordsReq) (*central.GetAuditLogAccessRecordsResp, error) {
	logger := s.Log(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)

	err := s.validateAuditLogReq(ctx, orgID, projectID, clusterID, nil)
	if err != nil {
		logger.Info("validate audit log req failed", zap.Error(err))
		return nil, errcode.NewInvalidInputErr(err)
	}

	api := s.clustersModel.InfraAPI()
	encryptor := s.clustersModel.Encryptor()
	var c entity.Cluster
	err = db.Transaction(func(tx *gorm.DB) error {
		c, err = dbutil.GetCluster(tx, clusterID)
		return err
	})
	if err != nil {
		logger.Error("get cluster failed", zap.Error(err))
		return nil, errcode.NewInvalidInputErr(errors.New(config.TEXT_PLEASE_RETRY))
	}

	sql := config.GetAllAuditLogAccessClause
	logger.Sugar().Infof("get audit log access record sql: %s", sql)

	data, err := cluster.QueryClusterWithSQL(ctx, api, encryptor, &c, tidbcloud.BuildClusterProxy, sql)
	if err != nil {
		logger.Error("get audit log access records failed", zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New(config.TEXT_PLEASE_RETRY))
	}

	// decode result
	var (
		resp   auditLogAccessRecordsResp
		result central.GetAuditLogAccessRecordsResp
	)
	if err = json.Unmarshal(data, &resp); err != nil {
		logger.Error("unmarshal audit log access records failed", zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New(config.TEXT_INTERNAL_ERR))
	}
	for i, id := range resp.ID {
		user, errUser := base64.StdEncoding.DecodeString(resp.User[i])
		db, errDB := base64.StdEncoding.DecodeString(resp.Db[i])
		table, errTbl := base64.StdEncoding.DecodeString(resp.Tbl[i])
		accessType, errAccessType := base64.StdEncoding.DecodeString(resp.AccessType[i])

		if errUser != nil || errDB != nil || errTbl != nil || errAccessType != nil {
			logger.Sugar().Errorf("decode audit log access record failed, err(user, db, table, access_type): %s, %s, %s, %s", errUser, errDB, errTbl, errAccessType)
			return nil, errcode.NewInternalErr(errors.New(config.TEXT_INTERNAL_ERR))
		}

		result.Records = append(result.Records, &central.AuditLogAccessRecord{
			Id:             uint64(id),
			UserExpr:       string(user),
			DbExpr:         string(db),
			TableExpr:      string(table),
			AccessTypeList: strings.Split(string(accessType), ","),
		})
	}

	logger.Sugar().Infof("get audit log access records success, records: %s, data: %s", result.Records, data)

	return &result, nil
}

func (s *clusterService) DeleteAuditLogAccessRecord(ctx context.Context, req *central.DeleteAuditLogAccessRecordReq) (*central.DeleteAuditLogAccessRecordResp, error) {
	logger := s.Log(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)

	err := s.validateAuditLogReq(ctx, orgID, projectID, clusterID, nil)
	if err != nil {
		logger.Info("validate audit log req failed", zap.Error(err))
		return nil, errcode.NewInvalidInputErr(err)
	}

	api := s.clustersModel.InfraAPI()
	encryptor := s.clustersModel.Encryptor()
	var c entity.Cluster
	err = db.Transaction(func(tx *gorm.DB) error {
		c, err = dbutil.GetCluster(tx, clusterID)
		return err
	})
	if err != nil {
		logger.Error("get cluster failed", zap.Error(err))
		return nil, errcode.NewInvalidInputErr(errors.New(config.TEXT_PLEASE_RETRY))
	}

	sql := fmt.Sprintf(config.DeleteAuditLogAccessClause, req.Id)
	logger.Sugar().Infof("delete audit log access record sql: %s", sql)

	err = util.ExecuteSqlCommon(ctx, &c, sql, api, encryptor, tidbcloud.BuildClusterProxy, config.AuditLogClusterproxyTimeout)
	if err != nil {
		logger.Error("delete audit log access record failed", zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New(config.TEXT_PLEASE_RETRY))
	}

	err = util.ExecuteSqlCommon(ctx, &c, config.FlushAuditLogAccessClause, api, encryptor, tidbcloud.BuildClusterProxy, config.AuditLogClusterproxyTimeout)
	if err != nil {
		logger.Error("flush audit log access record failed", zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New(config.TEXT_PLEASE_RETRY))
	}

	//clauses := []string{sql, config.FlushAuditLogAccessClause}

	//err = util.ExecuteMultiSqlCommon(ctx, &c, clauses, api, encryptor, tidbcloud.BuildClusterProxy, config.AuditLogClusterproxyTimeout)
	//if err != nil {
	//logger.Error("delete audit log access record failed", zap.Error(err))
	//return nil, errcode.NewInternalErr(errors.New("failed to delete audit log access record"))
	//}

	return &central.DeleteAuditLogAccessRecordResp{}, nil
}

func (s *clusterService) checkAuditLogInput(ctx context.Context, req *central.AuditLogAccessRecord) error {
	if req.Id != 0 {
		return errcode.NewInvalidInputErr(errors.New("id should not be set"))
	}

	var err error
	if _, err = regexp.Compile(req.UserExpr); err != nil {
		return errcode.NewInvalidInputErr(errors.New("User Expr is invalid"))
	}

	if _, err = regexp.Compile(req.DbExpr); err != nil {
		return errcode.NewInvalidInputErr(errors.New("DB Expr is invalid"))
	}

	if _, err = regexp.Compile(req.TableExpr); err != nil {
		return errcode.NewInvalidInputErr(errors.New("Table Expr is invalid"))
	}

	for _, accessType := range req.AccessTypeList {
		if _, exists := config.ValidAuditLogAccessTypeOption[accessType]; !exists {
			return errcode.NewInvalidInputErr(errors.Errorf("Access Type %s is invalid", accessType))
		}
	}

	return nil
}

func (s *clusterService) CreateAuditLogAccessRecord(ctx context.Context, req *central.CreateAuditLogAccessRecordReq) (*central.CreateAuditLogAccessRecordResp, error) {
	logger := s.Log(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)

	err := s.validateAuditLogReq(ctx, orgID, projectID, clusterID, nil)
	if err != nil {
		logger.Info("validate audit log req failed", zap.Error(err))
		return nil, errcode.NewInvalidInputErr(err)
	}

	if err = s.checkAuditLogInput(ctx, req.Record); err != nil {
		logger.Sugar().Infof("check audit log input failed %v", err)
		return nil, errcode.NewInvalidInputErr(err)
	}

	api := s.clustersModel.InfraAPI()
	encryptor := s.clustersModel.Encryptor()
	var c entity.Cluster
	err = db.Transaction(func(tx *gorm.DB) error {
		c, err = dbutil.GetCluster(tx, clusterID)
		return err
	})
	if err != nil {
		logger.Error("get cluster failed", zap.Error(err))
		return nil, errcode.NewInvalidInputErr(errors.New(config.TEXT_PLEASE_RETRY))
	}

	record := req.Record
	sql := fmt.Sprintf(config.CreateAuditLogAccessClause, record.UserExpr, record.DbExpr, record.TableExpr, strings.Join(record.AccessTypeList, ","))
	logger.Sugar().Infof("create audit log access record sql: %s", sql)

	err = util.ExecuteSqlCommon(ctx, &c, sql, api, encryptor, tidbcloud.BuildClusterProxy, config.AuditLogClusterproxyTimeout)
	if err != nil {
		logger.Error("create audit log access record failed", zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New(config.TEXT_DUPLICATED))
	}

	err = util.ExecuteSqlCommon(ctx, &c, config.FlushAuditLogAccessClause, api, encryptor, tidbcloud.BuildClusterProxy, config.AuditLogClusterproxyTimeout)
	if err != nil {
		logger.Error("flush audit log access record failed", zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New(config.TEXT_PLEASE_RETRY))
	}

	//clauses := []string{sql, config.FlushAuditLogAccessClause}

	//err = util.ExecuteMultiSqlCommon(ctx, &c, clauses, api, encryptor, tidbcloud.BuildClusterProxy, config.AuditLogClusterproxyTimeout)
	//if err != nil {
	//logger.Error("create audit log access record failed", zap.Error(err))
	//return nil, errcode.NewInternalErr(errors.New("create audit log access record failed, check if it's duplicated"))
	//}

	return &central.CreateAuditLogAccessRecordResp{}, nil
}