package scalebench

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"net/url"
	"strconv"
	"strings"
	"time"

	. "github.com/onsi/gomega"
	"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/workloadnode/tiup"
	"github.com/pingcap/endless/testcase/oltp/oltppkg"
	"github.com/pingcap/endless/testcase/oltp/systemtest/scale"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/core"
	"github.com/pingcap/test-infra/sdk/resource"
)

func ConnectDb(dbHost string, dbPort string) (*sql.DB, error) {
	dbURL, err := url.Parse(fmt.Sprintf("mysql://root:@%s:%s", dbHost, dbPort))
	if err != nil {
		return nil, err
	}
	dsn := resource.BuildDSN(dbURL, "")
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		return nil, err
	}
	return db, nil
}
func ImportData(ctx context.Context, tiupCluster *tiup.TiUPCluster, br resource.BR, caseCmdParms oltppkg.CaseCmdParms) error {
	pdURL := Try(tiupCluster.ServiceURL(resource.PDAddr)).(*url.URL)
	db := Try(tiupCluster.GetDB("")).(*sql.DB)
	defer db.Close()
	tidbClient := tidb.NewClient(db)

	s3Parms := oltppkg.S3Parms{S3Endpoint: "http://minio.pingcap.net:9000",
		S3AccessKeyID: "minioadmin", S3SecretAccessKey: "minioadmin"}
	brcmd := oltppkg.NewBr("/bin", "restore full", s3Parms.S3AccessKeyID, s3Parms.S3SecretAccessKey)
	brTpccCmd := brcmd.Pd(pdURL.String()).Storage("s3://benchmark/sysbench-32-10000000").
		S3endpoint(s3Parms.S3Endpoint).CheckReq(false).Build()
	exist := Try(tidbClient.IsTableExist(ctx, "test", "sbtest1")).(bool)
	if !exist {
		oltppkg.ExecWorkloadCmd(ctx, br, brTpccCmd, true, nil)
	}

	brTpccCmd = brcmd.Pd(pdURL.String()).Storage("s3://benchmark/tpcc-1000-4.0").
		S3endpoint(s3Parms.S3Endpoint).CheckReq(false).Build()
	exist = Try(tidbClient.IsTableExist(ctx, "tpcc", "new_order")).(bool)
	if !exist {
		oltppkg.ExecWorkloadCmd(ctx, br, brTpccCmd, true, nil)
	}
	return nil
}
func RunSysbenchWorkload(ctx core.TestContext, tiupCluster tiup.TiUPCluster, benchtoolset resource.WorkloadNode, caseCmdParms oltppkg.CaseCmdParms) {
	dbURL := Try(tiupCluster.ServiceURL(resource.DBAddr)).(*url.URL)
	dbName := "test"
	sysbench := oltppkg.NewSysbench("tpcc", "run")
	sysbenchcmd := sysbench.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Threads(64).
		Tables(32).TableSize(10000000).IgnoreError("2013,1213,1105,1205,8022,8027,8028,9004,9007")
	runcmd := sysbenchcmd.Build()
	oltppkg.ExecWorkloadCmd(ctx, benchtoolset, runcmd, false, nil)
}

func RunTpccWorkload(ctx context.Context, tiupCluster tiup.TiUPCluster, benchtoolset resource.WorkloadNode) {
	dbName := "tpcc"
	warehouse := 1000
	threads := 128
	tpcc := oltppkg.NewTpcc("run")
	dbURL, _ := tiupCluster.ServiceURL(resource.DBAddr)
	tpcccmd := tpcc.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Warehouses(warehouse).Threads(threads).Time("36000m").IgnoreError(IgnoreErrors)
	oltppkg.ExecWorkloadCmd(ctx, benchtoolset, tpcccmd.Build(), true, nil)
}

func ScaleInCluster(ctx context.Context, tiupCluster tiup.TiUPCluster, timeOut time.Duration) {
	Try(tiupCluster.ScaleIn("tikv2-peer:20160,tikv4-peer:20160,tikv6-peer:20160"))
	promURL := Try(tiupCluster.ServiceURL(resource.Prometheus)).(*url.URL)
	scale.WaitForTombstone(promURL, 3, 2*time.Minute, timeOut)
	// cmd := "tiup cluster prune tidb-test -y"
	// workloadCmd := workloadnode.CommandContext(ctx, cmd)
	// Try(workloadCmd.Run(controlerNode))
}

func ScaleOutCluster(ctx context.Context, tiupCluster tiup.TiUPCluster, timeOut time.Duration, withTidb bool, tidbTopo string) {
	Try(tiupCluster.ScaleOut(ScaleOutTopo))
	if withTidb {
		Try(tiupCluster.ScaleOut(tidbTopo))
	}
	WaitForBalance(ctx, tiupCluster, 0.1, 5*time.Minute, timeOut)
}

func WaitForBalance(ctx context.Context, tiupCluster tiup.TiUPCluster, threshhold float64, pollingInterval time.Duration, timeOut time.Duration) {
	Eventually(&tiupCluster, timeOut, pollingInterval).ShouldNot(SatisfyAny(
		StoreScoreDifferentExceeds(threshhold).WithCtx(ctx),
	))
}

