package insert_ap_test

import (
	"database/sql"
	"flag"
	"fmt"
	"math/rand"
	"net/url"
	"strconv"
	"strings"
	"time"

	_ "github.com/go-sql-driver/mysql"
	. "github.com/onsi/ginkgo"

	"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/workload/tpch"

	"github.com/pingcap/test-infra/sdk/resource"
	_ "github.com/pingcap/test-infra/sdk/resource/impl/k8s"
	"go.uber.org/zap"
)

// skip q16 for tiflash not support create view
const apQueries = "q1,q2,q3,q4,q5,q6,q7,q8,q9,q10,q11,q12,q13,q14,q15,q17,q18,q19,q20,q21,q22"

var _ = util.ParameterizedDescribe("TiFlashInsertAp", func(flagSet *flag.FlagSet) {
	defer zap.S().Sync()

	var (
		duration       time.Duration
		warmupDuration time.Duration
		sf             int
		threads        int
		replicas       int
	)

	flagSet.DurationVar(&duration, "duration", time.Hour*10, "Run duration.")
	flagSet.DurationVar(&warmupDuration, "warmupDuration", time.Second*60, "Run duration.")
	flagSet.IntVar(&threads, "threads", 2, "Threads number.")
	flagSet.IntVar(&sf, "sf", 100, "SF.")
	flagSet.IntVar(&replicas, "replicas", 1, "TiFlash replicas.")

	Context("TiFlash Ap Query under heavy insert workload", func() {
		var (
			tc    resource.TiDBCluster
			goTpc resource.WorkloadNode
			dbDSN string
		)
		ctx := suiteTestCtx
		BeforeEach(func() {
			tc = ctx.Resource("tc").(resource.TiDBCluster)
			goTpc = ctx.Resource("gotpc").(resource.WorkloadNode)
			dbURL := matcher.Try(tc.ServiceURL(resource.DBAddr)).(*url.URL)
			dbDSN = fmt.Sprintf("root@tcp(%s)/test?tidb_isolation_read_engines='tiflash'&tidb_mem_quota_query=10737418240&multiStatements=true", dbURL.Host)
		})

		It("Ap query should not fail", func() {
			db := matcher.Try(sql.Open("mysql", dbDSN)).(*sql.DB)
			tidbCli := tidb.NewClient(db)
			dsn := matcher.Try(tc.ServiceURL(resource.DBAddr)).(*url.URL)
			zap.L().Info("resources initialized.", zap.String("DBAddr", dsn.String()))
			go prepare(*dsn, goTpc, sf)
			time.Sleep(warmupDuration)
			zap.L().Info("connecting database", zap.String("DSN", dsn.String()), zap.String("correctDSN", dbDSN))

			db, err := sql.Open("mysql", dbDSN)

			if err != nil {
				zap.L().Error("errored when connecting to database", zap.Error(err))
				return
			}
			defer db.Close()
			tbls, err := tidbCli.ShowTableNames("test")
			if err != nil {
				zap.L().Error(err.Error())
				return
			}

			err = tidbCli.AlterBaseTablesSetTiFlashReplica("test", tbls, replicas, "")
			if err != nil {
				zap.L().Error(err.Error())
				return
			}

			qs := strings.Split(apQueries, ",")
			for i := 0; i < threads; i++ {
				go func() {
					for {
						start := time.Now()
						q := rand.Intn(len(qs))
						row, err := db.Query(tpch.Queries[q].Query)
						if row != nil {
							row.Close()
						}
						if err != nil {
							if strings.Contains(err.Error(), "Can not find access path matching 'tidb_isolation_read_engines'") {
								continue
							}
							zap.L().Error("executing AP query against TiFlash emitted error", zap.String("queryNo", "q"+strconv.Itoa(q+1)), zap.Error(err))
						} else {
							zap.L().Info("executing query successfully.", zap.String("queryNo", qs[q]), zap.String("duration", time.Now().Sub(start).String()))
						}
					}
				}()
			}
			time.Sleep(duration)
		})
	})
})

func prepare(dsn url.URL, goTpc resource.WorkloadNode, sf int) {
	options := resource.WorkloadNodeExecOptions{Command: fmt.Sprintf(`/go-tpc tpch --host %s -D test -P4000  -T 100 --sf %d prepare`, dsn.Hostname(), sf)}
	zap.L().Info("preparing TPC-H", zap.Any("options", options))

	stdout, stderr, ec, err := goTpc.Exec(options)
	if err != nil || ec != 0 {
		zap.L().Error("TPC-H prepare with error", zap.String("stdout", stdout), zap.String("stderr", stderr), zap.Int("exitcode", ec), zap.Error(err))
	}
}