package gc

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

	_ "github.com/go-sql-driver/mysql"
	. "github.com/onsi/gomega"
	"github.com/pingcap/endless/pkg/chaos"
	. "github.com/pingcap/endless/pkg/matcher"
	. "github.com/pingcap/endless/pkg/util"
	"github.com/pingcap/endless/testcase/oltp/oltppkg"
	"github.com/pingcap/endless/testcase/oltp/systemtest/pdrel"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"
	"golang.org/x/sync/errgroup"
	k8swait "k8s.io/apimachinery/pkg/util/wait"
)

func OLTP_GC_REL_GC_LEADER_DOWN( //gc leader 故障
	ctx context.Context,
	tc resource.TiDBCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms,
) {
	restartCheckParms := oltppkg.RestartCheckParameters{
		AllComponentRestartCheck: false,
		TiKVRestartCheck:         true,
		PDRestartCheck:           true,
		TiDBRestartCheck:         false,
	}
	qpsJitterCheckParms := oltppkg.QpsJitterCheckParameters{
		IsCheck:      true,
		Offset:       "2m",
		ToleratePerc: 0.8,
		TolerateDur:  120 * time.Second,
	}
	durJitterCheckParms := oltppkg.DurJitterCheckParameters{
		IsCheck:      true,
		Offset:       "2m",
		ToleratePerc: 2,
		TolerateDur:  600 * time.Second,
	}
	workloadCheckParms := oltppkg.WorkloadCheckParameters{
		IsCheck:   true,
		IgnoreErr: append(pdrel.GetPdIngoreErrors(caseCmdParms, "fun"), "9001"),
	}
	GcRelTest(ctx, tc, caseCmdParms, workloadRunParms, "gc-leader-down", restartCheckParms, qpsJitterCheckParms, durJitterCheckParms, workloadCheckParms)
}

func OLTP_GC_REL_TIKV_DOWN( // tikv instance 故障
	ctx context.Context,
	tc resource.TiDBCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms,
) {
	restartCheckParms := oltppkg.RestartCheckParameters{
		AllComponentRestartCheck: false,
		TiKVRestartCheck:         false,
		PDRestartCheck:           true,
		TiDBRestartCheck:         true,
	}
	qpsJitterCheckParms := oltppkg.QpsJitterCheckParameters{
		IsCheck:      false,
		Offset:       "2m",
		ToleratePerc: 0.8,
		TolerateDur:  120 * time.Second,
	}
	durJitterCheckParms := oltppkg.DurJitterCheckParameters{
		IsCheck:      false,
		Offset:       "2m",
		ToleratePerc: 2,
		TolerateDur:  600 * time.Second,
	}
	workloadCheckParms := oltppkg.WorkloadCheckParameters{
		IsCheck:   true,
		IgnoreErr: append(pdrel.GetPdIngoreErrors(caseCmdParms, "fun"), "9002"),
	}
	GcRelTest(ctx, tc, caseCmdParms, workloadRunParms, "tikv-down", restartCheckParms, qpsJitterCheckParms, durJitterCheckParms, workloadCheckParms)
}

func OLTP_GC_REL_TIDB_TIKV_NETWORK_DELAY( // network delay between tikv and tidb
	ctx context.Context,
	tc resource.TiDBCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms,
) {
	restartCheckParms := oltppkg.RestartCheckParameters{
		AllComponentRestartCheck: true,
	}
	qpsJitterCheckParms := oltppkg.QpsJitterCheckParameters{
		IsCheck:      true,
		Offset:       "2m",
		ToleratePerc: 0.8,
		TolerateDur:  120 * time.Second,
	}
	durJitterCheckParms := oltppkg.DurJitterCheckParameters{
		IsCheck:      true,
		Offset:       "2m",
		ToleratePerc: 2,
		TolerateDur:  600 * time.Second,
	}
	workloadCheckParms := oltppkg.WorkloadCheckParameters{
		IsCheck:   true,
		IgnoreErr: append(pdrel.GetPdIngoreErrors(caseCmdParms, "fun"), "9002"),
	}
	GcRelTest(ctx, tc, caseCmdParms, workloadRunParms, "tidb-tikv-network-delay", restartCheckParms, qpsJitterCheckParms, durJitterCheckParms, workloadCheckParms)
}

