package impl

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"strconv"
	"strings"

	"github.com/google/uuid"
	ticdc_v1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/dataflow-service-ng/app/event_bus/publisher/center"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	"github.com/tidbcloud/dataflow-service-ng/app/repo"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/dataflow-service-ng/app/utils"
	"github.com/tidbcloud/dataflow-service-ng/pkg/grpc/grpcli"
	regionalapi "github.com/tidbcloud/dataflow-service-ng/pkg/regional-api"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
)

const (
	allZonesPlaceHolder = "all zones"
)

type sinkEndpointImpl struct {
	transProvider          db.TransProvider
	cli                    regionalapi.TiCDCPrivateLinkClient
	bizPLEndpointRepo      repo.BizSinkEndpointRepo
	bizPLEndpointUsageRepo repo.BizSinkEndpointUsageRepo
	mgmtCli                grpcli.MgmtSvcGrpcClient
	eventPublisher         center.EventCenterPublisher
}

func NewSinkEndpointImpl(
	transProvider db.TransProvider,
	cli regionalapi.TiCDCPrivateLinkClient,
	bizPLEndpointRepo repo.BizSinkEndpointRepo,
	bizPLEndpointUsageRepo repo.BizSinkEndpointUsageRepo,
	mgmtCli grpcli.MgmtSvcGrpcClient,
	eventPublisher center.EventCenterPublisher,
) *sinkEndpointImpl {
	return &sinkEndpointImpl{
		transProvider:          transProvider,
		cli:                    cli,
		bizPLEndpointRepo:      bizPLEndpointRepo,
		bizPLEndpointUsageRepo: bizPLEndpointUsageRepo,
		mgmtCli:                mgmtCli,
		eventPublisher:         eventPublisher,
	}
}

func (m *sinkEndpointImpl) CreateSinkEndpoint(
	ctx context.Context, req *domain.CreateSinkEndpointReq,
) (*domain.CreateSinkEndpointResp, bizerr.BizError) {
	cluster, err := m.mgmtCli.GetClusterByID(ctx, req.ClusterID)
	if err != nil {
		return nil, err.Chain("get cluster")
	}
	clusterWrapper := utils.NewCluster(cluster)
	if req.EndpointType == entity.BizSinkEndpointTypeAmazonMSK {
		if err := m.preCheckCreateMSKSinkEndpointReq(ctx, req, clusterWrapper); err != nil {
			return nil, err.Chain("pre check create msk request")
		}
	} else {
		m.refineDNSInfo(req, entity.CloudProviderType(clusterWrapper.GetCloudProvider()))
		if err := m.preCheckCreateSinkEndpointReq(ctx, req, clusterWrapper); err != nil {
			return nil, err.Chain("pre check create request")
		}
	}
	endpoint := m.createSinkEndpointReq2Entity(clusterWrapper, req)
	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		return m.doCreateSinkEndpoint(ctx, clusterWrapper, endpoint)
	})
	if transErr != nil {
		return nil, convertTransactionError(transErr)
	}
	PublishPrivateEndpointEvent(
		ctx, m.eventPublisher, endpoint, entity.CreatePrivateEndpointForExternalServices,
	)
	return &domain.CreateSinkEndpointResp{SinkEndpoint: endpoint}, nil
}

func (m *sinkEndpointImpl) GetSinkEndpointAuthInfo(ctx context.Context,
	req *domain.GetSinkEndpointAuthInfoReq,
) (*domain.GetSinkEndpointAuthInfoResp, bizerr.BizError) {
	cluster, err := m.mgmtCli.GetClusterByID(ctx, req.ClusterID)
	if err != nil {
		return nil, err.Chain("get cluster")
	}
	clusterWrapper := utils.NewCluster(cluster)
	region := &regionalapi.Region{
		SharedPoolID: clusterWrapper.GetSharedPoolId(),
	}
	authReq := &ticdc_v1.GetSinkEndpointAuthInfoRequest{
		TenantId:     strconv.FormatUint(clusterWrapper.GetTenantID(), 10),
		SharedPoolId: clusterWrapper.GetSharedPoolId(),
	}
	resp, err := m.cli.GetPrivateLinkAuthInfo(ctx, region, authReq)
	if err != nil {
		return nil, err.Chain("get private link auth info")
	}
	return &domain.GetSinkEndpointAuthInfoResp{
		CloudAccountID: resp.CloudAccountId,
		AWSPrincipalID: resp.AwsPrincipalId,
	}, nil
}

