package drautosync

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

	. "github.com/onsi/gomega"
	. "github.com/pingcap/endless/pkg/matcher"
	. "github.com/pingcap/endless/pkg/util"
	"github.com/pingcap/endless/pkg/workload/bank"
	"github.com/pingcap/endless/pkg/workloadnode/tiup"
	"github.com/pingcap/endless/testcase/oltp/oltppkg"
	"github.com/pingcap/endless/testcase/oltp/tiuptest/tiuppkg"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"
	"golang.org/x/sync/errgroup"
	k8swait "k8s.io/apimachinery/pkg/util/wait"
)

type DrCaseOptions struct {
	DrRunTimeParms       DrRunTime
	WorkloadRunParms     oltppkg.WorkloadRunParms
	ChaoType             string
	SwitchReplicaMode    string
	IfVerify             bool
	IfVerifyData         bool
	IfResetTs            bool
	Tc                   *tiup.TiUPCluster
	DcNetworkDelay       string
	SwitchToBackFlag     bool
	IgnoreErr            []string
	BackupRole           string
	StopBank             bool
	workloadErrCheckChan chan oltppkg.WorkloadErrCheckChan
}

type DrRunTime struct {
	TotalRunTime       time.Duration
	PrimaryRunTime     time.Duration
	WorkloadWarmUptime time.Duration
	ChaosInjectTime    time.Duration
	ChaosRecoverTime   time.Duration
	BackupRunTime      time.Duration
}

