package impl

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

	"github.com/elliotchance/pie/v2"
	"github.com/google/uuid"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/models/private_link"
	"github.com/tidbcloud/dataflow-service/app/models/ticdc"
	"github.com/tidbcloud/dataflow-service/app/repo"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	"go.uber.org/zap"
)

const (
	privateLinkBillingCategory     = "PrivateLink"
	changefeedKernelIDPattern      = "changefeed-%d"
	privateLinkBillingSkuPattern   = "changefeed-pl-data-processing:%s:%s"
	privateLinkTrafficBillingRatio = 1.1
)

type privateLinkWorkerImpl struct {
	dbProvider                  db.TransProvider
	bizChangefeedRepo           repo.BizChangefeedRepo
	billingRecordRepo           repo.BillingRecordRepo
	coreCli                     grpcli.CoreGrpcClient
	changefeedTrafficRecordRepo repo.KernelChangefeedTrafficRecordRepo
	changefeedTrafficMeter      ticdc.ChangefeedTrafficMeter
}

func NewPrivateLinkWorker(
	dbProvider db.TransProvider,
	billingRecordRepo repo.BillingRecordRepo,
	coreCli grpcli.CoreGrpcClient,
	changefeedTrafficRecordRepo repo.KernelChangefeedTrafficRecordRepo,
	changefeedTrafficMeter ticdc.ChangefeedTrafficMeter,
	bizChangefeedRepo repo.BizChangefeedRepo,
) private_link.BizPrivateLinkWorker {
	return &privateLinkWorkerImpl{
		dbProvider:                  dbProvider,
		billingRecordRepo:           billingRecordRepo,
		coreCli:                     coreCli,
		changefeedTrafficRecordRepo: changefeedTrafficRecordRepo,
		changefeedTrafficMeter:      changefeedTrafficMeter,
		bizChangefeedRepo:           bizChangefeedRepo,
	}
}

type dataProcessingRecord struct {
	billingRecord entity.BillingRecord
	meterRecords  []entity.KernelChangefeedTrafficRecord
	clusterId     uint64
}

type changefeedWrapper struct {
	ID              uint64
	BizPLEndpointID uint64
	TenantID        uint64
	ProjectID       uint64
	ClusterID       uint64
}

func (p privateLinkWorkerImpl) MeterChangefeeds(
	ctx context.Context,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	logger.Info("start to meter changefeeds using private link.")

	changefeeds, err := p.getPLChangefeeds(ctx)
	if err != nil {
		return err
	}
	if len(changefeeds) == 0 {
		return nil
	}
	clusterIds := pie.Map(changefeeds, func(changefeed changefeedWrapper) uint64 {
		return changefeed.ClusterID
	})
	listClusterReq := &core.ListClustersReq{
		Filter: &core.ClusterFilter{
			ClusterIds: clusterIds,
		},
	}
	listClusterResp, err := p.coreCli.ListClusters(ctx, listClusterReq)
	if err != nil {
		logger.Error("fail to get pl changefeeds clusters", zap.Error(err))
		return err
	}
	for _, cf := range changefeeds {
		cluster, ok := listClusterResp.Clusters[cf.ClusterID]
		if !ok {
			logger.Error("fail to get pl changefeed corresponding cluster",
				zap.Uint64("changefeedId", cf.ID),
				zap.Uint64("clusterId", cf.ClusterID))
			continue
		}
		infoMap := map[string]string{
			"changefeedID": strconv.FormatUint(cf.ID, 10),
		}
		if cf.BizPLEndpointID != 0 {
			infoMap["bizPrivateLinkID"] = strconv.FormatUint(cf.BizPLEndpointID, 10)
		}
		req := &domain.MeterChangefeedTrafficReq{
			ChangefeedId:               cf.ID,
			ChangefeedKernelId:         fmt.Sprintf(changefeedKernelIDPattern, cf.ID),
			BizType:                    entity.KernelTrafficBizType_PrivateLink,
			PrimaryCluster:             cluster,
			InfoMap:                    infoMap,
			CloudProviderType:          entity.CloudProviderType(cluster.CloudProviderName),
			SecondaryCloudProviderType: entity.CloudProviderType(cluster.CloudProviderName),
		}
		if _, err = p.changefeedTrafficMeter.MeterChangefeedTraffic(ctx, req); err != nil {
			logger.Error("fail to meter changefeed", zap.Error(err))
		}
	}
	return nil
}