func (m *sinkEndpointImpl) preCheckCreateSinkEndpointReq(
	ctx context.Context, req *domain.CreateSinkEndpointReq, cluster *utils.Cluster,
) bizerr.BizError {
	// service_name is no longer enforced by proto validation (it is empty for the
	// amazon_msk type), so the vpc_endpoint_service path enforces it here.
	if req.ServiceName == "" {
		return bizerr.New(bizerr.ErrCodeParams, "endpoint service name is required")
	}
	if err := m.validateEndpointName(
		ctx, cluster.GetTenantID(), cluster.GetSharedPoolId(), cluster.GetRegionName(), req.Name); err != nil {
		return err
	}
	if err := m.validateRootDomainName(
		ctx, cluster.GetSharedPoolId(), req.RootDomainName); err != nil {
		return err
	}

	region := &regionalapi.Region{
		SharedPoolID: cluster.GetSharedPoolId(),
	}
	validateReq := &ticdc_v1.ValidateEndpointAvailabilityZonesRequest{
		TenantId:      strconv.FormatUint(cluster.GetTenantID(), 10),
		SharedPoolId:  cluster.GetSharedPoolId(),
		ServiceName:   req.ServiceName,
		EndpointZones: req.Zones,
	}
	return m.cli.ValidateEndpointAvailabilityZones(ctx, region, validateReq)
}

func (m *sinkEndpointImpl) preCheckCreateMSKSinkEndpointReq(
	ctx context.Context, req *domain.CreateSinkEndpointReq, cluster *utils.Cluster,
) bizerr.BizError {
	if cluster.GetCloudProvider() != string(entity.CloudProviderTypeAWS) {
		return bizerr.New(bizerr.ErrCodeParams, "amazon msk sink endpoint is only supported on AWS")
	}
	if err := m.validateEndpointName(
		ctx, cluster.GetTenantID(), cluster.GetSharedPoolId(), cluster.GetRegionName(), req.Name); err != nil {
		return err
	}
	if req.MSKArn == "" {
		return bizerr.New(bizerr.ErrCodeParams, "msk_arn is required for amazon msk sink endpoint")
	}
	// MSK connects via AWS-managed multi-VPC connectivity; the endpoint-service /
	// advertised-listener fields do not apply and must not be set (otherwise we'd
	// persist and echo misleading data for MSK endpoints).
	if req.ServiceName != "" || req.ServiceRegion != "" || req.RootDomainName != "" ||
		len(req.Zones) > 0 || len(req.DNSRecords) > 0 || req.IsCustomDomain {
		return bizerr.New(bizerr.ErrCodeParams,
			"service_name / availability zones / advertised listener must not be set for amazon msk sink endpoint")
	}
	// The VPC connection authentication scheme is decided once, at sink endpoint
	// creation time (AWS requires it and it is immutable). Only SASL/SCRAM is
	// implemented for now.
	if req.AuthType == "" {
		req.AuthType = entity.BizSinkEndpointAuthSASLScram
	}
	if req.AuthType != entity.BizSinkEndpointAuthSASLScram {
		return bizerr.New(bizerr.ErrCodeParams,
			"amazon msk auth type %s is not supported yet", req.AuthType)
	}
	return nil
}

func (m *sinkEndpointImpl) refineDNSInfo(req *domain.CreateSinkEndpointReq, provider entity.CloudProviderType) {
	if req.RootDomainName != "" {
		req.RootDomainName = m.refineDomainName(provider, req.RootDomainName)
	}
	for i := range req.DNSRecords {
		req.DNSRecords[i].RecordName = m.refineDomainName(provider, req.DNSRecords[i].RecordName)
	}
}

