package steps

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

	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/caselib/pkg/host"
	apiresource "github.com/pingcap/test-infra/common/model/resource"
	"github.com/pingcap/test-infra/sdk/core"
	"github.com/pingcap/test-infra/sdk/resource"
	v1 "github.com/prometheus/client_golang/api/prometheus/v1"
	"github.com/prometheus/common/model"
	"go.uber.org/zap"

	"github.com/montanaflynn/stats"
)

type initCDCComponents struct {
	upStream, downStream *apiresource.ResourceHandle
}

type TicdcSummaryLog  struct{
	CheckpointLag int
	SinkWriteRows int
}

type TidbSummaryLog  struct{
	QpsTotal int
	QpsWrite int
	P95Latency int
}

var TicdcMetrics TicdcSummaryLog
var TidbMetrics  TidbSummaryLog

func NewInitCDCComponents(source, sink *apiresource.ResourceHandle) Step {
	return &initCDCComponents{
		upStream:   source,
		downStream: sink,
	}
}

func (s *initCDCComponents) Execute(ctx context.Context) error {
	resolved, err := core.ResolveResourceHandle(s.upStream)
	if err != nil {
		return errors.Trace(err)
	}
	upstream := resolved.(resource.TiDBCluster)

	resolved, err = core.ResolveResourceHandle(s.downStream)
	if err != nil {
		return errors.Trace(err)
	}
	downstream := resolved.(resource.CDCSink)

	if err := GetInner(ctx).Board().SetTiDBCluster(upstream); err != nil {
		return errors.Trace(err)
	}

	if err := GetInner(ctx).Board().SetSink(downstream); err != nil {
		return errors.Trace(err)
	}

	return nil
}

func (s *initCDCComponents) String() string {
	return fmt.Sprintf("initCDCComponents: upstream=%+v, downstream=%+v", s.upStream, s.downStream)
}

type  removeCDCComponents struct {}

func NewRemoveCDCComponents() Step {
	return &removeCDCComponents{}
}

func (s *removeCDCComponents) Execute(ctx context.Context) error {
	GetInner(ctx).Board().RemoveTiDBCluster()
	GetInner(ctx).Board().RemoveSink()
	return nil
}

func (s *removeCDCComponents) String() string {
	return fmt.Sprintf("removeCDCComponents")
}

type getCDCVersion struct {
	source *apiresource.ResourceHandle
}

func (c *getCDCVersion) Execute(ctx context.Context) error {
	cdcHost, err := GetCDCHost(ctx)

	res, err := cdcHost.TiCDCVersion()
	if err != nil {
		return err
	}

	cdcVersion := res.Stdout[strings.Index(res.Stdout, ":")+2 : strings.Index(res.Stdout, "\n")]
	log.Info("get TiCDC version", zap.String("cdcVersion", cdcVersion))

	cdcHashCommitVersion := res.Stdout[strings.Index(res.Stdout, "Hash:")+6 : strings.Index(res.Stdout, "Git Branch")-1]
	log.Info("get TiCDC hash commit", zap.String("cdcHashCommitVersion", cdcHashCommitVersion))
	host.TestbedPara.TiCDCVersion = cdcVersion
	host.TestbedPara.TiCDCHashCommit = cdcHashCommitVersion

	resTiKV, err := cdcHost.TiKVVersion()
	if err != nil {
		return err
	}

	tikvVersion := resTiKV.Stdout[strings.Index(resTiKV.Stdout, ":")+4 : strings.Index(resTiKV.Stdout, "Edition:")-1]
	log.Info("get TiKV version", zap.String("tikvVersion", tikvVersion))
	tikvHashCommitVersion := resTiKV.Stdout[strings.Index(resTiKV.Stdout, "Hash:")+8 : strings.Index(resTiKV.Stdout, "Git Commit Branch")-1]
	log.Info("get TiKV hash commit", zap.String("tikvHashCommitVersion", tikvHashCommitVersion))
	host.TestbedPara.TiKVVersion = tikvVersion
	host.TestbedPara.TiKVHashCommit = tikvHashCommitVersion

	return nil
}

func (c *getCDCVersion) String() string {
	return fmt.Sprintf("cdc cluster: %s", *c.source)
}

func NewGetCDCVersion(source *apiresource.ResourceHandle) *getCDCVersion {
	return &getCDCVersion{
		source,
	}
}

type getCDCMetrics struct {
	source        *apiresource.ResourceHandle
	startTime	  *time.Time
}

func (c *getCDCMetrics) Execute(ctx context.Context) error {
	if err := c.recordMetrics(ctx, true, *c.startTime, time.Now()); err != nil {
		return errors.Trace(err)
	}
	log.Info("get cdc bench metrics")
	return nil
}

func (c *getCDCMetrics) String() string {
	return fmt.Sprintf("cdc metrics start from %s, end to %s", c.startTime, time.Now())
}

func NewGetCDCMetrics(source *apiresource.ResourceHandle, startTime *time.Time) *getCDCMetrics {
	return &getCDCMetrics{
		source,
		startTime,
	}
}

func recordMetric(ctx context.Context, api v1.API, query string, start, end time.Time) (int, error) {
	var values, _, err = api.QueryRange(ctx, query, v1.Range{
		Start: start,
		End:   end,
		Step:  time.Second * 30,
	})
	if err != nil {
		return 0, errors.Trace(err)
	}
	if err != nil {
		return 0, errors.Trace(err)
	}

	valMatrix := values.(model.Matrix)

	var inputs []float64
	for _, stream := range valMatrix {
		for _, sample := range stream.Values {
			inputs = append(inputs, float64(sample.Value))
		}
	}
	if len(inputs) == 0 {
		log.Info("no metrics value is returned")
		return 0, nil
	}
	// get max value during metrics range time
	max, err := stats.Max(inputs)
	if err != nil {
		return 0, err
	}
	return int(max), nil
}

func (s *getCDCMetrics) recordMetrics(ctx context.Context, cdcRunning bool, start, end time.Time) (error) {
	resolved, err := core.ResolveResourceHandle(s.source)
	if err != nil {
		return errors.Trace(err)
	}
	tc := resolved.(resource.TiDBCluster)
	prometheusAPI, err := tc.PrometheusAPI()
	if err != nil {
		return errors.Trace(err)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	defer cancel()

	TidbMetrics.QpsTotal, err = recordMetric(ctx, prometheusAPI, "sum(rate(tidb_executor_statement_total{}[1m])) by (type)", start, end)
	if err != nil {
		return errors.Trace(err)
	}
	TidbMetrics.QpsWrite, _ = recordMetric(ctx, prometheusAPI, "sum(rate(tidb_executor_statement_total{  type = 'insert|update|delete '}[1m])) by (type)", start, end)
	if err != nil {
		return errors.Trace(err)
	}
	TidbMetrics.P95Latency, _ = recordMetric(ctx, prometheusAPI, "histogram_quantile(0.95, sum(rate(tidb_server_handle_query_duration_seconds_bucket{}[1m])) by (le))", start, end)
	if err != nil {
		return errors.Trace(err)
	}

	TicdcMetrics.CheckpointLag, _ = recordMetric(ctx, prometheusAPI, "max(ticdc_owner_checkpoint_ts_lag) by (changefeed)", start, end)
	if err != nil {
		return errors.Trace(err)
	}
	TicdcMetrics.SinkWriteRows, _ = recordMetric(ctx, prometheusAPI, "sum(rate(ticdc_sink_txn_batch_size_sum{}[1m])) by (instance, changefeed)", start, end)
	if err != nil {
		return errors.Trace(err)
	}
	return  nil
}