package pdfun

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

	. "github.com/onsi/gomega"
	pdclient "github.com/pingcap/endless/pkg/client/pd"
	. "github.com/pingcap/endless/pkg/matcher"
	. "github.com/pingcap/endless/pkg/util"
	"github.com/pingcap/endless/pkg/workloadnode/cluster"
	"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_SPLIT_MERGE_FUN_001(
	ctx context.Context,
	tc cluster.BaseCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms,
) {
	oltppkg.LogAndBy("Split table")
	client := Try(pdclient.NewPDClient(tc)).(*pdclient.Client)
	Try(client.SetConfig("split-merge-interval", "10m"))
	Try(oltppkg.BasicClusterExecPdCtl(tc, "config set merge-schedule-limit 200"))

	defer func() {
		oltppkg.WaitForEmptyRegionMerge(ctx, tc, 2*time.Hour, 1*time.Minute, 100, 0.1)
		Try(oltppkg.BasicClusterExecPdCtl(tc, "config set merge-schedule-limit 8"))
		Try(client.SetConfig("split-merge-interval", "30m"))
	}()

	runTime := 1 * time.Hour
	ctxAll, cancelAll := context.WithTimeout(ctx, runTime)
	defer cancelAll()

	dbName := "dbsplit"
	tableNum := 100
	tableSize := 7000 * 10000
	oltppkg.RecreateDatabase(ctxAll, tc, dbName)
	oltppkg.CreateSysbenchTable(ctxAll, tc, dbName, tableNum, "", true)

	db := Try(tc.GetDB(dbName)).(*sql.DB)
	defer db.Close()

	for i := 1; i <= tableNum; i++ {
		tableName := fmt.Sprintf("sbtest%d", i)
		splitSqlTmp := "SPLIT TABLE %s BETWEEN (0) AND (1000000000) REGIONS 200"
		splitSql := fmt.Sprintf(splitSqlTmp, tableName)
		log.Info("Execute ", zap.String("splitSql", splitSql))
		_, err := db.ExecContext(ctxAll, splitSql)
		Expect(err).Should(BeNil())
		oltppkg.CheckRegionCountBalance(ctxAll, tc, dbName, tableName)
	}

	dbURL, _ := tc.ServiceURL(resource.DBAddr)
	sysbench := oltppkg.NewSysbench("oltp_insert", "run")
	sysbenchcmd := sysbench.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Threads(100).
		Tables(tableNum).Command("run").TableSize(tableSize).IgnoreError(caseCmdParms.SysbenchParms.IgnoreErrors)
	runcmd := sysbenchcmd.Build()

	g, ctxg := errgroup.WithContext(ctxAll)
	emptyRegionCheckParms := oltppkg.EmptyRegionCheckParameters{
		IsCheck:          false,
		RatioThreshhold:  0.8,
		NumberThreshhold: 10000,
	}
	oltppkg.FunCommonCheck(ctxg, tc, g, runTime, workloadRunParms.Cmdpr, emptyRegionCheckParms)

	g.Go(WithGinkgoRecover(func() {
		Try(oltppkg.ExecWorkloadCmd(ctxg, workloadRunParms.WorkloadNode, runcmd, false, resource.NopCloser(workloadRunParms.Cmdpw)))
		log.Info("run insert finished")
	}))
	g.Go(WithGinkgoRecover(func() {
		if strings.HasPrefix(caseCmdParms.WorkloadType.Type[0], "sysbench") {
			oltppkg.SplitTablesRepeatly(ctxg, tc, caseCmdParms.SysbenchParms.Db, 5*time.Minute)
		} else {
			oltppkg.SplitTablesRepeatly(ctxg, tc, caseCmdParms.TpccParms.Db, 5*time.Minute)
		}
		log.Info("split finished")
	}))

	//Increase imbalance check threshhold and ingore check for sysbench_select_random_points and sysbench_select_random_ranges
	//for  there is issue https://github.com/tikv/pd/issues/4529
	startTime := time.Now()
	oltppkg.WaitFor(ctxg, "Wait for warm up", 10*time.Minute)
	workloadtype := caseCmdParms.WorkloadType.Type[0]
	if workloadtype != "sysbench_select_random_points" && workloadtype != "sysbench_select_random_ranges" {
		oltppkg.WaitForWorkloadBalance(ctxg, tc, workloadtype, g, 2*time.Minute, 20*time.Minute, 2.0)
		oltppkg.LogAndBy("Monitor workload balance on all tikvs.")
		oltppkg.MonitorWorkloadBalance(ctxg, tc, workloadtype, g, runTime-time.Since(startTime), 2.0)
	}

	log.Info("wait err group finished")
	Expect(g.Wait()).Should(BeNil())
	log.Info("err group finished")

	for i := 1; i <= tableNum; i++ {
		tableName := fmt.Sprintf("sbtest%d", i)
		oltppkg.CheckRegionCountBalance(ctx, tc, dbName, tableName)
	}
	oltppkg.WaitForEmptyRegionMerge(ctx, tc, 1*time.Hour, 1*time.Minute, 100, 0.1)
}