package datarecovery

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

	_ "github.com/go-sql-driver/mysql"
	. "github.com/onsi/gomega"
	pdclient "github.com/pingcap/endless/pkg/client/pd"
	. "github.com/pingcap/endless/pkg/matcher"
	"github.com/pingcap/endless/testcase/oltp/oltppkg"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"

	// _ "github.com/pingcap/test-infra/sdk/resource/impl/k8s"
	"github.com/tidwall/gjson"
	"go.uber.org/zap"
)

func GetAllTiKVActiveInstanceId(tc resource.TiDBCluster) []int {
	pd := Try(pdclient.NewPDClient(tc)).(*pdclient.Client)
	path := "/pd/api/v1/stores"
	bytes := Try(pd.GetJsonBytes(path)).([]byte)
	result := gjson.GetBytes(bytes, "stores.#.store.address").Value()

	var instanceIdList []int
	for _, address := range result.([]interface{}) {
		instance := strings.Split(address.(string), ".")[0]
		instanceId, _ := strconv.Atoi(strings.Split(instance, "-")[2])
		instanceIdList = append(instanceIdList, instanceId)
	}
	log.Info("instanceIdList", zap.Ints("instanceIds: ", instanceIdList))
	return instanceIdList
}

func GetFailedStoresId(tc resource.TiDBCluster) []string {
	pd := Try(pdclient.NewPDClient(tc)).(*pdclient.Client)
	path := "/pd/api/v1/stores"
	bytes := Try(pd.GetJsonBytes(path)).([]byte)
	result := gjson.GetBytes(bytes, "stores.#.store.id").Value()

	var storesID []string
	for _, id := range result.([]interface{}) {
		storesID = append(storesID, strconv.Itoa(int(id.(float64))))
	}
	log.Info(fmt.Sprintf("storesID:%s", storesID))
	var failedStoresId []string
	for _, id := range storesID {
		path := "/pd/api/v1/store/" + id
		bytes := Try(pd.GetJsonBytes(path)).([]byte)
		result := gjson.GetBytes(bytes, "store.state_name").Value()
		if strings.ToLower(result.(string)) != "up" {
			failedStoresId = append(failedStoresId, id)
		}
	}
	log.Info(fmt.Sprintf("failedStoresId:%s", failedStoresId))
	return failedStoresId
}

func ExecuteOnlineUnsafeRecovery(tc resource.TiDBCluster, failedStoresId []string) error {
	len := len(failedStoresId)
	recoveryCmd := "unsafe remove-failed-store "
	for i, id := range failedStoresId {
		if i != len-1 {
			recoveryCmd = recoveryCmd + id + ","
		} else {
			recoveryCmd = recoveryCmd + id
		}
	}
	log.Info(recoveryCmd)
	r := Try(oltppkg.ExecPdCtl(tc, recoveryCmd)).(string)
	if strings.Contains(strings.ToLower(r), "success") {
		return nil
	}
	return errors.New(r)
}

func GetOnlineRecoveryResult(ctx context.Context, tc resource.TiDBCluster, interval time.Duration, timeout time.Duration) error {
	failedInfo := "unsafe recovery failed"
	successInfo := "unsafe recovery finished"
	timeOut := time.After(timeout)
	for {
		select {
		case <-ctx.Done():
			log.Info("ctx done.")
			return nil
		case <-timeOut:
			return errors.New("online recovery timeout.")
		default:
		}
		oltppkg.WaitFor(ctx, "wait for getting online recovery result", interval)
		bytes := []byte(Try(oltppkg.ExecPdCtl(tc, "unsafe remove-failed-stores show")).(string))
		recoveryInfo := gjson.GetBytes(bytes, "#.info").Value()
		var infoStr string
		for _, info := range recoveryInfo.([]interface{}) {
			infoStr = strings.ToLower(info.(string))
			if strings.Contains(infoStr, successInfo) || strings.Contains(infoStr, failedInfo) {
				break
			}
		}
		if strings.Contains(infoStr, successInfo) {
			return nil
		} else if strings.Contains(infoStr, failedInfo) {
			return errors.New(infoStr)
		}
	}
}

