package thanos

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

	"github.com/tidbcloud/metrics-integration/pipeline/internal/datasource/client"
	"go.opentelemetry.io/collector/pdata/pcommon"
	"go.opentelemetry.io/collector/pdata/pmetric"
	"go.uber.org/zap"
)

// Prepare the extra label apply functions.
// doNothing := func(pcommon.Map, client.LabelSet) {}
// applyErrTypeLabel := func(attrs pcommon.Map, ls client.LabelSet) {
// 	attrs.PutStr(metadata.A.ErrorType, ls.Type)
// }
// applySqlTypeLabel := func(attrs pcommon.Map, ls client.LabelSet) {
// 	attrs.PutStr(metadata.A.SQLType, ls.Type)
// }
// applySqlTypeLabelV2 := func(attrs pcommon.Map, ls client.LabelSet) {
// 	attrs.PutStr(metadata.A.SQLType, ls.SqlType)
// }
// applyResultLabel := func(attrs pcommon.Map, ls client.LabelSet) {
// 	attrs.PutStr(metadata.A.Result, ls.Result)
// }
// applyLabelsForTransaction := func(attrs pcommon.Map, ls client.LabelSet) {
// 	attrs.PutStr(metadata.A.TxnMode, ls.TxnMode)
// 	attrs.PutStr(metadata.A.ErrorType, ls.Type)
// }

type extraLabelFunc func(pcommon.Map, client.LabelSet)

func (q *centralThanosQuerier) queryThenWriteNumMetric(
	m pmetric.Metric,
	queryFunc func(ctx context.Context, clusterID uint64, metric string) (*client.Response, error),
	setNumFunc func(p pmetric.NumberDataPoint, v string) error,
	ctx context.Context,
	clusterID uint64,
	clusterName string,
	projectName string,
	t pcommon.Timestamp,
	extraLabelFunc extraLabelFunc) error {

	res, err := queryFunc(ctx, clusterID, m.Name())
	if err != nil || res.Status != "success" {
		return fmt.Errorf("failed to call queryFunc, err: %v, result: %+v", err, res)
	}

	for _, series := range res.Data.Result {
		// Parse results from the query request.
		labels, value := series.LabelSet, series.Value
		if len(value) != 2 {
			return fmt.Errorf("prometheus response failed: value(%+v)", value)
		}

		// No need to get pd metrics.
		if labels.Component == "pd" {
			continue
		}

		// Build labels.
		attrs := pcommon.NewMap()
		// attrs.PutStr(metadata.A.ProjectName, projectName)
		// attrs.PutStr(metadata.A.ClusterName, clusterName)
		// attrs.PutStr(metadata.A.ClusterID, strconv.FormatUint(clusterID, 10))
		// attrs.PutStr(metadata.A.Component, labels.Component)
		// attrs.PutStr(metadata.A.Instance, datasource.UnifyDBName(labels.Instance))
		extraLabelFunc(attrs, labels)

		// Write the data point and labels.
		var p pmetric.NumberDataPoint
		switch m.Type() {
		case pmetric.MetricTypeGauge:
			p = m.Gauge().DataPoints().AppendEmpty()
		case pmetric.MetricTypeSum:
			p = m.Sum().DataPoints().AppendEmpty()
		default:
			return fmt.Errorf("unsupported data type: %v", m.Type())
		}
		p.SetTimestamp(t)
		if err := setNumFunc(p, value[1].(string)); err != nil {
			return err
		}
		attrs.CopyTo(p.Attributes())
	}

	return nil
}

