package impl

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

	"github.com/google/uuid"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/models/private_link"
	"github.com/tidbcloud/dataflow-service/app/repo"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/utils"
	regionalapi "github.com/tidbcloud/dataflow-service/pkg/regional-api"
	ticdc_v1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
)

const (
	maxEndpointPerServicePerVPC = 5

	allZonesPlaceHolder = "all zones"
)

type sinkEndpointImpl struct {
	transProvider          db.TransProvider
	cli                    regionalapi.TiCDCPrivateLinkClient
	bizPLEndpointRepo      repo.BizPrivateLinkEndpointRepo
	bizPLEndpointUsageRepo repo.BizPrivateLinkEndpointUsageRepo
}

func NewSinkEndpointImpl(
	transProvider db.TransProvider,
	cli regionalapi.TiCDCPrivateLinkClient,
	bizPLEndpointRepo repo.BizPrivateLinkEndpointRepo,
	bizPLEndpointUsageRepo repo.BizPrivateLinkEndpointUsageRepo,
) private_link.SinkEndpointModel {
	return &sinkEndpointImpl{
		transProvider:          transProvider,
		cli:                    cli,
		bizPLEndpointRepo:      bizPLEndpointRepo,
		bizPLEndpointUsageRepo: bizPLEndpointUsageRepo,
	}
}

func (m *sinkEndpointImpl) CreateSinkEndpoint(
	ctx context.Context, req *domain.CreateSinkEndpointReq,
) (*domain.CreateSinkEndpointResp, bizerr.BizError) {
	m.refineDNSInfo(req)
	if err := m.preCheckCreateSinkEndpointReq(ctx, req); err != nil {
		return nil, err.Chain("pre check create request")
	}
	endpoint := m.createSinkEndpointReq2Entity(req)
	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		return m.doCreateSinkEndpoint(ctx, endpoint)
	})
	if transErr != nil {
		return nil, convertTransactionError(transErr)
	}
	return &domain.CreateSinkEndpointResp{SinkEndpoint: endpoint}, nil
}

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

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

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

func (m *sinkEndpointImpl) preCheckCreateSinkEndpointReq(
	ctx context.Context, req *domain.CreateSinkEndpointReq,
) bizerr.BizError {
	if err := m.validateEndpointName(
		ctx, req.TenantID, req.ProjectID, req.Provider, req.Region, req.Name); err != nil {
		return err
	}
	if err := m.validateEndpointService(
		ctx, req.TenantID, req.ProjectID, req.Provider, req.Region, req.ServiceName); err != nil {
		return err
	}
	if err := m.validateRootDomainName(
		ctx, req.TenantID, req.ProjectID, req.Provider, req.Region, req.RootDomainName); err != nil {
		return err
	}

	region := &regionalapi.Region{
		TenantID:      req.TenantID,
		Name:          req.Region,
		CloudProvider: string(req.Provider),
	}
	validateReq := &ticdc_v1.ValidateEndpointAvailabilityZonesRequest{
		TenantId:      strconv.FormatUint(req.TenantID, 10),
		ProjectId:     strconv.FormatUint(req.ProjectID, 10),
		ServiceName:   req.ServiceName,
		EndpointZones: req.Zones,
	}
	return m.cli.ValidateEndpointAvailabilityZones(ctx, region, validateReq)
}

