package cluster

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

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/pingcap/errcode"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/core-portal/internal/central/config"
	"github.com/tidbcloud/core-portal/internal/central/db"
	"github.com/tidbcloud/core-portal/internal/central/db/dbutil"
	"github.com/tidbcloud/core-portal/internal/central/grpc/grpcli"
	"github.com/tidbcloud/core-portal/internal/central/infra-api"
	"github.com/tidbcloud/core-portal/internal/central/model/billing"
	"github.com/tidbcloud/core-portal/internal/central/model/cluster"
	"github.com/tidbcloud/core-portal/internal/central/model/cluster/clusterproxy/tidbcloud"
	"github.com/tidbcloud/core-portal/internal/central/model/util"
	"github.com/tidbcloud/core-portal/internal/central/pkg/tidb/component"
	"github.com/tidbcloud/core-portal/internal/central/service"
	serviceUtils "github.com/tidbcloud/core-portal/internal/central/service/utils"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/central"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	entity "github.com/tidbcloud/core-svc/model"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/log"
	"go.opentelemetry.io/otel"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
	"google.golang.org/protobuf/types/known/timestamppb"
	"gorm.io/gorm"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/utils/pointer"
)

// TODO(zlj): delete when migrate `GetClusterCACertificate` to core-svc
const (
	// The bucket name of CA certificate in AWS s3
	caCertificateS3Bucket string = "cluster-certificates"

	// The key of CA certificate in bucket
	caCertificateS3Key string = "ca.pem"

	// Set response-content-disposition to 'attachment' to make browser don't attempt to render file
	respContentDispositionAttachment string = "attachment"

	// The default duration time to expires presigned URL
	defaultPresignExpires time.Duration = 1 * time.Minute
)

var (
	validClusterStatuses4AuditLog = []core.ClusterStatus{
		core.ClusterStatus_CLUSTER_STATUS_NORMAL, core.ClusterStatus_CLUSTER_STATUS_SCALING,
	}
)

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

	// AWS client
	// TODO(zlj): delete when migrate `GetClusterCACertificate` to core-svc
	s3PresignClient *s3.PresignClient
}

var _ central.ClusterServiceServer = (*clusterService)(nil)
var _ central.TenantServiceServer = (*tenantService)(nil)

type tenantService struct {
	central.UnimplementedTenantServiceServer
	service.OperationsService
	tenantModel cluster.Model
}

func (s *tenantService) userContext(ctx context.Context) cluster.UserContext {
	return s.tenantModel.User(ctx, s.CurrentUser(ctx))
}

func (s *tenantService) Log(ctx context.Context) *zap.Logger {
	return s.userContext(ctx).Log()
}

// NewService constructs a tenantService
func NewTenantService(s service.OperationsService, model cluster.Model) central.TenantServiceServer {
	return &tenantService{OperationsService: s, tenantModel: model}
}

var Tracer = otel.Tracer("central/service/cluster")

