package impl

import (
	"context"
	"regexp"
	"strconv"
	"sync"
	"time"

	"github.com/elliotchance/pie/v2"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/models/ticdc"
	"github.com/tidbcloud/dataflow-service/app/models/ticdc_spec"
	"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"
	"github.com/tidbcloud/pkgv2/db"
	"go.uber.org/zap"
)

var versionRegexp = regexp.MustCompile(`^v\d+\.\d+\.\d+`)

type clusterWorkerRegionalImpl struct {
	instanceTypeModel         ticdc_spec.CDCInstanceTypeModel
	ticdcClusterCli           regionalapi.TiCDCClusterClient
	transProvider             db.TransProvider
	ticdcClusterRepo          repo.TiCDCClusterRepo
	bizChangefeedResourceRepo repo.TiCDCBizChangefeedResourceRepo
}

func NewTiCDCClusterRegionalWorker(
	transProvider db.TransProvider,
	instanceTypeModel ticdc_spec.CDCInstanceTypeModel,
	ticdcClusterCli regionalapi.TiCDCClusterClient,
	ticdcClusterRepo repo.TiCDCClusterRepo,
	bizChangefeedResourceRepo repo.TiCDCBizChangefeedResourceRepo,
) ticdc.ClusterWorker {
	return &clusterWorkerRegionalImpl{
		instanceTypeModel:         instanceTypeModel,
		ticdcClusterCli:           ticdcClusterCli,
		transProvider:             transProvider,
		ticdcClusterRepo:          ticdcClusterRepo,
		bizChangefeedResourceRepo: bizChangefeedResourceRepo,
	}
}

type ticdcClusterWithResources struct {
	ticdcCluster        *entity.TiCDCCluster
	changefeedResources []*entity.TiCDCBizChangefeedResource
}

func (m *clusterWorkerRegionalImpl) SyncAllTiCDCClusters(ctx context.Context) bizerr.BizError {
	var ticdcClusters []*ticdcClusterWithResources
	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		clusters, err := m.listTiCDCClusters(ctx)
		if err != nil {
			return err
		}
		ticdcClusters = clusters
		return nil
	})
	if transErr != nil {
		return convertTransactionError(transErr)
	}

	regionTiCDCClusters := make(map[string][]*ticdcClusterWithResources)
	for _, ticdcCluster := range ticdcClusters {
		region := ticdcCluster.ticdcCluster.Region
		regionTiCDCClusters[region] = append(regionTiCDCClusters[region], ticdcCluster)
	}

	var wg sync.WaitGroup
	for region, oneRegionTiCDCClusters := range regionTiCDCClusters {
		wg.Add(1)
		go m.syncOneRegionTiCDCClusters(ctx, &wg, region, oneRegionTiCDCClusters)
	}
	wg.Wait()
	return nil
}

func (m *clusterWorkerRegionalImpl) listTiCDCClusters(
	ctx context.Context,
) ([]*ticdcClusterWithResources, bizerr.BizError) {
	ticdcClusters, err := m.ticdcClusterRepo.List(ctx,
		func(qs entity.TiCDCClusterQuerySet) entity.TiCDCClusterQuerySet {
			return qs.StatusIn(
				entity.TiCDCClusterStatusCreating,
				entity.TiCDCClusterStatusScaling,
				entity.TiCDCClusterStatusUpgrading,
				entity.TiCDCClusterStatusDeleting).
				IsRegionalEq(true)
		})
	if err != nil {
		return nil, err.Chain("list ticdc clusters")
	}

	ticdcClusterIDs := pie.Map(ticdcClusters, func(ticdcCluster entity.TiCDCCluster) entity.ID {
		return ticdcCluster.ID
	})
	changefeedResources, err := m.bizChangefeedResourceRepo.List(ctx,
		func(qs entity.TiCDCBizChangefeedResourceQuerySet) entity.TiCDCBizChangefeedResourceQuerySet {
			return qs.TiCDCClusterIDIn(ticdcClusterIDs...)
		})
	if err != nil {
		return nil, err.Chain("list biz changefeed resources")
	}

	ticdcClusterChangefeedResources := make(map[entity.ID][]*entity.TiCDCBizChangefeedResource, len(ticdcClusters))
	for i := range changefeedResources {
		r := &changefeedResources[i]
		ticdcClusterChangefeedResources[r.TiCDCClusterID] = append(
			ticdcClusterChangefeedResources[r.TiCDCClusterID], r)
	}

	clusters := make([]*ticdcClusterWithResources, 0, len(ticdcClusters))
	for i := range ticdcClusters {
		ticdcCluster := &ticdcClusters[i]
		clusters = append(clusters, &ticdcClusterWithResources{
			ticdcCluster:        ticdcCluster,
			changefeedResources: ticdcClusterChangefeedResources[ticdcCluster.ID],
		})
	}
	return clusters, nil
}

