package service

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

	"github.com/samber/lo"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
	"github.com/tidbcloud/serverless-service/internal/domain/repo"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	"github.com/tidbcloud/serverless-service/internal/infra/eventbus"
	event_constant "github.com/tidbcloud/serverless-service/pkg/event"
	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

const MaxAttachDomain = 10

var PrivateLinkConnectionVisibleErrCodes = []bizerr.ErrCode{
	bizerr.ErrCodeParams,
	bizerr.ErrCodeRecordNotFound,
}

type PrivateLinkConnectionService interface {
	// Get a private link connection by id.
	GetPrivateLinkConnection(
		ctx context.Context,
		clusterId, privateLinkConnectionId string,
	) (*entity.PrivateLinkConnection, bizerr.BizError)

	// List private link connections with pagination.
	ListPrivateLinkConnections(
		ctx context.Context,
		opts *vo.ListPrivateLinkConnectionOptions,
	) ([]*entity.PrivateLinkConnection, uint32, bizerr.BizError)

	// Create a private link connection.
	CreatePrivateLinkConnection(
		ctx context.Context,
		clusterId string,
		plc *entity.PrivateLinkConnection,
	) (*entity.PrivateLinkConnection, bizerr.BizError)

	// Delete a private link connection.
	DeletePrivateLinkConnection(
		ctx context.Context,
		clusterId, privateLinkConnectionId string,
	) (*entity.PrivateLinkConnection, bizerr.BizError)

	// Get account id and availability zones for a cluster.
	GetAvailabilityZones(ctx context.Context, clusterId string) (string, []string, bizerr.BizError)

	RegisterPrivateLinkConnection(
		ctx context.Context,
		clusterId, privateLinkConnectionId, bizId string, dryrun bool,
	) ([]string, bizerr.BizError)

	DeregisterPrivateLinkConnection(
		ctx context.Context,
		clusterId, privateLinkConnectionId, bizId string,
	) bizerr.BizError

	AttachDomains(
		ctx context.Context,
		plcd *entity.PLCAttachDomain,
		DryRun bool,
	) (*entity.PLCAttachDomain, bizerr.BizError)

	DetachDomains(
		ctx context.Context,
		clusterId, privateLinkConnectionId, attachDomainID string,
	) (*entity.PLCAttachDomain, bizerr.BizError)

	ListAttachDomains(
		ctx context.Context,
		filter entity.ListPLCAttachDomainsFilter,
	) (map[string][]*entity.PLCAttachDomain, bizerr.BizError)
}

type privateLinkConnectionService struct {
	mgmtSvc             facade.MgmtSvcFacade
	plcRepo             repo.PrivateLinkConnectionRepo
	plcLinkRepo         repo.PrivateLinkConnectionLinkRepo
	plcAttachDomainRepo repo.PLCAttachDomainRepo
	plcDomainRecordRepo repo.PLCDomainRecordRepo
	transProvider       db.TransProvider
	eventbusService     *eventbus.EventbusService
}

func NewPrivateLinkConnectionService(
	mgmtSvc facade.MgmtSvcFacade,
	plcRepo repo.PrivateLinkConnectionRepo,
	transProvider db.TransProvider,
	plcLinkRepo repo.PrivateLinkConnectionLinkRepo,
	plcAttachDomainRepo repo.PLCAttachDomainRepo,
	plcDomainRecordRepo repo.PLCDomainRecordRepo,
	eventbusService *eventbus.EventbusService,
) PrivateLinkConnectionService {
	return &privateLinkConnectionService{
		mgmtSvc:             mgmtSvc,
		plcRepo:             plcRepo,
		transProvider:       transProvider,
		plcLinkRepo:         plcLinkRepo,
		plcAttachDomainRepo: plcAttachDomainRepo,
		plcDomainRecordRepo: plcDomainRecordRepo,
		eventbusService:     eventbusService,
	}
}

func (s *privateLinkConnectionService) GetPrivateLinkConnection(
	ctx context.Context,
	clusterId, privateLinkConnectionId string,
) (*entity.PrivateLinkConnection, bizerr.BizError) {
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", clusterId), zap.String("private_link_connection_id", privateLinkConnectionId))
	plc, err := s.getPrivateLinkConnectionForCluster(ctx, clusterId, privateLinkConnectionId)
	if err != nil {
		logger.Error("failed to get private link connection", zap.Error(err))
		return nil, err
	}
	return plc, nil
}

func (s *privateLinkConnectionService) ListPrivateLinkConnections(
	ctx context.Context,
	opts *vo.ListPrivateLinkConnectionOptions,
) ([]*entity.PrivateLinkConnection, uint32, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", opts.ClusterID))
	cluster, err := s.getCluster(ctx, opts.ClusterID)
	if err != nil {
		logger.Error("failed to get cluster", zap.Error(err))
		return nil, 0, err
	}
	kubernetes := cluster.GetServerlessCluster().GetCluster().GetKubernetes()
	filter := &entity.ListPrivateLinkConnectionFilter{
		ClusterID: opts.ClusterID,
		Offset:    &opts.Offset,
		Limit:     &opts.Limit,
	}
	// Shared MSK private link connections are scoped by Kubernetes. Keep the
	// previous cluster-scoped behavior when Kubernetes is unavailable so that
	// listing existing non-MSK connections remains backward compatible.
	if kubernetes != "" {
		filter.TenantID = strconv.FormatUint(cluster.GetOrganizationId(), 10)
		filter.Region = cluster.GetRegion()
		filter.Kubernetes = kubernetes
		filter.IncludeSharedMSK = true
	}
	if opts.State != "" {
		filter.States = []constant.PrivateLinkConnectionState{opts.State}
	}

	plcs, err := s.plcRepo.ListByFilter(ctx, filter)
	if err != nil {
		logger.Error("failed to list private link connections", zap.Error(err))
		return nil, 0, err
	}

	count, err := s.plcRepo.CountByFilter(ctx, filter)
	if err != nil {
		logger.Error("failed to count private link connections", zap.Error(err))
		return nil, 0, err
	}

	return plcs, uint32(count), nil
}