func (opt *DrCaseOptions) InitClusterStartBank(ctxg1 context.Context,
	g1 *errgroup.Group, caseParams *oltppkg.CaseParams, warmUpTime time.Duration) (*DrClusterInfo, *bank.Bank2Client) {

	c := InitDrCluster(caseParams)

	if opt.DcNetworkDelay != "0ms" {
		c.InjectNetworkDelayBetweenDC(opt.DcNetworkDelay)
	}

	oltppkg.LogAndBy("Step: Start bank workload")
	dsn := resource.BuildDSN(c.DbURL, c.BankDbName)
	client := bank.Bank2Client{Config: &BankConfig}
	client.SetUp(ctxg1, dsn)

	g1.Go(WithGinkgoRecover(func() {
		client.Start(ctxg1, dsn)
		log.Info("finish bank tansfer start")
	}))
	g1.Go(WithGinkgoRecover(func() {
		client.StartVerify(ctxg1, dsn, `set @@session.tidb_replica_read='closest-replicas'`, `set @@tidb_read_staleness="-200"`)
		log.Info("Finish verify")
	}))
	return c, &client
}
func DrCommonCheckWithoutQPS(ctx context.Context, tc resource.MySQLCompatible, g *errgroup.Group, timeoutDuration time.Duration, pr *io.PipeReader) {
	qpsJitterCheckParms := oltppkg.QpsJitterCheckParameters{
		IsCheck: false,
	}
	durJitterCheckParms := oltppkg.DurJitterCheckParameters{
		IsCheck: false,
	}
	oltppkg.FunCommonCheck(ctx, tc, g, timeoutDuration, nil, qpsJitterCheckParms, durJitterCheckParms)
}
func (opt *DrCaseOptions) RunInPrimayClusterWithChaos(ctxAll context.Context,
	caseParams *oltppkg.CaseParams, c *DrClusterInfo,
) {
	runTime := opt.DrRunTimeParms.PrimaryRunTime
	ctx1, cancel1 := context.WithTimeout(ctxAll, runTime)
	defer cancel1()
	g1, ctxg1 := errgroup.WithContext(ctx1)
	oltppkg.LogAndBy("Step: Run workload on primary cluster")
	DrCommonCheckWithoutQPS(ctxg1, opt.Tc, g1, runTime, nil)

	oltppkg.WaitFor(ctxg1, "waiting for workload warm up", opt.DrRunTimeParms.WorkloadWarmUptime-3*time.Minute)
	oltppkg.BasicClusterExecPdCtl(c.Cluster, "config set replication-mode.dr-auto-sync.pause-region-split true")
	oltppkg.WaitFor(context.Background(), "waiting for split paused", 3*time.Minute)

	stopFlag := make(chan bool, 1)

	if opt.ChaoType == "down-backup-tikv-wait-for-sync" {
		g1.Go(WithGinkgoRecover(func() {
			chaoIds := Try(c.DownBackupDc(ctxg1)).(map[string]resource.ChaosIdentifier)
			oltppkg.WaitFor(ctxg1, "wait for some time to recover backup dc", opt.DrRunTimeParms.ChaosInjectTime)
			log.Info(fmt.Sprintf("delete chaos %v", chaoIds))
			for k, v := range chaoIds {
				Try(c.NodeWorkloadMap[k].ChaosClient().Delete(v))
			}
		}))
		g1.Go(WithGinkgoRecover(func() {
			oltppkg.WaitFor(ctxg1, "waiting for replication mode check", opt.DrRunTimeParms.ChaosInjectTime)
			WaitForReplicateStatus(caseParams.Tc, opt.DrRunTimeParms.ChaosRecoverTime, 1*time.Minute, "sync")
			stopFlag <- true
		}))
		Eventually(stopFlag).WithPolling(1 * time.Minute).WithTimeout(runTime - opt.DrRunTimeParms.WorkloadWarmUptime).Should(Receive())
		log.Info("Receive stop false, cancel context")
		cancel1()
	}
	if opt.ChaoType == "inject-network-partion-wait-for-sync" {
		g1.Go(WithGinkgoRecover(func() {
			chaoId := Try(c.InjectPartitionNetwork(strings.Join(c.DrTikvs, ","),
				strings.Join(append(c.PrimaryPds, c.PrimaryTikvs...), ","))).(resource.ChaosIdentifier)
			oltppkg.WaitFor(ctxg1, "wait for some time to recover tikv", opt.DrRunTimeParms.ChaosInjectTime)
			log.Info(fmt.Sprintf("delete chaos %v", chaoId))
			Try(c.NodeWorkloadMap["tikv4"].ChaosClient().Delete(chaoId))
		}))
		g1.Go(WithGinkgoRecover(func() {
			oltppkg.WaitFor(ctxg1, "waiting for replication mode check", opt.DrRunTimeParms.ChaosInjectTime)
			WaitForReplicateStatus(caseParams.Tc, opt.DrRunTimeParms.ChaosRecoverTime, 1*time.Minute, "sync")
			stopFlag <- true
		}))
		Eventually(stopFlag).WithPolling(1 * time.Minute).WithTimeout(runTime - opt.DrRunTimeParms.WorkloadWarmUptime).Should(Receive())
		log.Info("Receive stop false, cancel context")
		cancel1()
	}
	if opt.ChaoType == "inject-backup-chaos-wait-for-sync-recover" {
		g1.Go(WithGinkgoRecover(func() {
			chaoIds := Try(c.DownBackupDc(ctxg1)).(map[string]resource.ChaosIdentifier)
			oltppkg.WaitFor(ctxg1, "wait for some time to recover backup dc", opt.DrRunTimeParms.ChaosInjectTime)
			log.Info(fmt.Sprintf("delete chaos %v", chaoIds))
			for k, v := range chaoIds {
				Try(c.NodeWorkloadMap[k].ChaosClient().Delete(v))
			}
		}))
		g1.Go(WithGinkgoRecover(func() {
			WaitForReplicateStatus(caseParams.Tc, 5*time.Minute, 10*time.Second, "async")
		}))

		oltppkg.WaitFor(ctxg1, "waiting for replication change to sync_recover", 2*time.Minute)
		WaitForReplicateStatus(caseParams.Tc, opt.DrRunTimeParms.ChaosRecoverTime, 10*time.Second, "sync_recover")
		cancel1()
	}
	if opt.ChaoType == "random-chaos" {
		randomList := append(c.PrimaryPds, c.DrPD)
		randomList = append(randomList, c.PrimaryTikvs...)
		randomList = append(randomList, c.DrTikvs...)
		g1.Go(WithGinkgoRecover(func() {
			k8swait.PollImmediateUntilWithContext(ctxg1, 10*time.Minute, func(ctx context.Context) (done bool, err error) {
				randomPod := randomList[oltppkg.GetRandInt(0, len(randomList))]
				chaoId := Try(NetworkDownWorkloadNode(ctxg1, c.NodeWorkloadMap[randomPod])).(resource.ChaosIdentifier)
				faultDuration := oltppkg.GetRandDuration(time.Minute, 5)
				oltppkg.WaitFor(ctxg1, "wait for some time to recover chaos", faultDuration)
				log.Info(fmt.Sprintf("delete chaos %v", chaoId))
				Try(c.NodeWorkloadMap[randomPod].ChaosClient().Delete(chaoId))
				return false, nil
			})
		}))
	}

	Try(g1.Wait())
	log.Info("Finish running on primary dc.")
}
func (opt *DrCaseOptions) SwitchToBackupCluster(ctxAll context.Context,
	c *DrClusterInfo,
	bankClient *bank.Bank2Client,
) {
	log.Info("Start to run on backup cluster.")
	log.Info(fmt.Sprintf("%v", opt.DrRunTimeParms))
	ctx2, cancel2 := context.WithTimeout(ctxAll, opt.DrRunTimeParms.BackupRunTime)
	defer cancel2()
	g2, ctxg2 := errgroup.WithContext(ctx2)
	//resumeFlag := make(chan bool, 1)
	g2.Go(WithGinkgoRecover(func() {
		oltppkg.WaitFor(ctxg2, "waiting workload stop", 30*time.Second)
		bankClient.TearDown(ctxg2)
	}))
	g2.Go(WithGinkgoRecover(func() {
		oltppkg.WaitFor(ctxg2, "waiting workload stop", 30*time.Second)
		//opt.WorkloadRunParms.PauseWithResume(ctxg2, 5*time.Second, opt.DrRunTimeParms.BackupRunTime, resumeFlag)
	}))
	g2.Go(WithGinkgoRecover(func() {
		DownPrimaryAndVerifyACID(ctxAll, c, *bankClient, opt.IfVerifyData, opt.IfResetTs)
		//resumeFlag <- true
		dsn := resource.BuildDSN(c.DbURL, c.BankDbName)
		bankClient.Start(ctxg2, dsn)
		opt.PostCheckCluster(ctxAll, c)
	}))
	Try(g2.Wait())
	log.Info("Run in backup cluster finished.")
}
func (opt *DrCaseOptions) PostCheckCluster(ctx context.Context, clusterInfo *DrClusterInfo) {
	log.Info("Now run post check on backup cluster")
	ctx3, _ := context.WithTimeout(ctx, opt.DrRunTimeParms.BackupRunTime)
	g3, ctxg3 := errgroup.WithContext(ctx3)

	oltppkg.FunCommonCheck(ctxg3, clusterInfo.Cluster, g3, opt.DrRunTimeParms.BackupRunTime, nil)
	workloadCheckParms := oltppkg.WorkloadErrCheckChan{
		IfCheck:   true,
		IfPrint:   true,
		IgnoreErr: opt.IgnoreErr,
	}
	log.Info(fmt.Sprintf("Set workload ingore to %v", workloadCheckParms))
	opt.workloadErrCheckChan <- workloadCheckParms

	Try(g3.Wait())
	log.Info("Check in backup cluster finished.")
}