func (m *clusterWorkerRegionalImpl) syncOneRegionTiCDCClusters(
	ctx context.Context, wg *sync.WaitGroup, region string, ticdcClusters []*ticdcClusterWithResources,
) {
	defer wg.Done()
	logger := log.GetLogger(ctx).With(zap.String("region", region))
	logger.Info("start to sync ticdc clusters of one region")
	startTime := time.Now()
	defer func() {
		costDuration := time.Now().Sub(startTime)
		logger.Info("sync ticdc clusters of one region finished",
			zap.String("cost_time", costDuration.String()))
	}()

	tenantTiCDCClusters := make(map[uint64][]*ticdcClusterWithResources)
	for _, ticdcCluster := range ticdcClusters {
		tenantID := ticdcCluster.ticdcCluster.TenantID.Uint64()
		tenantTiCDCClusters[tenantID] = append(tenantTiCDCClusters[tenantID], ticdcCluster)
	}

	// we should query ticdc clusters by tenant because of the logical region.
	// there may be multiple logical regions in a physical region.
	for tenantID, oneTenantTiCDCClusters := range tenantTiCDCClusters {
		if err := m.syncOneRegionTenantTiCDCClusters(ctx, region, tenantID, oneTenantTiCDCClusters); err != nil {
			logger.Warn("sync one region ticdc clusters of one tenant error",
				zap.Uint64("tenant_id", tenantID), zap.Error(err))
			continue
		}
	}
}

func (m *clusterWorkerRegionalImpl) syncOneRegionTenantTiCDCClusters(
	ctx context.Context, region string, tenantID uint64, ticdcClusters []*ticdcClusterWithResources,
) bizerr.BizError {
	if len(ticdcClusters) == 0 {
		return nil
	}
	r := &regionalapi.Region{
		TenantID: tenantID,
		Name:     region,
		CloudProvider: string(domain.InfraProviderType2CloudProviderType(
			domain.InfraProviderType(ticdcClusters[0].ticdcCluster.InfraProviderType))),
	}
	req := &ticdc_v1.ListTiCDCClustersRequest{
		TenantId: strconv.FormatUint(tenantID, 10),
	}
	resp, err := m.ticdcClusterCli.ListTiCDCClusters(ctx, r, req)
	if err != nil {
		return err.Chain("list regional ticdc clusters")
	}

	regionalTiCDCClusters := make(map[string]*ticdc_v1.TiCDCCluster, len(resp.TicdcClusters))
	for _, ticdcCluster := range resp.TicdcClusters {
		regionalTiCDCClusters[ticdcCluster.Id] = ticdcCluster
	}

	for _, ticdcCluster := range ticdcClusters {
		regionalTiCDCCluster := regionalTiCDCClusters[ticdcCluster.ticdcCluster.Name]
		if ticdcCluster.ticdcCluster.Status == entity.TiCDCClusterStatusDeleting {
			m.syncOneDeletingTiCDCCluster(ctx, ticdcCluster, regionalTiCDCCluster)
		} else {
			m.syncOneNormalTiCDCCluster(ctx, ticdcCluster, regionalTiCDCCluster)
		}
	}
	return nil
}

