package impl

import (
	"context"
	"fmt"
	"math"
	"strings"
	"time"

	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/internal/infra_metrics"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/models/ticdc"
	"github.com/tidbcloud/dataflow-service/app/repo"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	"go.uber.org/zap"
)

const (
	queryMetricBeforeNowOffsetBySeconds = 5 * 60
	changefeedTrafficExprFormat         = "sum by (instance, changefeed)" +
		"(increase(%s{changefeed=\"%s\"}[%ds]))"
	changefeedTrafficFirstExprFormat = "sum by (instance, changefeed)(%s{changefeed=\"%s\"})"
	secondToMilliSecond              = 1000
	cdcSinkWriteMetricName           = "ticdc_sink_write_bytes_total"
	cdcSinkWriteMetricNameV65        = "ticdc_sinkv2_write_bytes_total"
)

type changefeedTrafficMeterImpl struct {
	transProvider               db.TransProvider
	infraMetrics                infra_metrics.InfraMetrics
	changefeedTrafficRecordRepo repo.KernelChangefeedTrafficRecordRepo
}

func NewChangefeedTrafficMeter(
	transProvider db.TransProvider,
	infraMetrics infra_metrics.InfraMetrics,
	changefeedTrafficRecordRepo repo.KernelChangefeedTrafficRecordRepo,
) ticdc.ChangefeedTrafficMeter {
	return &changefeedTrafficMeterImpl{
		transProvider:               transProvider,
		infraMetrics:                infraMetrics,
		changefeedTrafficRecordRepo: changefeedTrafficRecordRepo,
	}
}

func (m *changefeedTrafficMeterImpl) MeterChangefeedTraffic(
	ctx context.Context, req *domain.MeterChangefeedTrafficReq,
) (*domain.MeterChangefeedTrafficResp, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	logger.Info("start rg changefeed metering", zap.Uint64("changefeedId", req.ChangefeedId))

	expr, startTS, targetTime, err := m.generateQueryExpressionAndTargetTime(ctx, req.ChangefeedId, req.BizType,
		req.ChangefeedKernelId, req.PrimaryCluster)
	if err != nil {
		logger.Error("fail to generateQueryExpressionAndTargetTime", zap.Any("changefeedId", req.ChangefeedId),
			zap.String("expr", expr), zap.Error(err))
		return nil, err
	}
	rawRespBody, err := m.infraMetrics.QueryClusterMetrics(ctx, req.PrimaryCluster, expr, targetTime)
	if err != nil {
		logger.Error("get metrics failed", zap.Any("clusterId", req.PrimaryCluster.Id),
			zap.Any("changefeedId", req.ChangefeedId), zap.String("expr", expr), zap.Error(err))
		return nil, err
	}
	metrics, err := infra_metrics.DeserializeResponseToMetric(ctx, rawRespBody)
	if err != nil {
		logger.Error("fail to deserialize metrics", zap.Any("clusterId", req.PrimaryCluster.Id),
			zap.Any("changefeedId", req.ChangefeedId), zap.String("expr", expr))
		return nil, err
	}
	if len(metrics) == 0 {
		logger.Warn("get empty metric", zap.Any("clusterId", req.PrimaryCluster.Id),
			zap.Any("changefeedId", req.ChangefeedId), zap.String("expr", expr))
		return nil, nil
	}
	meterRecord, err := m.aggregatingAndMeteringChangefeed(ctx, startTS, metrics, req)
	if err != nil {
		return nil, err
	}
	resp := &domain.MeterChangefeedTrafficResp{
		MeterRecord: meterRecord,
	}
	return resp, nil
}

func (m *changefeedTrafficMeterImpl) generateQueryExpressionAndTargetTime(
	ctx context.Context,
	changefeedId uint64,
	kernelTrafficBizType entity.KernelTrafficBizType,
	changefeedKernelID string,
	cluster *core.Cluster,
) (string, time.Time, time.Time, bizerr.BizError) {
	queryLatestTime := time.Now().Add(time.Duration(-queryMetricBeforeNowOffsetBySeconds) * time.Second)
	meteringRecord, err := m.changefeedTrafficRecordRepo.Get(ctx,
		func(qs entity.KernelChangefeedTrafficRecordQuerySet) entity.KernelChangefeedTrafficRecordQuerySet {
			return qs.ChangefeedIDEq(entity.ID(changefeedId)).
				BusinessTypeEq(kernelTrafficBizType).
				OrderDescByEndTS()
		})
	if err != nil {
		if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
			expr := fmt.Sprintf(changefeedTrafficFirstExprFormat, m.getMetricName(cluster.Version), changefeedKernelID)
			return expr, queryLatestTime.Add(time.Duration(-5) * time.Second), queryLatestTime, nil
		} else {
			return "", time.Now(), time.Now(), err
		}
	}
	gapTimeBySeconds := queryLatestTime.Sub(*meteringRecord.EndTS).Seconds()
	queryDurationBySeconds := int64(math.Floor(gapTimeBySeconds))
	expr := fmt.Sprintf(changefeedTrafficExprFormat, m.getMetricName(cluster.Version),
		changefeedKernelID, queryDurationBySeconds)
	return expr, *meteringRecord.EndTS, queryLatestTime, nil
}

func (r *changefeedTrafficMeterImpl) getMetricName(version string) string {
	// for version 6.5.X, the metric is different.
	if strings.Contains(version, "6.5.") {
		return cdcSinkWriteMetricNameV65
	}
	return cdcSinkWriteMetricName
}

func (r *changefeedTrafficMeterImpl) aggregatingAndMeteringChangefeed(
	ctx context.Context,
	startTS time.Time,
	metrics []infra_metrics.Metric,
	req *domain.MeterChangefeedTrafficReq,
) (*entity.KernelChangefeedTrafficRecord, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	var total float64 = 0
	for _, metric := range metrics {
		total += float64(metric.Value.Value)
	}

	recordTime := time.Unix(int64(metrics[0].Value.Timestamp/secondToMilliSecond), 0)
	meteringRecord := &entity.KernelChangefeedTrafficRecord{
		TenantID:               entity.ID(req.PrimaryCluster.TenantId),
		ProjectID:              entity.ID(req.PrimaryCluster.ProjectId),
		ClusterID:              entity.ID(req.PrimaryCluster.Id),
		ChangefeedID:           entity.ID(req.ChangefeedId),
		Region:                 req.PrimaryCluster.RegionName,
		BusinessType:           req.BizType,
		SrcCloudProvider:       req.CloudProviderType,
		SecondaryCloudProvider: req.SecondaryCloudProviderType,
		SecondaryRegion:        req.SecondaryRegion,
		StartTS:                &startTS,
		EndTS:                  &recordTime,
		Traffic:                int64(total),
		Extension:              req.InfoMap,
	}
	if err := r.changefeedTrafficRecordRepo.Save(ctx, meteringRecord); err != nil {
		logger.Error("fail to save changefeed traffic records.", zap.Error(err))
		return nil, err
	}
	return meteringRecord, nil
}