func (opt *DrCaseOptions) DrAutoTest(ctx context.Context, caseParams *oltppkg.CaseParams) {
	ctxAll, cancelAll := context.WithTimeout(ctx, opt.DrRunTimeParms.TotalRunTime)
	defer cancelAll()
	g, ctxg := errgroup.WithContext(ctxAll)

	errCheck := oltppkg.WorkloadErrCheckChan{
		IfCheck:   true,
		IfPrint:   true,
		IgnoreErr: strings.Split(caseParams.CaseCmdParms.TpccParms.IgnoreErrors, ","),
	}
	if opt.ChaoType != "" {
		errCheck.IgnoreErr = append(errCheck.IgnoreErr, "9002")
	}
	opt.IgnoreErr = errCheck.IgnoreErr
	log.Info(fmt.Sprintf("Set workload check to %v", errCheck))
	opt.workloadErrCheckChan = make(chan oltppkg.WorkloadErrCheckChan, 3)
	opt.workloadErrCheckChan <- errCheck

	g.Go(WithGinkgoRecover(func() {
		oltppkg.WorkLoadErrLogCheckWithChan(ctxg, opt.WorkloadRunParms.Cmdpr, opt.DrRunTimeParms.TotalRunTime, opt.workloadErrCheckChan)
	}))
	c, bankClient := opt.InitClusterStartBank(ctxg, g, caseParams, opt.DrRunTimeParms.PrimaryRunTime)

	defer func() {
		DestroyPrimaryCluster(ctxAll, c.ClusterName, c.TiupNode)
	}()
	defer func() {
		if tc, ok := caseParams.Tc.(*tiup.TiUPCluster); ok {
			Try(tiuppkg.UploadClinic(tc, "--include", "monitor,log", "-f", "-6h", "-R", "tikv,pd"))
		}
	}()
	defer func() {
		log.Info("clean the chaos.")
		c.CleanChaos()
	}()

	opt.RunInPrimayClusterWithChaos(ctxg, caseParams, c)

	log.Info(fmt.Sprintf("Set workload check to %v", oltppkg.WorkloadErrCheckChan{IfCheck: false, IfPrint: false}))
	opt.workloadErrCheckChan <- oltppkg.WorkloadErrCheckChan{IfCheck: false, IfPrint: false}
	if opt.SwitchToBackFlag {
		opt.SwitchToBackupCluster(ctxAll, c, bankClient)
	}

	log.Info("Cancel context to tear down this test.")
	cancelAll()
	Try(g.Wait())
}

