package metrics

import (
	"bufio"
	"encoding/json"
	"fmt"
	"io"

	"github.com/pingcap/log"
	"github.com/prometheus/common/model"
	"github.com/prometheus/prometheus/prompb"
	"go.uber.org/zap"

	"github.com/pingcap-inc/clinic/pkg/metrics"
)

var (
	ClusterLabel = "tidb_cluster"
	HostLabel    = "hostname"

	SampleStreamStartToken  byte = '{'
	SampleStreamEndToken    byte = '}'
	SampleStreamsStartToken byte = '['
)

type metricIterator struct {
	clusterID string
	hostName  string
	packageID string

	reader *bufio.Reader
	path   string
	// len int
}

func (m *metricIterator) Next() (any, bool) {
	startBytes, err := m.next(SampleStreamStartToken)
	if err != nil || len(startBytes) == 0 {
		return nil, false
	}

	labelsBytes, err := m.next(SampleStreamEndToken)
	if err != nil || len(labelsBytes) == 0 {
		return nil, false
	}

	labels := make([]byte, 0)
	labels = append(labels, SampleStreamStartToken)
	labels = append(labels, labelsBytes...)

	values, err := m.next(SampleStreamEndToken)
	if err != nil || len(values) == 0 {
		return nil, false
	}

	tsBytes := append(labels, values...)

	var ss model.SampleStream
	err = json.Unmarshal(tsBytes, &ss)
	if err != nil {
		log.Error("MetricsIterator Unmarshal failed", zap.String("fileUUID", m.packageID), zap.String("path", m.path), zap.String("body", string(tsBytes)), zap.Error(err))
		return nil, false
	}

	ts := prompb.TimeSeries{
		Labels:  make([]prompb.Label, 0),
		Samples: make([]prompb.Sample, 0),
	}

	// dump labels
	for key, value := range ss.Metric {
		if string(key) == ClusterLabel {
			continue
		}
		ts.Labels = append(ts.Labels, prompb.Label{
			Name:  string(key),
			Value: string(value),
		})
	}

	// append label
	ts.Labels = append(ts.Labels, prompb.Label{
		Name:  ClusterLabel,
		Value: m.clusterID,
	})

	if m.hostName != "" {
		ts.Labels = append(ts.Labels, prompb.Label{
			Name:  HostLabel,
			Value: m.hostName,
		})
	}

	// dump samples
	for _, simple := range ss.Values {
		timestamp := simple.Timestamp
		if len(fmt.Sprintf("%d", timestamp)) == 10 {
			timestamp = timestamp * 1000
		}
		ts.Samples = append(ts.Samples, prompb.Sample{
			Value:     float64(simple.Value),
			Timestamp: int64(timestamp),
		})
	}
	metrics.SinkTimeSeriesDatapointCount.WithLabelValues(m.clusterID, m.packageID).Add(float64(len(ss.Values)))

	metrics.SinkTimeSeriesCount.WithLabelValues(m.clusterID, m.packageID).Inc()
	return ts, true
}

func (m *metricIterator) next(token byte) ([]byte, error) {
	bytes, err := m.reader.ReadBytes(token)
	if err != nil {
		if err == io.EOF {
			return nil, nil
		}
		log.Error("MetricsIterator read bytes failed", zap.String("fileUUID", m.packageID), zap.String("path", m.path), zap.Error(err))
		return nil, err
	}

	return bytes, nil
}

func (m *metricIterator) Len() (int, bool) {
	return 0, false
}

func NewMetricIterator(r io.Reader, path, clusterID, hostName, packageUUID string) (*metricIterator, error) {
	reader := bufio.NewReader(r)
	_, err := reader.ReadBytes(SampleStreamsStartToken)
	if err != nil {
		if err == io.EOF {
			return nil, nil
		}

		log.Error("read token failed", zap.Error(err))
		return nil, err
	}

	return &metricIterator{
		clusterID: clusterID,
		hostName:  hostName,
		packageID: packageUUID,
		reader:    reader,
		path:      path,
	}, nil
}