package service

import (
	"context"
	"fmt"
	"strconv"
	"strings"

	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	"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/utils"
	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	"go.uber.org/zap"
)

type PrivateLinkConnectionWorker interface {
	SyncPrivateLinkConnections(ctx context.Context) error
	DeletePrivateLinkConnections(ctx context.Context) error
	SyncPLCDomainRecords(ctx context.Context) error
}

type privateLinkConnectionWorker struct {
	mgmtSvc             facade.MgmtSvcFacade
	plcRepo             repo.PrivateLinkConnectionRepo
	plcAttachDomainRepo repo.PLCAttachDomainRepo
	plcDomainRecordRepo repo.PLCDomainRecordRepo
	transProvider       db.TransProvider
}

func mskRouteFields(
	plc *entity.PrivateLinkConnection,
) (globalcommonv1.CloudProvider_Name, uint64, globalcommonv1.ServicePlan, bizerr.BizError) {
	organizationID, err := strconv.ParseUint(plc.TenantID, 10, 64)
	if err != nil || organizationID == 0 {
		return globalcommonv1.CloudProvider_NAME_UNSPECIFIED, 0,
			globalcommonv1.ServicePlan_SERVICE_PLAN_UNSPECIFIED,
			bizerr.New(bizerr.ErrCodeParams, "invalid msk private link connection tenant id %q", plc.TenantID)
	}
	cloudProvider := getMSKCloudProvider(plc.CloudProvider)
	if cloudProvider == globalcommonv1.CloudProvider_NAME_UNSPECIFIED {
		return cloudProvider, 0, globalcommonv1.ServicePlan_SERVICE_PLAN_UNSPECIFIED,
			bizerr.New(bizerr.ErrCodeParams, "invalid msk private link connection cloud provider %q", plc.CloudProvider)
	}
	if plc.Region == "" || plc.Kubernetes == "" {
		return cloudProvider, 0, globalcommonv1.ServicePlan_SERVICE_PLAN_UNSPECIFIED,
			bizerr.New(bizerr.ErrCodeParams, "region and kubernetes are required for msk private link connection")
	}
	return cloudProvider, organizationID, globalcommonv1.ServicePlan_Essential, nil
}

func NewPrivateLinkConnectionWorker(
	mgmtSvc facade.MgmtSvcFacade,
	plcRepo repo.PrivateLinkConnectionRepo,
	plcAttachDomainRepo repo.PLCAttachDomainRepo,
	plcDomainRecordRepo repo.PLCDomainRecordRepo,
	transProvider db.TransProvider,
) PrivateLinkConnectionWorker {
	return &privateLinkConnectionWorker{
		mgmtSvc:             mgmtSvc,
		plcRepo:             plcRepo,
		plcAttachDomainRepo: plcAttachDomainRepo,
		plcDomainRecordRepo: plcDomainRecordRepo,
		transProvider:       transProvider,
	}
}

func (w *privateLinkConnectionWorker) SyncPrivateLinkConnections(ctx context.Context) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger.Info("sync private link connection")

	filter := &entity.ListPrivateLinkConnectionFilter{
		StatesNotIn: []constant.PrivateLinkConnectionState{
			constant.PrivateLinkConnectionStateDeleting,
		},
	}
	plcs, err := w.plcRepo.ListByFilter(ctx, filter)
	if err != nil {
		logger.Error("failed to list private link connections", zap.Error(err))
		return err
	}

	failed := 0
	for _, plc := range plcs {
		switch plc.Type {
		case constant.PrivateLinkConnectionAWSEndpointService:
			if err := w.syncAWSEndpointServicePLC(ctx, plc); err != nil {
				failed++
			}
		case constant.PrivateLinkConnectionAlicloudEndpointService:
			if err := w.syncAliCloudEndpointServicePLC(ctx, plc); err != nil {
				failed++
			}
		case constant.PrivateLinkConnectionAmazonMSKProvisioned:
			if err := w.syncAmazonMSKProvisionedPLC(ctx, plc); err != nil {
				failed++
			}
		}
	}

	logger.Info("sync private link connection done", zap.Int("total", len(plcs)), zap.Int("failed", failed))
	if failed > 0 {
		return fmt.Errorf("sync private link connection failed for %d/%d", failed, len(plcs))
	}
	return nil
}

