package tiupscale

import (
	"context"
	"net/url"
	"strings"
	"time"

	. "github.com/onsi/gomega"
	. "github.com/pingcap/endless/pkg/matcher"
	"github.com/pingcap/endless/pkg/workloadnode/tiup"
	"github.com/pingcap/endless/testcase/oltp/oltppkg"
	"github.com/pingcap/endless/testcase/oltp/systemtest/scale"

	//"github.com/pingcap/endless/testcase/oltp/tiuptest/tiuppkg"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/core"
	"github.com/pingcap/test-infra/sdk/resource"
	"golang.org/x/sync/errgroup"
)

type ScaleParameters struct {
	ScaleType       string
	ScaleOutTopo    string
	ScaleInList     []string
	QpsDropLessThan float64
	QpsCheckTime    time.Duration
}

func TiupScaleOutCluster(ctx context.Context, tc *tiup.TiUPCluster, timeOut time.Duration, scaleOutTopo string) {
	Try(tc.ScaleOut(scaleOutTopo))
	log.Info("Begion to scale out")
	promURL := Try(tc.ServiceURL(resource.Prometheus)).(*url.URL)
	scale.WaitForOnline(promURL, 0, 1*time.Minute, timeOut)
	oltppkg.WaitForBalance(ctx, tc, timeOut, 5*time.Minute, 0.1)
	log.Info("Finish scale out")
}
func TiupScaleIn(ctx context.Context, tc *tiup.TiUPCluster, timeOut time.Duration, scaleInList []string) {
	log.Info("Begin to scale in.")
	Try(tc.ScaleIn(strings.Join(scaleInList, ",")))
	promURL := Try(tc.ServiceURL(resource.Prometheus)).(*url.URL)
	scale.WaitForTombstone(promURL, 3, 1*time.Minute, timeOut)
	log.Info("Finish scale in")
}

func TiupRunScaleTest(ctx core.TestContext, tc *tiup.TiUPCluster, caseCmdParms oltppkg.CaseCmdParms, scaleParams ScaleParameters,
	workloadRunParms oltppkg.WorkloadRunParms) {

	scale.SetStoreLimitAccordingSize(ctx, tc)

	ctxAll, cancel := context.WithTimeout(ctx, caseCmdParms.RunTime.TotalRunTime)
	g, ctxg := errgroup.WithContext(ctxAll)

	oltppkg.FunCommonCheck(ctxg, tc, g, caseCmdParms.RunTime.TotalRunTime, workloadRunParms.Cmdpr)

	startTime := time.Now().Add(caseCmdParms.RunTime.WarmUpTime / 2)
	oltppkg.WaitFor(ctxg, "waiting for workload balance before scale", caseCmdParms.RunTime.WarmUpTime)
	if scaleParams.ScaleType == "in" {
		TiupScaleIn(ctxg, tc, caseCmdParms.RunTime.TotalRunTime, scaleParams.ScaleInList)
	} else {
		TiupScaleOutCluster(ctxg, tc, caseCmdParms.RunTime.TotalRunTime, scaleParams.ScaleOutTopo)
	}

	workloadRunParms.Pause(ctxg, 1*time.Minute)
	oltppkg.WaitFor(ctxg, "waiting for workload running after do scale", caseCmdParms.RunTime.RecoveryTime)
	cancel()
	Expect(g.Wait()).Should(BeNil())

	log.Info("begin to check the qps after do scale")
	oltppkg.RepeatQpsDurCheckWithAbsTime(ctx, scaleParams.QpsCheckTime, startTime, tc, scaleParams.QpsDropLessThan)

	// oltppkg.BasicClusterExecPdCtl(tc, "store store remove-tombstone")
	// oltppkg.CheckClusterHealth(tc)
}