func (s *privateLinkConnectionService) CreatePrivateLinkConnection(
	ctx context.Context,
	clusterId string,
	plc *entity.PrivateLinkConnection,
) (created *entity.PrivateLinkConnection, err bizerr.BizError) {
	defer func() {
		if err != nil || created == nil {
			return
		}
		s.publishPrivateLinkEvent(
			ctx,
			created.ClusterID,
			event_constant.EventTypeCreatePrivateLinkConnection,
			fmt.Sprintf("Private endpoint %s created.", created.DisplayName),
			event_constant.PrivateLinkConnectionDetail{
				ID:          created.ID,
				DisplayName: created.DisplayName,
				Type:        string(created.Type),
			},
		)
	}()
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", clusterId))
	logger.Info("CreatePrivateLinkConnection - start", zap.Any("payload", plc))

	cluster, err := s.getAndCheckCluster(ctx, clusterId)
	if err != nil {
		logger.Error("failed to get and check cluster", zap.Error(err))
		return nil, err
	}
	plc.FillWithCluster(cluster)
	plc.ID = entity.GenerateIDWithPrefix(constant.PrivateLinkConnectionIDPrefix)

	err = plc.Validate(cluster)
	if err != nil {
		logger.Error("invalid private link connection", zap.Error(err))
		return nil, err
	}

	if plc.Type == constant.PrivateLinkConnectionAmazonMSKProvisioned {
		if plc.Kubernetes == "" {
			return nil, bizerr.New(
				bizerr.ErrCodeParams,
				"kubernetes is required for amazon msk private link connection",
			)
		}
		if getMSKServicePlan(cluster) != globalcommonv1.ServicePlan_Essential {
			return nil, bizerr.New(
				bizerr.ErrCodeParams,
				"amazon msk private link connection only supports Essential clusters",
			)
		}
		existing, err := s.findReusableMSKPrivateLinkConnection(ctx, plc)
		if err != nil {
			logger.Error("failed to find reusable msk private link connection", zap.Error(err))
			return nil, err
		}
		if existing != nil {
			logger.Info("reuse existing msk private link connection", zap.String("plc_id", existing.ID))
			return existing, nil
		}
	}

	err = s.checkPrivateLinkConnection(ctx, plc)
	if err != nil {
		logger.Error("private link connection check failed", zap.Error(err))
		return nil, err
	}

	// Set a cluster custom status so the cluster cannot be deleted while the
	// private link connection still exists.
	//
	// MSK connections are excluded: they are shared at the kubernetes level
	// (see findReusableMSKPrivateLinkConnection), so they do not belong to the
	// cluster that happened to create them. Pinning the status to that cluster
	// would leave it undeletable while every cluster reusing the connection
	// stays unlocked. The "still in use" guarantee is already provided by the
	// changefeed's own CUSTOM_STATUS_TYPE_CDC status, which blocks cluster
	// deletion until the last changefeed is gone.
	if plc.Type != constant.PrivateLinkConnectionAmazonMSKProvisioned {
		uint64ClusterId, err1 := strconv.ParseUint(clusterId, 10, 64)
		if err1 != nil {
			logger.Error("failed to parse cluster id", zap.Error(err1))
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "invalid cluster id")
		}
		// NOTE: assign (not :=) so the deferred rollback below observes the
		// named return value rather than a shadowed copy.
		var statusId uint64
		statusId, err = s.mgmtSvc.SetClusterPrivateLinkConnectionStatus(ctx, uint64ClusterId, plc.ID)
		if err != nil {
			logger.Error("failed to set cluster custom status private link connection", zap.Error(err))
			return nil, err
		}
		plc.ClusterStatusId = statusId
		defer func() {
			if err != nil {
				s.deleteClusterCustomStatus(ctx, uint64ClusterId, statusId)
			}
		}()
	}

	switch plc.Type {
	case constant.PrivateLinkConnectionAWSEndpointService:
		return s.createAWSVpcEndpoint(ctx, clusterId, plc)
	case constant.PrivateLinkConnectionAlicloudEndpointService:
		return s.createAliCloudVpcEndpoint(ctx, clusterId, plc)
	case constant.PrivateLinkConnectionAmazonMSKProvisioned:
		return s.createAmazonMSKVPCConnection(ctx, clusterId, plc)
	default:
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}
}

