package tiup_upgrade

import (
	"database/sql"
	"fmt"
	"net/url"
	"strconv"
	"time"

	_ "github.com/go-sql-driver/mysql"
	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	. "github.com/pingcap/endless/pkg/workloadnode/tiup"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/core"
	"github.com/pingcap/test-infra/sdk/resource"
	_ "github.com/pingcap/test-infra/sdk/resource/impl/k8s" // register k8s impl

	_ "github.com/pingcap/endless/pkg/client/tidb"
	. "github.com/pingcap/endless/pkg/matcher"
	"go.uber.org/zap"
)

type record struct {
	a int
	b int
}

var cdcconfig = `
[filter] 
rules = ['test.*']
`

func SET_CDC_TASK(
	mycluster *TiUPCluster,
) {
	pdURL, err := mycluster.ServiceURL(resource.PDAddr)
	must(err)
	log.Info("pd info", zap.String("url", pdURL.String()), zap.String("ip:", pdURL.Host), zap.String("port:", pdURL.Port()))
	pdurl := pdURL.String()
	// pdurl := "http://pd3-peer:22379"
	resultFile := "cdc-test.txt"
	cdcCmd := ctx.Resource("cdc").(resource.WorkloadNode)
	cmdstr := fmt.Sprintf(`echo "[filter]" > /tmp/changefeed.toml;echo "rules = ['test.*','cdctest.*']" >> /tmp/changefeed.toml;cat /tmp/changefeed.toml`)
	res, _, _, err := cdcCmd.Exec(resource.WorkloadNodeExecOptions{Command: cmdstr})
	must(err)
	log.Info("task status", zap.String("result", res))

	cdcstr := fmt.Sprintf(`mkdir -p /root/result/; /tiup/deploy/cdc-8300/bin/cdc cli changefeed list --pd=%s > /root/result/%s`, pdurl, resultFile)
	res, _, _, err = cdcCmd.Exec(resource.WorkloadNodeExecOptions{Command: cdcstr})
	must(err)
	log.Info("task status", zap.String("result", res))
	// Expect(res).Should(ContainSubstring("normal"))

	cdcstr = fmt.Sprintf(`/tiup/deploy/cdc-8300/bin/cdc cli changefeed create --pd=%s --sink-uri=tidb://root:@%s/ --changefeed-id=simple-cdc-task  --config=/tmp/changefeed.toml --no-confirm >> /root/result/%s`, pdurl, cluster2Dsn, resultFile)
	Try(cdcCmd.Exec(resource.WorkloadNodeExecOptions{Command: cdcstr}))

	res, _, _, err = cdcCmd.Exec(resource.WorkloadNodeExecOptions{Command: "ls -ltr /root/result/"}) // 直接用
	must(err)
	Expect(res).Should(ContainSubstring(resultFile))
	cmdstr = fmt.Sprintf(`cat /root/result/%s `, resultFile)
	res, _, _, err = cdcCmd.Exec(resource.WorkloadNodeExecOptions{Command: cmdstr}) // 直接用
	must(err)
	Expect(res).Should(ContainSubstring("Create changefeed successfully!"))
	Expect(res).Should(ContainSubstring("normal"))
}

func CHECK_TASK_STATUS(
	mycluster *TiUPCluster,
) {
	pdURL, err := mycluster.ServiceURL(resource.PDAddr)
	must(err)
	log.Info("pd info", zap.String("url", pdURL.String()), zap.String("ip:", pdURL.Host), zap.String("port:", pdURL.Port()))
	pdurl := pdURL.String()
	// pdurl := "http://pd3-peer:22379"
	cdcCmd := ctx.Resource("cdc").(resource.WorkloadNode)
	cdcstr := fmt.Sprintf(`/tiup/deploy/cdc-8300/bin/cdc cli changefeed list --pd=%s `, pdurl)
	res, _, _, err := cdcCmd.Exec(resource.WorkloadNodeExecOptions{Command: cdcstr})
	must(err)
	log.Info("task status", zap.String("result", res))
	Expect(res).Should(ContainSubstring("normal"))
	Expect(res).ShouldNot(ContainSubstring("Err"))
}

func SYNC_CDC_DATA(
	mycluster *TiUPCluster,
) {
	dbURL = Try(mycluster.ServiceURL(resource.DBAddr)).(*url.URL)
	tidbIp = dbURL.Hostname()
	tidbPort = dbURL.Port()
	log.Info("db info", zap.String("url", dbURL.String()), zap.String("ip:", tidbIp), zap.String("port:", tidbPort), zap.String("host:", dbURL.Host))
	db := Try(mycluster.GetDB("test")).(*sql.DB)
	defer db.Close()
	Try(db.Query("drop database if exists cdctest;"))
	Try(db.Query("create database cdctest;"))
	Try(db.Query("create table cdctest.t(a int, b int, primary key(a), index idx(a));"))
	Try(db.Query("insert into cdctest.t values(1,1), (2,2);"))
}

func CHECK_CDC_DATA(
	mycluster *TiUPCluster,
	num int,
) {
	db2 := Try(mycluster.GetDB("test")).(*sql.DB)
	time.Sleep(60 * time.Second)
	defer db2.Close()
	rows := Try(db2.Query("select * from cdctest.t;")).(*sql.Rows)
	defer rows.Close()
	i := 0
	for rows.Next() {
		var r record
		err := rows.Scan(&r.a, &r.b)
		must(err)
		log.Info("records", zap.String("a", strconv.Itoa(r.a)), zap.String("b", strconv.Itoa(r.b)))
		i = i + 1
	}
	if i < num {
		Fail("cdc sync fail")
	}
}