func (w *privateLinkConnectionWorker) syncAWSEndpointServicePLC(
	ctx context.Context,
	plc *entity.PrivateLinkConnection,
) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger = logger.With(zap.String("plc_id", plc.ID))
	logger.Info("sync one private link connection")

	req := &globalserverlessv1.DescribeAWSVpcEndpointRequest{
		ClusterId:  plc.ClusterID,
		EndpointId: plc.AWSEndpointService.EndpointID,
	}
	resp, err := w.mgmtSvc.DescribeAWSVpcEndpoint(ctx, req)
	if err != nil {
		logger.Error("failed to describe aws vpc endpoint", zap.Error(err))
		return err
	}
	if resp == nil || resp.Endpoint == nil {
		logger.Error("aws vpc endpoint not found")
		return fmt.Errorf("vpc endpoints %s not found", plc.AWSEndpointService.EndpointID)
	}

	plc.SyncAWSEndpointService(resp.Endpoint)

	_, err = w.plcRepo.Save(ctx, plc)
	if err != nil {
		logger.Error("failed to save private link connection", zap.Error(err))
		return err
	}
	return nil
}

func (w *privateLinkConnectionWorker) syncAliCloudEndpointServicePLC(
	ctx context.Context,
	plc *entity.PrivateLinkConnection,
) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger = logger.With(zap.String("plc_id", plc.ID))
	logger.Info("sync one private link connection")

	req := &globalserverlessv1.DescribeAliCloudVpcEndpointRequest{
		ClusterId:  plc.ClusterID,
		EndpointId: plc.AlicloudEndpointService.EndpointID,
	}
	resp, err := w.mgmtSvc.DescribeAliCloudVpcEndpoint(ctx, req)
	if err != nil {
		logger.Error("failed to describe vpc endpoint", zap.Error(err))
		return err
	}
	if resp == nil || resp.Endpoint == nil {
		logger.Error("vpc endpoint not found")
		return fmt.Errorf("vpc endpoint %s not found", plc.AlicloudEndpointService.EndpointID)
	}

	plc.SyncAliCloudEndpointService(resp.Endpoint)
	if plc.State == constant.PrivateLinkConnectionStateAvailable && len(plc.AlicloudEndpointService.EndpointAzs) == 0 {
		zones, err := w.mgmtSvc.ListAliCloudVpcEndpointZones(
			ctx,
			&globalserverlessv1.ListAliCloudVpcEndpointZonesRequest{
				ClusterId:  plc.ClusterID,
				EndpointId: plc.AlicloudEndpointService.EndpointID,
			},
		)
		if err != nil {
			logger.Error("list vpc endpoint zones failed", zap.Error(err))
			return err
		}
		azs := make([]string, 0, len(zones.GetZones()))
		for _, zone := range zones.GetZones() {
			azs = append(azs, zone.ZoneId)
		}
		plc.AlicloudEndpointService.EndpointAzs = azs
	}

	_, err = w.plcRepo.Save(ctx, plc)
	if err != nil {
		logger.Error("failed to save private link connection", zap.Error(err))
		return err
	}
	return nil
}

// releaseMSKClusterStatus drops the SERVERLESS_PRIVATE_LINK_CONNECTION_EXIST
// cluster custom status left behind by MSK connections created before these
// connections became shared at the kubernetes level.
//
// Such a status pins the connection to whichever cluster happened to create it,
// so that cluster can never be deleted while any other cluster still reuses the
// connection. New MSK connections no longer set it (see
// CreatePrivateLinkConnection); this heals the existing rows in place so no
// manual cleanup is needed.
func (w *privateLinkConnectionWorker) releaseMSKClusterStatus(
	ctx context.Context,
	plc *entity.PrivateLinkConnection,
) error {
	if plc.ClusterStatusId == 0 {
		return nil
	}
	logger := utils.GetWorkerLoggerFromContext(ctx).With(zap.String("plc_id", plc.ID))

	clusterID, err := strconv.ParseUint(plc.ClusterID, 10, 64)
	if err != nil {
		logger.Error("failed to parse cluster id", zap.Error(err))
		return err
	}

	statusID := plc.ClusterStatusId
	logger.Info("releasing stale msk cluster custom status",
		zap.String("cluster_id", plc.ClusterID),
		zap.Uint64("cluster_status_id", statusID))

	// Clear the field first: a failure here leaves the status in place and the
	// next round retries, whereas clearing it after a successful delete could
	// crash in between and orphan the status with no pointer left to it.
	return w.transProvider.Transaction(ctx, func(ctx context.Context) error {
		plc.ClusterStatusId = 0
		if _, err := w.plcRepo.Save(ctx, plc); err != nil {
			logger.Error("failed to save private link connection", zap.Error(err))
			return err
		}
		if err := w.mgmtSvc.DeleteClusterCustomStatus(ctx, clusterID, statusID); err != nil {
			logger.Error("failed to delete msk cluster custom status", zap.Error(err))
			return err
		}
		return nil
	})
}

