package metrics

import (
	"context"
	"fmt"
	"github.com/pingcap/log"
	"github.com/pingcap/tep/pkg/apis/apiserver"
	"github.com/pingcap/tep/pkg/metrics/provider"
	"github.com/pingcap/tep/pkg/operator"
	store "github.com/pingcap/tep/pkg/registry/apiserver"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/wushilin/stream"
	"go.uber.org/zap"
	k8serrors "k8s.io/apimachinery/pkg/api/errors"
	metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"sort"
	"strings"
)

var (
	NetInUsage       = "sum(irate(container_network_receive_bytes_total{namespace=~'$$REPLACE_NAMESPACE$$'}[1m])) by (namespace)"
	NetOutUsage      = "sum(irate(container_network_transmit_bytes_total{namespace=~'$$REPLACE_NAMESPACE$$'}[1m])) by (namespace)"
	TopResourceUsage = map[string]string{
		"CPU":        TiDBClusterCPUQuery,
		"MEMORY":     fmt.Sprintf("%s/%s", TiDBClusterMemQuery, TiDBClusterMemLimit),
		"NETWORK_RX": NetInUsage,
		"NETWORK_TX": NetOutUsage,
	}

	QPSQuery = "sum(rate(tidb_server_query_total[1m]))"
	TPSQuery = "sum(rate(tidb_session_transaction_total[1m]))"

	TopQuery = map[string]string{
		"TPS":  TPSQuery,
		"QPS":  QPSQuery,
		"DISK": fmt.Sprintf("%s/%s", TiDBClusterDiskQuery, TiDBClusterDiskLimit),
	}
)

type TopInstanceStorage struct {
	*store.Store
}

func NewTopInstanceStorage(store *store.Store) *TopInstanceStorage {
	return &TopInstanceStorage{
		Store: store,
	}
}

func (s *TopInstanceStorage) Kind() string {
	return "TopInstances"
}

func (s *TopInstanceStorage) New() runtime.Object {
	return &apiserver.MetricValue{}
}

func (s *TopInstanceStorage) NewList() runtime.Object {
	return &apiserver.MetricValueList{}
}

func (s *TopInstanceStorage) NamespaceScoped() bool {
	return false
}

func (s *TopInstanceStorage) NewExtraListOptions() runtime.Object {
	return &apiserver.TopInstanceMetricsListOptions{}
}

func (s *TopInstanceStorage) List(ctx context.Context, options *metainternalversion.ListOptions, extraOptions runtime.Object) (runtime.Object, error) {
	o := extraOptions.(*apiserver.TopInstanceMetricsListOptions)
	if o.Top <= 0 {
		o.Top = 5
	}

	queryType := strings.ToUpper(o.Type)
	query, exist := TopResourceUsage[queryType]
	if exist {
		instances, err := s.buildInstances()
		if err != nil {
			return nil, err
		}

		var queryErr error
		dataSet := SampleIncreaseRatioList{}
		valueList := &apiserver.MetricValueList{}

		stream.FromArray(instances).Map(func(i instance) string {
			return i.namespace
		}).Map(func(ns string) []apiserver.MetricValue {
			r, err := s.K8SMetricsProvider.GetMetricSeries(strings.ReplaceAll(query, "$$REPLACE_NAMESPACE$$", ns), o.StartTimeInSec, o.EndTimeInSec, provider.MetricInfo{
				Metric:       "query_range",
				StepInSec:    o.Step,
				AppendLabels: map[string]string{"__name__": fmt.Sprintf("top_%s", strings.ToLower(o.Type)), "instance": ns},
			})

			if err != nil {
				log.Warn("query data failed", zap.Error(err))
				queryErr = err
				return nil
			}

			valueList.TypeMeta = r.TypeMeta
			valueList.ListMeta = r.ListMeta
			return r.Items

		}).Filter(func(items []apiserver.MetricValue) bool {
			return items != nil && len(items) > 0
		}).Each(func(items []apiserver.MetricValue) {
			dataSet = append(dataSet, newSampleIncreaseRatio(items[0]))
		})

		if len(dataSet) == 0 && queryErr != nil {
			return nil, queryErr
		}

		sort.Sort(sort.Reverse(dataSet))

		items := make([]apiserver.MetricValue, 0)
		for i := 0; i < len(dataSet) && i < o.Top; i++ {
			items = append(items, dataSet[i].Metric)
		}

		valueList.Items = items
		return valueList, nil
	}

	query, exist = TopQuery[queryType]
	if !exist {
		return nil, k8serrors.NewBadRequest("Unknown type - " + o.Type)
	}

	return s.topInstanceQuery(query, o)
}

func (s *TopInstanceStorage) NeedParameter() bool {
	return true
}