func (s *privateLinkConnectionService) checkPrivateLinkConnection(
	ctx context.Context,
	plc *entity.PrivateLinkConnection,
) bizerr.BizError {
	switch plc.Type {
	case constant.PrivateLinkConnectionAWSEndpointService:
		serviceName := plc.AWSEndpointService.EndpointServiceName
		count, err := s.plcRepo.CountByFilter(ctx, &entity.ListPrivateLinkConnectionFilter{
			AWSEndpointServiceName: serviceName,
			ClusterID:              plc.ClusterID,
		})
		if err != nil {
			return err
		}
		if count > 0 {
			return bizerr.New(bizerr.ErrCodeParams,
				"aws endpoint service %s already exists in this cluster",
				serviceName)
		}
	case constant.PrivateLinkConnectionAlicloudEndpointService:
		serviceName := plc.AlicloudEndpointService.EndpointServiceName
		count, err := s.plcRepo.CountByFilter(ctx, &entity.ListPrivateLinkConnectionFilter{
			AlicloudEndpointServiceName: serviceName,
			ClusterID:                   plc.ClusterID,
		})
		if err != nil {
			return err
		}
		if count > 0 {
			return bizerr.New(bizerr.ErrCodeParams,
				"alicloud endpoint service %s already exists in this cluster",
				serviceName)
		}
	case constant.PrivateLinkConnectionAmazonMSKProvisioned:
		if plc.AmazonMSKProvisioned == nil || plc.AmazonMSKProvisioned.MSKArn == "" {
			return bizerr.New(bizerr.ErrCodeParams, "amazon msk arn is required")
		}
		mskArn := plc.AmazonMSKProvisioned.MSKArn
		deletingCount, err := s.plcRepo.CountByFilter(ctx, &entity.ListPrivateLinkConnectionFilter{
			AmazonMSKProvisionedArn: mskArn,
			TenantID:                plc.TenantID,
			Region:                  plc.Region,
			Kubernetes:              plc.Kubernetes,
			Type:                    constant.PrivateLinkConnectionAmazonMSKProvisioned,
			States:                  []constant.PrivateLinkConnectionState{constant.PrivateLinkConnectionStateDeleting},
		})
		if err != nil {
			return err
		}
		if deletingCount > 0 {
			return bizerr.New(bizerr.ErrCodeParams,
				"amazon msk provisioned %s is being deleted in this kubernetes, please retry later",
				mskArn)
		}
		activeCount, err := s.plcRepo.CountByFilter(ctx, &entity.ListPrivateLinkConnectionFilter{
			AmazonMSKProvisionedArn: mskArn,
			TenantID:                plc.TenantID,
			Region:                  plc.Region,
			Kubernetes:              plc.Kubernetes,
			Type:                    constant.PrivateLinkConnectionAmazonMSKProvisioned,
			States: []constant.PrivateLinkConnectionState{
				constant.PrivateLinkConnectionStateCreating,
				constant.PrivateLinkConnectionStatePendingAcceptance,
				constant.PrivateLinkConnectionStateAvailable,
			},
		})
		if err != nil {
			return err
		}
		if activeCount > 0 {
			return bizerr.New(bizerr.ErrCodeParams,
				"amazon msk provisioned %s already exists in this kubernetes",
				mskArn)
		}
	}
	return nil
}

func (s *privateLinkConnectionService) findReusableMSKPrivateLinkConnection(
	ctx context.Context,
	plc *entity.PrivateLinkConnection,
) (*entity.PrivateLinkConnection, bizerr.BizError) {
	if plc.AmazonMSKProvisioned == nil || plc.AmazonMSKProvisioned.MSKArn == "" {
		return nil, nil
	}
	plcs, err := s.plcRepo.ListByFilter(ctx, &entity.ListPrivateLinkConnectionFilter{
		TenantID:                plc.TenantID,
		Region:                  plc.Region,
		Kubernetes:              plc.Kubernetes,
		Type:                    constant.PrivateLinkConnectionAmazonMSKProvisioned,
		AmazonMSKProvisionedArn: plc.AmazonMSKProvisioned.MSKArn,
		States: []constant.PrivateLinkConnectionState{
			constant.PrivateLinkConnectionStateCreating,
			constant.PrivateLinkConnectionStatePendingAcceptance,
			constant.PrivateLinkConnectionStateAvailable,
		},
		Limit: lo.ToPtr(int32(1)),
	})
	if err != nil {
		return nil, err
	}
	if len(plcs) == 0 {
		return nil, nil
	}
	return plcs[0], nil
}