func (p privateLinkWorkerImpl) BillChangefeeds(
	ctx context.Context,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	unSyncedMeteringRecords, err := p.changefeedTrafficRecordRepo.List(ctx,
		func(qs entity.KernelChangefeedTrafficRecordQuerySet) entity.KernelChangefeedTrafficRecordQuerySet {
			return qs.BusinessTypeEq(entity.KernelTrafficBizType_PrivateLink).SyncedEq(false)
		})
	if err != nil {
		logger.Error("fail to get changefeed traffic record", zap.Error(err))
		return err
	}
	dataProcessingRecords, err := p.aggregateDataProcessingRecordsByChangefeed(unSyncedMeteringRecords)
	if err != nil {
		return err
	}

	for _, dataProcessingRecord := range dataProcessingRecords {
		if err := p.dbProvider.Transaction(ctx, func(ctx context.Context) error {
			for _, meteringRecord := range dataProcessingRecord.meterRecords {
				meteringRecord.Synced = true
				if err := p.changefeedTrafficRecordRepo.Save(ctx, &meteringRecord); err != nil {
					return err
				}
			}
			dataProcessingRecord.billingRecord.Total = int64(float64(dataProcessingRecord.billingRecord.Total) *
				privateLinkTrafficBillingRatio)
			if err := p.billingRecordRepo.Save(ctx, &dataProcessingRecord.billingRecord); err != nil {
				return err
			}
			return nil
		}); err != nil {
			logger.Error("fail to save billing records", zap.Error(err))
			return bizerr_wrapper.NewInternalErr("fail to save billing records")
		}
	}
	return nil
}

func (p privateLinkWorkerImpl) aggregateDataProcessingRecordsByChangefeed(
	unSyncedMeteringRecords []entity.KernelChangefeedTrafficRecord,
) (map[uint64]*dataProcessingRecord, bizerr.BizError) {
	dataProcessingRecords := make(map[uint64]*dataProcessingRecord)
	for _, meteringRecord := range unSyncedMeteringRecords {
		key := uint64(meteringRecord.ChangefeedID)
		record, ok := dataProcessingRecords[key]
		if !ok {
			sku := fmt.Sprintf(privateLinkBillingSkuPattern, meteringRecord.SrcCloudProvider, meteringRecord.Region)
			record = &dataProcessingRecord{
				billingRecord: entity.BillingRecord{
					TenantID:    meteringRecord.TenantID,
					ProjectID:   meteringRecord.ProjectID,
					ClusterID:   meteringRecord.ClusterID,
					SKU:         sku,
					Synced:      false,
					Category:    privateLinkBillingCategory,
					ExtraLabels: entity.BillingRecordExtraLabel(meteringRecord.Extension),
					CostID:      uuid.NewString(),
					Total:       0,
					MeasuredTS:  *meteringRecord.EndTS,
				},
				meterRecords: []entity.KernelChangefeedTrafficRecord{},
			}
			dataProcessingRecords[key] = record
		}
		if meteringRecord.EndTS.After(record.billingRecord.MeasuredTS) {
			record.billingRecord.MeasuredTS = *meteringRecord.EndTS
		}
		record.billingRecord.Total += meteringRecord.Traffic
		record.meterRecords = append(record.meterRecords, meteringRecord)
	}
	return dataProcessingRecords, nil
}

func (p privateLinkWorkerImpl) getPLChangefeeds(ctx context.Context) ([]changefeedWrapper, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	bizWrappers, err := p.getBizChangefeedWrappers(ctx)
	if err != nil {
		logger.Error("fail to get pl biz_changefeed", zap.Error(err))
		return nil, err
	}
	changefeedIds := strings.Join(pie.Map(bizWrappers, func(t changefeedWrapper) string {
		return fmt.Sprintf("%d", t.ID)
	}), ",")
	logger.Info("traffic of pl biz_changefeed will be metered", zap.String("changefeeds", changefeedIds))
	return bizWrappers, nil
}

func (p privateLinkWorkerImpl) getBizChangefeedWrappers(ctx context.Context) ([]changefeedWrapper, bizerr.BizError) {
	bizChangefeeds, err := p.bizChangefeedRepo.List(ctx,
		func(qs entity.BizChangefeedQuerySet) entity.BizChangefeedQuerySet {
			return qs.OperationStatusNotIn(
				entity.BizChangefeedOperationStatusCreating,
				entity.BizChangefeedOperationStatusDeleting,
				entity.BizChangefeedOperationStatusDeleted)
		})
	if err != nil {
		return nil, err
	}
	bizChangefeeds = pie.Filter(bizChangefeeds, func(cf entity.BizChangefeed) bool {
		if cf.BizPLEndpointID != 0 {
			return true
		}
		return cf.NetworkType == entity.DownstreamNetworkTypeManuallyPrivateLink ||
			cf.NetworkType == entity.DownstreamNetworkTypeManuallyPrivateServiceConnect
	})
	bizCfWrappers := pie.Map(bizChangefeeds, func(cf entity.BizChangefeed) changefeedWrapper {
		return changefeedWrapper{
			ID:              cf.ID.Uint64(),
			BizPLEndpointID: cf.BizPLEndpointID.Uint64(),
			TenantID:        cf.TenantID.Uint64(),
			ProjectID:       cf.ProjectID.Uint64(),
			ClusterID:       cf.ClusterID.Uint64(),
		}
	})
	return bizCfWrappers, nil
}