package impl

//nolint:gci
import (
	"context"
	"fmt"
	"maps"
	"slices"
	"strconv"
	"sync"
	"time"

	"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/pkg/log"
	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"
	"go.uber.org/zap"
)

type sinkEndpointWorkerImpl struct {
	cli                    regionalapi.TiCDCPrivateLinkClient
	bizPLEndpointRepo      repo.BizPrivateLinkEndpointRepo
	bizPLEndpointUsageRepo repo.BizPrivateLinkEndpointUsageRepo
	sinkEndpointModel      private_link.SinkEndpointModel
}

func NewSinkEndpointWorkerImpl(
	cli regionalapi.TiCDCPrivateLinkClient,
	bizPLEndpointRepo repo.BizPrivateLinkEndpointRepo,
	bizPLEndpointUsageRepo repo.BizPrivateLinkEndpointUsageRepo,
	sinkEndpointModel private_link.SinkEndpointModel,
) private_link.SinkEndpointWorker {
	return &sinkEndpointWorkerImpl{
		cli:                    cli,
		bizPLEndpointRepo:      bizPLEndpointRepo,
		bizPLEndpointUsageRepo: bizPLEndpointUsageRepo,
		sinkEndpointModel:      sinkEndpointModel,
	}
}

type tenantRegionEndpointsGroup struct {
	tenantID      uint64
	region        string
	cloudProvider string
	endpoints     []entity.BizPrivateLinkEndpoint
}

func (m *sinkEndpointWorkerImpl) SyncAllEndpoints(ctx context.Context) error {
	endpointsGroups, err := m.listEndpointsGroupByTenantRegion(ctx)
	if err != nil {
		log.GetLogger(ctx).Error("list endpoints groups error", zap.Error(err))
		return err
	}

	var wg sync.WaitGroup
	for _, group := range endpointsGroups {
		wg.Add(1)
		go m.syncOneTenantRegionEndpoints(ctx, &wg, group)
	}
	wg.Wait()
	return nil
}

func (m *sinkEndpointWorkerImpl) listEndpointsGroupByTenantRegion(
	ctx context.Context,
) ([]*tenantRegionEndpointsGroup, bizerr.BizError) {
	filter := func(qs entity.BizPrivateLinkEndpointQuerySet) entity.BizPrivateLinkEndpointQuerySet {
		return qs.StatusNotIn(entity.BizPLEndpointStatusDeleted)
	}
	endpoints, err := m.bizPLEndpointRepo.List(ctx, filter)
	if err != nil {
		return nil, err
	}

	groupsMap := make(map[string]*tenantRegionEndpointsGroup)
	for _, endpoint := range endpoints {
		k := m.genTenantRegionKey(endpoint.TenantID.Uint64(), endpoint.Region)
		group, ok := groupsMap[k]
		if !ok {
			group = &tenantRegionEndpointsGroup{
				tenantID:      endpoint.TenantID.Uint64(),
				region:        endpoint.Region,
				cloudProvider: string(endpoint.CloudProvider),
				endpoints:     []entity.BizPrivateLinkEndpoint{},
			}
			groupsMap[k] = group
		}
		group.endpoints = append(group.endpoints, endpoint)
	}

	return slices.Collect(maps.Values(groupsMap)), nil
}

func (m *sinkEndpointWorkerImpl) genTenantRegionKey(tenantID uint64, region string) string {
	return fmt.Sprintf("%d-%s", tenantID, region)
}

func (m *sinkEndpointWorkerImpl) syncOneTenantRegionEndpoints(
	ctx context.Context, wg *sync.WaitGroup, endpointsGroup *tenantRegionEndpointsGroup,
) {
	defer wg.Done()

	logger := log.GetLogger(ctx).With(
		zap.Uint64("tenantID", endpointsGroup.tenantID), zap.String("region", endpointsGroup.region))
	region := &regionalapi.Region{
		TenantID:      endpointsGroup.tenantID,
		Name:          endpointsGroup.region,
		CloudProvider: endpointsGroup.cloudProvider,
	}
	req := &ticdc_v1.ListTiCDCSinkEndpointsRequest{
		TenantId: strconv.FormatUint(endpointsGroup.tenantID, 10),
	}
	resp, err := m.cli.ListTiCDCPrivateLinks(ctx, region, req)
	if err != nil {
		logger.Error("list regional sink endpoints error", zap.Error(err))
		return
	}

	regionalID2Endpoints := make(map[string]*ticdc_v1.TiCDCSinkEndpoint, len(resp.TicdcSinkEndpoints))
	for _, endpoint := range resp.TicdcSinkEndpoints {
		regionalID2Endpoints[endpoint.Id] = endpoint
	}

	for i := range endpointsGroup.endpoints {
		endpoint := &endpointsGroup.endpoints[i]
		if endpoint.Status == entity.BizPLEndpointStatusDeleting {
			m.syncOneDeletingEndpoint(ctx, logger, endpoint, regionalID2Endpoints[endpoint.RegionalEndpointID])
		} else {
			m.syncOneNormalEndpoint(ctx, logger, endpoint, regionalID2Endpoints[endpoint.RegionalEndpointID])
		}
	}
}