func (s *privateLinkConnectionService) createAWSVpcEndpoint(
	ctx context.Context,
	clusterId string,
	plc *entity.PrivateLinkConnection,
) (*entity.PrivateLinkConnection, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", clusterId))
	logger.Info("start to create aws vpc endpoint")
	if err := s.transProvider.Transaction(ctx, func(ctx context.Context) error {
		resp, err := s.mgmtSvc.CreateAWSVpcEndpoint(ctx, &globalserverlessv1.CreateAWSVpcEndpointRequest{
			ClusterId:             clusterId,
			EndpointServiceName:   plc.AWSEndpointService.EndpointServiceName,
			EndpointServiceRegion: plc.AWSEndpointService.EndpointServiceRegion,
		})
		if err != nil {
			logger.Error("failed to create vpc endpoint via mgmtsvc", zap.Error(err))
			return parseVisibleErr(err)
		}

		plc.State = constant.PrivateLinkConnectionStatePendingAcceptance
		plc.AWSEndpointService.EndpointID = resp.Endpoint.EndpointId

		plc, err = s.plcRepo.Create(ctx, plc)
		if err != nil {
			logger.Error("failed to create plc in db", zap.Error(err))
			return err
		}
		return nil
	}); err != nil {
		var e bizerr.BizError
		if errors.As(err, &e) {
			return nil, e
		}
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	return plc, nil
}

func (s *privateLinkConnectionService) createAliCloudVpcEndpoint(
	ctx context.Context,
	clusterId string,
	plc *entity.PrivateLinkConnection,
) (*entity.PrivateLinkConnection, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", clusterId))
	logger.Info("start to create alicloud vpc endpoint")
	if err := s.transProvider.Transaction(ctx, func(ctx context.Context) error {
		resp, err := s.mgmtSvc.CreateAliCloudVpcEndpoint(ctx, &globalserverlessv1.CreateAliCloudVpcEndpointRequest{
			ClusterId:           clusterId,
			EndpointServiceName: plc.AlicloudEndpointServiceName,
		})
		if err != nil {
			logger.Error("failed to create vpc endpoint via mgmtsvc", zap.Error(err))
			return parseVisibleErr(err)
		}

		plc.State = constant.PrivateLinkConnectionStateCreating
		plc.AlicloudEndpointService.EndpointID = resp.Endpoint.EndpointId
		plc, err = s.plcRepo.Create(ctx, plc)
		if err != nil {
			logger.Error("failed to create plc in db", zap.Error(err))
			return err
		}
		return nil
	}); err != nil {
		var e bizerr.BizError
		if errors.As(err, &e) {
			return nil, e
		}
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	return plc, nil
}

func (s *privateLinkConnectionService) createAmazonMSKVPCConnection(
	ctx context.Context,
	clusterId string,
	plc *entity.PrivateLinkConnection,
) (*entity.PrivateLinkConnection, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", clusterId))
	logger.Info("start to create amazon msk vpc connection")
	if plc.AmazonMSKProvisioned == nil || plc.AmazonMSKProvisioned.MSKArn == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "amazon msk arn is required")
	}
	if err := s.transProvider.Transaction(ctx, func(ctx context.Context) error {
		cloudProvider, organizationID, servicePlan, routeErr := mskRouteFields(plc)
		if routeErr != nil {
			return routeErr
		}
		resp, err := s.mgmtSvc.CreateMSKVPCConnection(ctx, &globalserverlessv1.CreateMSKVPCConnectionRequest{
			ClusterId:       "",
			Kubernetes:      plc.Kubernetes,
			Region:          plc.Region,
			CloudProviderId: cloudProvider,
			OrganizationId:  organizationID,
			ServicePlan:     servicePlan,
			MskArn:          plc.AmazonMSKProvisioned.MSKArn,
		})
		if err != nil {
			logger.Error("failed to create msk vpc connection via mgmtsvc", zap.Error(err))
			return parseVisibleErr(err)
		}
		plc.State = constant.PrivateLinkConnectionStateCreating
		plc.AmazonMSKProvisioned.MSKVPCConnectionArn = resp.MskVpcConnection.MskVpcConnectionArn
		plc.AmazonMSKProvisioned.BootstrapBroker = resp.MskVpcConnection.BootstrapBroker
		plc.AmazonMSKProvisioned.MSKVPCConnectionState = resp.MskVpcConnection.State.String()

		plc, err = s.plcRepo.Create(ctx, plc)
		if err != nil {
			logger.Error("failed to create plc in db", zap.Error(err))
			return err
		}
		return nil
	}); err != nil {
		var e bizerr.BizError
		if errors.As(err, &e) {
			return nil, e
		}
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	return plc, nil
}

func (s *privateLinkConnectionService) DeletePrivateLinkConnection(
	ctx context.Context,
	clusterId, privateLinkConnectionId string,
) (plc *entity.PrivateLinkConnection, err bizerr.BizError) {
	defer func() {
		if err != nil || plc == nil {
			return
		}
		s.publishPrivateLinkEvent(
			ctx,
			plc.ClusterID,
			event_constant.EventTypeDeletePrivateLinkConnection,
			fmt.Sprintf("Private endpoint %s deleted.", plc.DisplayName),
			event_constant.PrivateLinkConnectionDetail{
				ID:          plc.ID,
				DisplayName: plc.DisplayName,
				Type:        string(plc.Type),
			},
		)
	}()
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", clusterId), zap.String("private_link_connection_id", privateLinkConnectionId))
	logger.Info("DeletePrivateLinkConnection - start")

	plc, err = s.getPrivateLinkConnectionForCluster(ctx, clusterId, privateLinkConnectionId)
	if err != nil {
		logger.Error("failed to get private link connection before delete", zap.Error(err))
		return nil, err
	}

	// precheck private link connection links
	linkClusterID := clusterId
	if plc.Type == constant.PrivateLinkConnectionAmazonMSKProvisioned {
		linkClusterID = ""
	}
	links, err := s.plcLinkRepo.ListByPrivateLinkConnectionID(ctx, linkClusterID, plc.ID)
	if err != nil {
		logger.Error("failed to list plc links", zap.Error(err))
		return nil, err
	}
	if len(links) > 0 {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"cannot delete private link connection, used by: %v",
			lo.Map(links, func(item *entity.PrivateLinkConnectionLink, _ int) string {
				return item.BizID
			}),
		)
	}

	if plc.Type != constant.PrivateLinkConnectionAmazonMSKProvisioned {
		attachDomains, err := s.plcAttachDomainRepo.ListByFilter(ctx, entity.ListPLCAttachDomainsFilter{
			ClusterID:               clusterId,
			PrivateLinkConnectionID: privateLinkConnectionId,
		})
		if err != nil {
			logger.Error("failed to list attach domains", zap.Error(err))
			return nil, err
		}
		if len(attachDomains) > 0 {
			return nil, bizerr.New(
				bizerr.ErrCodeParams,
				"please detach attach domains before deleting private link connection",
			)
		}
	}

	switch plc.Type {
	case constant.PrivateLinkConnectionAWSEndpointService:
		return s.deleteAWSVpcEndpoint(ctx, clusterId, plc)
	case constant.PrivateLinkConnectionAlicloudEndpointService:
		return s.deleteAliCloudVpcEndpoint(ctx, clusterId, plc)
	case constant.PrivateLinkConnectionAmazonMSKProvisioned:
		return s.deleteAmazonMSKVPCConnection(ctx, clusterId, plc)
	default:
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}
}

func (s *privateLinkConnectionService) deleteClusterCustomStatus(
	ctx context.Context,
	clusterId uint64,
	statusId uint64,
) {
	logger := log.WithContextL(ctx).With(zap.Uint64("cluster_id", clusterId))
	logger.Info("start to delete plc cluster custom status", zap.Uint64("status_id", statusId))
	if err := s.mgmtSvc.DeleteClusterCustomStatus(ctx, clusterId, statusId); err != nil {
		logger.Error("failed to delete plc cluster custom status, please handle it manually", zap.Error(err))
	}
}

