package gc

import (
	"context"
	"strings"
	"time"

	"database/sql"
	"net/url"

	_ "github.com/go-sql-driver/mysql"
	. "github.com/onsi/gomega"
	"github.com/pingcap/endless/pkg/chaos"
	"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/util"
	"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"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
)

func GetRunParms() *oltppkg.RunTimePara {
	runTimePara := oltppkg.NewRunTimePara(30*time.Minute, 10*time.Minute, oltppkg.GetRandDuration(time.Minute, 10), 10*time.Minute)
	return runTimePara
}
func DropOneDatabase(ctx context.Context, tc resource.TiDBCluster, caseCmdParms oltppkg.CaseCmdParms) float64 {
	var dropSize float64
	if strings.HasPrefix(caseCmdParms.WorkloadType.Type[0], "sysbench") || strings.HasPrefix(caseCmdParms.WorkloadType.Type[0], "ycsb") {
		dropSize = oltppkg.GetDatabaseSize(ctx, tc, caseCmdParms.TpccParms.Db)
		Try(oltppkg.RecreateDatabase(ctx, tc, caseCmdParms.TpccParms.Db))
	} else {
		dropSize = oltppkg.GetDatabaseSize(ctx, tc, caseCmdParms.SysbenchParms.Db)
		Try(oltppkg.RecreateDatabase(ctx, tc, caseCmdParms.SysbenchParms.Db))
	}
	return dropSize
}
func GetGcLeaderChaos(ctx context.Context, tc resource.TiDBCluster) chaos.Chaos {
	instanceId := Try(oltppkg.GetGcLeader(ctx, tc)).(int)
	c := chaos.Chaos{
		Name:           "gc leader",
		Tc:             tc,
		ExperimentSpec: &resource.FailureExperimentSpec{},
		SelectorOpts: resource.TiDBClusterSelectorOpts{
			TiDBClusterTiDBSelectorSpec: &resource.TiDBClusterTiDBSelectorSpec{
				Peer: oltppkg.ToIntPtr(instanceId),
			},
		},
	}
	return c
}

func GcPostCheck(ctx context.Context, tc resource.TiDBCluster, checkTimeOut time.Duration, dropSize float64, initSumSize float64) {
	log.Info("Current storage size:", zap.Float64("initSumSize", initSumSize), zap.Float64("dropSize", dropSize))
	log.Info("Test finished, check the store size to see if gc works fine.")

	ctx2, cancel2 := context.WithTimeout(ctx, checkTimeOut)
	defer cancel2()
	g, ctxg := errgroup.WithContext(ctx2)
	g.Go(WithGinkgoRecover(func() {
		ConsistentlyWithMetrics(tc, checkTimeOut).
			ShouldNot(GCSafePointIncrease().For(20 * time.Minute).WithCtx(ctxg))
	}))
	g.Go(WithGinkgoRecover(func() {
		Eventually(func() float64 {
			_, curSumSize, err := oltppkg.GetStoreSizeByType(tc, "used_size")
			Expect(err).Should(BeNil())
			log.Info("CurSize", zap.Float64("curSumSize", curSumSize))
			return initSumSize - curSumSize
		}, checkTimeOut, 1*time.Minute).Should(BeNumerically(">=", dropSize/2))
	}))
}

func BrRestoreDataWithoutCheck(ctx context.Context, tc resource.TiDBCluster, br resource.BR, s3Parms oltppkg.S3Parms, caseCmdParms oltppkg.CaseCmdParms) {
	Try(oltppkg.ExecPdCtl(tc, "log debug"))
	defer func() {
		Try(oltppkg.ExecPdCtl(tc, "log info"))
	}()
	ctxbr, cancel := context.WithTimeout(ctx, 6*time.Hour)
	defer cancel()
	db := Try(tc.GetDB("")).(*sql.DB)
	tidbClient := tidb.NewClient(db)
	defer db.Close()

	pdURL := Try(tc.ServiceURL(resource.PDAddr)).(*url.URL)
	brcmd := oltppkg.NewBr("", "restore full", s3Parms.S3AccessKeyID, s3Parms.S3SecretAccessKey)
	oltppkg.LogAndBy("BR restore data for tpcc")
	brTpccCmd := brcmd.Pd(pdURL.String()).Storage(caseCmdParms.TpccParms.Storage).S3endpoint(s3Parms.S3Endpoint).CheckReq(false).Build()
	exist := Try(tidbClient.IsTableExist(ctxbr, caseCmdParms.TpccParms.Db, "new_order")).(bool)
	if !exist {
		Try(oltppkg.ExecWorkloadCmd(ctxbr, br, brTpccCmd, true, nil))
	}

	oltppkg.LogAndBy("BR restore data for sysbench")
	brcmd = oltppkg.NewBr("", "restore full", s3Parms.S3AccessKeyID, s3Parms.S3SecretAccessKey)
	exist = Try(tidbClient.IsTableExist(ctxbr, caseCmdParms.SysbenchParms.Db, "sbtest1")).(bool)
	brSysbenchCmd := brcmd.Pd(pdURL.String()).Storage(caseCmdParms.SysbenchParms.Storage).S3endpoint(s3Parms.S3Endpoint).CheckReq(false).Build()
	if !exist {
		Try(oltppkg.ExecWorkloadCmd(ctxbr, br, brSysbenchCmd, true, nil))
	}
}