func AdjustPDSchedules(tc resource.TiDBCluster, operator string) {
	if operator == "close" {
		Try(oltppkg.ExecPdCtl(tc, "config show"))
		r := Try(oltppkg.ExecPdCtl(tc, "config set region-schedule-limit 0")).(string)
		Expect(strings.Contains(strings.ToLower(r), "success")).To(BeTrue())
		r = Try(oltppkg.ExecPdCtl(tc, "config set replica-schedule-limit 0")).(string)
		Expect(strings.Contains(strings.ToLower(r), "success")).To(BeTrue())
		r = Try(oltppkg.ExecPdCtl(tc, "config set merge-schedule-limit 0")).(string)
		Expect(strings.Contains(strings.ToLower(r), "success")).To(BeTrue())
	} else if operator == "recover" {
		Try(oltppkg.ExecPdCtl(tc, "config show"))
		r := Try(oltppkg.ExecPdCtl(tc, "config set region-schedule-limit 2048")).(string)
		Expect(strings.Contains(strings.ToLower(r), "success")).To(BeTrue())
		r = Try(oltppkg.ExecPdCtl(tc, "config set replica-schedule-limit 64")).(string)
		Expect(strings.Contains(strings.ToLower(r), "success")).To(BeTrue())
		r = Try(oltppkg.ExecPdCtl(tc, "config set merge-schedule-limit 8")).(string)
		Expect(strings.Contains(strings.ToLower(r), "success")).To(BeTrue())
	}
}

func GetTombstoneNum(tc resource.TiDBCluster) float64 {
	promURL := Try(tc.ServiceURL(resource.Prometheus)).(*url.URL)
	query := `sum(pd_cluster_status{type="store_tombstone_count"})`
	return Try(oltppkg.GetMetric(promURL, query, time.Now())).(float64)
}

func RemoveTombstone(ctx context.Context, tc resource.TiDBCluster, pollingInterval time.Duration, timeout time.Duration) {
	promURL := Try(tc.ServiceURL(resource.Prometheus)).(*url.URL)
	r := Try(oltppkg.ExecPdCtl(tc, "store remove-tombstone")).(string)
	Expect(strings.Contains(strings.ToLower(r), "success")).To(BeTrue())
	query := `sum(pd_cluster_status{type="store_tombstone_count"})`
	Eventually(func() float64 {
		curTombstone := Try(oltppkg.GetMetric(promURL, query, time.Now())).(float64)
		log.Info("current curTombstone:", zap.Float64("curTombstone", curTombstone))
		return curTombstone
	}, timeout, pollingInterval).Should(Equal(0))
}

func WaitForStoreRegionsToZero(ctx context.Context, tc resource.TiDBCluster, storesId []string, pollingInterval time.Duration, timeout time.Duration) {
	pd := Try(pdclient.NewPDClient(tc)).(*pdclient.Client)
	pathPrefix := "/pd/api/v1/regions/store/"
	Eventually(func() bool {
		var countToZero bool = true
		for _, storeId := range storesId {
			path := pathPrefix + storeId
			bytes := Try(pd.GetJsonBytes(path)).([]byte)
			count := gjson.GetBytes(bytes, "count").Value().(float64)
			log.Info(fmt.Sprintf("Regions of Store %s: %d", storeId, int(math.Ceil(count))))
			if int(math.Ceil(count)) != 0 {
				countToZero = false
			}
		}
		return countToZero
	}, timeout, pollingInterval).Should(Equal(true))
}

func ExecTiKVCtl(tc resource.TiDBCluster, cmd string) (string, error) {
	tikvCtlCmd := fmt.Sprintf("/tikv-ctl %s", cmd)
	stdout, stderr, exitcode, err := tc.Exec(resource.TiDBClusterExecOptions{Target: resource.TiKV, Command: tikvCtlCmd})
	if err != nil {
		log.Info(fmt.Sprintf("stdout:%s\n stderr:%s\n exitCode:%d\n", stdout, stderr, exitcode))
		return "", err
	}
	fmt.Printf("%s\n%v\n", tikvCtlCmd, stdout)
	return stdout, nil
}