func (m *sinkEndpointImpl) refineDomainName(provider entity.CloudProviderType, domainName string) string {
	if domainName == "" {
		return ""
	}

	if provider == entity.CloudProviderTypeAzure || provider == entity.CloudProviderTypeAliCloud {
		return strings.TrimRight(domainName, ".")
	}
	if !strings.HasSuffix(domainName, ".") {
		return domainName + "."
	}
	return domainName
}

func (m *sinkEndpointImpl) validateRootDomainName(
	ctx context.Context, sharedPoolID string, rootDomainName string,
) bizerr.BizError {
	if rootDomainName == "" {
		return nil
	}

	filter := func(qs entity.BizSinkEndpointQuerySet) entity.BizSinkEndpointQuerySet {
		return qs.SharedPoolIDEq(sharedPoolID).DeletedTimeEq(0)
	}
	endpoints, err := m.bizPLEndpointRepo.List(ctx, filter)
	if err != nil {
		return err
	}
	for _, endpoint := range endpoints {
		if endpoint.RootDomainName == "" {
			continue
		}
		if strings.HasSuffix(endpoint.RootDomainName, rootDomainName) ||
			strings.HasSuffix(rootDomainName, endpoint.RootDomainName) {
			return bizerr.New(bizerr.ErrCodeParams,
				"the domain name is conflict with domain name %s of endpoint %d",
				endpoint.RootDomainName, endpoint.ID)
		}
	}
	return nil
}

func (m *sinkEndpointImpl) doCreateSinkEndpoint(
	ctx context.Context, clusterWrapper *utils.Cluster, endpoint *entity.BizSinkEndpoint,
) bizerr.BizError {
	if err := m.bizPLEndpointRepo.Save(ctx, endpoint); err != nil {
		return err.Chain("save private link endpoint entity")
	}
	if err := m.createRegionalEndpointIfNotExists(ctx, clusterWrapper, endpoint); err != nil {
		return err.Chain("create regional endpoint")
	}
	return nil
}

func (m *sinkEndpointImpl) createRegionalEndpointIfNotExists(
	ctx context.Context, clusterWrapper *utils.Cluster, endpointEnt *entity.BizSinkEndpoint,
) bizerr.BizError {
	region := &regionalapi.Region{
		SharedPoolID: endpointEnt.SharedPoolID,
	}
	req := &ticdc_v1.CreateTiCDCSinkEndpointRequest{
		TicdcSinkEndpointId: endpointEnt.RegionalEndpointID,
		TicdcSinkEndpoint:   m.convertRegionalSinkEndpoint(endpointEnt),
	}
	if _, err := m.cli.CreateTiCDCPrivateLink(ctx, region, req); err != nil {
		if err.ErrCode() == bizerr.ErrCodeRecordExist {
			return nil
		}
		return err
	}
	return nil
}

func (m *sinkEndpointImpl) createSinkEndpointReq2Entity(
	clusterWrapper *utils.Cluster, req *domain.CreateSinkEndpointReq,
) *entity.BizSinkEndpoint {
	endpointType := req.EndpointType
	if endpointType == "" {
		endpointType = entity.BizSinkEndpointTypeVPCEndpointService
	}
	endpoint := &entity.BizSinkEndpoint{
		TenantID:       entity.ID(clusterWrapper.GetTenantID()),
		CloudProvider:  entity.CloudProviderType(clusterWrapper.GetCloudProvider()),
		SharedPoolID:   clusterWrapper.GetSharedPoolId(),
		Region:         clusterWrapper.GetRegionName(),
		Name:           req.Name,
		Status:         entity.BizPLEndpointStatusCreating,
		EndpointType:   endpointType,
		EventClusterID: entity.ID(clusterWrapper.GetClusterID()),
		EventProjectID: entity.ID(clusterWrapper.GetProjectID()),
		EventAuthorID:  req.UserID,
		EventAuthor:    req.UserEmail,
	}
	// The regional endpoint id must be unique within the shared pool. MSK and the
	// endpoint-service model carry disjoint fields, so only persist the ones that
	// belong to the chosen type and keep the rest empty.
	if endpointType == entity.BizSinkEndpointTypeAmazonMSK {
		endpoint.MSKArn = req.MSKArn
		endpoint.AuthType = req.AuthType
		endpoint.RegionalEndpointID = getRegionalEndpointID(clusterWrapper.GetSharedPoolId(), req.MSKArn)
		return endpoint
	}
	endpoint.Zones = req.Zones
	endpoint.ServiceName = req.ServiceName
	endpoint.ServiceRegion = req.ServiceRegion
	endpoint.IsCustomDomain = req.IsCustomDomain
	endpoint.RootDomainName = req.RootDomainName
	endpoint.DNSRecords = req.DNSRecords
	endpoint.RegionalEndpointID = getRegionalEndpointID(clusterWrapper.GetSharedPoolId(), req.ServiceName)
	return endpoint
}