func (m *sinkEndpointImpl) validateEndpointName(
	ctx context.Context, tenantID, projectID uint64, provider entity.CloudProviderType, region, endpointName string,
) bizerr.BizError {
	filter := func(qs entity.BizPrivateLinkEndpointQuerySet) entity.BizPrivateLinkEndpointQuerySet {
		return qs.TenantIDEq(entity.ID(tenantID)).ProjectIDEq(entity.ID(projectID)).
			CloudProviderEq(provider).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) validateEndpointService(
	ctx context.Context, tenantID, projectID uint64, provider entity.CloudProviderType, region, serviceName string,
) bizerr.BizError {
	filter := func(qs entity.BizPrivateLinkEndpointQuerySet) entity.BizPrivateLinkEndpointQuerySet {
		return qs.TenantIDEq(entity.ID(tenantID)).ProjectIDEq(entity.ID(projectID)).
			CloudProviderEq(provider).RegionEq(region).ServiceNameEq(serviceName).DeletedTimeEq(0)
	}
	endpoints, err := m.bizPLEndpointRepo.List(ctx, filter)
	if err != nil {
		return err
	}

	if len(endpoints) >= maxEndpointPerServicePerVPC {
		return bizerr.New(bizerr.ErrCodeParams,
			"you can only create at most %d endpoints for the service %s per cluster",
			maxEndpointPerServicePerVPC, serviceName)
	}
	return nil
}

func (m *sinkEndpointImpl) validateRootDomainName(
	ctx context.Context, tenantID, projectID uint64, provider entity.CloudProviderType, region, rootDomainName string,
) bizerr.BizError {
	rootDomainName = strings.TrimRight(rootDomainName, ".")
	if rootDomainName == "" {
		return nil
	}

	filter := func(qs entity.BizPrivateLinkEndpointQuerySet) entity.BizPrivateLinkEndpointQuerySet {
		return qs.TenantIDEq(entity.ID(tenantID)).ProjectIDEq(entity.ID(projectID)).
			CloudProviderEq(provider).RegionEq(region).DeletedTimeEq(0)
	}
	endpoints, err := m.bizPLEndpointRepo.List(ctx, filter)
	if err != nil {
		return err
	}
	for _, endpoint := range endpoints {
		endpointDomainName := strings.TrimRight(endpoint.RootDomainName, ".")
		if endpointDomainName == "" {
			continue
		}
		if strings.HasSuffix(endpointDomainName, rootDomainName) ||
			strings.HasSuffix(rootDomainName, endpointDomainName) {
			return bizerr.New(bizerr.ErrCodeParams,
				"the domain name is conflict with domain name %s of endpoint %s",
				endpoint.RootDomainName, endpoint.Name)
		}
	}
	return nil
}

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

func (m *sinkEndpointImpl) createRegionalEndpointIfNotExists(
	ctx context.Context, endpointEnt *entity.BizPrivateLinkEndpoint,
) bizerr.BizError {
	region := &regionalapi.Region{
		TenantID:      endpointEnt.TenantID.Uint64(),
		Name:          endpointEnt.Region,
		CloudProvider: string(endpointEnt.CloudProvider),
	}
	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(
	req *domain.CreateSinkEndpointReq,
) *entity.BizPrivateLinkEndpoint {
	return &entity.BizPrivateLinkEndpoint{
		TenantID:           entity.ID(req.TenantID),
		ProjectID:          entity.ID(req.ProjectID),
		CloudProvider:      req.Provider,
		Region:             req.Region,
		Zones:              req.Zones,
		Name:               req.Name,
		ServiceName:        req.ServiceName,
		ServiceRegion:      req.ServiceRegion,
		RegionalEndpointID: getRegionalSinkEndpointID(req.ProjectID, req.ServiceName),
		Status:             entity.BizPLEndpointStatusCreating,
		IsCustomDomain:     req.IsCustomDomain,
		RootDomainName:     req.RootDomainName,
		DNSRecords:         req.DNSRecords,
		Properties:         req.Properties,
	}
}

func (m *sinkEndpointImpl) convertRegionalSinkEndpoint(
	endpointEnt *entity.BizPrivateLinkEndpoint,
) *ticdc_v1.TiCDCSinkEndpoint {
	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),
		ProjectId:         strconv.FormatUint(endpointEnt.ProjectID.Uint64(), 10),
		ServiceName:       endpointEnt.ServiceName,
		RootDomainName:    endpointEnt.RootDomainName,
		DnsRecords:        records,
		AvailabilityZones: endpointEnt.Zones,
	}
}

func (m *sinkEndpointImpl) ListSinkEndpoint(
	ctx context.Context, req *domain.ListSinkEndpointsReq,
) (*domain.ListSinkEndpointsResp, bizerr.BizError) {
	filter := m.genEndpointFilterByListSinkEndpointReq(req)
	endpoints, err := m.bizPLEndpointRepo.List(ctx, filter)
	if err != nil {
		return nil, err
	}

	sinkEndpointsWithUsage := make([]domain.SinkEndpointWithUsage, 0, len(endpoints))
	for i := range endpoints {
		usageInfo := m.getSinkEndpointUsageInfo(ctx, &endpoints[i])
		sinkEndpointsWithUsage = append(sinkEndpointsWithUsage, domain.SinkEndpointWithUsage{
			Endpoint:  &endpoints[i],
			UsageInfo: usageInfo,
		})
	}
	return &domain.ListSinkEndpointsResp{
		SinkEndpoints: sinkEndpointsWithUsage,
	}, nil
}