func OLTP_GC_REL_GC_LEADER_NETWORK_LOSS( // network loss in gc leader
	ctx context.Context,
	tc resource.TiDBCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms,
) {
	restartCheckParms := oltppkg.RestartCheckParameters{
		AllComponentRestartCheck: true,
	}
	qpsJitterCheckParms := oltppkg.QpsJitterCheckParameters{
		IsCheck:      true,
		Offset:       "2m",
		ToleratePerc: 0.8,
		TolerateDur:  120 * time.Second,
	}
	durJitterCheckParms := oltppkg.DurJitterCheckParameters{
		IsCheck:      true,
		Offset:       "2m",
		ToleratePerc: 2,
		TolerateDur:  600 * time.Second,
	}
	workloadCheckParms := oltppkg.WorkloadCheckParameters{
		IsCheck:   true,
		IgnoreErr: append(pdrel.GetPdIngoreErrors(caseCmdParms, "fun"), "9001", "9002"),
	}
	GcRelTest(ctx, tc, caseCmdParms, workloadRunParms, "gc-leader-network-loss", restartCheckParms, qpsJitterCheckParms, durJitterCheckParms, workloadCheckParms)
}
func OLTP_GC_REL_TIKV_NETWORK_PARTITION( // gc leader network partition
	ctx context.Context,
	tc resource.TiDBCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms,
) {
	restartCheckParms := oltppkg.RestartCheckParameters{
		AllComponentRestartCheck: true,
	}
	qpsJitterCheckParms := oltppkg.QpsJitterCheckParameters{
		IsCheck:      false,
		Offset:       "2m",
		ToleratePerc: 0.8,
		TolerateDur:  120 * time.Second,
	}
	durJitterCheckParms := oltppkg.DurJitterCheckParameters{
		IsCheck:      false,
		Offset:       "2m",
		ToleratePerc: 2,
		TolerateDur:  600 * time.Second,
	}
	workloadCheckParms := oltppkg.WorkloadCheckParameters{
		IsCheck:   true,
		IgnoreErr: append(pdrel.GetPdIngoreErrors(caseCmdParms, "fun"), "9002"),
	}
	GcRelTest(ctx, tc, caseCmdParms, workloadRunParms, "tikv-network-partition", restartCheckParms, qpsJitterCheckParms, durJitterCheckParms, workloadCheckParms)
}

func GcRelTest(ctx context.Context,
	tc resource.TiDBCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms,
	chaoType string,
	commonCheckParams ...interface{}) {

	//to w/a pr https://github.com/pingcap/tidb/pull/32726
	db := Try(tc.GetDB("")).(*sql.DB)
	defer db.Close()
	if oltppkg.CompareVersionWithCurrent(ctx, tc, "6.1.0") >= 0 {
		log.Info("set @@global.tidb_gc_max_wait_time")
		Try(db.Exec("set @@global.tidb_gc_max_wait_time=900"))
	}

	defer func() {
		db := Try(tc.GetDB("")).(*sql.DB)
		defer db.Close()
		if oltppkg.CompareVersionWithCurrent(ctx, tc, "6.1.0") >= 0 {
			oltppkg.WaitFor(ctx, "waiting for tidb recover", 2*time.Minute)
			Try(db.Exec("set @@global.tidb_gc_max_wait_time=86400"))
		}
	}()

	runTimeParms := oltppkg.RunTimePara{
		TotalRunTime:  60 * time.Minute,
		FaultDuration: oltppkg.GetRandDuration(1*time.Second, 600),
		WarmUpTime:    5 * time.Minute,
		RecoveryTime:  10 * time.Minute,
		RunInterval:   25 * time.Minute,
	}

	oltppkg.LogAndBy("Drop database")
	_, initSumSize, err := oltppkg.GetStoreSizeByType(tc, "used_size")
	Expect(err).Should(BeNil())

	dropSize := DropOneDatabase(ctx, tc, caseCmdParms)

	ctx1, cancel1 := context.WithTimeout(ctx, runTimeParms.TotalRunTime)
	defer cancel1()
	ctx2, cancel2 := context.WithTimeout(ctx, runTimeParms.TotalRunTime-runTimeParms.RecoveryTime-10*time.Minute)
	defer cancel2()
	g1, ctxg1 := errgroup.WithContext(ctx1)
	log.Info(fmt.Sprintf("%v", commonCheckParams))
	oltppkg.FunCommonCheck(ctxg1, tc, g1, runTimeParms.TotalRunTime, workloadRunParms.Cmdpr, commonCheckParams...)

	g2, ctxg2 := errgroup.WithContext(ctx2)
	startTime := time.Now().Add(runTimeParms.WarmUpTime / 2)

	oltppkg.WaitFor(ctxg2, "waiting for workload running", runTimeParms.WarmUpTime)
	oltppkg.LogAndBy("Inject chaos")

	g2.Go(WithGinkgoRecover(func() {
		//Inject chaos periodly
		conditionWrapper := GetGcRelConditionWrapper(ctxg2, tc, workloadRunParms, chaoType)
		k8swait.PollImmediateUntilWithContext(ctxg2, runTimeParms.RunInterval, conditionWrapper)
	}))

	Expect(g2.Wait()).Should(BeNil())

	workloadRunParms.Pause(ctxg1, 1*time.Minute)
	oltppkg.WaitFor(ctxg1, "workload recover", runTimeParms.RecoveryTime)
	oltppkg.RepeatQpsDurCheckWithAbsTime(ctxg1, 5*time.Minute, startTime, tc, 0.5)

	GcPostCheck(context.Background(), tc, 30*time.Minute, dropSize, initSumSize)
	Expect(g1.Wait()).Should(BeNil())
}

