package impl

import (
	"context"
	"fmt"
	"sync"
	"time"

	ticdc_v1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/dataflow-service-ng/app/models/ticdc"
	"github.com/tidbcloud/dataflow-service-ng/app/repo"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/dataflow-service-ng/pkg/log"
	regionalapi "github.com/tidbcloud/dataflow-service-ng/pkg/regional-api"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
)

type changefeedWorkerRegionalImpl struct {
	changefeedCli    regionalapi.ChangefeedClient
	ticdcClusterRepo repo.TiCDCClusterRepo
	changefeedRepo   repo.TiCDCChangefeedRepo
}

func NewChangefeedRegionalWorker(
	changefeedCli regionalapi.ChangefeedClient,
	ticdcClusterRepo repo.TiCDCClusterRepo,
	changefeedRepo repo.TiCDCChangefeedRepo,
) ticdc.LongTermSynchronization {
	return &changefeedWorkerRegionalImpl{
		changefeedCli:    changefeedCli,
		ticdcClusterRepo: ticdcClusterRepo,
		changefeedRepo:   changefeedRepo,
	}
}

type ticdcClusterChangefeeds struct {
	cluster     *entity.TiCDCCluster
	changefeeds []*entity.TiCDCChangefeed
}

func (m *changefeedWorkerRegionalImpl) SyncAllChangefeed(ctx context.Context) bizerr.BizError {
	clusterChangefeedsList, err := m.listChangefeedsGroupByTiCDCCluster(ctx)
	if err != nil {
		return err
	}

	var wg sync.WaitGroup
	for i := range clusterChangefeedsList {
		wg.Add(1)
		oneClusterChangefeeds := clusterChangefeedsList[i]
		go m.syncOneTiCDCClusterChangefeeds(ctx, &wg, oneClusterChangefeeds)
	}
	wg.Wait()
	return nil
}

func (m *changefeedWorkerRegionalImpl) listChangefeedsGroupByTiCDCCluster(
	ctx context.Context,
) ([]*ticdcClusterChangefeeds, bizerr.BizError) {
	ticdcClusters, err := m.ticdcClusterRepo.List(ctx,
		func(qs entity.TiCDCClusterQuerySet) entity.TiCDCClusterQuerySet {
			return qs.StatusNe(entity.TiCDCClusterStatusDeleted)
		})
	if err != nil {
		return nil, err.Chain("list ticdc clusters")
	}
	changefeeds, err := m.changefeedRepo.List(ctx,
		func(qs entity.TiCDCChangefeedQuerySet) entity.TiCDCChangefeedQuerySet {
			return qs.DeletedAtIsNull()
		})
	if err != nil {
		return nil, err.Chain("list changefeeds")
	}

	id2TiCDCclusters := make(map[entity.ID]*entity.TiCDCCluster, len(ticdcClusters))
	for i := range ticdcClusters {
		ticdcCluster := &ticdcClusters[i]
		id2TiCDCclusters[ticdcCluster.ID] = ticdcCluster
	}
	ticdcClusterID2Changefeeds := make(map[entity.ID][]*entity.TiCDCChangefeed, len(ticdcClusters))
	for i := range changefeeds {
		changefeed := &changefeeds[i]
		ticdcClusterID2Changefeeds[changefeed.TiCDCClusterID] = append(
			ticdcClusterID2Changefeeds[changefeed.TiCDCClusterID], changefeed)
	}

	ticdcClusterChangefeedsList := make([]*ticdcClusterChangefeeds, 0, len(ticdcClusters))
	for ticdcClusterID, oneClusterChangefeeds := range ticdcClusterID2Changefeeds {
		ticdcCluster, ok := id2TiCDCclusters[ticdcClusterID]
		if !ok {
			log.GetLogger(ctx).Info("ticdc cluster of changefeeds is not in regional or deleted",
				zap.Uint64("ticdc_cluster_id", ticdcClusterID.Uint64()))
			continue
		}
		clusterChangefeeds := &ticdcClusterChangefeeds{
			cluster:     ticdcCluster,
			changefeeds: oneClusterChangefeeds,
		}
		ticdcClusterChangefeedsList = append(ticdcClusterChangefeedsList, clusterChangefeeds)
	}
	return ticdcClusterChangefeedsList, nil
}