func BuildWorkloadRuncmd(ctx context.Context, tc tiup.TiUPCluster,
	benchToolset resource.WorkloadNode, caseCmdParms *oltppkg.CaseCmdParms) string {
	timestamp := strconv.FormatInt(time.Now().UnixNano()/1e6, 10)
	// var runcmd string
	dbURL, _ := tc.ServiceURL(resource.DBAddr)
	db := Try(tc.GetDB("")).(*sql.DB)
	defer db.Close()
	var (
		runcmd       string
		dbName       string
		threads      int
		tableNum     int
		tableSize    int
		warehouse    int
		ignoreErrors string
	)
	workload := caseCmdParms.WorkloadType.Type[0]
	if strings.HasPrefix(workload, "tpcc") {
		dbName = caseCmdParms.TpccParms.Db
		threads = caseCmdParms.TpccParms.Threads
		warehouse = caseCmdParms.TpccParms.Warehouse
		ignoreErrors = caseCmdParms.TpccParms.IgnoreErrors
	} else if strings.HasPrefix(workload, "sysbench") {
		dbName = caseCmdParms.SysbenchParms.Db
		threads = caseCmdParms.SysbenchParms.Threads
		tableNum = caseCmdParms.SysbenchParms.TableNum
		tableSize = caseCmdParms.SysbenchParms.TableSize
		ignoreErrors = caseCmdParms.SysbenchParms.IgnoreErrors
	} else if strings.HasPrefix(workload, "ycsb") {
	} else {
		err := errors.New("invalid workloadType")
		Expect(err).Should(Succeed())
	}
	switch {
	case workload == "ycsb":
		runcmd = BuildWorkloadYcsb(ctx, tc, *caseCmdParms)
	case workload == "sysbench_prepare":
		caseCmdParms.SysbenchParms.Db = fmt.Sprintf("%s_%s", "sysbenchprepare", timestamp)
		dbName = caseCmdParms.SysbenchParms.Db
		Try(RecreateDatabase(ctx, tc, dbName))
		sysbench := oltppkg.NewSysbench("oltp_insert", "prepare")
		sysbenchcmd := sysbench.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Threads(threads).
			Tables(tableNum).TableSize(tableSize).IgnoreError(ignoreErrors)
		runcmd = sysbenchcmd.Build()
	case strings.HasPrefix(workload, "sysbench_"):
		sysbench := oltppkg.NewSysbench(workload[9:], "run")
		sysbenchcmd := sysbench.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Threads(threads).
			Tables(tableNum).TableSize(tableSize).IgnoreError(ignoreErrors)
		runcmd = sysbenchcmd.Build()
	case workload == "tpcc_prepare":
		caseCmdParms.TpccParms.Db = fmt.Sprintf("%s_%s", "tpccprepare", timestamp)
		dbName = caseCmdParms.TpccParms.Db
		Try(RecreateDatabase(ctx, tc, dbName))
		tpcc := oltppkg.NewTpcc("prepare")
		tpcccmd := tpcc.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Warehouses(warehouse).Threads(threads).Time("36000m").IgnoreError(ignoreErrors)
		runcmd = tpcccmd.Build()
	case workload == "tpcc":
		tpcc := oltppkg.NewTpcc("run")
		tpcccmd := tpcc.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Warehouses(warehouse).Threads(threads).Time("36000m").IgnoreError(ignoreErrors)
		runcmd = tpcccmd.Build()
	default:
		tpcc := oltppkg.NewTpcc("run")
		tpcccmd := tpcc.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Warehouses(warehouse).Threads(threads).Time("36000m").IgnoreError(ignoreErrors)
		runcmd = tpcccmd.Build()
	}
	return runcmd
}

func RecreateDatabase(ctx context.Context, tc tiup.TiUPCluster, dbname string) error {
	dbUrl := Try(tc.ServiceURL(resource.DBAddr)).(*url.URL)
	db, err := sql.Open("mysql", resource.BuildDSN(dbUrl, ""))
	if err != nil {
		log.Warn("Fail to connect to db.")
		return err
	}
	log.Info(fmt.Sprintf("drop database if exists %s", dbname))
	_, err = db.ExecContext(ctx, fmt.Sprintf("drop database if exists %s", dbname))
	if err != nil {
		return err
	}
	log.Info(fmt.Sprintf("create database %s", dbname))
	_, err = db.ExecContext(ctx, fmt.Sprintf("create database %s", dbname))
	if err != nil {
		return err
	}
	return nil
}
func BuildWorkloadYcsb(ctx context.Context, tc tiup.TiUPCluster, caseCmdParms oltppkg.CaseCmdParms) string {
	dbURL, _ := tc.ServiceURL(resource.DBAddr)
	ycsbCmdTpl := `go-ycsb run mysql -P /ycsb/workloads/%s -p mysql.host=%s -p mysql.port=%s -p recordcount=%d -p operationcount=%d -p workload=core -p fieldcount=10 -p fieldlength=100 -p threadcount=%d -p requestdistribution=zipfian`
	cmd := fmt.Sprintf(ycsbCmdTpl,
		caseCmdParms.YcsbParms.SubTypes[0],
		dbURL.Hostname(),
		dbURL.Port(),
		caseCmdParms.YcsbParms.RecordCount,
		caseCmdParms.YcsbParms.OperationCount,
		caseCmdParms.YcsbParms.ThreadCount,
	)
	return cmd
}