func (s *privateLinkConnectionService) deleteAWSVpcEndpoint(
	ctx context.Context,
	clusterId string,
	plc *entity.PrivateLinkConnection,
) (*entity.PrivateLinkConnection, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", clusterId))
	logger.Info("start to delete aws vpc endpoint")
	if err := s.transProvider.Transaction(ctx, func(ctx context.Context) error {
		err := s.mgmtSvc.DeleteAWSVpcEndpoint(ctx, &globalserverlessv1.DeleteAWSVpcEndpointRequest{
			ClusterId:  clusterId,
			EndpointId: plc.AWSEndpointService.EndpointID,
		})
		if err != nil {
			logger.Error("failed to delete vpc endpoint via mgmtsvc", zap.Error(err))
			return parseVisibleErr(err)
		}
		plc.State = constant.PrivateLinkConnectionStateDeleting
		plc, err = s.plcRepo.Save(ctx, plc)
		if err != nil {
			logger.Error("failed to delete plc in db", zap.Error(err))
			return err
		}
		return nil
	}); err != nil {
		var e bizerr.BizError
		if errors.As(err, &e) {
			return nil, e
		}
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	return plc, nil
}

// AliCloud VPC endpoint will be deleted sync
func (s *privateLinkConnectionService) deleteAliCloudVpcEndpoint(
	ctx context.Context,
	clusterId string,
	plc *entity.PrivateLinkConnection,
) (*entity.PrivateLinkConnection, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", clusterId))
	logger.Info("start to delete alicloud vpc endpoint")

	plc.State = constant.PrivateLinkConnectionStateDeleting
	plc, err := s.plcRepo.Save(ctx, plc)
	if err != nil {
		logger.Error("failed to delete plc in db", zap.Error(err))
		return nil, err
	}
	return plc, err
}

func (s *privateLinkConnectionService) deleteAmazonMSKVPCConnection(
	ctx context.Context,
	clusterId string,
	plc *entity.PrivateLinkConnection,
) (*entity.PrivateLinkConnection, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", clusterId))
	logger.Info("start to delete amazon msk vpc connection")
	if plc.AmazonMSKProvisioned == nil || plc.AmazonMSKProvisioned.MSKVPCConnectionArn == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "msk vpc connection arn is required")
	}
	if err := s.transProvider.Transaction(ctx, func(ctx context.Context) error {
		cloudProvider, organizationID, servicePlan, routeErr := mskRouteFields(plc)
		if routeErr != nil {
			return routeErr
		}
		_, err := s.mgmtSvc.DeleteMSKVPCConnection(ctx, &globalserverlessv1.DeleteMSKVPCConnectionRequest{
			ClusterId:           "",
			Kubernetes:          plc.Kubernetes,
			Region:              plc.Region,
			CloudProviderId:     cloudProvider,
			OrganizationId:      organizationID,
			ServicePlan:         servicePlan,
			MskVpcConnectionArn: plc.AmazonMSKProvisioned.MSKVPCConnectionArn,
		})
		if err != nil {
			logger.Error("failed to delete msk vpc connection via mgmtsvc", zap.Error(err))
			return parseVisibleErr(err)
		}
		plc.State = constant.PrivateLinkConnectionStateDeleting
		plc, err = s.plcRepo.Save(ctx, plc)
		if err != nil {
			logger.Error("failed to delete plc in db", zap.Error(err))
			return err
		}
		return nil
	}); err != nil {
		var e bizerr.BizError
		if errors.As(err, &e) {
			return nil, e
		}
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	return plc, nil
}

func (s *privateLinkConnectionService) GetAvailabilityZones(
	ctx context.Context,
	clusterId string,
) (string, []string, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", clusterId))
	logger.Info("GetAvailabilityZones - start")

	cluster, err := s.getAndCheckCluster(ctx, clusterId)
	if err != nil {
		logger.Error("failed to get and check cluster", zap.Error(err))
		return "", nil, err
	}

	resp, err := s.mgmtSvc.GetZoneIds(ctx, &globalserverlessv1.GetZoneIdsRequest{
		ClusterId:       clusterId,
		CloudProviderId: getMSKCloudProvider(cluster.CloudProvider),
		OrganizationId:  cluster.OrganizationId,
		ServicePlan:     globalcommonv1.ServicePlan_Essential,
		Region:          cluster.Region,
		Kubernetes:      cluster.GetServerlessCluster().GetCluster().GetKubernetes(),
	})
	if err != nil {
		return "", nil, parseVisibleErr(err)
	}

	accountID := resp.GetAccountId()
	if accountID == "" {
		accountID = getAccountID(cluster.CloudProvider)
	}
	return accountID, resp.ZoneIds, nil
}

// RegisterPrivateLinkConnection registers a private link connection to a biz id.
// dryrun: if dryrun is true, biz id can be empty in order to test the private link connection.
// return:
// - aws_endpoint_service: return all domains, the regional domain without az will be the first one
// - alicloud_endpoint_service: return one domain
func (s *privateLinkConnectionService) RegisterPrivateLinkConnection(
	ctx context.Context,
	clusterId, privateLinkConnectionId, bizId string, dryrun bool,
) ([]string, bizerr.BizError) {
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", clusterId)).
		With(zap.String("private_link_connection_id", privateLinkConnectionId)).
		With(zap.String("biz_id", bizId)).
		With(zap.Bool("dry_run", dryrun))
	logger.Info("RegisterPrivateLinkConnection - start")

	plc, err := s.getPrivateLinkConnectionForCluster(ctx, clusterId, privateLinkConnectionId)
	if err != nil {
		logger.Error("failed to get private link connection", zap.Error(err))
		if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return nil, bizerr.New(
				bizerr.ErrCodeRecordNotFound,
				"private link connection %s is not found",
				privateLinkConnectionId,
			)
		}
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}

	if plc.State != constant.PrivateLinkConnectionStateAvailable {
		logger.Error("private link connection is not in active state")
		return nil, bizerr.New(bizerr.ErrCodeParams,
			"private link connection %s is not in active state", privateLinkConnectionId)
	}

	if !dryrun {
		if bizId == "" {
			logger.Error("biz id is required")
		}
		link := &entity.PrivateLinkConnectionLink{
			ClusterID:               clusterId,
			PrivateLinkConnectionID: privateLinkConnectionId,
			BizID:                   bizId,
		}
		_, err = s.plcLinkRepo.Create(ctx, link)
		if err != nil {
			logger.Error("failed to create private link connection link", zap.Error(err))
			if err.ErrCode() == bizerr.ErrCodeRecordExist {
				return nil, bizerr.New(bizerr.ErrCodeRecordExist, "biz id %s already binded", bizId)
			}
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
		}
	}

	switch plc.Type {
	case constant.PrivateLinkConnectionAlicloudEndpointService:
		return []string{plc.AlicloudEndpointService.Domain}, nil
	case constant.PrivateLinkConnectionAWSEndpointService:
		// sort to ensure the regional domain is the first one
		return sortByLength(plc.AWSEndpointService.Domains), nil
	case constant.PrivateLinkConnectionAmazonMSKProvisioned:
		return []string{plc.AmazonMSKProvisioned.BootstrapBroker}, nil
	default:
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "unknown cloud provider")
	}
}