func (m *changefeedWorkerRegionalImpl) syncOneTiCDCClusterChangefeeds(
	ctx context.Context, wg *sync.WaitGroup, clusterChangefeeds *ticdcClusterChangefeeds,
) {
	defer wg.Done()
	ticdcCluster := clusterChangefeeds.cluster
	changefeedEnts := clusterChangefeeds.changefeeds
	logger := log.GetLogger(ctx).With(zap.Uint64("ticdc_cluster_id", ticdcCluster.ID.Uint64()))
	logger.Info("start to sync changefeeds of one ticdc cluster")
	startTime := time.Now()
	defer func() {
		costDuration := time.Now().Sub(startTime)
		logger.Info("sync changefeeds of one ticdc cluster finished",
			zap.String("cost_time", costDuration.String()))
	}()

	region := &regionalapi.Region{
		SharedPoolID: ticdcCluster.SharedPoolID,
	}
	tidbClusterIDMap := make(map[entity.ID]struct{})
	for _, changefeedEnt := range changefeedEnts {
		tidbClusterIDMap[changefeedEnt.UpStreamClusterID] = struct{}{}
	}
	kernelID2ChangefeedEnts := make(map[string]*entity.TiCDCChangefeed, len(changefeedEnts))
	for _, changefeedEnt := range changefeedEnts {
		kernelID2ChangefeedEnts[changefeedEnt.KernelChangefeedID] = changefeedEnt
	}

	for tidbClusterID := range tidbClusterIDMap {
		listChangefeedReq := &ticdc_v1.ListChangefeedsRequest{
			TicdcClusterId: ticdcCluster.Name,
			TidbClusterId:  fmt.Sprintf("%d", tidbClusterID.Uint64()),
		}
		listChangefeedResp, err := m.changefeedCli.ListChangefeeds(ctx, region, listChangefeedReq)
		if err != nil {
			logger.Warn("list changefeeds from regional error", zap.Error(err))
			return
		}

		for _, kernelChangefeed := range listChangefeedResp.Items {
			kernelID := kernelChangefeed.Id
			changefeedEnt, ok := kernelID2ChangefeedEnts[kernelID]
			if !ok {
				logger.Error("kernel changefeed is not found in changefeed entities",
					zap.String("kernel_changefeed_id", kernelID))
				continue
			}
			logger := logger.With(zap.String("kernel_changefeed_id", kernelID),
				zap.Uint64("changefeed_id", changefeedEnt.ID.Uint64()))
			if err := m.syncOneChangefeedStatus(ctx, changefeedEnt, kernelChangefeed); err != nil {
				logger.Warn("sync changefeed status error", zap.Error(err))
			} else {
				logger.Info("sync changefeed status successfully")
			}
		}
	}
}

func (m *changefeedWorkerRegionalImpl) syncOneChangefeedStatus(
	ctx context.Context, changefeedEnt *entity.TiCDCChangefeed, kernelChangefeed *ticdc_v1.ChangefeedCommonInfo,
) bizerr.BizError {
	updatedAt := time.Now()
	checkpointTs := kernelChangefeed.CheckpointTso
	changefeedEnt.KernelMetrics = entity.TiCDCChangefeedKernelMetrics{
		ID:            kernelChangefeed.Id,
		FeedState:     entity.ChangefeedKernelStatus(kernelChangefeed.State),
		CheckpointTSO: checkpointTs,
		UpdatedAt:     &updatedAt,
	}
	if changefeedErr := kernelChangefeed.Error; changefeedErr != nil {
		changefeedEnt.KernelMetrics.RunningError = &entity.TiCDCChangefeedKernelRunningError{
			Addr:    changefeedErr.Addr,
			Code:    changefeedErr.Code,
			Message: changefeedErr.Message,
		}
	}
	if err := m.changefeedRepo.Save(ctx, changefeedEnt); err != nil {
		return err.Chain("save changefeed entity")
	}
	return nil
}