func (m *sinkEndpointImpl) convertRegionalSinkEndpoint(
	endpointEnt *entity.BizSinkEndpoint,
) *ticdc_v1.TiCDCSinkEndpoint {
	if endpointEnt.EndpointType == entity.BizSinkEndpointTypeAmazonMSK {
		return &ticdc_v1.TiCDCSinkEndpoint{
			TenantId:     strconv.FormatUint(endpointEnt.TenantID.Uint64(), 10),
			SharedPoolId: endpointEnt.SharedPoolID,
			EndpointType: ticdc_v1.TiCDCSinkEndpoint_AMAZON_MSK,
			MskArn:       endpointEnt.MSKArn,
			AuthType:     mskAuthTypeToRegional(endpointEnt.AuthType),
		}
	}
	records := make([]*ticdc_v1.TiCDCSinkEndpoint_DNSRecord, 0, len(endpointEnt.DNSRecords))
	for _, r := range endpointEnt.DNSRecords {
		records = append(records, &ticdc_v1.TiCDCSinkEndpoint_DNSRecord{
			AvailabilityZone: r.AvailabilityZone,
			RecordName:       r.RecordName,
		})
	}
	return &ticdc_v1.TiCDCSinkEndpoint{
		TenantId:          strconv.FormatUint(endpointEnt.TenantID.Uint64(), 10),
		SharedPoolId:      endpointEnt.SharedPoolID,
		EndpointType:      ticdc_v1.TiCDCSinkEndpoint_VPC_ENDPOINT_SERVICE,
		ServiceName:       endpointEnt.ServiceName,
		RootDomainName:    endpointEnt.RootDomainName,
		DnsRecords:        records,
		AvailabilityZones: endpointEnt.Zones,
	}
}

func mskAuthTypeToRegional(t entity.BizSinkEndpointAuthType) ticdc_v1.TiCDCSinkEndpoint_MskAuthType {
	switch t {
	case entity.BizSinkEndpointAuthSASLScram:
		return ticdc_v1.TiCDCSinkEndpoint_SASL_SCRAM
	case entity.BizSinkEndpointAuthSASLIAM:
		return ticdc_v1.TiCDCSinkEndpoint_SASL_IAM
	default:
		return ticdc_v1.TiCDCSinkEndpoint_MSK_AUTH_TYPE_UNSPECIFIED
	}
}

func (m *sinkEndpointImpl) ListSinkEndpoint(
	ctx context.Context, req *domain.ListSinkEndpointsReq,
) (*domain.ListSinkEndpointsResp, bizerr.BizError) {
	cluster, err := m.mgmtCli.GetClusterByID(ctx, req.ClusterID)
	if err != nil {
		return nil, err.Chain("get cluster")
	}
	clusterWrapper := utils.NewCluster(cluster)
	filter := m.genEndpointFilterByListSinkEndpointReq(clusterWrapper, req)
	endpoints, err := m.bizPLEndpointRepo.List(ctx, filter)
	if err != nil {
		return nil, err
	}
	return &domain.ListSinkEndpointsResp{
		SinkEndpoints: endpoints,
	}, nil
}

