package scale

import (
	"context"
	"database/sql"
	"fmt"
	"io"
	"strings"
	"time"

	. "github.com/onsi/gomega"
	pdclient "github.com/pingcap/endless/pkg/client/pd"
	"github.com/pingcap/endless/pkg/client/tidb"
	. "github.com/pingcap/endless/pkg/matcher"
	. "github.com/pingcap/endless/pkg/matcher/metrics"
	. "github.com/pingcap/endless/pkg/util"
	"github.com/pingcap/endless/testcase/oltp/oltppkg"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
)

func OLTP_SCALE_IN_FUN_001(
	ctx context.Context,
	tc resource.TiDBCluster,
	cmdpr *io.PipeReader,
	caseCmdParms oltppkg.CaseCmdParms,
) {
	runTime := 3 * time.Hour
	ctxAll, cancelAll := context.WithTimeout(ctx, runTime)
	defer cancelAll()

	//Start oom and output check before all.
	oltppkg.LogAndBy("Start oom and output check before all.")
	g0, ctxg0 := errgroup.WithContext(ctxAll)

	workloadCheckParms := oltppkg.WorkloadCheckParameters{
		IsCheck: true,
		//9001: PD server timeout
		//8027: Information schema is out of date: schema failed to update in 1 lease
		IgnoreErr: append(strings.Split(caseCmdParms.TpccParms.IgnoreErrors, ","), "8027", "9001", "fail", "invalid connection", "err dial tcp"),
	}
	oltppkg.FunCommonCheck(ctxg0, tc, g0, runTime, cmdpr, workloadCheckParms)

	//Init parameters and wait cluster to balance
	oltppkg.LogAndBy("Step 1: Set store limit and waiting cluster balance.")
	pd := Try(pdclient.NewPDClient(tc)).(*pdclient.Client)
	_, sumSize, err := oltppkg.GetStoreSizeByType(tc, "used_size")
	Expect(err).Should(BeNil())
	//check if > 2T
	if sumSize > 2*1024*1024 {
		Try(pd.SetAllStoreLimit(200))
	} else {
		Try(pd.SetAllStoreLimit(60))
	}

	//Try(pd.DeleteTombstone())
	defer pd.SetAllStoreLimit(15)

	bcthreshhold := 0.5
	oltppkg.LogAndBy("Wait for region size balance.")
	Eventually(tc, 30*time.Minute, 5*time.Minute).ShouldNot(StoreRegionSizeDifferentExceeds(bcthreshhold).WithCtx(ctxg0))

	db := Try(tc.GetDB("")).(*sql.DB)
	tidbClient := tidb.NewClient(db)
	initTidbNum := Try(tidbClient.GetInstanceNum("tidb")).(uint32)
	initPdNum := Try(tidbClient.GetInstanceNum("pd")).(uint32)
	initTikvNum := Try(tidbClient.GetInstanceNum("tikv")).(uint32)

	log.Info("initial number of instance", zap.Uint32("tidb-num", initTidbNum), zap.Uint32("pd-num", initPdNum), zap.Uint32("tikv-num", initTikvNum))

	targetNumber := map[string]uint32{
		"tikv": initTikvNum - 1,
		"tidb": initTidbNum - 1,
		"pd":   initPdNum - 1}

	oltppkg.LogAndBy("Step 2: Scale-in the cluster")
	startTime := time.Now().Add(10 * time.Minute)
	oltppkg.WaitFor(ctxg0, "Warm workload", 10*time.Minute)

	for comp, num := range targetNumber {
		ctx1, cancel1 := context.WithTimeout(ctxg0, runTime)
		g1, ctxg1 := errgroup.WithContext(ctx1)
		var qpsJitterCheckParms oltppkg.QpsJitterCheckParameters
		switch comp {
		case "tidb":
			qpsJitterCheckParms = oltppkg.QpsJitterCheckParameters{
				IsCheck:      true,
				Offset:       "2m",
				ToleratePerc: 0.8,
				TolerateDur:  120 * time.Second,
			}
		case "pd":
			qpsJitterCheckParms = oltppkg.QpsJitterCheckParameters{
				IsCheck:      true,
				Offset:       "2m",
				ToleratePerc: 0.8,
				TolerateDur:  120 * time.Second,
			}
		case "tikv":
			qpsJitterCheckParms = oltppkg.QpsJitterCheckParameters{
				IsCheck:      true,
				Offset:       "2m",
				ToleratePerc: 0.5,
				TolerateDur:  120 * time.Second,
			}
		}
		oltppkg.FunCommonCheck(ctxg1, tc, g1, runTime, nil, qpsJitterCheckParms)
		g1.Go(WithGinkgoRecover(func() {
			oltppkg.WaitFor(ctxg1, "Wait for workload running", 5*time.Minute)
			Try(ScaleClusterByNumber(ctxg1, tc, map[string]uint32{comp: num}, 4*time.Hour))
			cancel1()
		}))
		Expect(g1.Wait()).Should(BeNil())

		oltppkg.WaitFor(ctxg0, "Waiting for workload recover", 5*time.Minute)
		oltppkg.RepeatQpsDurCheckWithAbsTime(ctxg0, 5*time.Minute, startTime, tc, 0.8)
	}

	cancelAll()
	Expect(g0.Wait()).Should(BeNil())
}