func (m *sinkEndpointWorkerImpl) syncOneDeletingEndpoint(
	ctx context.Context, logger *zap.Logger,
	endpointEnt *entity.BizPrivateLinkEndpoint, regionalEndpoint *ticdc_v1.TiCDCSinkEndpoint,
) {
	if regionalEndpoint != nil {
		return
	}
	endpointEnt.DeletedTime = time.Now().Unix()
	endpointEnt.Status = entity.BizPLEndpointStatusDeleted
	if err := m.bizPLEndpointRepo.Save(ctx, endpointEnt); err != nil {
		logger.Warn("set sink endpoint deleted error",
			zap.Error(err), zap.Uint64("id", endpointEnt.ID.Uint64()))
		return
	}
	logger.Info("set sink endpoint deleted successfully", zap.Uint64("id", endpointEnt.ID.Uint64()))
}

func (m *sinkEndpointWorkerImpl) syncOneNormalEndpoint(
	ctx context.Context, logger *zap.Logger,
	endpointEnt *entity.BizPrivateLinkEndpoint, regionalEndpoint *ticdc_v1.TiCDCSinkEndpoint,
) {
	if regionalEndpoint == nil {
		logger.Error("regional endpoint is not found", zap.Uint64("endpointId", endpointEnt.ID.Uint64()))
		return
	}

	updated := false
	status := m.endpointStatusRegional2Entity(regionalEndpoint.Phase)
	if endpointEnt.Status != status {
		endpointEnt.Status = status
		updated = true
	}
	if endpointEnt.EndpointAddress != regionalEndpoint.EndpointAddress {
		endpointEnt.EndpointAddress = regionalEndpoint.EndpointAddress
		updated = true
	}
	if len(regionalEndpoint.ZonesAddresses) > 0 && len(endpointEnt.ZonesAddresses) == 0 {
		endpointEnt.ZonesAddresses = make(map[string]string, len(regionalEndpoint.ZonesAddresses))
	}
	for zone, addr := range regionalEndpoint.ZonesAddresses {
		if endpointEnt.ZonesAddresses[zone] != addr {
			endpointEnt.ZonesAddresses[zone] = addr
			updated = true
		}
	}
	if endpointEnt.CloudEndpointID != regionalEndpoint.EndpointId {
		endpointEnt.CloudEndpointID = regionalEndpoint.EndpointId
		updated = true
	}
	if endpointEnt.CloudAccountID != regionalEndpoint.CloudAccountId {
		endpointEnt.CloudAccountID = regionalEndpoint.CloudAccountId
		updated = true
	}

	if !updated {
		return
	}
	if err := m.bizPLEndpointRepo.Save(ctx, endpointEnt); err != nil {
		logger.Warn("update sink endpoint info error",
			zap.Error(err), zap.Uint64("id", endpointEnt.ID.Uint64()))
		return
	}
	logger.Info("update sink endpoint info successfully", zap.Uint64("id", endpointEnt.ID.Uint64()))
}

func (m *sinkEndpointWorkerImpl) endpointStatusRegional2Entity(
	phase ticdc_v1.TiCDCSinkEndpoint_Phase,
) entity.BizPLEndpointStatus {
	switch phase {
	case ticdc_v1.TiCDCSinkEndpoint_CREATING, ticdc_v1.TiCDCSinkEndpoint_PHASE_UNSPECIFIED:
		return entity.BizPLEndpointStatusCreating
	case ticdc_v1.TiCDCSinkEndpoint_PENDING_ACCEPTANCE:
		return entity.BizPLEndpointStatusPendingAcceptance
	case ticdc_v1.TiCDCSinkEndpoint_AVAILABLE:
		return entity.BizPLEndpointStatusAvailable
	case ticdc_v1.TiCDCSinkEndpoint_REJECTED:
		return entity.BizPLEndpointStatusRejected
	case ticdc_v1.TiCDCSinkEndpoint_FAILED:
		return entity.BizPLEndpointStatusFailed
	default:
		return ""
	}
}

// CleanupUnusedEndpoints cleans up unused BizPrivateLinkEndpoint resources.
// If an endpoint is not referenced by any BizPrivateLinkEndpointUsage, it will be deleted after UnusedEndpointRetentionDays.
func (m *sinkEndpointWorkerImpl) CleanupUnusedEndpoints(ctx context.Context) error {
	logger := log.GetLogger(ctx)

	endpoints, err := m.listActiveEndpoints(ctx)
	if err != nil {
		logger.Error("list active endpoints error", zap.Error(err))
		return err
	}

	if len(endpoints) == 0 {
		return nil
	}

	stats := &cleanupStats{}
	now := time.Now()
	retentionThreshold := now.AddDate(0, 0, -private_link.UnusedEndpointRetentionDays)

	for i := range endpoints {
		m.processEndpointForCleanup(ctx, &endpoints[i], now, retentionThreshold, stats)
	}

	stats.logSummary(logger)
	return nil
}