func (m *sinkEndpointImpl) genEndpointFilterByListSinkEndpointReq(
	clusterWrapper *utils.Cluster, req *domain.ListSinkEndpointsReq,
) repo.BizPrivateLinkEndpointFilterOP {
	return func(qs entity.BizSinkEndpointQuerySet) entity.BizSinkEndpointQuerySet {
		qs = qs.TenantIDEq(entity.ID(clusterWrapper.GetTenantID())).SharedPoolIDEq(clusterWrapper.GetSharedPoolId()).
			StatusNotIn(entity.BizPLEndpointStatusDeleting, entity.BizPLEndpointStatusDeleted)
		if req.Status != "" {
			qs = qs.StatusEq(req.Status)
		}
		return qs
	}
}

func (m *sinkEndpointImpl) QuerySinkEndpoint(
	ctx context.Context, req *domain.QuerySinkEndpointReq,
) (*domain.QuerySinkEndpointResp, bizerr.BizError) {
	endpoint, err := m.querySinkEndpoint(ctx, req)
	if err != nil {
		return nil, err
	}
	return &domain.QuerySinkEndpointResp{
		SinkEndpoint: endpoint,
	}, nil
}

func (m *sinkEndpointImpl) DeleteSinkEndpoint(ctx context.Context, req *domain.DeleteSinkEndpointReq) bizerr.BizError {
	endpoint, err := m.querySinkEndpoint(ctx, &domain.QuerySinkEndpointReq{
		ID:        req.ID,
		ClusterID: req.ClusterID,
	})
	if err != nil {
		if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return nil
		}
		return err.Chain("load private link endpoint entity")
	}
	if err := m.preCheckDeleteSinkEndpointReq(ctx, req); err != nil {
		return err.Chain("pre check delete request")
	}

	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		endpoint.EventClusterID = entity.ID(req.ClusterID)
		endpoint.EventProjectID = entity.ID(req.ProjectID)
		endpoint.EventAuthorID = req.UserID
		endpoint.EventAuthor = req.UserEmail
		return m.doDeleteSinkEndpoint(ctx, endpoint)
	})
	if bizErr := convertTransactionError(transErr); bizErr != nil {
		return bizErr
	}
	PublishPrivateEndpointEvent(
		ctx, m.eventPublisher, endpoint, entity.DeletePrivateEndpointForExternalServices,
	)
	return nil
}

func (m *sinkEndpointImpl) querySinkEndpoint(
	ctx context.Context, req *domain.QuerySinkEndpointReq,
) (*entity.BizSinkEndpoint, bizerr.BizError) {
	if req.ClusterID == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster id is required for sink endpoint query")
	}

	cluster, err := m.mgmtCli.GetClusterByID(ctx, req.ClusterID)
	if err != nil {
		return nil, err.Chain("get cluster")
	}
	clusterWrapper := utils.NewCluster(cluster)
	filter := func(qs entity.BizSinkEndpointQuerySet) entity.BizSinkEndpointQuerySet {
		return qs.IDEq(entity.ID(req.ID)).
			TenantIDEq(entity.ID(clusterWrapper.GetTenantID())).
			SharedPoolIDEq(clusterWrapper.GetSharedPoolId()).
			DeletedTimeEq(0)
	}
	endpoints, err := m.bizPLEndpointRepo.List(ctx, filter)
	if err != nil {
		return nil, err
	}
	if len(endpoints) == 0 {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "sink endpoint %d not found", req.ID)
	}
	return &endpoints[0], nil
}

func (m *sinkEndpointImpl) preCheckDeleteSinkEndpointReq(
	ctx context.Context, req *domain.DeleteSinkEndpointReq,
) bizerr.BizError {
	usageFilter := func(qs entity.BizSinkEndpointUsageQuerySet) entity.BizSinkEndpointUsageQuerySet {
		return qs.BizSinkEndpointIDEq(entity.ID(req.ID))
	}
	usages, err := m.bizPLEndpointUsageRepo.List(ctx, usageFilter)
	if err != nil {
		return err.Chain("list private link endpoint usages")
	}
	if len(usages) == 0 {
		return nil
	}

	usageDisplayList := make([]string, 0, len(usages))
	for _, usage := range usages {
		usageDisplayList = append(usageDisplayList, fmt.Sprintf("%s(%s)", usage.UsageType, usage.OwnerID))
	}
	sort.Strings(usageDisplayList)
	return bizerr.New(
		bizerr.ErrCodeParams,
		"the sink endpoint is used by following resources, please delete them in advance\n%s",
		strings.Join(usageDisplayList, "\n"))
}