func ExecTiKVCmd(tc resource.TiDBCluster, peer int, cmd string) (string, error) {
	stdout, stderr, exitcode, err := tc.Exec(resource.TiDBClusterExecOptions{
		Target:  resource.TiKV,
		Command: cmd,
		Peer:    peer,
	})
	if err != nil {
		fmt.Printf("%s\n%s\nexitCode:%d\n", cmd, stderr, exitcode)
		return stderr, err
	}
	fmt.Printf("%s\n%v\n", cmd, stdout)
	return stdout, nil
}

func DownLoadSSTChaos(tc resource.TiDBCluster, peer int) {
	// wget http://httpfileserver-0.httpfileserver-peer.glh-master-55wfj.svc:80/sst_chaos/ldb
	// httpFileserver := ctx.Resource("httpfileserver").(resource.WorkloadNode)
	queryLdb, _ := ExecTiKVCmd(tc, peer, "ls ldb")
	querySSTChaos, _ := ExecTiKVCmd(tc, peer, "ls sst-chaos")
	ns := tc.Handle().Testbed
	format := "wget http://httpfileserver-0.httpfileserver-peer.%s:80/%s"
	if queryLdb != "ldb" {
		ldbDownLoadCmd := fmt.Sprintf(format, ns, "sst_chaos/ldb")
		Try(ExecTiKVCmd(tc, peer, ldbDownLoadCmd))
		ldbPath := Try(ExecTiKVCmd(tc, peer, "ls ldb")).(string)
		Expect(ldbPath).To(Equal("ldb"))
		Try(ExecTiKVCmd(tc, peer, "chmod 777 ldb"))
	}
	if querySSTChaos != "sst-chaos" {
		sstChaosDownLoadCmd := fmt.Sprintf(format, ns, "sst_chaos/sst-chaos")
		Try(ExecTiKVCmd(tc, peer, sstChaosDownLoadCmd))
		sstChaosPath := Try(ExecTiKVCmd(tc, peer, "ls sst-chaos")).(string)
		Expect(sstChaosPath).To(Equal("sst-chaos"))
		Try(ExecTiKVCmd(tc, peer, "chmod 777 sst-chaos"))
	}
}

// tool path: 172.16.5.31:/root/lxt/sst-chaos/sst-chaos-bins.tar.gz
// ./ldb manifest_dump --path=/var/lib/tikv/data/db/MANIFEST-000006 | less
// ./sst-chaos /ldb /var/lib/tikv/data/db/MANIFEST-000006
// use absolute path
func InjectSSTErr(tc resource.TiDBCluster, peer int) error {
	DownLoadSSTChaos(tc, peer)
	manifestFilePath := Try(ExecTiKVCmd(tc, peer, "ls /var/lib/tikv/data/db/MANIFEST-*")).(string)
	sstChaosPath := "/sst-chaos"
	ldbPath := "/ldb"
	injectCmd := fmt.Sprintf(".%s %s %s", sstChaosPath, ldbPath, manifestFilePath)
	Try(ExecTiKVCmd(tc, peer, injectCmd))
	return nil
}

// /tikv-ctl --pd http://tc-pd.testbed.svc:2379 compact-cluster
func CompactCluster(tc resource.TiDBCluster) error {
	pdStr := (Try(pdclient.NewPDClient(tc)).(*pdclient.Client)).GetPDAddr()
	cmd := fmt.Sprintf("--pd %s compact-cluster", pdStr)
	result := Try(ExecTiKVCtl(tc, cmd)).(string)
	resultList := strings.Split(result, "\n")
	for _, r := range resultList {
		log.Info(r)
		if !strings.Contains(strings.ToLower(r), "success") {
			return errors.New("compact failed")
		}
	}
	return nil
}

func CheckConfigOnline(tc resource.TiDBCluster, compType resource.TiDBCompType, config string, targetValue string) {
	configRows := oltppkg.GetConfigByName(tc, compType.String(), config)
	currentValue := configRows[0].Value
	Expect(currentValue).To(Equal(targetValue))
}