package gc

import (
	"context"
	"database/sql"
	"flag"
	"fmt"
	"net/url"
	"time"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	. "github.com/pingcap/endless/pkg/matcher"
	"github.com/pingcap/endless/pkg/workloadnode"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"
	"go.uber.org/zap"

	. "github.com/pingcap/endless/pkg/util"
)

var _ = ParameterizedDescribe("InternalTxnAffectGCTest", func(flagSet *flag.FlagSet) {
	var (
		s3Endpoint        string
		s3AccessKeyID     string
		s3SecretAccessKey string

		storage   string
		dbname    string
		warehouse int
		threads   int
		duration  time.Duration
	)

	flagSet.StringVar(&s3Endpoint, "s3-endpoing", "http://minio.pingcap.net:9000", "")
	flagSet.StringVar(&s3AccessKeyID, "s3-key", "minioadmin", "")
	flagSet.StringVar(&s3SecretAccessKey, "s3-secret", "minioadmin", "")

	flagSet.StringVar(&storage, "storage", "s3://benchmark/tpcc-1000-4.0", "")
	flagSet.StringVar(&dbname, "db", "tpcc", "")
	flagSet.IntVar(&warehouse, "warehouse", 1000, "")
	flagSet.IntVar(&threads, "threads", 1024, "Threads number.")
	flagSet.DurationVar(&duration, "duration", 3*time.Hour, "Run duration.")

	ctx := suiteTestCtx
	var (
		tc    resource.TiDBCluster
		br    resource.BR
		goTpc resource.WorkloadNode
	)
	// Testbed is testbed/testbed_multizone.yaml.
	Context("with one AZ Down", func() {

		brCmdTpl := `
export AWS_ACCESS_KEY_ID=%s AWS_SECRET_ACCESS_KEY=%s
/br restore full --pd "%s" \
	--storage "%s" \
	--s3.endpoint "%s" \
	--send-credentials-to-tikv=true --check-requirements=false
`
		goTpcCmdTpl := `/go-tpc tpcc run --host %s -P4000 --warehouses %d -D %s -T %d`

		BeforeEach(func() {
			tc = ctx.Resource("tc").(resource.TiDBCluster)
			br = ctx.Resource("br").(resource.BR)
			goTpc = ctx.Resource("gotpc").(resource.WorkloadNode)
		})

		Context("GC test", func() {
			TestCase("internal txn prevent safepoint advance until tidb_gc_max_wait_time", "gc_test_1", func() {
				ctx2, cancel := context.WithTimeout(ctx, duration)
				defer cancel()

				By("get db")
				dbURL := Try(tc.ServiceURL(resource.DBAddr)).(*url.URL)
				db := Try(tc.GetDB("")).(*sql.DB)

				By("set system variable")
				_, err := db.Exec("set global tidb_gc_max_wait_time=1200")
				Expect(err).Should(BeNil())
				var gcMaxWaitTime int
				err = db.QueryRow("select @@tidb_gc_max_wait_time").Scan(&gcMaxWaitTime)
				Expect(err).Should(BeNil())
				Expect(gcMaxWaitTime).To(Equal(1200))

				By("check safepoint")
				CheckSafePoint(ctx2, db, 20*time.Minute)

				By("Preparing TPC-C data")
				_, err = db.ExecContext(ctx2, fmt.Sprintf("drop database if exists %s", dbname))
				Expect(err).Should(BeNil())
				_, err = db.ExecContext(ctx2, fmt.Sprintf("create database %s", dbname))
				Expect(err).Should(BeNil())
				pdURL := Try(tc.ServiceURL(resource.PDAddr)).(*url.URL)
				s := fmt.Sprintf(brCmdTpl, s3AccessKeyID, s3SecretAccessKey, pdURL.Host, storage, s3Endpoint)
				cmd := workloadnode.CommandContext(ctx2, s)
				Try(cmd.Run(br))

				By("run TPC-C")
				s = fmt.Sprintf(goTpcCmdTpl, dbURL.Hostname(), warehouse, dbname, threads)
				cmd = workloadnode.CommandContext(ctx2, s)
				Try(cmd.Run(goTpc))

				By("set system variable default value")
				_, err = db.Exec("set global tidb_gc_max_wait_time=86400")
				Expect(err).Should(BeNil())
				err = db.QueryRow("select @@tidb_gc_max_wait_time").Scan(&gcMaxWaitTime)
				Expect(err).Should(BeNil())
				Expect(gcMaxWaitTime).To(Equal(86400))
			})
		})
	})
})

func CheckSafePoint(ctx context.Context, db *sql.DB, checkIntervalTime time.Duration) {
	// get gc safepoint
	var firstTikvGCSafePoint string
	err := db.QueryRow("select VARIABLE_VALUE from mysql.tidb where VARIABLE_NAME='tikv_gc_safe_point'").Scan(&firstTikvGCSafePoint)
	Expect(err).Should(BeNil())
	log.Info("", zap.Any("firstTikvGCSafePoint: ", firstTikvGCSafePoint))

	go func() {
		for {
			select {
			case <-ctx.Done():
				return
			default:
			}

			// get gc safepoint every checkIntervalTime
			time.Sleep(checkIntervalTime)
			var nextTikvGCSafePoint string
			err := db.QueryRow("select VARIABLE_VALUE from mysql.tidb where VARIABLE_NAME='tikv_gc_safe_point'").Scan(&nextTikvGCSafePoint)
			Expect(err).Should(BeNil())
			log.Info("", zap.Any("nextTikvGCSafePoint: ", nextTikvGCSafePoint))

			// check if gc safepoint is advancing
			time1, err := time.Parse("20060102-15:04:05 +0000", firstTikvGCSafePoint)
			Expect(err).Should(BeNil())
			time2, err := time.Parse("20060102-15:04:05 +0000", nextTikvGCSafePoint)
			Expect(err).Should(BeNil())
			if time2.Sub(time1) <= 0 {
				log.Error("safe point do not proceed.", zap.Any("firstTikvGCSafePoint", firstTikvGCSafePoint), zap.Any("nextTikvGCSafePoint", nextTikvGCSafePoint))
				Expect(time2).Should(BeTemporally(">", time1))
			}
			firstTikvGCSafePoint = nextTikvGCSafePoint
		}
	}()
}