package set_config

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

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	"github.com/pingcap/endless/pkg/client/tidb"
	. "github.com/pingcap/endless/pkg/matcher"
	"github.com/pingcap/endless/testcase/tiflash/pkg"
	"github.com/pingcap/log"
	logsearch "github.com/pingcap/test-infra/sdk/pkg/logs"
	"github.com/pingcap/test-infra/sdk/resource"
	_ "github.com/pingcap/test-infra/sdk/resource/impl/k8s" // register k8s impl
)

var _ = Describe("set config test", func() {
	var (
		tc         resource.TiDBCluster
		tidbClient *tidb.Client
		db         *sql.DB
		duration   time.Duration
	)
	Describe("TiFlash", func() {
		Context("test set config", func() {
			BeforeEach(func() {
				tc = Try(suiteTestCtx.Resource("tc")).(resource.TiDBCluster)
				duration = Try(time.ParseDuration(suiteTestCtx.Param("duration", "1h"))).(time.Duration)
				db = Try(pkg.OpenDB(tc, "")).(*sql.DB)
				log.Info("create database and table")
				Try(db.Exec("drop database if exists config"))
				Try(db.Exec("create database config"))
				db = Try(pkg.OpenDB(tc, "config")).(*sql.DB)
				Try(db.Exec("create table config_0 (id int)"))
				tidbClient = tidb.NewClient(db)
				log.Info("alter tiflash replica")
				Try(tidbClient.AlterDatabaseSetTiFlashReplica("config", 1, ""))
				log.Info("insert data")
				Try(db.Exec("insert into config_0 values (1);"))
				Try(db.Exec("insert into config_0 select * from config_0;"))
				Try(db.Exec("insert into config_0 select * from config_0;"))
				Try(db.Exec("insert into config_0 select * from config_0;"))
				Try(db.Exec("insert into config_0 select * from config_0;"))
				Try(db.Exec("insert into config_0 select * from config_0;"))
				Try(db.Exec("insert into config_0 select * from config_0;"))
				Try(db.Exec("insert into config_0 select * from config_0;"))
				Try(db.Exec("insert into config_0 select * from config_0;"))
				Try(db.Exec("insert into config_0 select * from config_0;"))
				Try(db.Exec("insert into config_0 select * from config_0;"))
				Try(db.Exec("insert into config_0 select * from config_0;"))
				Try(db.Exec("insert into config_0 select * from config_0;"))
				Try(db.Exec("insert into config_0 select * from config_0;"))
				Try(db.Exec("insert into config_0 select * from config_0;"))
				Try(db.Exec("split table config_0 between (0) and (30000) regions 20;"))
				Try(db.Exec("insert into config_0 values (2);"))
				Eventually(func() bool {
					return Try(tidbClient.IsDBTiFlashReplicaAvailable("config")).(bool)
				}, 10*time.Minute, 3*time.Second).Should(BeTrue())
				Try(tidbClient.ShowClusterVersionInformation())
			})

			It("sql will success #set_config_test1#", func() {
				ctx, _ := context.WithTimeout(suiteTestCtx, duration)
				Try(setConfigTest(ctx, db, tc))
			})
		})
	})
})

func setConfigTest(ctx context.Context, db *sql.DB, tc resource.TiDBCluster) error {

	tiflashNodesNum := len(Try(tc.GetNodes(resource.TiFlash)).([]resource.TiDBComponentNode))

	mppResult, err := searchTargetTiFlashLog(ctx, tc, db, "set @@tidb_max_tiflash_threads=-1", "set context setting max_threads to -1", true, tiflashNodesNum)
	if err != nil {
		return err
	}

	notMppResult, err := searchTargetTiFlashLog(ctx, tc, db, "set @@tidb_max_tiflash_threads=-1", "set context setting max_threads to -1", false, tiflashNodesNum)
	if err != nil {
		return err
	}

	if mppResult != nil || notMppResult != nil {
		errMsg := "max_threads -1 should not be pushed to TiFlash"
		log.Error(errMsg)
		return errors.New(errMsg)
	} else {
		log.Info("set max_threads -1 success")
	}

	threads := []int{0, 1, 15, 100}
	for thread, _ := range threads {
		query := fmt.Sprintf("set @@tidb_max_tiflash_threads=%d", thread)
		logString := fmt.Sprintf("set context setting max_threads to %d", thread)
		mppResult, err := searchTargetTiFlashLog(ctx, tc, db, query, logString, true, tiflashNodesNum)
		if err != nil {
			return err
		}

		notMppResult, err := searchTargetTiFlashLog(ctx, tc, db, query, logString, false, tiflashNodesNum)
		if err != nil {
			return err
		}
		if len(mppResult) < tiflashNodesNum || len(notMppResult) < tiflashNodesNum {
			errMsg := fmt.Sprintf("max_threads %d not be pushed to all TiFlash nodes", thread)
			log.Error(errMsg)
			return errors.New(errMsg)
		} else {
			log.Info(fmt.Sprintf("set max_threads %d success", thread))
		}
	}

	return nil
}

func searchTargetTiFlashLog(ctx context.Context, tc resource.TiDBCluster, db *sql.DB, query string, logString string, allowMpp bool, tiflashNodesNum int) ([]string, error) {

	startTime := time.Now()
	// 10m is long enough to execute sql and do not limit the end time for search log.
	endTime := startTime.Add(10 * time.Minute)
	var fetchLog []string
	fetchLog = []string{""}
	logopt := resource.TiDBClusterSearchLogOptions{
		Target:    resource.TiFlash,
		Levels:    []logsearch.LogLevel{logsearch.LogLevelInfo},
		StartTime: startTime,
		EndTime:   endTime,
		Patterns:  fetchLog,
		Limit:     10,
	}

	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return nil, err
	}

	if _, err = tx.Exec("set @@tidb_isolation_read_engines='tiflash';"); err != nil {
		return nil, err
	}

	if _, err = tx.Exec(fmt.Sprintf("set @@tidb_allow_mpp=%t;", allowMpp)); err != nil {
		return nil, err
	}

	if _, err = tx.Exec(fmt.Sprintf("set @@tidb_enforce_mpp=%t;", allowMpp)); err != nil {
		return nil, err
	}

	logopt.Patterns = []string{logString}
	if _, err = tx.Exec(query); err != nil {
		return nil, err
	}
	time.Sleep(1 * time.Second)
	if _, _, err = pkg.Query(ctx, tx, "select * from config_0", false); err != nil {
		return nil, err
	}

	defer Try(tx.Commit())

	if logs, err := tc.SearchLog(ctx, logopt); err != nil {
		return nil, err
	} else {
		return logs, err
	}
}