func (w *privateLinkConnectionWorker) syncAmazonMSKProvisionedPLC(
	ctx context.Context,
	plc *entity.PrivateLinkConnection,
) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger = logger.With(zap.String("plc_id", plc.ID))
	logger.Info("sync one private link connection")

	if err := w.releaseMSKClusterStatus(ctx, plc); err != nil {
		return err
	}

	cloudProvider, organizationID, servicePlan, routeErr := mskRouteFields(plc)
	if routeErr != nil {
		return routeErr
	}

	resp, err := w.mgmtSvc.GetMSKVPCConnection(ctx, &globalserverlessv1.GetMSKVPCConnectionRequest{
		ClusterId:           "",
		Kubernetes:          plc.Kubernetes,
		Region:              plc.Region,
		CloudProviderId:     cloudProvider,
		OrganizationId:      organizationID,
		ServicePlan:         servicePlan,
		MskVpcConnectionArn: plc.AmazonMSKProvisioned.MSKVPCConnectionArn,
	})
	if err != nil {
		logger.Error("failed to get msk vpc connection", zap.Error(err))
		return err
	}
	if resp == nil || resp.MskVpcConnection == nil {
		logger.Error("msk vpc connection not found")
		return fmt.Errorf("msk vpc connection %s not found", plc.AmazonMSKProvisioned.MSKVPCConnectionArn)
	}

	plc.SyncAmazonMSKProvisioned(resp.MskVpcConnection)
	_, err = w.plcRepo.Save(ctx, plc)
	if err != nil {
		logger.Error("failed to save private link connection", zap.Error(err))
		return err
	}
	return nil
}

func (w *privateLinkConnectionWorker) DeletePrivateLinkConnections(ctx context.Context) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger.Info("delete private link connection")
	filter := &entity.ListPrivateLinkConnectionFilter{
		States: []constant.PrivateLinkConnectionState{
			constant.PrivateLinkConnectionStateDeleting,
		},
	}
	plcs, err := w.plcRepo.ListByFilter(ctx, filter)
	if err != nil {
		logger.Error("failed to list private link connections", zap.Error(err))
		return err
	}

	failed := 0
	for _, plc := range plcs {
		// try to remove cluster custom status first
		if plc.ClusterStatusId != 0 {
			if err := w.transProvider.Transaction(ctx, func(ctx context.Context) error {
				uint64ClusterId, err := strconv.ParseUint(plc.ClusterID, 10, 64)
				if err != nil {
					logger.Error("failed to parse cluster id", zap.Error(err))
					return err
				}

				clusterStatusId := plc.ClusterStatusId
				plc.ClusterStatusId = 0
				_, err = w.plcRepo.Save(ctx, plc)
				if err != nil {
					logger.Error("failed to save private link connection", zap.Error(err))
					return err
				}

				if err := w.mgmtSvc.DeleteClusterCustomStatus(ctx, uint64ClusterId, clusterStatusId); err != nil {
					logger.Error("failed to delete plc cluster custom status before deleting plc", zap.Error(err))
					return err
				}
				return nil
			}); err != nil {
				return err
			}
		}

		switch plc.Type {
		case constant.PrivateLinkConnectionAWSEndpointService:
			if err := w.deleteAWSEndpointServicePLC(ctx, plc); err != nil {
				failed++
			}
		case constant.PrivateLinkConnectionAlicloudEndpointService:
			if err := w.deleteAliCloudEndpointServicePLC(ctx, plc); err != nil {
				failed++
			}
		case constant.PrivateLinkConnectionAmazonMSKProvisioned:
			if err := w.deleteAmazonMSKProvisionedPLC(ctx, plc); err != nil {
				failed++
			}
		}
	}

	logger.Info("delete private link connection done", zap.Int("total", len(plcs)), zap.Int("failed", failed))
	if failed > 0 {
		return fmt.Errorf("delete private link connection failed for %d/%d", failed, len(plcs))
	}
	return nil
}

func (w *privateLinkConnectionWorker) SyncPLCDomainRecords(ctx context.Context) error {
	// sync creating records
	createErr := w.createPLCDomainRecords(ctx)
	// sync deleting records
	deleteErr := w.deletePLCDomainRecords(ctx)
	if createErr != nil && deleteErr != nil {
		return fmt.Errorf("sync create records failed: %v; sync delete records failed: %v", createErr, deleteErr)
	}
	if createErr != nil {
		return createErr
	}
	if deleteErr != nil {
		return deleteErr
	}
	return nil
}

func (w *privateLinkConnectionWorker) createPLCDomainRecords(ctx context.Context) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger.Info("sync creating plc domain records")

	records, err := w.plcDomainRecordRepo.ListByFilter(ctx, &entity.ListPLCDomainRecordsFilter{
		States: []constant.PLCDomainRecordState{constant.PLCDomainRecordStateCreating},
	})
	if err != nil {
		logger.Error("failed to list plc domain records", zap.Error(err))
		return err
	}

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

	failed := 0
	for attachDomainID, attachRecords := range recordsByAttachID {
		if err := w.createOnePLCDomainRecords(ctx, attachDomainID, attachRecords); err != nil {
			failed++
		}
	}

	logger.Info(
		"sync creating plc domain records done",
		zap.Int("total", len(recordsByAttachID)),
		zap.Int("failed", failed),
	)
	if failed > 0 {
		return fmt.Errorf("sync creating plc domain records failed for %d/%d", failed, len(recordsByAttachID))
	}
	return nil
}