// NewService constructs a ClusterService.
func NewService(s service.OperationsService, model cluster.Model, billingMode *billing.Model, awsConfig aws.Config) central.ClusterServiceServer {
	return &clusterService{
		OperationsService: s,
		clustersModel:     model,
		billingModel:      billingMode,
		// TODO(zlj): delete when migrate `GetClusterCACertificate` to core-svc
		s3PresignClient: s3.NewPresignClient(
			s3.NewFromConfig(awsConfig)),
	}
}

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(dc *cluster.DeletedCluster) *central.DeletedCluster {
	return &central.DeletedCluster{
		Id:                       dc.ID,
		Name:                     dc.Name,
		Provider:                 dc.Provider,
		Region:                   dc.Region,
		RegionName:               dc.RegionName,
		DeleteTime:               timestamppb.New(dc.DeleteTime),
		LatestBackupSize:         dc.LatestBackupSize,
		CrossAccountDeploymentId: dc.XAccountDeployId,
	}
}

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,
		RegionName:      cluster.RegionName,
		CreatedAt:       timestamppb.New(cluster.CreatedAt),
		CreatedBy:       nil,
		Endpoint: &central.Endpoint{
			Ready:   cluster.Endpoints.MySQL.Ready,
			Address: cluster.Endpoints.MySQL.Host,
			Port:    cluster.Endpoints.MySQL.Port,
			Token:   cluster.Endpoints.MySQL.Token,
		},
		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
	}

	if cluster.CreatedBy != nil {
		// TODO(Louis): Replace it when CBP supports querying api key information by id.
		creatorEmail := cluster.CreatedBy.Email
		if creatorEmail == dbutil.TenantSystemUserEmail(cluster.CreatedBy.TenantID) {
			creatorEmail = "OpenAPI"
		}

		clusterItem.CreatedBy = &central.User{
			Email:  creatorEmail,
			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 {
			clusterItem.CreatedBy.InviteTime = timestamppb.New(*cluster.CreatedBy.InvitedAt)
		}
	}

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

		if resource.ComponentName == component.TiDB {
			clusterComponents.Tidb = &clusterComponent
		}
		if resource.ComponentName == component.TiKV {
			if capy.Disk != nil {
				clusterComponent.NodeProfile.StorageGi = capy.Disk.SizeGi
				clusterItem.Capacity = clusterComponent.NodeProfile.StorageGi * clusterComponent.Replicas
			}
			clusterComponents.Tikv = &clusterComponent
		}
		if resource.ComponentName == component.TiFlash {
			if capy.Disk != nil {
				clusterComponent.NodeProfile.StorageGi = capy.Disk.SizeGi
			}
			clusterComponents.Tiflash = &clusterComponent
		}
	}

	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, _, err := context.ListV2(ctx, projectID, entity.Cloud(req.Provider))
	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"))
	}

	// status filter
	sf := make(map[central.Cluster_Status]struct{})
	if req.Filter != nil {
		for _, status := range req.Filter.Statuses {
			sf[status] = struct{}{}
		}
	}
	if len(sf) > 0 {
		newClusters := make([]cluster.Cluster, 0)
		for _, cluster := range clusters {
			if _, ok := sf[cluster.Status]; !ok {
				continue
			}
			newClusters = append(newClusters, cluster)
		}
		clusters = newClusters
	}

	// Since userClusterStatus cannot be converted to core.ClusterStatus equivalently, there is currently
	// no way to fitler with userClusterStatus by calling core-svc with polulating core.ListClustersReq.Filter
	items, err := s.paginateClustersList(ctx, clusters, req.Page, req.PerPage)
	if err != nil {
		return nil, err
	}

	return &central.Clusters{
		Total:    int32(len(clusters)),
		Items:    items,
		BaseResp: &central.BaseResp{},
	}, nil
}

func (s *clusterService) paginateClustersList(ctx context.Context, clusters []cluster.Cluster, page, perPage uint32) ([]*central.Cluster, bizerr.BizError) {

	uint32Min := func(x, y uint32) uint32 {
		if x < y {
			return x
		}
		return y
	}

	startIndex := uint32(0)
	endIndex := uint32(len(clusters))
	if page > 0 {
		startIndex = (page - 1) * perPage
		endIndex = uint32Min(startIndex+perPage, uint32(len(clusters)))
	} else if perPage > 0 {
		// page == 0 && perPage > 0
		return nil, bizerr.New(bizerr.ErrCodeParams, "pagination param error, page: %d, per_page: %d", page, perPage)
	}
	items := make([]*central.Cluster, 0)
	for i := startIndex; i < endIndex; i++ {
		items = append(items, s.clusterOutput(s.Log(ctx), &clusters[i]))
	}
	return items, 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(item)
	}
	return &central.DeletedClusters{
		Total: int32(len(items)),
		Items: service.Pagination(req.Page, req.PerPage, items).([]*central.DeletedCluster),
	}, nil
}