func sortByLength(s []string) []string {
	if len(s) <= 1 {
		return s
	}
	out := append([]string(nil), s...)
	sort.SliceStable(out, func(i, j int) bool {
		if len(out[i]) == len(out[j]) {
			return out[i] < out[j]
		}
		return len(out[i]) < len(out[j])
	})
	return out
}

func (s *privateLinkConnectionService) DeregisterPrivateLinkConnection(
	ctx context.Context,
	clusterId, privateLinkConnectionId, bizId string,
) bizerr.BizError {
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", clusterId)).
		With(zap.String("private_link_connection_id", privateLinkConnectionId)).
		With(zap.String("biz_id", bizId))
	logger.Info("DeregisterPrivateLinkConnection - start")

	err := s.plcLinkRepo.Delete(ctx, clusterId, privateLinkConnectionId, bizId)
	if err != nil {
		logger.Error("failed to delete private link connection link", zap.Error(err))
		if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return nil
		}
		return bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	return nil
}

func (s *privateLinkConnectionService) AttachDomains(
	ctx context.Context,
	plcd *entity.PLCAttachDomain,
	dryRun bool,
) (*entity.PLCAttachDomain, bizerr.BizError) {
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", plcd.ClusterID)).
		With(zap.String("private_link_connection_id", plcd.PrivateLinkConnectionID))

	logger.Info("AttachDomains - start",
		zap.String("unique_id", plcd.UniqueID),
		zap.String("type", string(plcd.Type)),
		zap.Bool("dry_run", dryRun))

	plc, err := s.plcRepo.GetById(ctx, plcd.ClusterID, plcd.PrivateLinkConnectionID)
	if err != nil {
		logger.Error("failed to get private link connection", zap.Error(err))
		return nil, err
	}

	err = plcd.Validate(plc)
	if err != nil {
		logger.Error("validate attach domain fail", zap.Error(err))
		return nil, err
	}

	plcd.FillInField(plc)

	err = s.precheckAttachDomain(ctx, plcd)
	if err != nil {
		logger.Error("attach domain precheck failed", zap.Error(err))
		return nil, err
	}

	plcRecords := generateDomainRecords(plc, plcd)
	plcd.Records = plcRecords

	if dryRun {
		for _, record := range plcRecords {
			record.State = ""
		}
		return plcd, nil
	}

	if err := s.transProvider.Transaction(ctx, func(ctx context.Context) error {
		plcd, err = s.plcAttachDomainRepo.Create(ctx, plcd)
		if err != nil {
			if err.ErrCode() == bizerr.ErrCodeRecordExist {
				return bizerr.New(bizerr.ErrCodeParams,
					"conflict unique name %s, please provide other unique name",
					plcd.UniqueID)
			}
			logger.Error("failed to create attach domain", zap.Error(err))
			return err
		}

		err = s.plcDomainRecordRepo.BatchInsert(ctx, plcRecords)
		if err != nil {
			logger.Error("failed to create domain records", zap.Error(err))
			return err
		}
		return nil
	}); err != nil {
		var e bizerr.BizError
		if errors.As(err, &e) {
			return nil, e
		}
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}

	s.publishPrivateLinkEvent(
		ctx,
		plcd.ClusterID,
		event_constant.EventTypeAttachPrivateLinkConnectionDomains,
		fmt.Sprintf("Private endpoint domains attached to %s.", plcd.UniqueID),
		event_constant.PrivateLinkConnectionDomainDetail{
			PrivateLinkConnectionID: plcd.PrivateLinkConnectionID,
			AttachID:                plcd.ID,
			UniqueID:                plcd.UniqueID,
			Type:                    string(plcd.Type),
		},
	)
	return plcd, nil
}

func buildRecordName(azID string, plcd *entity.PLCAttachDomain) string {
	hostedZoneName := plcd.HostedZone
	if plcd.Type == constant.PLCAttachDomainTypeTiDBCloudManaged {
		if azID != "" {
			return fmt.Sprintf("*.%s.%s.%s", azID, plcd.UniqueID, hostedZoneName)
		}
		return fmt.Sprintf("*.%s.%s", plcd.UniqueID, hostedZoneName)
	}
	if azID != "" {
		return fmt.Sprintf("*.%s.%s", azID, hostedZoneName)
	}
	return fmt.Sprintf("*.%s", hostedZoneName)
}