func (m *sinkEndpointImpl) doDeleteSinkEndpoint(
	ctx context.Context, endpointEnt *entity.BizSinkEndpoint,
) bizerr.BizError {
	endpointEnt.Status = entity.BizPLEndpointStatusDeleting
	if err := m.bizPLEndpointRepo.Save(ctx, endpointEnt); err != nil {
		return err.Chain("save private link endpoint entity")
	}
	if err := m.deleteRegionalEndpointIfExists(ctx, endpointEnt); err != nil {
		return err.Chain("delete regional endpoint")
	}
	return nil
}

func (m *sinkEndpointImpl) deleteRegionalEndpointIfExists(
	ctx context.Context, endpointEnt *entity.BizSinkEndpoint,
) bizerr.BizError {
	region := &regionalapi.Region{
		SharedPoolID: endpointEnt.SharedPoolID,
	}
	req := &ticdc_v1.DeleteTiCDCSinkEndpointRequest{
		TicdcSinkEndpointId: endpointEnt.RegionalEndpointID,
		SharedPoolId:        endpointEnt.SharedPoolID,
	}
	if err := m.cli.DeleteTiCDCPrivateLink(ctx, region, req); err != nil {
		if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return nil
		}
		return err
	}
	return nil
}

func (m *sinkEndpointImpl) GetAvailabilityZones(
	ctx context.Context, req *domain.GetAvailabilityZonesReq,
) (*domain.GetAvailabilityZonesResp, bizerr.BizError) {
	cluster, err := m.mgmtCli.GetClusterByID(ctx, req.ClusterID)
	if err != nil {
		return nil, err
	}
	clusterWrapper := utils.NewCluster(cluster)
	region := &regionalapi.Region{
		SharedPoolID: clusterWrapper.GetSharedPoolId(),
	}
	zoneReq := &ticdc_v1.GetAvailableZonesRequest{
		TenantId:     strconv.FormatUint(clusterWrapper.GetTenantID(), 10),
		SharedPoolId: clusterWrapper.GetSharedPoolId(),
	}
	zoneResp, err := m.cli.GetAvailableZones(ctx, region, zoneReq)
	if err != nil {
		return nil, err
	}
	return &domain.GetAvailabilityZonesResp{
		Zones: zoneResp.Zones,
	}, nil
}

func (m *sinkEndpointImpl) DescribeEndpointServiceAvailabilityZones(
	ctx context.Context, req *domain.DescribeEndpointServiceAvailabilityZonesReq,
) (*domain.DescribeEndpointServiceAvailabilityZonesResp, bizerr.BizError) {
	cluster, err := m.mgmtCli.GetClusterByID(ctx, req.ClusterID)
	if err != nil {
		return nil, err
	}
	clusterWrapper := utils.NewCluster(cluster)
	region := &regionalapi.Region{
		SharedPoolID: clusterWrapper.GetSharedPoolId(),
	}
	zoneReq := &ticdc_v1.DescribeEndpointServiceAvailabilityZonesRequest{
		SharedPoolId:  clusterWrapper.GetSharedPoolId(),
		ServiceName:   req.ServiceName,
		ServiceRegion: normalizeEndpointServiceRegion(clusterWrapper.GetCloudProvider(), req.ServiceRegion),
	}
	zoneResp, err := m.cli.DescribeEndpointServiceAvailabilityZones(ctx, region, zoneReq)
	if err != nil {
		return nil, err
	}

	zones := make([]domain.EndpointServiceAvailabilityZone, 0, len(zoneResp.Zones))
	for _, zone := range zoneResp.Zones {
		zones = append(zones, domain.EndpointServiceAvailabilityZone{
			AvailabilityZone:   zone.AvailabilityZone,
			AvailabilityZoneID: zone.AvailabilityZoneId,
		})
	}
	return &domain.DescribeEndpointServiceAvailabilityZonesResp{
		Zones: zones,
	}, nil
}