func (m *clusterWorkerRegionalImpl) syncOneDeletingTiCDCCluster(ctx context.Context,
	ticdcCluster *ticdcClusterWithResources,
	regionalTiCDCCluster *ticdc_v1.TiCDCCluster,
) {
	logger := log.GetLogger(ctx).With(
		zap.Uint64("id", ticdcCluster.ticdcCluster.ID.Uint64()),
		zap.String("name", ticdcCluster.ticdcCluster.Name))
	if regionalTiCDCCluster == nil {
		logger.Info("ticdc cluster is not exist in region control plane, mark it as deleted")
		if err := m.markTiCDCClusterDeleted(ctx, ticdcCluster); err != nil {
			logger.Warn("mark ticdc cluster deleted error",
				zap.Error(err))
		}
	}
	log.GetLogger(ctx).Info("ticdc cluster is still running, skip it")
}

func (m *clusterWorkerRegionalImpl) syncOneNormalTiCDCCluster(
	ctx context.Context, ticdcCluster *ticdcClusterWithResources, regionalTiCDCCluster *ticdc_v1.TiCDCCluster,
) {
	clusterEnt := ticdcCluster.ticdcCluster
	logger := log.GetLogger(ctx).With(zap.Uint64("id", clusterEnt.ID.Uint64()))
	logger.Info("sync ticdc cluster")

	if regionalTiCDCCluster == nil {
		logger.Warn("regional ticdc cluster is not found, mark it as deleted")

		if err := m.markTiCDCClusterDeleted(ctx, ticdcCluster); err != nil {
			log.GetLogger(ctx).Warn("mark ticdc cluster deleted error",
				zap.Error(err))
		}
		return
	}

	if regionalTiCDCCluster.Phase != ticdc_v1.TiCDCCluster_RUNNING {
		logger.Info("regional ticdc cluster is not running, skip it",
			zap.String("phase", regionalTiCDCCluster.Phase.String()))
		return
	}
	lastAppliedSpec := regionalTiCDCCluster.LastAppliedNodeSpec
	if lastAppliedSpec == nil {
		logger.Info("last applied spec is nil, skip it")
		return
	}

	providerType := domain.InfraProviderType2CloudProviderType(domain.InfraProviderType(clusterEnt.InfraProviderType))
	clusterSpecMatched, err := m.checkTiCDCClusterSpecMatched(
		ctx, clusterEnt.SpecificationCode,
		lastAppliedSpec.InstanceType, lastAppliedSpec.Count,
		providerType, clusterEnt.Region)
	if err != nil {
		logger.Error("check ticdc cluster spec matched", zap.Error(err))
		return
	}
	if !clusterSpecMatched {
		logger.Info("ticdc cluster spec and regional ticdc cluster last applied spec are not matched",
			zap.String("spec", clusterEnt.SpecificationCode),
			zap.String("regional_instance_type", lastAppliedSpec.InstanceType),
			zap.Uint32("regional_node_count", lastAppliedSpec.Count))
		return
	}

	if clusterEnt.Status == entity.TiCDCClusterStatusUpgrading {
		version := getVersionFromTag(regionalTiCDCCluster.Configuration.Tag)
		if version != "" {
			clusterEnt.Version = version
		}
	}
	if err := m.syncTiCDCClusterRunning(ctx, ticdcCluster); err != nil {
		logger.Error("sync ticdc cluster running", zap.Error(err))
	}
}

func (m *clusterWorkerRegionalImpl) syncTiCDCClusterRunning(
	ctx context.Context, ticdcCluster *ticdcClusterWithResources,
) bizerr.BizError {
	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		return m.markTiCDCClusterRunning(ctx, ticdcCluster)
	})
	if transErr != nil {
		return convertTransactionError(transErr)
	}
	log.GetLogger(ctx).Info(
		"mark ticdc cluster running successfully",
		zap.Uint64("ticdc_cluster_id", ticdcCluster.ticdcCluster.ID.Uint64()),
		zap.String("ticdc_cluster_name", ticdcCluster.ticdcCluster.Name))
	return nil
}