func generateDomainRecords(
	plc *entity.PrivateLinkConnection,
	plcd *entity.PLCAttachDomain,
) []*entity.PLCDomainRecord {
	records := make([]*entity.PLCDomainRecord, 0)
	switch plc.Type {
	case constant.PrivateLinkConnectionAWSEndpointService:
		for azId, azName := range plc.AWSEndpointService.EndpointAzIdToName {
			// try to find the correct azs
			value := plc.AWSEndpointService.Domains[0]
			for _, domain := range plc.AWSEndpointService.Domains {
				if strings.Contains(domain, azName) {
					value = domain
					break
				}
			}
			record := &entity.PLCDomainRecord{
				ClusterID:               plcd.ClusterID,
				CloudProvider:           plcd.CloudProvider,
				PrivateLinkConnectionID: plcd.PrivateLinkConnectionID,
				AttachDomainID:          plcd.ID,
				Record:                  buildRecordName(azId, plcd),
				Value:                   value,
				State:                   constant.PLCDomainRecordStateCreating,
			}
			records = append(records, record)
		}
		// add the boostrap domain
		record := &entity.PLCDomainRecord{
			ClusterID:               plcd.ClusterID,
			CloudProvider:           plcd.CloudProvider,
			PrivateLinkConnectionID: plcd.PrivateLinkConnectionID,
			AttachDomainID:          plcd.ID,
			Record:                  buildRecordName("", plcd),
			Value:                   plc.AWSEndpointService.Domains[0],
			State:                   constant.PLCDomainRecordStateCreating,
		}
		records = append(records, record)
	case constant.PrivateLinkConnectionAlicloudEndpointService:
		record := &entity.PLCDomainRecord{
			ClusterID:               plcd.ClusterID,
			CloudProvider:           plcd.CloudProvider,
			PrivateLinkConnectionID: plcd.PrivateLinkConnectionID,
			AttachDomainID:          plcd.ID,
			Record:                  buildRecordName("", plcd),
			Value:                   plc.AlicloudEndpointService.Domain,
			State:                   constant.PLCDomainRecordStateCreating,
		}
		records = append(records, record)
	}
	return records
}

func (s *privateLinkConnectionService) precheckAttachDomain(
	ctx context.Context,
	plcd *entity.PLCAttachDomain,
) bizerr.BizError {
	count, err := s.plcAttachDomainRepo.CountByPLC(ctx, plcd.ClusterID, plcd.PrivateLinkConnectionID)
	if err != nil {
		return err
	}
	if count >= MaxAttachDomain {
		return bizerr.New(bizerr.ErrCodeParams,
			"cannot attach more domains, max attach domain is %d", MaxAttachDomain)
	}

	isExist, err := s.plcAttachDomainRepo.CheckHostedZone(ctx, plcd.HostedZone, plcd.UniqueID)
	if err != nil {
		return err
	}
	if isExist {
		return bizerr.New(bizerr.ErrCodeParams,
			"conflict unique name %s, please provide other unique name",
			plcd.UniqueID)
	}
	return nil
}

func (s *privateLinkConnectionService) DetachDomains(
	ctx context.Context,
	clusterId, privateLinkConnectionId, attachDomainID string,
) (*entity.PLCAttachDomain, bizerr.BizError) {
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", clusterId)).
		With(zap.String("private_link_connection_id", privateLinkConnectionId)).
		With(zap.String("attach_domain_id", attachDomainID))

	logger.Info("DetachDomains - start")

	plcd, err := s.plcAttachDomainRepo.Get(ctx, clusterId, privateLinkConnectionId, attachDomainID)
	if err != nil {
		logger.Error("failed to get private link connection attach domain", zap.Error(err))
		return nil, err
	}

	records, err := s.plcDomainRecordRepo.ListByFilter(ctx, &entity.ListPLCDomainRecordsFilter{
		ClusterID:               clusterId,
		PrivateLinkConnectionID: privateLinkConnectionId,
		AttachDomainID:          attachDomainID,
	})
	if err != nil {
		logger.Error("failed to list private link connection domain records", zap.Error(err))
		return nil, err
	}

	plcd.Records = records
	if len(records) == 0 {
		if err := s.plcAttachDomainRepo.DeleteByID(ctx, clusterId, privateLinkConnectionId, attachDomainID); err != nil {
			return nil, err
		}
		s.publishDetachDomainsEvent(ctx, plcd)
		return plcd, nil
	}

	for _, record := range records {
		record.State = constant.PLCDomainRecordStateDeleting
	}
	if err := s.plcDomainRecordRepo.UpdateStateByAttachDomainID(
		ctx,
		clusterId,
		privateLinkConnectionId,
		attachDomainID,
		constant.PLCDomainRecordStateDeleting,
		"",
	); err != nil {
		logger.Error("failed to mark private link connection domain records deleting", zap.Error(err))
		return nil, err
	}
	logger.Info("DetachDomains - end")
	s.publishDetachDomainsEvent(ctx, plcd)
	return plcd, nil
}

func (s *privateLinkConnectionService) ListAttachDomains(
	ctx context.Context,
	filter entity.ListPLCAttachDomainsFilter,
) (map[string][]*entity.PLCAttachDomain, bizerr.BizError) {
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", filter.ClusterID)).
		With(zap.String("private_link_connection_id", filter.PrivateLinkConnectionID))

	plcds, err := s.plcAttachDomainRepo.ListByFilter(ctx, entity.ListPLCAttachDomainsFilter{
		ClusterID:                filter.ClusterID,
		PrivateLinkConnectionID:  filter.PrivateLinkConnectionID,
		PrivateLinkConnectionIDs: filter.PrivateLinkConnectionIDs,
	})
	if err != nil {
		logger.Error("failed to list private link connection attach domain", zap.Error(err))
		return nil, err
	}

	records, err := s.plcDomainRecordRepo.ListByFilter(ctx, &entity.ListPLCDomainRecordsFilter{
		ClusterID:                filter.ClusterID,
		PrivateLinkConnectionID:  filter.PrivateLinkConnectionID,
		PrivateLinkConnectionIDs: filter.PrivateLinkConnectionIDs,
	})
	if err != nil {
		logger.Error("failed to list private link connection attach domain", zap.Error(err))
		return nil, err
	}

	recordsByAttachID := make(map[string][]*entity.PLCDomainRecord)
	for _, record := range records {
		recordsByAttachID[record.AttachDomainID] = append(recordsByAttachID[record.AttachDomainID], record)
	}

	plcdsByPrivateLinkConnectionID := make(map[string][]*entity.PLCAttachDomain)
	for _, plcd := range plcds {
		plcd.Records = recordsByAttachID[plcd.ID]
		plcdsByPrivateLinkConnectionID[plcd.PrivateLinkConnectionID] = append(
			plcdsByPrivateLinkConnectionID[plcd.PrivateLinkConnectionID], plcd)
	}

	return plcdsByPrivateLinkConnectionID, nil
}