func (w *privateLinkConnectionWorker) createOnePLCDomainRecords(
	ctx context.Context,
	attachDomainID string,
	attachRecords []*entity.PLCDomainRecord,
) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger = logger.With(zap.String("attach_domain_id", attachDomainID))

	record := attachRecords[0]
	attach, err := w.plcAttachDomainRepo.Get(ctx, record.ClusterID, record.PrivateLinkConnectionID, attachDomainID)
	if err != nil {
		logger.Error(
			"failed to get plc attach domain",
			zap.Error(err),
			zap.String("attach_domain_id", attachDomainID),
		)
		return err
	}

	if err = w.ensureHostedZone(ctx, attach); err != nil {
		logger.Error("failed to ensure hosted zone", zap.Error(err), zap.String("attach_domain_id", attachDomainID))
		return err
	}

	if err = w.createDNSRecordsInRegional(ctx, attachRecords, attach); err != nil {
		logger.Error(
			"failed to create dns records in regional",
			zap.Error(err),
			zap.String("attach_domain_id", attachDomainID),
		)
		return err
	}

	if err := w.plcDomainRecordRepo.ActiveByAttachDomainID(
		ctx,
		record.ClusterID,
		record.PrivateLinkConnectionID,
		attachDomainID,
		record.AWSRoute53CreateChangeID,
		record.AliDNSRecordID,
	); err != nil {
		logger.Error(
			"failed to active records, please check manually",
			zap.Error(err),
			zap.String("attach_domain_id", attachDomainID),
		)
		return err
	}
	return nil
}

func (w *privateLinkConnectionWorker) deletePLCDomainRecords(ctx context.Context) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger.Info("sync deleting plc domain records")

	records, err := w.plcDomainRecordRepo.ListByFilter(ctx, &entity.ListPLCDomainRecordsFilter{
		States: []constant.PLCDomainRecordState{
			constant.PLCDomainRecordStateDeleting,
			constant.PLCDomainRecordStateRecordsDeleted,
		},
	})
	if err != nil {
		logger.Error("failed to list plc domain records", zap.Error(err))
		return err
	}

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

	failed := 0
	for attachDomainID, attachRecords := range recordsByAttachID {
		if err := w.deleteOnePLCDomainRecords(ctx, attachDomainID, attachRecords); err != nil {
			failed++
		}
	}

	logger.Info(
		"sync deleting plc domain records done",
		zap.Int("total", len(recordsByAttachID)),
		zap.Int("failed", failed),
	)
	if failed > 0 {
		return fmt.Errorf("sync deleting plc domain records failed for %d/%d", failed, len(recordsByAttachID))
	}
	return nil
}

func (w *privateLinkConnectionWorker) deleteOnePLCDomainRecords(
	ctx context.Context,
	attachDomainID string,
	attachRecords []*entity.PLCDomainRecord,
) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger = logger.With(zap.String("attach_domain_id", attachDomainID))

	record := attachRecords[0]
	attach, err := w.plcAttachDomainRepo.Get(ctx, record.ClusterID, record.PrivateLinkConnectionID, attachDomainID)
	if err != nil {
		logger.Error(
			"failed to get plc attach domain",
			zap.Error(err),
			zap.String("attach_domain_id", attachDomainID),
		)
		return err
	}
	if err := w.ensureDeleteRecords(ctx, attach, attachRecords); err != nil {
		logger.Error(
			"failed to ensure delete records",
			zap.Error(err),
			zap.String("attach_domain_id", attachDomainID),
		)
		return err
	}

	if err := w.ensureDeleteHostedZone(ctx, attach); err != nil {
		logger.Error("failed to delete hosted zone", zap.Error(err), zap.String("attach_domain_id", attachDomainID))
		return err
	}

	if err := w.transProvider.Transaction(ctx, func(ctx context.Context) error {
		if err := w.plcAttachDomainRepo.DeleteByID(ctx, record.ClusterID, record.PrivateLinkConnectionID, attachDomainID); err != nil {
			return err
		}
		return w.plcDomainRecordRepo.SoftDeleteByID(ctx, record.ClusterID, record.PrivateLinkConnectionID, attachDomainID)
	}); err != nil {
		logger.Error(
			"failed to delete plc attach domain and records",
			zap.Error(err),
			zap.String("attach_domain_id", attachDomainID),
		)
		return err
	}
	return nil
}