func OLTP_SCALE_IN_FUN_002(
	ctx context.Context,
	tc resource.TiDBCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms,
) {
	runTimeParms := oltppkg.NewRunTimePara(2*time.Hour, 10*time.Minute, oltppkg.GetRandDuration(time.Minute, 10), 10*time.Minute)
	ctxAll, cancelAll := context.WithTimeout(ctx, runTimeParms.TotalRunTime)
	g0, ctxg0 := errgroup.WithContext(ctxAll)
	defer cancelAll()

	scaleInPercent := 0.5
	defer func() {
		Try(oltppkg.ExecPdCtl(tc, "store limit all 15"))
		Try(oltppkg.ExecPdCtl(tc, "store remove-tombstone"))
	}()

	InitForScale(ctxg0, g0, tc, workloadRunParms, caseCmdParms, runTimeParms)

	oltppkg.LogAndBy("Step 3: Scale in tikv parallel")
	g1, ctxg1 := errgroup.WithContext(ctxg0)
	g1.Go(WithGinkgoRecover(func() {
		log.Info(fmt.Sprintf("initial number of instance %v", GetComponentNum(tc)))
		ScaleInTiKVParallel(ctxg1, tc, float32(scaleInPercent), runTimeParms.TotalRunTime, true)
	}))
	Expect(g1.Wait()).Should(BeNil())
}
func OLTP_SCALE_IN_FUN_003(
	ctx context.Context,
	tc resource.TiDBCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms,
) {
	runTimeParms := oltppkg.NewRunTimePara(2*time.Hour, 10*time.Minute, oltppkg.GetRandDuration(time.Minute, 10), 60*time.Minute)
	ctxAll, cancelAll := context.WithTimeout(ctx, runTimeParms.TotalRunTime)
	defer cancelAll()
	g0, ctxg0 := errgroup.WithContext(ctxAll)

	scaleInPercent := 0.5
	defer func() {
		Try(oltppkg.ExecPdCtl(tc, "store limit all 15"))
		Try(oltppkg.ExecPdCtl(tc, "store remove-tombstone"))
	}()

	InitForScale(ctxg0, g0, tc, workloadRunParms, caseCmdParms, runTimeParms)

	oltppkg.LogAndBy("Step 3: Scale in tikv parallel")
	log.Info(fmt.Sprintf("initial number of instance %v", GetComponentNum(tc)))
	g1, ctxg1 := errgroup.WithContext(ctxg0)
	var delIDs []string
	g1.Go(WithGinkgoRecover(func() {
		delIDs = ScaleInTiKVParallel(ctxg1, tc, float32(scaleInPercent), runTimeParms.TotalRunTime, false)
	}))
	g1.Go(WithGinkgoRecover(func() {
		oltppkg.WaitFor(ctxg1, "waiting for deleting store", 5*time.Minute)
		for _, id := range delIDs {
			Try(oltppkg.ExecPdCtl(tc, fmt.Sprintf("store cancel-delete %s", id)))
		}
		oltppkg.WaitForBalance(ctxg1, tc, runTimeParms.RecoveryTime, 2*time.Minute, 0.05)
	}))
	Expect(g1.Wait()).Should(BeNil())
}