func (q *centralThanosQuerier) queryThenWriteDurationHistogramMetric(
	m pmetric.Metric,
	queryFunc func(ctx context.Context, clusterID uint64, metric string) (*client.Response, error),
	ctx context.Context,
	clusterID uint64,
	clusterName string,
	projectName string,
	t pcommon.Timestamp,
	extraLabelFunc extraLabelFunc) error {

	res, err := queryFunc(ctx, clusterID, m.Name())
	if err != nil || res.Status != "success" {
		return fmt.Errorf("failed to call queryFunc, err: %v, result: %+v", err, res)
	}

	histoMap := q.calculateBuckets(res.Data.Result)

	for desc, data := range histoMap {
		// No need to get pd metrics.
		if desc.Component == "pd" {
			continue
		}

		attrs := pcommon.NewMap()
		// attrs.PutStr(metadata.A.ProjectName, projectName)
		// attrs.PutStr(metadata.A.ClusterID, strconv.FormatUint(clusterID, 10))
		// attrs.PutStr(metadata.A.ClusterName, clusterName)
		// attrs.PutStr(metadata.A.Component, desc.Component)
		// attrs.PutStr(metadata.A.Instance, datasource.UnifyDBName(desc.Instance))
		extraLabelFunc(attrs, desc)

		p := m.Histogram().DataPoints().AppendEmpty()
		p.SetTimestamp(t)
		p.SetCount(data.count)
		p.SetSum(data.sum)
		//p.SetBucketCounts(pcommon.NewImmutableUInt64Slice(data.bucketCounts))
		//p.SetExplicitBounds(pcommon.NewImmutableFloat64Slice(durationBuckets))
		attrs.CopyTo(p.Attributes())
	}

	return nil
}

func setIntMetric(p pmetric.NumberDataPoint, v string) error {
	num, err := strconv.ParseInt(v, 10, 64)
	if err != nil {
		f64, err := strconv.ParseFloat(v, 64)
		if err != nil {
			return fmt.Errorf("failed to parse value: %v", v)
		}
		num = int64(f64)
	}
	p.SetIntValue(num)
	return nil
}

func setDoubleMetric(p pmetric.NumberDataPoint, v string) error {
	num, err := strconv.ParseFloat(v, 64)
	if err != nil {
		return fmt.Errorf("failed to parse value: %v", v)
	}
	p.SetDoubleValue(num)
	return nil
}

type histoData struct {
	t            time.Time
	sum          float64
	count        uint64
	bucketCounts []uint64
}

// calculateBuckets group raw series by histoDesc.
// It assumes results contains _sum, _count and _bucket series,
// And _bucket series contain the le label.
func (q *centralThanosQuerier) calculateBuckets(res []client.Result) map[client.LabelSet]*histoData {
	histoMap := make(map[client.LabelSet]*histoData)
	for _, series := range res {
		// For each series, build a LabelSet as the group key.
		desc := client.LabelSet{
			// KubernetesNamespace contains cluster id.
			KubernetesNamespace: series.KubernetesNamespace,
			Component:           series.Component,
			Instance:            series.Instance,
			SqlType:             series.SqlType,
			Type:                series.Type,
			InTxn:               series.InTxn,
			TxnMode:             series.TxnMode,
		}

		if _, ok := histoMap[desc]; !ok {
			histoMap[desc] = &histoData{
				sum:          0,
				count:        0,
				bucketCounts: make([]uint64, len(durationBuckets)+1),
			}
		}
		data := histoMap[desc]

		// Check series data point.
		if len(series.Value) != 2 {
			// Go to the next series.
			q.logger.Warn("calculateBuckets: prometheus response failed", zap.Any("value", series.Value))
			continue
		}

		// Reduce series to histoData.
		if strings.HasSuffix(series.MetricName, "_sum") {
			num, err := strconv.ParseFloat(series.Value[1].(string), 64)
			if err != nil {
				// Go to the next series.
				q.logger.Warn("calculateBuckets: parsing _sum failed", zap.Any("value", series.Value))
				continue
			}
			data.sum = num

			// always use `_sum` metric timestamp as the timestamp for the histogram
			tsFloat, _ := series.Value[0].(float64)
			nanos := int64(tsFloat-float64(int64(tsFloat))) * 1e9
			data.t = time.Unix(int64(tsFloat), nanos)

		} else if strings.HasSuffix(series.MetricName, "_count") {
			num, err := strconv.ParseFloat(series.Value[1].(string), 64)
			if err != nil {
				q.logger.Warn("calculateBuckets: parsing _count failed", zap.Any("value", series.Value))
				continue
			}
			data.count = uint64(num)
		} else if strings.HasSuffix(series.MetricName, "_bucket") {
			le, err := strconv.ParseFloat(series.Le, 64)
			if err != nil {
				q.logger.Warn("calculateBuckets: parsing _bucket le failed", zap.String("le", series.Le))
				continue
			}
			num, err := strconv.ParseFloat(series.Value[1].(string), 64)
			if err != nil {
				q.logger.Warn("calculateBuckets: parsing _bucket value failed", zap.Any("value", series.Value))
				continue
			}
			data.bucketCounts[findBucketIdx(le)] = uint64(num)
		}
	}

	// Caution: Buckets in prometheus are cumulative but buckets in otel are not!
	for _, data := range histoMap {
		// The first bucket is not cumulative.
		adjusted := make([]uint64, len(data.bucketCounts))
		adjusted[0] = data.bucketCounts[0]
		for i := 1; i < len(data.bucketCounts); i++ {
			adjusted[i] = data.bucketCounts[i] - data.bucketCounts[i-1]
		}
		data.bucketCounts = adjusted
	}

	return histoMap
}

