package provider

import (
	"context"
	"fmt"
	"github.com/pingcap/log"
	"github.com/pingcap/tep/pkg/apis/apiserver"
	"github.com/pingcap/tep/pkg/metrics/client"
	pmodel "github.com/prometheus/common/model"
	"github.com/wushilin/stream"
	"go.uber.org/zap"
	apierr "k8s.io/apimachinery/pkg/api/errors"
	"time"
)

type PrometheusProvider struct {
	promClient      client.Client
	metricConverter MetricConverter
}

func (p *PrometheusProvider) GetMetric(query string, info MetricInfo) (*apiserver.MetricValueList, error) {
	// Here is where we're making the query, need to be before here xD
	queryResults, err := p.promClient.Query(context.TODO(), pmodel.Now(), client.Selector(query))

	if err != nil {
		log.Error("unable to fetch metrics from prometheus: %v", zap.Error(err))
		// don't leak implementation details to the user
		return nil, apierr.NewInternalError(fmt.Errorf("unable to fetch metrics"))
	}
	list, err := p.metricConverter.Convert(info, queryResults)
	if err != nil {
		return nil, apierr.NewInternalError(err)
	}
	return list, err
}

func (p *PrometheusProvider) Series(match string, startTimeInSec, endTimeInSec int64, info MetricInfo) (*apiserver.MetricValueList, error) {
	interval := pmodel.Interval{
		Start: pmodel.TimeFromUnix(startTimeInSec),
		End:   pmodel.TimeFromUnix(endTimeInSec),
	}
	series, err := p.promClient.Series(context.TODO(), interval, client.Selector(match))
	if err != nil {
		log.Error("unable to fetch series from prometheus", zap.Error(err))
		return nil, apierr.NewInternalError(fmt.Errorf("unable to fetch series"))
	}

	list := &apiserver.MetricValueList{}
	stream.FromArray(series).Map(func(s client.Series) apiserver.MetricValue {
		ml := make(map[string]string)
		for key, value := range s.Labels {
			ml[string(key)] = string(value)
		}
		return apiserver.MetricValue{
			MetricLabels: ml,
		}
	}).Each(func(value apiserver.MetricValue) {
		list.Items = append(list.Items, value)
	})

	return list, nil
}

func (p *PrometheusProvider) GetMetricSeries(query string, startTimeInSec, endTimeInSec int64, info MetricInfo) (*apiserver.MetricValueList, error) {
	// Here is where we're making the query, need to be before here xD
	queryResults, err := p.promClient.QueryRange(context.TODO(), client.Range{
		Start: pmodel.TimeFromUnix(startTimeInSec),
		End:   pmodel.TimeFromUnix(endTimeInSec),
		Step:  time.Duration(info.StepInSec) * time.Second,
	}, client.Selector(query))

	if err != nil {
		log.Error("unable to fetch metrics from prometheus", zap.Error(err))
		// don't leak implementation details to the user
		return nil, apierr.NewInternalError(fmt.Errorf("unable to fetch metrics"))
	}
	list, err := p.metricConverter.Convert(info, queryResults)
	if err != nil {
		return nil, apierr.NewInternalError(err)
	}
	return list, err
}

// NewPrometheusProvider creates an ExternalMetricsProvider capable of responding to Kubernetes requests for external metric data
func NewPrometheusProvider(promClient client.Client) *PrometheusProvider {
	metricConverter := NewMetricConverter()
	return &PrometheusProvider{
		promClient:      promClient,
		metricConverter: metricConverter,
	}
}

// GetPrometheusDNSName return prometheus url based namespace and name
func GetTiDBPrometheusDNSName(namespace string, name string) string {
	return GetPromDNSName(namespace, name+"-prometheus")
}

func GetPromDNSName(namespace string, name string) string {
	return fmt.Sprintf("http://%s.%s.svc:9090", name, namespace)
}