package operator_upgrade

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

	. "github.com/onsi/ginkgo"
	"github.com/onsi/gomega"
	"github.com/pingcap/endless/testcase/oltp/oltppkg"
	"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/datagen"
	"github.com/pingcap/endless/pkg/matcher"
	"github.com/pingcap/endless/pkg/util"
	"github.com/pingcap/endless/testcase/tiflash/pkg"
)

const prepareRows = 100 * 100

var _ = Describe("TiFlash", func() {
	Context("tiflash", func() {
		ctx := suiteTestCtx
		var (
			tc                resource.TiDBCluster
			dbName            = "test"
			replicas          int
			duration          time.Duration
			ti                *datagen.TableInfo
			socketConnTimeout func(error) error
		)
		BeforeEach(func() {
			duration = matcher.Try(time.ParseDuration(ctx.Param("duration", "2h"))).(time.Duration)
			// duration = matcher.Try(time.ParseDuration("20m")).(time.Duration)
			replicas = matcher.Try(strconv.Atoi(ctx.Param("replicas", "2"))).(int)
			tc = ctx.Resource("tc").(resource.TiDBCluster)
			socketConnTimeout = func(err error) error {
				if err != nil {
					if strings.Contains(err.Error(), "socks connect tcp") {
						return nil
					}
				}
				return err
			}
			db := matcher.Try(pkg.OpenDB(tc, dbName)).(*sql.DB)
			tidbCli := tidb.NewClient(db)
			matcher.Try(tidbCli.ShowClusterVersionInformation())
			ti = datagen.NewTableInfo(datagen.AutoIncrWideTable, db, false)
			matcher.Try(tidbCli.AlterBaseTablesSetTiFlashReplica(dbName, []string{ti.GetTableName()}, replicas, ""))
			gomega.Eventually(func() bool {
				return matcher.Try(tidbCli.IsTiFlashReplicaAvailable(dbName, []string{ti.GetTableName()})).(bool)
			}, 3*time.Minute, 3*time.Second).Should(gomega.BeTrue())
			count := matcher.Try(pkg.Query(suiteTestCtx, db, fmt.Sprintf("select count(*) from %s", ti.GetTableName()), false)).([]string)
			if matcher.Try(strconv.Atoi(count[0])).(int) > 0 {
				log.Info("skip prepare data", zap.String("rows count", count[0]))
				return
			}
			log.Info("prepare data begin", zap.Int("prepare rows", prepareRows))
			matcher.Try(ti.Run(&datagen.Request{
				Ctx: context.Background(),
				PrepareRequest: &datagen.PrepareRequest{
					TotalRows:   prepareRows,
					InsertBatch: 10,
					Threads:     10,
				}}))
			log.Info("prepare data done", zap.Int("prepare rows", prepareRows))
		})
		/*
			From junshen
			升级相关，tiflash 的冒烟测试流程包含下面几步应该会比较好：
			1. 在旧版本上创建表 t0 导入 1000 条数据，建 tiflash 副本，等 available，通过隔离读指定 tiflash 核对查询结果（条数、某个列的 sum）
			2. 升级 tiflash，并观察 tiflash 是否有 crash
			3. 再进行查询 t0，通过隔离读指定 tiflash 核对查询结果
			4. 往 t0 插入或更新一些数据，核对查询结果
			5. 创建另外一个表 t1 导入 1000 条数据，建 tiflash replica 等 available，通过隔离读指定 tiflash 核对查询结果
		*/
		// 期望不出现 invalid connection，socket timeout 等之外的问题
		It("[tiflash]should be consistency with data and available for the write and read #001", func() {
			ctx, _ := context.WithTimeout(suiteTestCtx, duration)
			g, c := errgroup.WithContext(ctx)
			g.Go(util.WithGinkgoRecover(func() {
				matcher.Try(ti.Run(&datagen.Request{
					Ctx: c,
					DmlReqs: []*datagen.DMLRequest{
						{
							DmlType:      datagen.Insert,
							Threads:      10,
							InsertBatch:  10,
							ErrorHandler: pkg.ErrorHandler(pkg.CtxDoneHandler, pkg.InvalidConnectionHandler, socketConnTimeout),
						},
						{
							DmlType:      datagen.Update,
							Threads:      5,
							Limit:        1,
							ErrorHandler: pkg.ErrorHandler(pkg.CtxDoneHandler, pkg.InvalidConnectionHandler, socketConnTimeout),
						},
						{
							DmlType:      datagen.Delete,
							Threads:      1,
							Limit:        1,
							ErrorHandler: pkg.ErrorHandler(pkg.CtxDoneHandler, pkg.InvalidConnectionHandler, socketConnTimeout),
						},
					}}))
			}))
			g.Go(util.WithGinkgoRecover(func() {
				db2 := matcher.Try(pkg.OpenDB(tc, dbName)).(*sql.DB)
				for {
					err := pkg.DiffTiFlashTiKV(c, db2, fmt.Sprintf("select count(*) from %s", ti.GetTableName()), true)
					if err != nil {
						log.Error("check consistence error", zap.Error(err))
						if pkg.CtxDoneHandler(err) == nil {
							return
						}
						matcher.Try(pkg.ErrorHandler(pkg.InvalidConnectionHandler, socketConnTimeout)(err))
					}
				}
			}))
			g.Go(util.WithGinkgoRecover(func() {
				log.Info("stable run 20 minutes before upgrade")
				time.Sleep(20 * time.Minute)
				oltppkg.CheckClusterHealth(tc)
			}))
			matcher.Try(g.Wait())

		})

	})
})