func (m *sinkEndpointImpl) genEndpointFilterByListSinkEndpointReq(
	req *domain.ListSinkEndpointsReq,
) repo.BizPrivateLinkEndpointFilterOP {
	return func(qs entity.BizPrivateLinkEndpointQuerySet) entity.BizPrivateLinkEndpointQuerySet {
		qs = qs.TenantIDEq(entity.ID(req.TenantID)).ProjectIDEq(entity.ID(req.ProjectID)).
			StatusNotIn(entity.BizPLEndpointStatusDeleting, entity.BizPLEndpointStatusDeleted).OrderDescByCreatedAt()
		if req.Provider != "" {
			qs = qs.CloudProviderEq(req.Provider)
		}
		if req.Region != "" {
			qs = qs.RegionEq(req.Region)
		}
		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.bizPLEndpointRepo.Load(ctx, req.ID)
	if err != nil {
		return nil, err
	}
	usageInfo := m.getSinkEndpointUsageInfo(ctx, endpoint)
	return &domain.QuerySinkEndpointResp{
		SinkEndpoint: endpoint,
		UsageInfo:    usageInfo,
	}, nil
}

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

	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		return m.doDeleteSinkEndpoint(ctx, endpoint)
	})
	return convertTransactionError(transErr)
}

func (m *sinkEndpointImpl) preCheckDeleteSinkEndpointReq(
	ctx context.Context, req *domain.DeleteSinkEndpointReq,
) bizerr.BizError {
	usageFilter := func(qs entity.BizPrivateLinkEndpointUsageQuerySet) entity.BizPrivateLinkEndpointUsageQuerySet {
		return qs.BizPrivateLinkEndpointIDEq(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: %d", usage.BusinessType, usage.BusinessID))
	}
	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.BizPrivateLinkEndpoint,
) 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.BizPrivateLinkEndpoint,
) bizerr.BizError {
	region := &regionalapi.Region{
		TenantID:      endpointEnt.TenantID.Uint64(),
		Name:          endpointEnt.Region,
		CloudProvider: string(endpointEnt.CloudProvider),
	}
	req := &ticdc_v1.DeleteTiCDCSinkEndpointRequest{
		TicdcSinkEndpointId: endpointEnt.RegionalEndpointID,
	}
	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) {
	region := &regionalapi.Region{
		TenantID:      req.TenantID,
		Name:          req.Region,
		CloudProvider: string(req.Provider),
	}
	zoneReq := &ticdc_v1.GetAvailableZonesRequest{
		TenantId:  strconv.FormatUint(req.TenantID, 10),
		ProjectId: strconv.FormatUint(req.ProjectID, 10),
	}
	zoneResp, err := m.cli.GetAvailableZones(ctx, region, zoneReq)
	if err != nil {
		return nil, err
	}
	return &domain.GetAvailabilityZonesResp{
		Zones: zoneResp.Zones,
	}, nil
}