func (s *privateLinkConnectionService) getPrivateLinkConnectionForCluster(
	ctx context.Context,
	clusterId, privateLinkConnectionId string,
) (*entity.PrivateLinkConnection, bizerr.BizError) {
	plc, err := s.plcRepo.GetById(ctx, clusterId, privateLinkConnectionId)
	if err == nil {
		return plc, nil
	}
	if err.ErrCode() != bizerr.ErrCodeRecordNotFound {
		return nil, err
	}

	plc, getErr := s.plcRepo.GetByID(ctx, privateLinkConnectionId)
	if getErr != nil {
		if getErr.ErrCode() != bizerr.ErrCodeRecordNotFound {
			return nil, getErr
		}
		return nil, err
	}
	if plc.Type != constant.PrivateLinkConnectionAmazonMSKProvisioned {
		return nil, err
	}
	if checkErr := s.checkMSKPrivateLinkConnectionForCluster(ctx, clusterId, plc); checkErr != nil {
		return nil, checkErr
	}
	return plc, nil
}

func (s *privateLinkConnectionService) checkMSKPrivateLinkConnectionForCluster(
	ctx context.Context,
	clusterId string,
	plc *entity.PrivateLinkConnection,
) bizerr.BizError {
	cluster, err := s.getCluster(ctx, clusterId)
	if err != nil {
		return err
	}
	if plc.TenantID != strconv.FormatUint(cluster.GetOrganizationId(), 10) ||
		plc.CloudProvider != cluster.GetCloudProvider() ||
		plc.Region != cluster.GetRegion() ||
		plc.Kubernetes != cluster.GetServerlessCluster().GetCluster().GetKubernetes() {
		return bizerr.New(
			bizerr.ErrCodeRecordNotFound,
			"private link connection %s is not found",
			plc.ID,
		)
	}
	return nil
}

func (s *privateLinkConnectionService) publishDetachDomainsEvent(ctx context.Context, plcd *entity.PLCAttachDomain) {
	if plcd == nil {
		return
	}
	s.publishPrivateLinkEvent(
		ctx,
		plcd.ClusterID,
		event_constant.EventTypeDetachPrivateLinkConnectionDomains,
		fmt.Sprintf("Private endpoint domains detached from %s.", plcd.UniqueID),
		event_constant.PrivateLinkConnectionDomainDetail{
			PrivateLinkConnectionID: plcd.PrivateLinkConnectionID,
			AttachID:                plcd.ID,
			UniqueID:                plcd.UniqueID,
			Type:                    string(plcd.Type),
		},
	)
}

func (s *privateLinkConnectionService) publishPrivateLinkEvent(
	ctx context.Context,
	clusterID, eventType string,
	summary string,
	detail any,
) {
	publishDisplayEventAsync(
		ctx,
		s.eventbusService,
		func(ctx context.Context) *tidbmgmtservicev1.Cluster {
			return loadClusterForEvent(ctx, s.mgmtSvc, clusterID)
		},
		eventType,
		summary,
		detail,
	)
}

func (s *privateLinkConnectionService) getCluster(
	ctx context.Context,
	clusterId string,
) (*tidbmgmtservicev1.Cluster, bizerr.BizError) {
	cid, err := strconv.ParseUint(clusterId, 10, 64)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid clusterID:%s", clusterId)
	}
	return s.mgmtSvc.GetCluster(ctx, cid, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
}

func (s *privateLinkConnectionService) getAndCheckCluster(
	ctx context.Context,
	clusterId string,
) (*tidbmgmtservicev1.Cluster, bizerr.BizError) {
	cluster, err := s.getCluster(ctx, clusterId)
	if err != nil {
		return nil, err
	}
	if cluster.GetServerlessCluster().
		GetUserStatus() !=
		globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster %s is not in normal status", clusterId)
	}
	return cluster, nil
}

func getMSKCloudProvider(provider string) globalcommonv1.CloudProvider_Name {
	switch strings.ToLower(strings.TrimSpace(provider)) {
	case "aws", "eks":
		return globalcommonv1.CloudProvider_AWS
	case "gcp", "gke":
		return globalcommonv1.CloudProvider_GCP
	case "azure", "aks":
		return globalcommonv1.CloudProvider_AZURE
	case "alicloud", "aliyun", "ack":
		return globalcommonv1.CloudProvider_ALICLOUD
	default:
		return globalcommonv1.CloudProvider_NAME_UNSPECIFIED
	}
}

func getMSKServicePlan(cluster *tidbmgmtservicev1.Cluster) globalcommonv1.ServicePlan {
	autoScaling := cluster.GetServerlessCluster().GetCluster().GetAutoScaling()
	if autoScaling.GetMinRcu() > 0 || autoScaling.GetMaxRcu() > 0 {
		return globalcommonv1.ServicePlan_Essential
	}
	return globalcommonv1.ServicePlan_Starter
}

func parseVisibleErr(err bizerr.BizError) bizerr.BizError {
	if err == nil {
		return nil
	}
	if lo.SomeBy(PrivateLinkConnectionVisibleErrCodes, func(item bizerr.ErrCode) bool {
		return err.ErrCode() == item
	}) {
		return err
	}
	return bizerr.New(bizerr.ErrCodeUnknown, "")
}

func getAccountID(provider string) string {
	stage := config.Global().ServerConfig.Stage
	switch provider {
	case config.ProviderAliCloud:
		if stage == "prod" {
			return "5108739859105002"
		}
		return "5654539846535456"
	case config.ProviderAWS:
		if stage == "prod" {
			return "380838443567"
		}
		return "385595570414"
	}
	return ""
}