func GetGcRelConditionWrapper(
	ctx context.Context,
	tc resource.TiDBCluster,
	workloadRunParms oltppkg.WorkloadRunParms,
	chaoType string,
) (conditionWrapper k8swait.ConditionWithContextFunc) {
	var c chaos.Chaos
	switch chaoType {
	case "gc-leader-down":
		c = GetGcLeaderChaos(ctx, tc)
	case "tikv-down":
		instanceType := resource.TiKV
		instanceIdList := oltppkg.GetInstanceIdList(tc, instanceType)
		instanceId, _ := oltppkg.GetRandInstance(tc, instanceType)
		experimentSpec := &resource.FailureExperimentSpec{}
		c = Try(oltppkg.InitChaos(tc, instanceType, "failure", experimentSpec, &instanceIdList[instanceId])).(chaos.Chaos)
	case "tidb-tikv-network-delay":
		instanceId := Try(oltppkg.GetGcLeader(ctx, tc)).(int)
		tikvInstanceId, _ := oltppkg.GetRandInstance(tc, resource.TiKV)
		c = chaos.Chaos{
			Name:           "tikv tidb network delay",
			Tc:             tc,
			TargetTc:       tc,
			ExperimentSpec: &resource.NetworkLatencySpec{Latency: "200ms"},
			SelectorOpts: resource.TiDBClusterSelectorOpts{
				TiDBClusterTiDBSelectorSpec: &resource.TiDBClusterTiDBSelectorSpec{
					Peer: oltppkg.ToIntPtr(instanceId),
				},
			},
			TargetSelectorOpts: &resource.TiDBClusterSelectorOpts{
				TiDBClusterTiKVSelectorSpec: &resource.TiDBClusterTiKVSelectorSpec{
					Peer: oltppkg.ToIntPtr(tikvInstanceId),
				},
			},
		}
	case "gc-leader-network-loss":
		instanceId := Try(oltppkg.GetGcLeader(ctx, tc)).(int)
		c = chaos.Chaos{
			Name:           "gc leader network loss",
			Tc:             tc,
			ExperimentSpec: &resource.NetworkLossSpec{Loss: "75", Correlation: "50"},
			SelectorOpts: resource.TiDBClusterSelectorOpts{
				TiDBClusterTiDBSelectorSpec: &resource.TiDBClusterTiDBSelectorSpec{
					Peer: oltppkg.ToIntPtr(instanceId),
				},
			},
		}

	case "tikv-network-partition":
		gcInstanceId := Try(oltppkg.GetGcLeader(ctx, tc)).(int)
		c = chaos.Chaos{
			Name:           "tikv tidb network partition",
			Tc:             tc,
			TargetTc:       tc,
			ExperimentSpec: &resource.NetworkPartitionSpec{},
			SelectorOpts: resource.TiDBClusterSelectorOpts{
				TiDBClusterTiKVSelectorSpec: &resource.TiDBClusterTiKVSelectorSpec{
					All: true,
				},
			},
			TargetSelectorOpts: &resource.TiDBClusterSelectorOpts{
				TiDBClusterTiDBSelectorSpec: &resource.TiDBClusterTiDBSelectorSpec{
					Peer: oltppkg.ToIntPtr(gcInstanceId),
				},
			},
		}
	default:
		return nil
	}
	conditionWrapper = func(ctx context.Context) (bool, error) {
		faultDuration := oltppkg.GetRandDuration(1*time.Second, 600)
		oltppkg.InjectFaultByChaos(&c, ctx, faultDuration)
		randSecond := int(oltppkg.GetRandNum(10, 100))
		oltppkg.WaitFor(ctx, "gc leader recover", time.Duration(randSecond)*time.Second)
		return false, nil
	}
	return conditionWrapper
}