package impl

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

	"github.com/tidbcloud/dataflow-service-ng/app/internal/infra_metrics"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	"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"
	"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"
)

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.ChangefeedKernelId)
	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.GetClusterID()),
			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.GetClusterID()),
			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.GetClusterID()),
			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,
	changefeedKernelID string,
) (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)).
				OrderDescByEndTS()
		})
	if err != nil {
		if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
			expr := fmt.Sprintf(changefeedTrafficFirstExprFormat, m.getMetricName(), 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(),
		changefeedKernelID, queryDurationBySeconds)
	return expr, *meteringRecord.EndTS, queryLatestTime, nil
}

func (r *changefeedTrafficMeterImpl) getMetricName() string {
	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.GetTenantID()),
		ClusterID:     entity.ID(req.PrimaryCluster.GetClusterID()),
		ChangefeedID:  entity.ID(req.ChangefeedId),
		Region:        req.PrimaryCluster.GetRegionName(),
		CloudProvider: entity.CloudProviderType(req.PrimaryCluster.GetCloudProvider()),
		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
}