func normalizeEndpointServiceRegion(provider string, serviceRegion string) string {
	serviceRegion = strings.TrimSpace(serviceRegion)
	if strings.EqualFold(provider, string(entity.CloudProviderTypeAWS)) {
		return strings.TrimPrefix(serviceRegion, "aws-")
	}
	if strings.EqualFold(provider, string(entity.CloudProviderTypeAliCloud)) {
		return strings.TrimPrefix(serviceRegion, "alicloud-")
	}
	return serviceRegion
}

func (m *sinkEndpointImpl) GenerateSinkAdvertisedListener(
	ctx context.Context, req *domain.GenerateSinkAdvertisedListenerReq,
) (*domain.GenerateSinkAdvertisedListenerResp, bizerr.BizError) {
	cluster, err := m.mgmtCli.GetClusterByID(ctx, req.ClusterID)
	if err != nil {
		return nil, err
	}
	clusterWrapper := utils.NewCluster(cluster)

	rootDomainName := m.genRootDomainName(clusterWrapper, req)
	if err := m.validateRootDomainName(
		ctx, clusterWrapper.GetSharedPoolId(), rootDomainName); err != nil {
		return nil, err
	}

	records := m.genDNSRecords(rootDomainName, req.Zones)
	return &domain.GenerateSinkAdvertisedListenerResp{
		RootDomainName: rootDomainName,
		DNSRecords:     records,
	}, nil
}

func (m *sinkEndpointImpl) genRootDomainName(
	clusterWrapper *utils.Cluster,
	req *domain.GenerateSinkAdvertisedListenerReq,
) string {
	switch clusterWrapper.GetCloudProvider() {
	case string(entity.CloudProviderTypeAzure):
		return GenAzureRootDomainName(req.DownstreamID, clusterWrapper.GetRegionName(),
			clusterWrapper.GetCloudProvider(),
			clusterWrapper.GetSharedPoolId())
	default:
		return GenRootDomainName(req.DownstreamID, clusterWrapper.GetRegionName(),
			clusterWrapper.GetCloudProvider(), clusterWrapper.GetSharedPoolId())
	}
}

func (m *sinkEndpointImpl) genDNSRecords(rootDomainName string, zones []string) []entity.BizSinkEndpointDNSRecord {
	records := make([]entity.BizSinkEndpointDNSRecord, 0, len(zones))
	if len(zones) > 0 {
		for _, zone := range zones {
			records = append(records, entity.BizSinkEndpointDNSRecord{
				AvailabilityZone: zone,
				RecordName:       fmt.Sprintf("%s.%s", zone, rootDomainName),
			})
		}
	} else {
		records = []entity.BizSinkEndpointDNSRecord{
			{
				AvailabilityZone: allZonesPlaceHolder,
				RecordName:       rootDomainName,
			},
		}
	}
	return records
}

func (m *sinkEndpointImpl) ValidateSinkAdvertisedListenerDomain(
	ctx context.Context, req *domain.ValidateSinkAdvertisedListenerDomainReq,
) (*domain.ValidateSinkAdvertisedListenerDomainResp, bizerr.BizError) {
	cluster, err := m.mgmtCli.GetClusterByID(ctx, req.ClusterID)
	if err != nil {
		return nil, err
	}
	clusterWrapper := utils.NewCluster(cluster)
	if err := m.validateRootDomainName(
		ctx, clusterWrapper.GetSharedPoolId(), req.RootDomainName); err != nil {
		return nil, err
	}
	records := m.genDNSRecords(req.RootDomainName, req.Zones)
	return &domain.ValidateSinkAdvertisedListenerDomainResp{
		DNSRecords: records,
	}, nil
}