func SYNC_CDC_DATA2(
	mycluster *TiUPCluster,
) {
	dbURL = Try(mycluster.ServiceURL(resource.DBAddr)).(*url.URL)
	tidbIp = dbURL.Hostname()
	tidbPort = dbURL.Port()
	log.Info("db info", zap.String("url", dbURL.String()), zap.String("ip:", tidbIp), zap.String("port:", tidbPort), zap.String("host:", dbURL.Host))
	db := Try(mycluster.GetDB("test")).(*sql.DB)
	defer db.Close()
	Try(db.Query("insert into cdctest.t values(3,3), (4,4);"))

}
func SYNC_DIFF_CHECK_MySQL(
	db string,
) {
	tidbtest := ctx.Resource("wkload").(resource.WorkloadNode)
	cmd := fmt.Sprintf(`sed -i 's/bank/%s/g' /tools/sync-diff/config-mysql.toml`, db)
	Try(tidbtest.Exec(resource.WorkloadNodeExecOptions{Command: cmd}))

	Try(tidbtest.Exec(resource.WorkloadNodeExecOptions{Command: "/tools/sync-diff/run.sh mysql"}))
	time.Sleep(20 * time.Second)
	res, _, _, err := tidbtest.Exec(resource.WorkloadNodeExecOptions{Command: "tail /tools/sync-diff/mysql/sync_diff.log"})
	must(err)
	Expect(res).Should(ContainSubstring("check pass!!!"))
}

func SYNC_DIFF_CHECK_TiDB(
	db string,
) {
	tidbtest := ctx.Resource("wkload").(resource.WorkloadNode)
	cmd := fmt.Sprintf(`sed -i 's/bank/%s/g' /tools/sync-diff/config-tidb.toml`, db)
	Try(tidbtest.Exec(resource.WorkloadNodeExecOptions{Command: cmd}))

	Try(tidbtest.Exec(resource.WorkloadNodeExecOptions{Command: "/tools/sync-diff/run.sh tidb"}))
	time.Sleep(10 * time.Second)
	res, _, _, err := tidbtest.Exec(resource.WorkloadNodeExecOptions{Command: "tail /tools/sync-diff/tidb/sync_diff.log"})
	must(err)
	Expect(res).Should(ContainSubstring("check pass!!!"))
}

func BR_BACKUP(
	workloadNode resource.WorkloadNode,
	db string,
) {
	pdURL, err := cluster.ServiceURL(resource.PDAddr)
	must(err)
	log.Info("pd info", zap.String("url", pdURL.String()), zap.String("ip:", pdURL.Host), zap.String("port:", pdURL.Port()))
	pdDSN := pdURL.Host
	// pdDSN := "pd3-peer:22379"
	brstr := fmt.Sprintf(`cd /root/result; tiup br backup db --pd="%s" --db %s --storage "local:///home/tidb" --ratelimit 256 --log-file backtable.log`, pdDSN, db)
	res, _, _, err := workloadNode.Exec(resource.WorkloadNodeExecOptions{Command: brstr})
	must(err)
	log.Info("br backup:", zap.String("result", res))

}

func INSTALL_MYSQL_CLIENT(
	ctx core.TestContext,
) {
	dbURL := Try(cluster.ServiceURL(resource.DBAddr)).(*url.URL)
	host := dbURL.Hostname()
	port := dbURL.Port()
	tidbtest := ctx.Resource("dbdeployer").(resource.WorkloadNode)
	Try(tidbtest.Exec(resource.WorkloadNodeExecOptions{Command: "mkdir -p /root/opt/mysql;cp dbdeployer-1.64.0.linux /usr/local/bin/dbdeployer"}))
	Try(tidbtest.Exec(resource.WorkloadNodeExecOptions{Command: "dbdeployer --version"}))
	Try(tidbtest.Exec(resource.WorkloadNodeExecOptions{Command: "dbdeployer downloads get-unpack mysql-5.0.96.tar.xz"}))
	Try(tidbtest.Exec(resource.WorkloadNodeExecOptions{Command: "dbdeployer downloads get-unpack mysql-5.1.72.tar.xz"}))
	// Try(tidbtest.Exec(resource.WorkloadNodeExecOptions{Command: "dbdeployer downloads get-unpack mysql-5.5.62.tar.xz"}))
	// Try(tidbtest.Exec(resource.WorkloadNodeExecOptions{Command: "dbdeployer downloads get-unpack mysql-5.6.44.tar.xz"}))
	Try(tidbtest.Exec(resource.WorkloadNodeExecOptions{Command: "dbdeployer downloads get-unpack mysql-5.7.26.tar.xz"}))
	Try(tidbtest.Exec(resource.WorkloadNodeExecOptions{Command: "dbdeployer downloads get-unpack mysql-8.0.17-linux-x86_64-minimal.tar.xz"}))
	Try(tidbtest.Exec(resource.WorkloadNodeExecOptions{Command: "dbdeployer downloads get-unpack mysql-8.0.22-linux-glibc2.17-x86_64-minimal.tar.xz"}))
	Try(tidbtest.Exec(resource.WorkloadNodeExecOptions{Command: "dbdeployer downloads get-unpack mysql-8.0.25-linux-glibc2.17-x86_64-minimal.tar.xz"}))
	Try(tidbtest.Exec(resource.WorkloadNodeExecOptions{Command: "dbdeployer downloads get-unpack mysql-8.0.27-linux-glibc2.17-x86_64-minimal.tar.xz"}))

	cmd := fmt.Sprintf("./tidb_client_test -host %s -port %s --verbose > result.log", host, port)
	log.Info("cmd str", zap.String("cmd", cmd))
	defer func() {
		log.Info("mysql client test fail")
	}()
	Try(tidbtest.Exec(resource.WorkloadNodeExecOptions{Command: cmd}))
}