package comp_test

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

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	"go.uber.org/zap"

	"github.com/pingcap/test-infra/sdk/resource"

	_ "github.com/go-sql-driver/mysql"
	_ "github.com/pingcap/endless/pkg/client/tidb"
)

type NopWriterCloser struct {
}

// Write is empty operation
func (*NopWriterCloser) Write(p []byte) (n int, err error) {
	return len(p), nil
}

// Close is empty operation
func (*NopWriterCloser) Close() error {
	return nil
}

const (
	metaDsn  = "root:@tcp(172.16.6.53:4000)/comp?parseTime=True"
	mysqlDsn = "root:Pingcap$%^456@tcp(172.16.4.90:53306)/new_data?"
)

func execCommand(workloadName string, cmd string) error {
	ctx := suiteTestCtx
	ctx.Logger().Info("workload exec command", zap.String("workload", workloadName), zap.String("cmd", cmd))

	wk := ctx.Resource(workloadName).(resource.WorkloadNode)
	c, err := wk.ExecAsync(&NopWriterCloser{}, &NopWriterCloser{}, resource.WorkloadNodeExecOptions{
		Command: cmd})
	Expect(err).Should(Succeed())

	e := c.Wait()
	if e.ExitCode != 0 {
		return fmt.Errorf("%d", e.ExitCode)
	}
	return nil
}

var _ = Describe("comp_upgrade", func() {
	Context("new cluster", func() {
		It("update config", func() {
			ctx := suiteTestCtx

			loadBalancer := ctx.Resource("lb").(resource.LoadBalancer)
			dbHost, port := loadBalancer.Endpoint()
			dbPort := fmt.Sprint(port)

			version := getVersion(fmt.Sprintf("root:@tcp(%s:%s)/test", dbHost, dbPort))
			if version >= "5.3.0" && version < "6.1.0" {
				config := `
new_collations_enabled_on_first_bootstrap = true
[prepared-plan-cache]
enabled=true`
				// 5.3.0 GA plan-cache
				// >= 6.1.0 use variable and the default is on
				tc := ctx.Resource("tc").(resource.TiDBCluster)
				err := tc.UpdateConfig(resource.TiDBClusterUpdateConfigOptions{Target: resource.TiDB, Config: config, Timeout: 20 * time.Minute})
				Expect(err).Should(Succeed())
			}
		})
	})
})

func getVersion(dsn string) string {
	db, err := sql.Open("mysql", dsn)
	Expect(err).Should(Succeed())
	defer db.Close()

	r, err := db.Query("select version()")
	Expect(err).Should(Succeed())

	r.Next()
	var version string
	Expect(r.Scan(&version)).Should(Succeed())
	r.Close()

	version = versionFromTiDB(version)
	return version
}

// ./comp run prepare
var _ = Describe("comp_prepare", func() {
	Context("new cluster", func() {
		It("create tables and insert data ", func() {
			ctx := suiteTestCtx

			loadBalancer := ctx.Resource("lb").(resource.LoadBalancer)
			dbHost, port := loadBalancer.Endpoint()
			dbPort := fmt.Sprint(port)

			version := getVersion(fmt.Sprintf("root:@tcp(%s:%s)/test", dbHost, dbPort))
			db, err := sql.Open("mysql", fmt.Sprintf("root:@tcp(%s:%s)/test", dbHost, dbPort))
			Expect(err).Should(Succeed())

			ctx.Logger().Info("version number", zap.String("version", version))
			if version >= "5.4.0" {
				// GA index-merge
				_, err := db.Exec("set global tidb_enable_index_merge = true;")
				Expect(err).Should(Succeed())
			}

			globalVars := ctx.Param("globalvars")
			if len(globalVars) != 0 {
				for _, item := range strings.Split(globalVars, "&") {
					_, err := db.Exec("set global " + item)
					Expect(err).Should(Succeed())
				}
			}
			db.Close()

			dsn := fmt.Sprintf("root:@tcp(%s:%s)/coms", dbHost, dbPort)
			workloadName := "comp_workload"
			fileDir := ctx.Param("filedir", "")
			if len(fileDir) == 0 {
				fileDir = strings.ReplaceAll(version, ".", "")
			}

			featureVars := ctx.Param("featurevars")
			featureMap := map[string]struct{}{}
			if len(featureVars) != 0 {
				for _, item := range strings.Split(featureVars, "&") {
					featureMap[item] = struct{}{}
				}
			}

			err = execCommand(workloadName, fmt.Sprintf("mkdir -p /data/sdf/comp/%s/", fileDir))
			Expect(err).Should(Succeed())

			cmd := fmt.Sprintf(`/comp --file-dir /data/sdf/comp/%s --log-level warn --meta-dsn "%s" --mysql-dsn "%s" --tidb-dsn "%s" --report-email "%s" --report-interval %s --threads 50 run prepare --db-type tidb`,
				fileDir, metaDsn,
				mysqlDsn, dsn,
				ctx.Param("reportemail", "chenpeng@pingcap.com"),
				ctx.Param("reportinterval", "6"))
			if _, ok := featureMap["tiflash"]; ok {
				cmd += " --tiflash"
			}
			cmd += fmt.Sprintf(" >> /data/sdf/comp/%s/init.log 2>&1", fileDir)
			ctx.Logger().Info("generate command", zap.String("cmd", cmd))
			err = execCommand(workloadName, cmd)
			Expect(err).Should(Succeed())

			if _, ok := featureMap["table_cache"]; ok {
				cmd := fmt.Sprintf(`/comp --file-dir /data/sdf/comp/%s --log-level warn --meta-dsn "%s" --mysql-dsn "%s" --tidb-dsn "%s" --report-email "%s" --report-interval %s --threads 50 run ddl "alter table %%s cache" >> /data/sdf/comp/%s-tablecache.log 2>&1`,
					fileDir, metaDsn,
					mysqlDsn, dsn,
					ctx.Param("reportemail", "chenpeng@pingcap.com"),
					ctx.Param("reportinterval", "6"), fileDir)
				err = execCommand(workloadName, cmd)
				Expect(err).Should(Succeed())
			}
		})
	})
})