func (s *TopInstanceStorage) buildInstances() ([]instance, error) {
	opts := metav1.ListOptions{}
	opts.LabelSelector = fmt.Sprintf("%s=%s", operator.CreatedByLabelKey, operator.CreatedByLabelVal)

	tc, err := s.Controller().GetTidbClusters("", opts)
	if err != nil {
		return nil, k8serrors.NewInternalError(err)
	}

	instances := make([]instance, len(tc.Items))
	stream.FromArray(tc.Items).Map(func(tc v1alpha1.TidbCluster) instance {
		return instance{
			name:      tc.Name,
			namespace: tc.Namespace,
		}
	}).CollectTo(instances)

	return instances, nil
}

func sortUsage(result *apiserver.MetricValueList) SampleIncreaseRatioList {
	usageList := make(SampleIncreaseRatioList, len(result.Items))
	stream.FromArray(result.Items).Map(func(value apiserver.MetricValue) *SampleIncreaseRatio {
		return newSampleIncreaseRatio(value)
	}).CollectTo(usageList)

	sort.Sort(sort.Reverse(usageList))

	return usageList
}

func (s *TopInstanceStorage) topInstanceQuery(query string, o *apiserver.TopInstanceMetricsListOptions) (*apiserver.MetricValueList, error) {
	list, err := s.buildInstances()
	if err != nil {
		return nil, err
	}

	var queryErr error
	dataSet := SampleIncreaseRatioList{}
	valueList := &apiserver.MetricValueList{}

	context := ""
	stream.FromArray(list).Map(func(i instance) *provider.PrometheusProvider {
		p, err := store.NewMetricsProvider(provider.GetTiDBPrometheusDNSName(i.namespace, i.name))
		if err != nil {
			log.Warn("create provider failed", zap.Error(err))
			queryErr = err
			return nil
		}

		context = i.namespace
		return p
	}).Filter(func(p *provider.PrometheusProvider) bool {
		return p != nil
	}).Map(func(p *provider.PrometheusProvider) []apiserver.MetricValue {
		r, err := p.GetMetricSeries(query, o.StartTimeInSec, o.EndTimeInSec, provider.MetricInfo{
			Metric:       "query_range",
			StepInSec:    o.Step,
			AppendLabels: map[string]string{"instance": context, "__name__": fmt.Sprintf("top_%s", strings.ToLower(o.Type))},
		})
		if err != nil {
			log.Warn("query data failed", zap.Error(err))
			queryErr = err
			return nil
		}

		valueList.TypeMeta = r.TypeMeta
		valueList.ListMeta = r.ListMeta
		return r.Items
	}).Filter(func(items []apiserver.MetricValue) bool {
		return items != nil && len(items) > 0
	}).Each(func(items []apiserver.MetricValue) {
		dataSet = append(dataSet, newSampleIncreaseRatio(items[0]))
	})

	if len(dataSet) == 0 && queryErr != nil {
		return nil, queryErr
	}

	sort.Sort(sort.Reverse(dataSet))

	items := make([]apiserver.MetricValue, 0)
	for i := 0; i < len(dataSet) && i < o.Top; i++ {
		items = append(items, dataSet[i].Metric)
	}

	valueList.Items = items
	return valueList, nil
}

type instance struct {
	name      string
	namespace string
}

// SampleIncreaseRatio is the increase ratio of metric value from the first sample
// to the last sample in the SampleStream
type SampleIncreaseRatio struct {
	Ratio  float64 `json:"ratio"`
	Metric apiserver.MetricValue
}

func newSampleIncreaseRatio(metric apiserver.MetricValue) (r *SampleIncreaseRatio) {
	values := metric.Values
	if len(values) <= 1 {
		return &SampleIncreaseRatio{
			Ratio:  0.0,
			Metric: metric,
		}
	}

	firstVal := float64(values[0].Value)
	lastVal := float64(values[len(values)-1].Value)

	if firstVal != 0.0 {
		return &SampleIncreaseRatio{
			Ratio:  (lastVal - firstVal) / firstVal,
			Metric: metric,
		}
	}

	if lastVal > 0.0 {
		return &SampleIncreaseRatio{
			Ratio:  1.0,
			Metric: metric,
		}
	}

	if lastVal < 0.0 {
		return &SampleIncreaseRatio{
			Ratio:  -1.0,
			Metric: metric,
		}
	}

	return &SampleIncreaseRatio{
		Ratio:  0.0,
		Metric: metric,
	}
}

type SampleIncreaseRatioList []*SampleIncreaseRatio

func (s SampleIncreaseRatioList) Len() int           { return len(s) }
func (s SampleIncreaseRatioList) Less(i, j int) bool { return s[i].Ratio < s[j].Ratio }
func (s SampleIncreaseRatioList) Swap(i, j int)      { s[i], s[j] = s[j], s[i] }