func VerifyACID(ctx context.Context, c *DrClusterInfo, bankClient bank.Bank2Client) {
	oltppkg.LogAndBy("Step: Verify bank ACID.")
	db2 := Try(c.ConnectDB(c.BankDbName)).(*sql.DB)
	defer db2.Close()
	//bankClient.Verify(ctx, db2, `set @@session.tidb_replica_read='closest-replicas'`, `set @@tidb_read_staleness="-200"`)
	bankClient.Verify(ctx, db2)
	Try(c.GetAllTablesCounts(ctx, c.BankDbName))
	Try(c.GetAllTablesCounts(ctx, "test"))
	Try(c.GetAllTablesCounts(ctx, "information_schema"))
	log.Info("verify ACID finished")
}
func VerifyData(ctx context.Context, c *DrClusterInfo, bankClient bank.Bank2Client, bankCount map[string]int64, sysbenchCount map[string]int64) {
	oltppkg.LogAndBy("Step: verify data consistent.")
	db2 := Try(c.ConnectDB(c.BankDbName)).(*sql.DB)
	defer db2.Close()
	bankCount2 := c.GetAllTablesCounts(ctx, c.BankDbName)
	sysbenchCount2 := c.GetAllTablesCounts(ctx, "test")
	Expect(CompareTwoMap(bankCount, bankCount2)).Should(BeNumerically(">=", 0))
	Expect(CompareTwoMap(sysbenchCount, sysbenchCount2)).Should(BeNumerically(">=", 0))
	log.Info("test finished")
}

func DownPrimaryAndVerifyACID(ctx context.Context, c *DrClusterInfo, bankClient bank.Bank2Client, ifVerifyData bool, ifResetTs bool) {
	var sysbenchCount, bankCount map[string]int64
	if ifVerifyData {
		sysbenchCount, bankCount = GetDataInPrimaryCluster(ctx, c)
	} else {
		c.DownPrimaryDc(ctx)
	}

	c.RestoreBackDc(ctx)
	c.RestartBackDc(ctx)
	if ifResetTs {
		c.ResetToMinResolvedTs(ctx)
	}

	VerifyACID(ctx, c, bankClient)
	if ifVerifyData {
		VerifyData(ctx, c, bankClient, bankCount, sysbenchCount)
	}
}

func GetDataInPrimaryCluster(ctxAll context.Context, c *DrClusterInfo) (map[string]int64, map[string]int64) {
	oltppkg.LogAndBy("Step: Read data from primary cluster")
	c.GetAllTablesCounts(ctxAll, "test")
	c.GetAllTablesCounts(ctxAll, "bank")

	oltppkg.LogAndBy("Step: Down dr cluster to block write")
	c.TiupExecPdCtl(ctxAll, "config set replication-mode.dr-auto-sync.wait-store-timeout 30m")
	for _, id := range c.DrStoreIds {
		c.TiupExecPdCtl(ctxAll, fmt.Sprintf("scheduler add evict-leader-scheduler %s", id))
	}
	oltppkg.WaitForLeaderToZero(c.PromURL, "tikv4-peer", 10*time.Second, 2*time.Minute)

	// to avoid issue https://github.com/tikv/tikv/issues/12914
	c.TiupExecPdCtl(ctxAll, "config set leader-schedule-limit 0")
	oltppkg.WaitFor(ctxAll, "waiting leader balance finished.", 5*time.Minute)

	backupDcChaos := Try(c.DownBackupDc(ctxAll)).(map[string]resource.ChaosIdentifier)

	oltppkg.LogAndBy("Step: Read data from primary cluster")
	sysbenchCount := c.GetAllTablesCounts(ctxAll, "test")
	bankCount := c.GetAllTablesCounts(ctxAll, "bank")

	oltppkg.LogAndBy("Step: Down primary")
	c.DownPrimaryDc(ctxAll)
	for node, chaoId := range backupDcChaos {
		Try(c.NodeWorkloadMap[node].ChaosClient().Delete(chaoId))
	}
	return sysbenchCount, bankCount
}