// ./comp test case
var _ = Describe("comp_test", func() {
	Context("cluster with data", func() {
		It("test cases", func() {
			ctx := suiteTestCtx

			loadBalancer := ctx.Resource("lb").(resource.LoadBalancer)
			dbHost, port := loadBalancer.Endpoint()
			dbPort := fmt.Sprint(port)
			dsn := fmt.Sprintf("root:@tcp(%s:%s)/coms", dbHost, dbPort)

			workloadName := ctx.Param("workload")
			if len(workloadName) == 0 {
				workloadName = "comp_workload"
			}
			comp := ctx.Resource(workloadName).(resource.WorkloadNode)

			fileDir := ctx.Param("filedir")
			if len(fileDir) == 0 {
				version := getVersion(fmt.Sprintf("root:@tcp(%s:%s)/test", dbHost, dbPort))
				fileDir = strings.ReplaceAll(version, ".", "")
			}
			binaryProtocol := ctx.Param("binaryprotocol")
			if len(binaryProtocol) == 0 {
				binaryProtocol = "false"
			}
			labels := ctx.Param("caselabel")
			if len(labels) == 0 {
				panic("need caselabel")
			}
			labelDir := strings.ReplaceAll(labels, " ", "")
			labelDir = strings.ReplaceAll(labelDir, ",", "-")

			runclusterid := ctx.Param("runclusterid", "0")
			recordclusterid := ctx.Param("recordclusterid", "0")

			err := execCommand(workloadName, fmt.Sprintf("mkdir -p /data/sdf/comp/%s/", fileDir))
			Expect(err).Should(Succeed())

			cmd := fmt.Sprintf(`/comp --file-dir /data/sdf/comp/%s/%s --log-level warn --meta-dsn "%s" --mysql-dsn "%s" --tidb-dsn "%s" --report-email "%s" --report-interval %s --threads 300 test case --case-label "%s" --run-cluster-id %s --record-cluster-id %s --binary-protocol=%s >> /data/sdf/comp/%s/%stest.log 2>&1`,
				fileDir, labelDir, metaDsn,
				mysqlDsn, dsn,
				ctx.Param("reportemail", "chenpeng@pingcap.com"),
				ctx.Param("reportinterval", "6"),
				labels,
				runclusterid,
				recordclusterid,
				binaryProtocol,
				fileDir, labelDir)
			ctx.Logger().Info("generate command", zap.String("cmd", cmd))

			run := true
			for i := 1; run; i++ {
				ctx.Logger().Warn("run", zap.Int("times", i))
				c, err := comp.ExecAsync(&NopWriterCloser{}, &NopWriterCloser{}, resource.WorkloadNodeExecOptions{
					Command: cmd})
				if err != nil {
					ctx.Logger().Error("exec async error", zap.Error(err))
					time.Sleep(1 * time.Minute)
					continue
				}

				e := c.Wait()
				if e.ExitCode != 0 {
					ctx.Logger().Error("comp exited", zap.Int("exit code", e.ExitCode))
					time.Sleep(1 * time.Minute)
					continue
				}
				run = false
			}
		})
	})
})