// findBucketIdx find the position in bucketCounts for a given upper bound.
func findBucketIdx(le float64) int {
	for i, f := range durationBuckets {
		if le > f {
			continue
		}
		return i
	}
	return len(durationBuckets)
}

func mergeMetrics(mergeTarget pmetric.Metrics, mergeSource pmetric.Metrics) pmetric.Metrics {
	metricsMap := make(map[string]pmetric.MetricSlice)
	srms := mergeSource.ResourceMetrics()
	for i := 0; i < srms.Len(); i++ {
		rm := srms.At(i)
		attrs := rm.Resource().Attributes()
		_id, ok := attrs.Get("cluster_id")
		if !ok {
			continue
		}
		id := _id.AsString()
		// assume cluster_id for each ResourceMetrics is unique
		if rm.ScopeMetrics().Len() == 0 {
			continue
		}

		// assume only the scope at index 0 is used
		mtcs := rm.ScopeMetrics().At(0).Metrics()
		metricsMap[id] = mtcs
	}

	trms := mergeTarget.ResourceMetrics()
	for i := 0; i < trms.Len(); i++ {
		trm := trms.At(i)
		attrs := trm.Resource().Attributes()
		_id, ok := attrs.Get("cluster_id")
		if !ok {
			continue
		}
		id := _id.AsString()
		smtcs, ok := metricsMap[id]
		if !ok {
			continue
		}

		// assume only the scope at index 0 is used
		for j := 0; j < smtcs.Len(); j++ {
			mtc := smtcs.At(j)
			newMtc := trm.ScopeMetrics().At(0).Metrics().AppendEmpty()
			mtc.CopyTo(newMtc)
		}

	}
	return mergeTarget
}

func mergeMetricsToNewScope(mergeTarget pmetric.Metrics, mergeSource pmetric.Metrics) pmetric.Metrics {
	metricsMap := make(map[string]pmetric.ScopeMetrics)
	srms := mergeSource.ResourceMetrics()
	for i := 0; i < srms.Len(); i++ {
		rm := srms.At(i)
		attrs := rm.Resource().Attributes()
		_id, ok := attrs.Get("cluster_id")
		if !ok {
			continue
		}
		id := _id.AsString()
		// assume cluster_id for each ResourceMetrics is unique
		if rm.ScopeMetrics().Len() == 0 {
			continue
		}

		// assume only the scope at index 0 is used
		mtcs := rm.ScopeMetrics().At(0)
		metricsMap[id] = mtcs
	}

	trms := mergeTarget.ResourceMetrics()
	for i := 0; i < trms.Len(); i++ {
		trm := trms.At(i)
		attrs := trm.Resource().Attributes()
		_id, ok := attrs.Get("cluster_id")
		if !ok {
			continue
		}
		id := _id.AsString()
		smtcs, ok := metricsMap[id]
		if !ok {
			continue
		}

		newScopeMtcs := trm.ScopeMetrics().AppendEmpty()
		smtcs.CopyTo(newScopeMtcs)
	}

	return mergeTarget
}