func (s *clusterService) ChangePassword(ctx context.Context, req *central.ChangePasswordReq) (_ *central.DefaultResp, rErr error) {
	defer func() {
		serviceUtils.AuditLog(ctx, rErr, log.AuditEventChangeClusterRootPassword, zap.Uint64("clusterId", req.ClusterId))
	}()

	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
	}

	coreReq := &core.ChangeClusterPasswdReq{
		ClusterId: req.ClusterId,
		Passwd:    req.Password,
	}
	_, err = grpcli.ChangeClusterPasswd(ctx, coreReq)
	if err != nil {
		return nil, err
	}
	return &central.DefaultResp{
		BaseResp: &central.BaseResp{},
	}, nil
}

// Delete marks a cluster as deleted in the DB.
func (s *clusterService) Delete(ctx context.Context, req *central.ClusterIdReq) (_ *emptypb.Empty, rErr error) {
	defer func() {
		serviceUtils.AuditLog(ctx, rErr, log.AuditEventDeleteCluster, zap.Uint64("clusterId", req.ClusterId))
	}()

	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 &emptypb.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) (*emptypb.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 &emptypb.Empty{}, nil
}

func (s *clusterService) GetTrafficFilter(ctx context.Context, req *central.ClusterIdReq) (_ *central.TrafficFilterResp, rErr error) {
	defer func() {
		serviceUtils.AuditLog(ctx, rErr, log.AuditEventGetTrafficFilter, zap.Uint64("clusterId", req.ClusterId))
	}()

	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 := grpcli.GetCluster(ctx, req.OrgId, req.ProjectId, req.ClusterId)
	if err != nil {
		return nil, err
	}
	resp := &central.TrafficFilterResp{
		Total:          int32(len(cluster.TrafficFilter.Items)),
		TrafficFilters: &central.TrafficFilterList{},
	}
	for _, each := range cluster.TrafficFilter.Items {
		resp.TrafficFilters.Cidrs = append(resp.TrafficFilters.Cidrs, &central.TrafficFilter{
			Cidr:        each.Cidr,
			Description: each.Description,
		})
	}

	return resp, nil
}

func (s *clusterService) UpdateTrafficFilter(ctx context.Context, req *central.UpdateTrafficFilterReq) (_ *central.UpdateTrafficFilterResp, rErr error) {
	defer func() {
		serviceUtils.AuditLog(ctx, rErr, log.AuditEventUpdateTrafficFilter, zap.Int64("clusterId", req.ClusterId))
	}()

	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
	}

	coreReq := &core.UpdateClusterTrafficFilterReq{
		ClusterId: uint64(req.ClusterId),
		Filter:    &core.TrafficFilter{},
	}
	for _, each := range req.TrafficFilters.Cidrs {
		coreReq.Filter.Items = append(coreReq.Filter.Items, &core.TrafficFilterItem{
			Cidr:        each.Cidr,
			Description: each.Description,
		})
	}

	if _, err := grpcli.UpdateClusterTrafficFilter(ctx, coreReq); err != nil {
		logger := s.Log(ctx)
		logger.Error("failed to update traffic filter", zap.Int64("clusterID", int64(clusterID.Uint64())), zap.Error(err))
		return nil, err
	}
	return &central.UpdateTrafficFilterResp{}, 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) GetProjectCIDR(ctx context.Context, req *central.GetCIDRReq) (_ *central.ProjectCIDR, rErr error) {
	defer func() {
		serviceUtils.AuditLog(ctx, rErr, log.AuditEventGetProjectCIDR, zap.Uint64("projectId", req.ProjectId))
	}()

	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 := cluster.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 := cluster.GetProjectNetwork(ctx, projectID, provider.ID)
	switch err {
	case nil:
		projectCIDR.Cidr = projectNetwork.CIDR
		projectCIDR.IsDefault = false
	case gorm.ErrRecordNotFound:
		projectCIDR.Cidr = cluster.GetDefaultCIDR()
		projectCIDR.IsDefault = true
	default:
		return nil, err
	}
	projectCIDR.RegionCidr = regionCIDR

	return &projectCIDR, nil
}