func (w *privateLinkConnectionWorker) createDNSRecordsInRegional(
	ctx context.Context,
	records []*entity.PLCDomainRecord,
	attach *entity.PLCAttachDomain,
) bizerr.BizError {
	if len(records) == 0 {
		return bizerr.New(bizerr.ErrCodeUnknown, "no records to create")
	}

	var reqHostedZoneName, reqHostedZoneID string
	if attach.Type == constant.PLCAttachDomainTypeTiDBCloudManaged {
		reqHostedZoneName = attach.HostedZone
	} else {
		reqHostedZoneID = attach.HostedZoneID
	}

	clusterId := records[0].ClusterID
	provider := records[0].CloudProvider
	logger := utils.GetWorkerLoggerFromContext(ctx).
		With(zap.String("cluster_id", clusterId)).
		With(zap.String("private_link_connection_id", records[0].PrivateLinkConnectionID))

	switch provider {
	case config.ProviderAWS:
		recordsMap := make(map[string]string)
		for _, record := range records {
			recordsMap[record.Record] = record.Value
		}
		resp, err := w.mgmtSvc.CreateAWSRoute53Records(ctx, &globalserverlessv1.CreateAWSRoute53RecordsRequest{
			ClusterId:      clusterId,
			HostedZoneName: reqHostedZoneName,
			HostedZoneId:   reqHostedZoneID,
			Records:        recordsMap,
		})
		if err != nil {
			logger.Error("failed to create route53 records", zap.Error(err))
			return err
		}
		for _, record := range records {
			record.AWSRoute53CreateChangeID = resp.ChangeId
		}
	case config.ProviderAliCloud:
		if len(records) > 1 {
			return bizerr.New(bizerr.ErrCodeUnknown, "alicloud dns only support create one record at a time")
		}
		// Alicloud DNS record does not need hosted zone name suffix
		record := strings.TrimSuffix(records[0].Record, "."+attach.HostedZone)
		resp, err := w.mgmtSvc.CreateAliCloudDNSRecord(ctx, &globalserverlessv1.CreateAliCloudDNSRecordRequest{
			ClusterId:      clusterId,
			HostedZoneName: reqHostedZoneName,
			HostedZoneId:   reqHostedZoneID,
			Record:         record,
			Value:          records[0].Value,
		})
		if err != nil {
			logger.Error("failed to create alicloud dns record", zap.Error(err), zap.String("record", record))
			return err
		}
		records[0].AliDNSRecordID = resp.RecordId
	}
	logger.Info("create dns records in regional done")
	return nil
}

func (w *privateLinkConnectionWorker) deleteDNSRecordsInRegional(
	ctx context.Context,
	records []*entity.PLCDomainRecord,
	attach *entity.PLCAttachDomain,
) bizerr.BizError {
	if len(records) == 0 {
		return bizerr.New(bizerr.ErrCodeUnknown, "no records to delete")
	}

	clusterId := records[0].ClusterID
	provider := records[0].CloudProvider
	logger := utils.GetWorkerLoggerFromContext(ctx).
		With(zap.String("cluster_id", clusterId)).
		With(zap.String("private_link_connection_id", records[0].PrivateLinkConnectionID))

	switch provider {
	case config.ProviderAWS:
		recordsMap := make(map[string]string)
		for _, record := range records {
			recordsMap[record.Record] = record.Value
		}
		var reqHostedZoneName, reqHostedZoneID string
		if attach.Type == constant.PLCAttachDomainTypeTiDBCloudManaged {
			reqHostedZoneName = attach.HostedZone
		} else {
			reqHostedZoneID = attach.HostedZoneID
		}
		resp, err := w.mgmtSvc.DeleteAWSRoute53Records(ctx, &globalserverlessv1.DeleteAWSRoute53RecordsRequest{
			ClusterId:      clusterId,
			HostedZoneName: reqHostedZoneName,
			HostedZoneId:   reqHostedZoneID,
			Records:        recordsMap,
		})
		if err != nil {
			logger.Error("failed to delete route53 records", zap.Error(err))
			return err
		}
		for _, record := range records {
			record.AWSRoute53DeleteChangeID = resp.ChangeId
		}
	case config.ProviderAliCloud:
		if len(records) > 1 {
			return bizerr.New(bizerr.ErrCodeUnknown, "alicloud dns only support delete one record at a time")
		}
		err := w.mgmtSvc.DeleteAliCloudDNSRecord(ctx, &globalserverlessv1.DeleteAliCloudDNSRecordRequest{
			ClusterId: clusterId,
			RecordId:  records[0].AliDNSRecordID,
		})
		if err != nil {
			logger.Error("failed to delete alicloud dns record", zap.Error(err))
			return err
		}
	}
	return nil
}