func (m *clusterWorkerRegionalImpl) markTiCDCClusterRunning(
	ctx context.Context, ticdcCluster *ticdcClusterWithResources,
) bizerr.BizError {
	clusterEnt := ticdcCluster.ticdcCluster
	clusterEnt.Status = entity.TiCDCClusterStatusRunning
	if err := m.ticdcClusterRepo.Save(ctx, clusterEnt); err != nil {
		return err.Chain("save ticdc cluster entity")
	}

	for _, changefeedResource := range ticdcCluster.changefeedResources {
		if changefeedResource.ActualSpecification == changefeedResource.ExpectedSpecification {
			continue
		}

		changefeedResource.ActualSpecification = changefeedResource.ExpectedSpecification
		if err := m.bizChangefeedResourceRepo.Save(ctx, changefeedResource); err != nil {
			return err.Chain("save biz changefeed resource %d", changefeedResource.ID.Uint64())
		}
	}
	return nil
}

func (m *clusterWorkerRegionalImpl) syncTiCDCClusterDeleted(
	ctx context.Context, ticdcCluster *ticdcClusterWithResources,
) bizerr.BizError {
	for _, changefeedResource := range ticdcCluster.changefeedResources {
		if changefeedResource.ExpectedSpecification != "" {
			return bizerr.New(
				bizerr.ErrCodeUnknown,
				"expected spec of changefeed resource %d is not empty",
				changefeedResource.ID.Uint64())
		}
	}
	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		return m.markTiCDCClusterDeleted(ctx, ticdcCluster)
	})
	if transErr != nil {
		return convertTransactionError(transErr)
	}

	log.GetLogger(ctx).Info(
		"mark ticdc cluster deleted successfully",
		zap.Uint64("ticdc_cluster_id", ticdcCluster.ticdcCluster.ID.Uint64()),
		zap.String("ticdc_cluster_name", ticdcCluster.ticdcCluster.Name))
	return nil
}

func (m *clusterWorkerRegionalImpl) markTiCDCClusterDeleted(
	ctx context.Context, ticdcCluster *ticdcClusterWithResources,
) bizerr.BizError {
	clusterEnt := ticdcCluster.ticdcCluster
	clusterEnt.Status = entity.TiCDCClusterStatusDeleted
	deletedAt := time.Now()
	clusterEnt.DeletedAt = &deletedAt
	clusterEnt.DeletedTime = deletedAt.Unix()
	if err := m.ticdcClusterRepo.Save(ctx, clusterEnt); err != nil {
		return err.Chain("save ticdc cluster entity")
	}

	for _, changefeedResource := range ticdcCluster.changefeedResources {
		if changefeedResource.ActualSpecification == changefeedResource.ExpectedSpecification {
			continue
		}

		changefeedResource.ActualSpecification = changefeedResource.ExpectedSpecification
		if err := m.bizChangefeedResourceRepo.Save(ctx, changefeedResource); err != nil {
			return err.Chain("save biz changefeed resource %d", changefeedResource.ID.Uint64())
		}
	}
	return nil
}

func (m *clusterWorkerRegionalImpl) checkTiCDCClusterSpecMatched(
	ctx context.Context, specCode string,
	regionalInstanceType string, regionalNodeCount uint32,
	cloudProvider domain.CloudProviderType, region string,
) (bool, bizerr.BizError) {
	instanceTypeReq := &domain.GetCDCInstanceTypeReq{
		Name:          specCode,
		CloudProvider: cloudProvider,
		Region:        region,
	}
	instanceTypeResp, err := m.instanceTypeModel.GetCDCInstanceType(ctx, instanceTypeReq)
	if err != nil {
		return false, err.Chain("get instance type")
	}

	instanceType := instanceTypeResp.InstanceType
	return regionalInstanceType == instanceType.MachineType && regionalNodeCount == instanceType.Replicas, nil
}

func getVersionFromTag(tag string) string {
	return versionRegexp.FindString(tag)
}