func (s *clusterService) GetProjectCIDRWithoutRegion(ctx context.Context, req *central.GetCIDRReq) (*central.ProjectCIDR, error) {
	return s.GetProjectCIDR(ctx, req) // proxy to GetProjectCIDR
}

// 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 := cluster.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, rErr error) {
	defer func() {
		serviceUtils.AuditLog(ctx, rErr, log.AuditEventCreateProjectCIDR, zap.Uint64("projectId", req.ProjectId))
	}()

	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 := cluster.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, rErr error) {
	defer func() {
		serviceUtils.AuditLog(ctx, rErr, log.AuditEventGetProjectRegionCIDR, zap.Uint64("projectId", req.ProjectId))
	}()

	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) (*emptypb.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 &emptypb.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, rErr error) {
	defer func() {
		serviceUtils.AuditLog(ctx, rErr, log.AuditEventUpdateAuditLogStatus, zap.Uint64("clusterId", req.ClusterId), zap.Bool("enableAuditLog", req.Enabled))
	}()

	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)
	}
	if err := ensureClusterStatus(ctx, req.ClusterId, validClusterStatuses4AuditLog); err != nil {
		return nil, err
	}

	// pre-check only if enable audit-log
	if req.Enabled {
		// 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)
	}
	if err := ensureClusterStatus(ctx, req.ClusterId, validClusterStatuses4AuditLog); err != nil {
		return nil, 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)
	}
	if err := ensureClusterStatus(ctx, req.ClusterId, validClusterStatuses4AuditLog); err != nil {
		return nil, 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)
	}
	if err := ensureClusterStatus(ctx, req.ClusterId, validClusterStatuses4AuditLog); err != nil {
		return nil, 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, rErr error) {
	defer func() {
		serviceUtils.AuditLog(ctx, rErr, log.AuditEventDeleteAuditLogAccessRecord, zap.Uint64("clusterId", req.ClusterId))
	}()

	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 := ensureClusterStatus(ctx, req.ClusterId, validClusterStatuses4AuditLog); err != nil {
		return nil, 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, rErr error) {
	defer func() {
		serviceUtils.AuditLog(ctx, rErr, log.AuditEventCreateAuditLogAccessRecord, zap.Uint64("clusterId", req.ClusterId))
	}()

	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 := ensureClusterStatus(ctx, req.ClusterId, validClusterStatuses4AuditLog); err != nil {
		return nil, 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
}

// TODO(zlj): migrate it to core service
// GetClusterCACertificate get the download URL to download the CA certitficate of the cluster
func (s *clusterService) GetClusterCACertificate(ctx context.Context, req *central.GetClusterCACertificateReq) (*central.GetClusterCACertificateResp, error) {
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	context := s.userCluster(ctx)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	// Call AWS to get presign URL
	resp, err := s.s3PresignClient.PresignGetObject(ctx, &s3.GetObjectInput{
		Bucket:                     pointer.StringPtr(caCertificateS3Bucket),
		Key:                        pointer.StringPtr(caCertificateS3Key),
		ResponseContentDisposition: pointer.StringPtr(respContentDispositionAttachment),
	}, s3.WithPresignExpires(defaultPresignExpires))
	if err != nil {
		return nil, err
	}

	return &central.GetClusterCACertificateResp{
		DownloadUrl: resp.URL,
	}, nil
}

func ensureClusterStatus(ctx context.Context, clusterId uint64, expectedStatuses []core.ClusterStatus) bizerr.BizError {
	cluster, err := grpcli.GetClusterWithoutPermCheck(ctx, clusterId)
	if err != nil {
		return err
	}

	for _, each := range expectedStatuses {
		if cluster.Status == each {
			return nil
		}
	}
	return bizerr.New(bizerr.ErrCodeParams, "the cluster's status is invalid: %s", cluster.Status)
}