func (w *privateLinkConnectionWorker) ensureHostedZone(
	ctx context.Context,
	attach *entity.PLCAttachDomain,
) bizerr.BizError {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger = logger.With(zap.String("attach_domain_id", attach.ID))
	// TiDB Cloud Managed type does not need to create hosted zone
	if attach.Type == constant.PLCAttachDomainTypeTiDBCloudManaged {
		return nil
	}
	if attach.HostedZoneID == "" {
		switch attach.CloudProvider {
		case config.ProviderAWS:
			resp, err := w.mgmtSvc.CreateAWSRoute53PrivateHostedZone(
				ctx,
				&globalserverlessv1.CreateAWSRoute53PrivateHostedZoneRequest{
					ClusterId:      attach.ClusterID,
					HostedZoneName: attach.HostedZone,
				},
			)
			if err != nil {
				return err
			}
			attach.HostedZoneID = resp.HostedZoneId
			// aws hosted zone can be created with vpc attached
			attach.VPCAttached = true
		case config.ProviderAliCloud:
			resp, err := w.mgmtSvc.CreateAliCloudDNSPrivateHostedZone(
				ctx,
				&globalserverlessv1.CreateAliCloudDNSPrivateHostedZoneRequest{
					ClusterId:      attach.ClusterID,
					HostedZoneName: attach.HostedZone,
				},
			)
			if err != nil {
				return err
			}
			attach.HostedZoneID = resp.HostedZoneId
		default:
			return bizerr.New(bizerr.ErrCodeUnknown, "unsupported cloud provider %s", attach.CloudProvider)
		}
		if _, err := w.plcAttachDomainRepo.Save(ctx, attach); err != nil {
			return err
		}
		logger.Info("ensure hosted zone done", zap.String("hosted_zone_id", attach.HostedZoneID))
	}
	if !attach.VPCAttached {
		switch attach.CloudProvider {
		case config.ProviderAliCloud:
			if _, err := w.mgmtSvc.AttachVpcToAliCloudDNSPrivateHostedZone(ctx, &globalserverlessv1.AttachVpcToAliCloudDNSPrivateHostedZoneRequest{
				ClusterId:    attach.ClusterID,
				HostedZoneId: attach.HostedZoneID,
			}); err != nil {
				return err
			}
			attach.VPCAttached = true
		case config.ProviderAWS:
			return bizerr.New(bizerr.ErrCodeUnknown, "aws provider must have vpc attached")
		default:
			return bizerr.New(bizerr.ErrCodeUnknown, "unsupported cloud provider %s", attach.CloudProvider)
		}
		if _, err := w.plcAttachDomainRepo.Save(ctx, attach); err != nil {
			return err
		}
		logger.Info("ensure hosted zone vpc attached done", zap.String("hosted_zone_id", attach.HostedZoneID))
	}
	return nil
}

func (w *privateLinkConnectionWorker) ensureDeleteHostedZone(
	ctx context.Context,
	attach *entity.PLCAttachDomain,
) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger = logger.With(zap.String("attach_domain_id", attach.ID))
	// TiDB Cloud Managed type does not need to delete hosted zone
	if attach.Type == constant.PLCAttachDomainTypeTiDBCloudManaged {
		return nil
	}
	// hosted zone id is empty, no need to delete
	if attach.HostedZoneID == "" {
		return nil
	}
	hostedZoneID := attach.HostedZoneID
	switch attach.CloudProvider {
	case config.ProviderAWS:
		if err := w.transProvider.Transaction(ctx, func(ctx context.Context) error {
			hostedZoneID := attach.HostedZoneID
			attach.HostedZoneID = ""
			attach.VPCAttached = false
			if _, err := w.plcAttachDomainRepo.Save(ctx, attach); err != nil {
				logger.Error("failed to save plc attach domain", zap.Error(err))
				return err
			}
			if _, err := w.mgmtSvc.DeleteAWSRoute53PrivateHostedZone(ctx, &globalserverlessv1.DeleteAWSRoute53PrivateHostedZoneRequest{
				ClusterId:    attach.ClusterID,
				HostedZoneId: hostedZoneID,
			}); err != nil {
				logger.Error("failed to delete aws route53 private hosted zone", zap.Error(err))
				return err
			}
			return nil
		}); err != nil {
			return err
		}
	case config.ProviderAliCloud:
		if attach.VPCAttached {
			if err := w.transProvider.Transaction(ctx, func(ctx context.Context) error {
				attach.VPCAttached = false
				if _, err := w.plcAttachDomainRepo.Save(ctx, attach); err != nil {
					logger.Error("failed to save plc attach domain", zap.Error(err))
					return err
				}
				if _, err := w.mgmtSvc.DetachVpcFromAliCloudDNSPrivateHostedZone(ctx, &globalserverlessv1.DetachVpcFromAliCloudDNSPrivateHostedZoneRequest{
					ClusterId:    attach.ClusterID,
					HostedZoneId: attach.HostedZoneID,
				}); err != nil {
					logger.Error("failed to detach vpc from alicloud dns private hosted zone", zap.Error(err))
					return err
				}
				return nil
			}); err != nil {
				return err
			}
		}

		if err := w.transProvider.Transaction(ctx, func(ctx context.Context) error {
			hostedZoneID := attach.HostedZoneID
			attach.HostedZoneID = ""
			if _, err := w.plcAttachDomainRepo.Save(ctx, attach); err != nil {
				logger.Error("failed to save plc attach domain", zap.Error(err))
				return err
			}
			if _, err := w.mgmtSvc.DeleteAliCloudDNSPrivateHostedZone(ctx, &globalserverlessv1.DeleteAliCloudDNSPrivateHostedZoneRequest{
				ClusterId:    attach.ClusterID,
				HostedZoneId: hostedZoneID,
			}); err != nil {
				logger.Error("failed to delete alicloud dns private hosted zone", zap.Error(err))
				return err
			}
			return nil
		}); err != nil {
			return err
		}
	default:
		return bizerr.New(bizerr.ErrCodeUnknown, "unsupported cloud provider %s", attach.CloudProvider)
	}
	logger.Info(
		"ensure delete hosted zone done",
		zap.String("attach_domain_id", attach.ID),
		zap.String("hosted_zone_id", hostedZoneID),
	)
	return nil
}

