package high_currency_test

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

	. "github.com/onsi/ginkgo"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"

	"github.com/pingcap/endless/pkg/client/tidb"
	"github.com/pingcap/endless/pkg/matcher"
	"github.com/pingcap/endless/pkg/util"
	"github.com/pingcap/endless/testcase/tiflash/pkg"
	"github.com/pingcap/endless/testcase/tiflash/pkg/checker"
	workload2 "github.com/pingcap/endless/testcase/tiflash/pkg/workload"
	"github.com/pingcap/endless/testcase/tiflash/pkg/workload/workloads"
)

var _ = Describe("TiFlash", func() {
	var (
		tc resource.TiDBCluster
		wl workload2.Workload
	)
	BeforeEach(func() {
		tc = suiteTestCtx.Resource("tc").(resource.TiDBCluster)
		wlType := workloads.WorkloadType(suiteTestCtx.Param("workload", string(workloads.TpchWorkloadType)))
		if w, ok := workloads.HtapWorkloads[wlType]; ok {
			wl = w(suiteTestCtx)
		} else {
			Skip("don't have the workload " + string(wlType))
		}
		tc = suiteTestCtx.Resource("tc").(resource.TiDBCluster)
		matcher.Try(pkg.DisableExpressionPushdown(tc, []string{"date_add"}, []string{"tiflash"}))
		wl.Prepare(suiteTestCtx)
		// the database is created after prepare
		db := matcher.Try(tc.GetDB(wl.DBName())).(*sql.DB)
		tidbClient := tidb.NewClient(db)
		defer tidbClient.Close()
		matcher.Try(tidbClient.ShowClusterVersionInformation())
	})
	// 高并发，query有一定复杂度（有多个join），数据量小，整体持续一段时间（有pattern，并发一段时间，再静止一段时间），观测点是 thread pool频繁的伸缩，功能正常不crash
	// 正常的高并发(固定高并发数目），查询不限制内存，且在一个低一点 currency
	Context("when inject high currency query", func() {
		It("should not crash #001#", func() {
			round := matcher.Try(strconv.Atoi(suiteTestCtx.Param("round", "20"))).(int)
			runTime := matcher.Try(time.ParseDuration(suiteTestCtx.Param("runtime", "10m"))).(time.Duration)
			intervalTime := matcher.Try(time.ParseDuration(suiteTestCtx.Param("interval", "6m"))).(time.Duration)
			ctx, cancel := context.WithCancel(suiteTestCtx)
			g, _ := errgroup.WithContext(ctx)
			g.Go(util.WithGinkgoRecover(func() {
				defer cancel()
				matcher.Try(wl.RunTP(ctx, pkg.ErrorHandler(
					pkg.TiDBStaleSchemaHandler,
					pkg.DeadlockFoundHandler,
					pkg.LockWaitTimeoutHandler)))
			}))
			g.Go(util.WithGinkgoRecover(func() {
				defer cancel()
				for i := 0; i < round; i++ {
					log.Info("high currency query start", zap.Int("round", i))
					ctx1, _ := context.WithTimeout(context.Background(), runTime)
					// olap_threads param
					matcher.Try(wl.RunAP(ctx1, pkg.ErrorHandler(pkg.CtxDoneHandler)))
					log.Info("high currency query end and wait some time", zap.Int("round", i), zap.Duration("wait time", intervalTime))
					time.Sleep(intervalTime)
				}
				log.Info("high currency query stop")
			}))
			matcher.Try(g.Wait())
		})
		// tpch 高并发 20 -> 50 -> 100
		// 同时设置内存限制，这样就会触发 cancel query
		// 预期是 1. tiflash 不会 crash; 2. tiflash mvcc snapshot lifetime 不会超过 30 分钟; 3. 查询不会超时（默认 1h 超时）
		It("should not crash #002#", func() {
			enableMpp := matcher.Try(strconv.ParseBool(suiteTestCtx.Param("mpp", "true"))).(bool)
			if enableMpp {
				wl.SetApConnParam(append(pkg.OlapConParams, "tidb_allow_mpp=on"))
			} else {
				wl.SetApConnParam(append(pkg.OlapConParams, "tidb_allow_mpp=off"))
			}
			// wlName -> tpch
			currency := []int{20, 40, 60, 80, 100, 200}
			ctx, cancel := context.WithCancel(suiteTestCtx)
			g, _ := errgroup.WithContext(ctx)
			g.Go(util.WithGinkgoRecover(func() {
				defer cancel()
				for _, c := range currency {
					log.Info("high currency query start", zap.Int("currency", c))
					ctx1, _ := context.WithTimeout(context.Background(), 1*time.Hour)
					wl.SetApThreads(c)
					// ignore out of memory quota error
					matcher.Try(wl.RunAP(ctx1, pkg.SkipErrorHandler))
					log.Info("high currency query end and wait some time", zap.Int("currency", c), zap.Duration("wait time", 20*time.Minute))
					time.Sleep(10 * time.Minute)
				}
				log.Info("high currency query stop")
			}))
			checker.CommonCheckWithoutLatencyAndRestart(ctx, tc, g, 7*time.Hour)
			matcher.Try(g.Wait())
		})
	})
})