func (m *sinkEndpointImpl) GenerateSinkAdvertisedListener(
	ctx context.Context, req *domain.GenerateSinkAdvertisedListenerReq,
) (*domain.GenerateSinkAdvertisedListenerResp, bizerr.BizError) {
	rootDomainName := m.genRootDomainName(req)
	if err := m.validateRootDomainName(
		ctx, req.TenantID, req.ProjectID, req.Provider, req.Region, rootDomainName); err != nil {
		return nil, err
	}

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

func (m *sinkEndpointImpl) genRootDomainName(req *domain.GenerateSinkAdvertisedListenerReq) string {
	switch req.Provider {
	case entity.CloudProviderTypeAzure:
		return GenAzureRootDomainName(req.DownstreamID, req.Region, req.Provider, req.ProjectID)
	default:
		return GenRootDomainName(req.DownstreamID, req.Region, req.Provider, req.ProjectID)
	}
}

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

func (m *sinkEndpointImpl) ValidateSinkAdvertisedListenerDomain(
	ctx context.Context, req *domain.ValidateSinkAdvertisedListenerDomainReq,
) (*domain.ValidateSinkAdvertisedListenerDomainResp, bizerr.BizError) {
	if err := m.validateRootDomainName(
		ctx, req.TenantID, req.ProjectID, req.Provider, req.Region, req.RootDomainName); err != nil {
		return nil, err
	}

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

func (m *sinkEndpointImpl) GetSinkEndpointAuthInfo(
	ctx context.Context, req *domain.GetSinkEndpointAuthInfoReq,
) (*domain.GetSinkEndpointAuthInfoResp, bizerr.BizError) {
	region := &regionalapi.Region{
		TenantID:      req.TenantID,
		Name:          req.Region,
		CloudProvider: string(req.Provider),
	}
	authReq := &ticdc_v1.GetSinkEndpointAuthInfoRequest{
		TenantId:  strconv.FormatUint(req.TenantID, 10),
		ProjectId: strconv.FormatUint(req.ProjectID, 10),
	}
	authResp, err := m.cli.GetPrivateLinkAuthInfo(ctx, region, authReq)
	if err != nil {
		return nil, err
	}
	return &domain.GetSinkEndpointAuthInfoResp{
		Network:        authResp.Network,
		CloudAccountID: authResp.CloudAccountId,
		AWSPrincipalID: authResp.AwsPrincipalId,
	}, 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.BizPrivateLinkEndpointUsage {
	return &entity.BizPrivateLinkEndpointUsage{
		BizPrivateLinkEndpointID: entity.ID(req.SinkEndpointID),
		BusinessType:             req.BusinessType,
		BusinessID:               entity.ID(req.BusinessID),
	}
}

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.BizPrivateLinkEndpointUsageQuerySet) entity.BizPrivateLinkEndpointUsageQuerySet {
		return qs.BizPrivateLinkEndpointIDEq(entity.ID(req.SinkEndpointID)).
			BusinessTypeEq(req.BusinessType).BusinessIDEq(entity.ID(req.BusinessID))
	}
}

func getRegionalSinkEndpointID(projectID uint64, serviceName string) string {
	return strings.ToLower(fmt.Sprintf("endpoint-%d-%s", projectID, utils.FnvHash8(serviceName+uuid.NewString())))
}

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())
}

func GenRootDomainName(kafkaID, region string, cloudVendor entity.CloudProviderType, projectID uint64) string {
	return strings.ToLower(fmt.Sprintf("%s.%s.%s.%d.tidbcloud.com.", kafkaID, region, cloudVendor, projectID))
}

// getSinkEndpointUsageInfo retrieves usage information for a sink endpoint
func (m *sinkEndpointImpl) getSinkEndpointUsageInfo(
	ctx context.Context, endpoint *entity.BizPrivateLinkEndpoint,
) *domain.SinkEndpointUsageInfo {
	usageFilter := func(qs entity.BizPrivateLinkEndpointUsageQuerySet) entity.BizPrivateLinkEndpointUsageQuerySet {
		return qs.BizPrivateLinkEndpointIDEq(endpoint.ID)
	}
	usages, err := m.bizPLEndpointUsageRepo.List(ctx, usageFilter)
	if err != nil || len(usages) == 0 {
		// If there's an error or no usages, check if UnusedAt is set
		if endpoint.UnusedAt != nil {
			return &domain.SinkEndpointUsageInfo{
				ChangefeedIds: []uint64{},
				LastUsedAt:    endpoint.UnusedAt,
			}
		}
		return &domain.SinkEndpointUsageInfo{
			ChangefeedIds: []uint64{},
		}
	}

	changefeedIds := make([]uint64, 0, len(usages))
	var lastUsedAt time.Time
	for _, usage := range usages {
		if usage.BusinessType == entity.BizPLEndpointUsageBusinessTypeChangefeed {
			changefeedIds = append(changefeedIds, usage.BusinessID.Uint64())
		}
		// Use CreatedAt as an approximation of last used time
		if lastUsedAt.IsZero() || usage.CreatedAt.After(lastUsedAt) {
			lastUsedAt = usage.CreatedAt
		}
	}

	usageInfo := &domain.SinkEndpointUsageInfo{
		ChangefeedIds: changefeedIds,
	}

	// Set last used time: prefer UnusedAt, otherwise use the latest CreatedAt from usage records
	if endpoint.UnusedAt != nil {
		usageInfo.LastUsedAt = endpoint.UnusedAt
	} else if !lastUsedAt.IsZero() {
		usageInfo.LastUsedAt = &lastUsedAt
	}

	return usageInfo
}

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