func (w *privateLinkConnectionWorker) ensureDeleteRecords(
	ctx context.Context,
	attach *entity.PLCAttachDomain,
	records []*entity.PLCDomainRecord,
) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger = logger.With(zap.String("attach_domain_id", attach.ID))
	if len(records) == 0 {
		return nil
	}
	record := records[0]
	if record.State == constant.PLCDomainRecordStateRecordsDeleted {
		logger.Info("records already deleted, skip")
		return nil
	}
	needDelete := false
	switch attach.CloudProvider {
	case config.ProviderAWS:
		needDelete = record.AWSRoute53CreateChangeID != ""
	case config.ProviderAliCloud:
		needDelete = record.AliDNSRecordID != 0
	default:
		return bizerr.New(bizerr.ErrCodeUnknown, "unsupported cloud provider %s", attach.CloudProvider)
	}
	return w.transProvider.Transaction(ctx, func(ctx context.Context) error {
		if err := w.plcDomainRecordRepo.UpdateStateByAttachDomainID(
			ctx,
			record.ClusterID,
			record.PrivateLinkConnectionID,
			record.AttachDomainID,
			constant.PLCDomainRecordStateRecordsDeleted,
			"",
		); err != nil {
			return err
		}
		if needDelete {
			if err := w.deleteDNSRecordsInRegional(ctx, records, attach); err != nil {
				return err
			}
			logger.Info("delete dns records in regional done")
		}
		return nil
	})
}

// only sync state for aws endpoint service, because we already delete it.
func (w *privateLinkConnectionWorker) deleteAWSEndpointServicePLC(
	ctx context.Context,
	plc *entity.PrivateLinkConnection,
) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger = logger.With(zap.String("plc_id", plc.ID))
	logger.Info("delete one private link connection", zap.String("type", string(plc.Type)))

	req := &globalserverlessv1.DescribeAWSVpcEndpointRequest{
		ClusterId:  plc.ClusterID,
		EndpointId: plc.AWSEndpointService.EndpointID,
	}
	resp, err := w.mgmtSvc.DescribeAWSVpcEndpoint(ctx, req)
	if err != nil {
		if err.ErrCode() != bizerr.ErrCodeRecordNotFound {
			logger.Error("failed to describe vpc endpoint", zap.Error(err))
			return err
		}
	}

	// deleted
	if resp == nil || resp.Endpoint == nil ||
		resp.Endpoint.State == regionalserverlessv1.AWSVpcEndpointState_DELETED {
		err = w.plcRepo.DeleteById(ctx, plc.ClusterID, plc.ID)
		if err != nil {
			logger.Error("failed to delete private link connection", zap.Error(err))
			return err
		}
		return nil
	}

	// still deleting
	if resp.Endpoint.State != regionalserverlessv1.AWSVpcEndpointState_DELETING {
		logger.Error("aws endpoint state is not deleting", zap.String("state", resp.Endpoint.State.String()))
		return fmt.Errorf("aws endpoint state is not deleting, state is %s", resp.Endpoint.State.String())
	}
	return nil
}