func (m *sinkEndpointImpl) validateEndpointName(
	ctx context.Context, tenantID uint64, sharedPoolID string,
	region, endpointName string,
) bizerr.BizError {
	filter := func(qs entity.BizSinkEndpointQuerySet) entity.BizSinkEndpointQuerySet {
		return qs.TenantIDEq(entity.ID(tenantID)).SharedPoolIDEq(sharedPoolID).
			RegionEq(region).NameEq(endpointName).DeletedTimeEq(0)
	}
	endpoints, err := m.bizPLEndpointRepo.List(ctx, filter)
	if err != nil {
		return err
	}
	if len(endpoints) > 0 {
		return bizerr.New(bizerr.ErrCodeParams, "endpoint name %s is duplicate", endpointName)
	}
	return nil
}

func (m *sinkEndpointImpl) CreateSinkEndpointUsage(
	ctx context.Context, req *domain.CreateSinkEndpointUsageReq,
) (*domain.CreateSinkEndpointUsageResp, bizerr.BizError) {
	usageEnt := m.createSinkEndpointUsageReq2Entity(req)
	if err := m.bizPLEndpointUsageRepo.Save(ctx, usageEnt); err != nil {
		return nil, err
	}
	return &domain.CreateSinkEndpointUsageResp{
		SinkEndpointUsage: usageEnt,
	}, nil
}

func (m *sinkEndpointImpl) createSinkEndpointUsageReq2Entity(
	req *domain.CreateSinkEndpointUsageReq,
) *entity.BizSinkEndpointUsage {
	return &entity.BizSinkEndpointUsage{
		BizSinkEndpointID: entity.ID(req.SinkEndpointID),
		OwnerID:           req.OwnerID,
		UsageType:         req.UsageType,
	}
}

func (m *sinkEndpointImpl) DeleteSinkEndpointUsage(
	ctx context.Context, req *domain.DeleteSinkEndpointUsageReq,
) bizerr.BizError {
	filter := m.deleteSinkEndpointUsageReq2Filter(req)
	usages, err := m.bizPLEndpointUsageRepo.List(ctx, filter)
	if err != nil {
		return err.Chain("list private link endpoint usages")
	}
	if len(usages) == 0 {
		return nil
	}

	if err := m.bizPLEndpointUsageRepo.Delete(ctx, usages[0].ID.Uint64()); err != nil {
		return err.Chain("delete private link endpoint usage")
	}
	return nil
}

func (m *sinkEndpointImpl) deleteSinkEndpointUsageReq2Filter(
	req *domain.DeleteSinkEndpointUsageReq,
) repo.BizPrivateLinkEndpointUsageFilterOP {
	return func(qs entity.BizSinkEndpointUsageQuerySet) entity.BizSinkEndpointUsageQuerySet {
		return qs.BizSinkEndpointIDEq(entity.ID(req.SinkEndpointID)).
			OwnerIDEq(req.OwnerID).
			UsageTypeEq(req.UsageType)
	}
}

func convertTransactionError(err error) bizerr.BizError {
	if err == nil {
		return nil
	}

	var bizErr bizerr.BizError
	if errors.As(err, &bizErr) {
		return bizErr
	}
	return bizerr.New(bizerr.ErrCodeDB, "execute transaction error: %s", err.Error())
}

// getRegionalEndpointID returns the regional endpoint id is used to identify the endpoint in the regional api
// the is regional endpoint id is unique in the sharedPoolID
func getRegionalEndpointID(sharedPoolID, serviceName string) string {
	return strings.ToLower(fmt.Sprintf("%s-%s-%s", sharedPoolID,
		utils.FnvHash8(serviceName), utils.FnvHash8(uuid.New().String())))
}

func GenRootDomainName(downStreamID, region string, cloudVendor string, sharedPoolID string) string {
	return strings.ToLower(fmt.Sprintf("%s.%s.%s.%s.tidbcloud.com.", downStreamID, region, cloudVendor, sharedPoolID))
}

// GenAzureRootDomainName generates the root domain name for Azure private link. the private dns zone name is can not end with "."
func GenAzureRootDomainName(downStreamID, region string, cloudVendor string, sharedPoolID string) string {
	return strings.TrimRight(GenRootDomainName(downStreamID, region, cloudVendor, sharedPoolID), ".")
}