// listActiveEndpoints returns all endpoints that are not in deleting or deleted status
func (m *sinkEndpointWorkerImpl) listActiveEndpoints(
	ctx context.Context,
) ([]entity.BizPrivateLinkEndpoint, bizerr.BizError) {
	filter := func(qs entity.BizPrivateLinkEndpointQuerySet) entity.BizPrivateLinkEndpointQuerySet {
		return qs.StatusNotIn(entity.BizPLEndpointStatusDeleting, entity.BizPLEndpointStatusDeleted)
	}
	return m.bizPLEndpointRepo.List(ctx, filter)
}

// processEndpointForCleanup processes a single endpoint for cleanup
func (m *sinkEndpointWorkerImpl) processEndpointForCleanup(
	ctx context.Context,
	endpoint *entity.BizPrivateLinkEndpoint,
	now time.Time,
	retentionThreshold time.Time,
	stats *cleanupStats,
) {
	logger := log.GetLogger(ctx).With(zap.Uint64("endpoint_id", endpoint.ID.Uint64()))

	isReferenced, err := m.isEndpointReferenced(ctx, endpoint.ID)
	if err != nil {
		logger.Warn("check endpoint reference error", zap.Error(err))
		return
	}

	if isReferenced {
		m.clearUnusedAtIfSet(ctx, endpoint, logger)
		return
	}

	// Endpoint is not referenced
	if endpoint.UnusedAt == nil {
		m.markEndpointAsUnused(ctx, endpoint, now, logger, stats)
	} else if endpoint.UnusedAt.Before(retentionThreshold) {
		m.deleteUnusedEndpoint(ctx, endpoint, logger, stats)
	}
}

// isEndpointReferenced checks if the endpoint is referenced by any BizPrivateLinkEndpointUsage
func (m *sinkEndpointWorkerImpl) isEndpointReferenced(
	ctx context.Context,
	endpointID entity.ID,
) (bool, bizerr.BizError) {
	usageFilter := func(qs entity.BizPrivateLinkEndpointUsageQuerySet) entity.BizPrivateLinkEndpointUsageQuerySet {
		return qs.BizPrivateLinkEndpointIDEq(endpointID)
	}
	count, err := m.bizPLEndpointUsageRepo.Count(ctx, usageFilter)
	if err != nil {
		return false, err
	}
	return count > 0, nil
}

// markEndpointAsUnused marks an endpoint as unused by setting UnusedAt to current time
func (m *sinkEndpointWorkerImpl) markEndpointAsUnused(
	ctx context.Context,
	endpoint *entity.BizPrivateLinkEndpoint,
	now time.Time,
	logger *zap.Logger,
	stats *cleanupStats,
) {
	endpoint.UnusedAt = &now
	if err := m.bizPLEndpointRepo.Save(ctx, endpoint); err != nil {
		logger.Warn("mark endpoint as unused error", zap.Error(err))
		return
	}
	stats.markedUnused++
	logger.Info("marked endpoint as unused")
}

// deleteUnusedEndpoint deletes an endpoint that has been unused for more than retention period
func (m *sinkEndpointWorkerImpl) deleteUnusedEndpoint(
	ctx context.Context,
	endpoint *entity.BizPrivateLinkEndpoint,
	logger *zap.Logger,
	stats *cleanupStats,
) {
	deleteReq := &domain.DeleteSinkEndpointReq{
		ID: endpoint.ID.Uint64(),
	}
	if err := m.sinkEndpointModel.DeleteSinkEndpoint(ctx, deleteReq); err != nil {
		logger.Warn("delete endpoint error", zap.Error(err))
		return
	}
	stats.deleted++
	logger.Info("triggered endpoint deletion", zap.Time("unused_at", *endpoint.UnusedAt))
}

// clearUnusedAtIfSet clears the UnusedAt field if it's set, indicating the endpoint is now in use
func (m *sinkEndpointWorkerImpl) clearUnusedAtIfSet(
	ctx context.Context,
	endpoint *entity.BizPrivateLinkEndpoint,
	logger *zap.Logger,
) {
	if endpoint.UnusedAt == nil {
		return
	}

	endpoint.UnusedAt = nil
	if err := m.bizPLEndpointRepo.Save(ctx, endpoint); err != nil {
		logger.Warn("clear endpoint unused_at error", zap.Error(err))
		return
	}
	logger.Info("cleared endpoint unused_at")
}

// cleanupStats tracks statistics during cleanup process
type cleanupStats struct {
	markedUnused int
	deleted      int
}

func (s *cleanupStats) logSummary(logger *zap.Logger) {
	if s.markedUnused > 0 || s.deleted > 0 {
		logger.Info("cleanup unused endpoints completed",
			zap.Int("marked_unused", s.markedUnused),
			zap.Int("deleted", s.deleted))
	}
}