func (w *privateLinkConnectionWorker) deleteAliCloudEndpointServicePLC(
	ctx context.Context,
	plc *entity.PrivateLinkConnection,
) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger = logger.With(zap.String("plc_id", plc.ID))
	logger.Info("delete one private link connection", zap.String("type", string(plc.Type)))

	// delete alicloud endpoint
	// 1. check endpoint state. if not exist, mark plc as deleted; if exist and active, proceed to delete.
	// 2. delete zones one by one until no zone remains
	// 3. delete endpoint
	// 4. if timeout, retry in next worker loop

	// describe endpoint
	resp, err := w.mgmtSvc.DescribeAliCloudVpcEndpoint(ctx, &globalserverlessv1.DescribeAliCloudVpcEndpointRequest{
		ClusterId:  plc.ClusterID,
		EndpointId: plc.AlicloudEndpointService.EndpointID,
	})
	if err != nil {
		if err.ErrCode() != bizerr.ErrCodeRecordNotFound {
			logger.Error("failed to describe vpc endpoint", zap.Error(err))
			return err
		}
	}

	// endpoint has been deleted
	if err != nil || resp == nil || resp.Endpoint == nil {
		logger.Info("alicloud vpc endpoint deleted successfully")
		err = w.plcRepo.DeleteById(ctx, plc.ClusterID, plc.ID)
		if err != nil {
			logger.Error("failed to delete private link connection", zap.Error(err))
			return err
		}
		return nil
	}

	// only active endpoint can handle delete operation, other state:
	// 1. endpoint is deleting
	// 2. endpoint is in creating
	// 3. endpoint is in pending (zone is in deleting)
	if resp.Endpoint.State != regionalserverlessv1.AliCloudVpcEndpointState_ACTIVE {
		logger.Info(
			"endpoint not in active state, wait and retry",
			zap.String("state", resp.Endpoint.State.String()),
		)
		return nil
	}

	// delete when endpoint is active
	zonesResp, err := w.mgmtSvc.ListAliCloudVpcEndpointZones(
		ctx,
		&globalserverlessv1.ListAliCloudVpcEndpointZonesRequest{
			ClusterId:  plc.ClusterID,
			EndpointId: plc.AlicloudEndpointService.EndpointID,
		},
	)
	if err != nil {
		logger.Error("list vpc endpoint zones failed", zap.Error(err))
		return err
	}
	azs := zonesResp.GetZones()
	// delete zone
	if len(azs) > 0 {
		logger.Info("delete zone from alicloud vpc endpoint", zap.String("zone_id", azs[0].ZoneId))
		err = w.mgmtSvc.RemoveAliCloudZoneFromVpcEndpoint(
			ctx,
			&globalserverlessv1.RemoveAliCloudZoneFromVpcEndpointRequest{
				ClusterId:  plc.ClusterID,
				EndpointId: plc.AlicloudEndpointService.EndpointID,
				ZoneId:     azs[0].ZoneId,
			},
		)
		if err != nil {
			logger.Error("remove zone from alicloud vpc endpoint failed", zap.Error(err))
			return err
		}
		return nil
	}
	// delete endpoint when no zone remains
	logger.Info("delete alicloud vpc endpoint")
	err = w.mgmtSvc.DeleteAliCloudVpcEndpoint(ctx, &globalserverlessv1.DeleteAliCloudVpcEndpointRequest{
		ClusterId:  plc.ClusterID,
		EndpointId: plc.AlicloudEndpointService.EndpointID,
	})
	if err != nil {
		logger.Error("delete alicloud vpc endpoint failed", zap.Error(err))
		return err
	}
	return nil
}

func (w *privateLinkConnectionWorker) deleteAmazonMSKProvisionedPLC(
	ctx context.Context,
	plc *entity.PrivateLinkConnection,
) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger = logger.With(zap.String("plc_id", plc.ID))
	logger.Info("delete one private link connection", zap.String("type", string(plc.Type)))
	cloudProvider, organizationID, servicePlan, routeErr := mskRouteFields(plc)
	if routeErr != nil {
		return routeErr
	}

	resp, err := w.mgmtSvc.GetMSKVPCConnection(ctx, &globalserverlessv1.GetMSKVPCConnectionRequest{
		ClusterId:           "",
		Kubernetes:          plc.Kubernetes,
		Region:              plc.Region,
		CloudProviderId:     cloudProvider,
		OrganizationId:      organizationID,
		ServicePlan:         servicePlan,
		MskVpcConnectionArn: plc.AmazonMSKProvisioned.MSKVPCConnectionArn,
	})
	if err != nil {
		if err.ErrCode() != bizerr.ErrCodeRecordNotFound {
			logger.Error("failed to get msk vpc connection", zap.Error(err))
			return err
		}
	}

	if resp == nil || resp.MskVpcConnection == nil {
		logger.Info("msk vpc connection deleted successfully")
		err = w.plcRepo.DeleteById(ctx, plc.ClusterID, plc.ID)
		if err != nil {
			logger.Error("failed to delete private link connection", zap.Error(err))
			return err
		}
		return nil
	}

	if resp.MskVpcConnection.State != regionalserverlessv1.MSKVpcConnectionState_DELETING {
		logger.Error("msk vpc connection is not deleting", zap.String("state", resp.MskVpcConnection.State.String()))
		return fmt.Errorf("msk vpc connection is not deleting, state is %s", resp.MskVpcConnection.State.String())
	}
	return nil
}