package concurrent_ddl_test

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

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	_ "github.com/pingcap/endless/pkg/client/tidb"
	. "github.com/pingcap/endless/pkg/matcher"
	"github.com/pingcap/endless/pkg/util"
	. "github.com/pingcap/endless/testcase/upgrade/utils"
	"github.com/pingcap/endless/testcase/upgrade/workload"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	"golang.org/x/sync/errgroup"
)

type ddlJobs struct {
	jobId       int
	dbName      string
	tableName   string
	jobType     string
	schemaState string
	schemaId    int
	tableId     int
	rowCount    int
	createTime  []uint8
	startTime   []uint8
	endTime     []uint8
	state       string
}

var _ = Describe("Cluster", func() {
	Describe("[precheck]", func() {
		var (
			tc  resource.TiDBCluster
			db  *sql.DB
			ctx context.Context
		)
		BeforeEach(func() {
			tc = suiteTestCtx.Resource("tc").(resource.TiDBCluster)
			db, _ = tc.GetDB("test")
			ctx, _ = context.WithTimeout(suiteTestCtx, 60*time.Minute)
		})

		It("should execute all kind of ddl success before upgrade", func() {
			if CheckVersion(db, ">= 5.3.0") {
				supportDDL = append(supportDDL, placeSql...)
			}
			for _, sql := range supportDDL {
				log.Info("execute sql", zap.String("sql", sql))
				_, err := db.Exec(sql)
				Expect(err).ShouldNot(HaveOccurred())
			}
		})

		It("can cancel add index and column before upgrade", func() {
			Try(db.Exec("drop database if exists sbtest1;"))
			wload := workload.NewWorkload(suiteTestCtx, "sbtest1")
			wload.Prepare(ctx)
			g, _ := errgroup.WithContext(ctx)

			for i := 1; i < 5; i++ {
				indexSql := fmt.Sprintf("alter table sbtest1.sbtest%d add index idx(k);", i)
				g.Go(util.WithGinkgoRecover(func() {
					runSql(db, indexSql)
				}))
			}
			// some version (before 5.1) unsupported column type from char to varchar
			if CheckVersion(db, ">= 5.1.0") {
				for i := 1; i < 10; i++ {
					columnSql := fmt.Sprintf("alter table sbtest1.sbtest%d modify c varchar(130)", i)
					g.Go(util.WithGinkgoRecover(func() {
						runSql(db, columnSql)
					}))
				}
			}
			//wait some time to cancel the ddl, so there will have running and finished jobs
			time.Sleep(2 * time.Minute)
			count := getDDlInfo(ctx, db, "")
			for _, jobInfo := range count {
				//FIXME: maybe it become synced
				if jobInfo.state != "synced" {
					log.Info("cancel ddl", zap.Object("job_info", &jobInfo))
					cancelSql := fmt.Sprintf("admin cancel ddl jobs %d", jobInfo.jobId)
					log.Info("execute sql", zap.String("sql", cancelSql))
					_, err := db.Exec(cancelSql)
					Expect(err).ShouldNot(HaveOccurred())
				}
			}
			time.Sleep(1 * time.Minute)
			dhs := getDDlInfo(ctx, db, "admin show ddl jobs where state in ('cancelled', 'rollback done');")
			Expect(len(dhs) > 0).To(BeTrue(), "admin cancel ddl should works")
		})
	})

	Describe("[postcheck]", func() {
		var (
			tc     resource.TiDBCluster
			db     *sql.DB
			ctx    context.Context
			cancel context.CancelFunc
		)
		BeforeEach(func() {
			tc = suiteTestCtx.Resource("tc").(resource.TiDBCluster)
			db, _ = tc.GetDB("test")
			ctx, cancel = context.WithTimeout(suiteTestCtx, 60*time.Minute)
		})
		AfterEach(func() {
			cancel()
		})

		It("cluster should works well and ddl history is correct", func() {
			count := getDDlInfo(ctx, db, "")
			Expect(len(count)).To(Equal(10))
		})

		It("ddl can work concurrent run", func() {
			g, _ := errgroup.WithContext(ctx)
			var now string
			db.QueryRow("select now()").Scan(&now)

			for i := 1; i < 11; i++ {
				columnSql := fmt.Sprintf("alter table sbtest1.sbtest%d change column c d varchar(120)", i)
				g.Go(util.WithGinkgoRecover(func() {
					runSql(db, columnSql)
				}))
			}
			for i := 1; i < 11; i++ {
				indexSql := fmt.Sprintf("alter table sbtest1.sbtest%d add index idx1(pad)", i)
				g.Go(util.WithGinkgoRecover(func() {
					runSql(db, indexSql)
				}))
			}
			// both add index and change column is re-org job
			time.Sleep(30 * time.Second)
			err := CheckConcurrentRunning(ctx, db)
			Expect(err).ShouldNot(HaveOccurred())

			syncedSql := fmt.Sprintf("admin show ddl jobs 100 where state = 'synced' and db_name = 'sbtest1' and start_time > '%s';", now)
			synced := getDDlInfo(ctx, db, syncedSql)
			Expect(len(synced)).To(Equal(20), "the ddl should synced")
		})

		It("should execute all kind of ddl success after upgrade", func() {
			if CheckVersion(db, ">= 5.3.0") {
				supportDDL = append(supportDDL, placeSql...)
			}
			for _, sql := range supportDDL {
				log.Info("execute sql", zap.String("sql", sql))
				_, err := db.Exec(sql)
				Expect(err).ShouldNot(HaveOccurred())
			}
		})

		It("can cancel ddl after upgrade", func() {
			g, _ := errgroup.WithContext(ctx)
			for i := 1; i < 5; i++ {
				indexSql := fmt.Sprintf("alter table sbtest1.sbtest%d add index idx2(k, pad);", i)
				g.Go(util.WithGinkgoRecover(func() {
					runSql(db, indexSql)
				}))
			}
			for i := 1; i < 5; i++ {
				columnSql := fmt.Sprintf("alter table sbtest1.sbtest%d change column d d varchar(125)", i)
				g.Go(util.WithGinkgoRecover(func() {
					runSql(db, columnSql)
				}))
			}
			//wait some time to cancel the ddl, so there will have running and finished jobs
			time.Sleep(1 * time.Minute)
			count := getDDlInfo(ctx, db, "")
			for _, jobInfo := range count {
				//FIXME: maybe it become synced
				if jobInfo.state != "synced" {
					log.Info("cancel ddl", zap.Object("job_info", &jobInfo))
					cancelSql := fmt.Sprintf("admin cancel ddl jobs %d", jobInfo.jobId)
					log.Info("execute sql", zap.String("sql", cancelSql))
					_, err := db.Exec(cancelSql)
					Expect(err).ShouldNot(HaveOccurred())
				}
			}
			time.Sleep(1 * time.Minute)
			dhs := getDDlInfo(ctx, db, "admin show ddl jobs 10 where state in ('cancelled', 'rollback done');")
			Expect(len(dhs) > 0).To(BeTrue(), "admin cancel ddl should works")
		})
	})
})

var supportDDL = []string{
	// database
	"create database ddl1;",
	"drop database ddl1;",
	"drop database if exists ddl2;",
	"create database ddl2;",
	"use ddl2;",
	"alter database ddl2 default character set = utf8mb4 collate utf8mb4_bin;",

	//table
	"create table t(a bigint primary key, b char(10));",
	"create table t1 like t;",
	"rename table t1 to t2;",

	//view
	"create view v1 as select a from t;",
	"drop view v1;",

	// column
	"alter table t add column c int not null;",
	"alter table t add column c1 int not null after c;",
	"alter table t drop column b;",
	"alter table t modify column c bigint;",
	"alter table t change column c d bigint;",

	//stats
	"analyze table t;",
	"analyze table t with 10 buckets;",
	// "analyze table t columns a;",

	//binding
	"create session binding for select * from t where d = 10 using select * from t ignore index (primary) where d = 10;",
	"create global binding for select * from t where d > 100 using select * from t use index (primary) where d > 100;",
	"drop session binding for select * from t where d = 10;",

	// index
	"alter table t add index idx (c1);",
	"alter table t alter index idx invisible;",
	"alter table t rename index idx to idx1;",
	"alter table t drop index idx1;",
	"create index idx2 on t(d);",
	"drop index idx2 on t;",

	//sequence
	"CREATE SEQUENCE seq2 start 3 increment 2 minvalue 1 maxvalue 10 cache 3;",
	"drop sequence seq2;",

	// tiflash
	// "alter table t set tiflash replica 1;",
	// "alter table t compact tiflash replica;",
}

//since 5.3
var placeSql = []string{
	// placement rule
	"create placement policy p1 followers = 4;",
	"alter placement policy p1 followers = 5;",
	"drop placement policy p1;",
}

func (di *ddlJobs) MarshalLogObject(enc zapcore.ObjectEncoder) error {
	enc.AddInt("job_id", di.jobId)
	enc.AddInt("schema_id", di.schemaId)
	enc.AddString("db_name", di.dbName)
	enc.AddInt("table_id", di.tableId)
	enc.AddString("table_name", di.tableName)
	enc.AddInt("row_count", di.rowCount)
	enc.AddString("job_type", di.jobType)
	enc.AddString("schema_state", di.schemaState)
	enc.AddString("state", di.state)
	enc.AddString("create_time", string(di.createTime))
	enc.AddString("end_time", string(di.endTime))
	enc.AddString("start_time", string(di.startTime))

	return nil
}

func getDDlInfo(ctx context.Context, db *sql.DB, sql string) []ddlJobs {
	if sql == "" {
		sql = "admin show ddl jobs;"
	}
	rows, err := db.QueryContext(ctx, sql)
	var jobInfo ddlJobs
	var count []ddlJobs
	Expect(err).ShouldNot(HaveOccurred())
	for rows.Next() {
		if CheckVersion(db, ">= 6.0.0") {
			err = rows.Scan(&jobInfo.jobId, &jobInfo.dbName, &jobInfo.tableName, &jobInfo.jobType, &jobInfo.schemaState, &jobInfo.schemaId, &jobInfo.tableId, &jobInfo.rowCount, &jobInfo.createTime, &jobInfo.startTime, &jobInfo.endTime, &jobInfo.state)
		} else {
			err = rows.Scan(&jobInfo.jobId, &jobInfo.dbName, &jobInfo.tableName, &jobInfo.jobType, &jobInfo.schemaState, &jobInfo.schemaId, &jobInfo.tableId, &jobInfo.rowCount, &jobInfo.startTime, &jobInfo.endTime, &jobInfo.state)
		}
		Expect(err).ShouldNot(HaveOccurred())
		Expect(jobInfo).ShouldNot(BeNil())
		count = append(count, jobInfo)
	}
	return count
}

func runSql(db *sql.DB, sql string) {
	log.Info("execute sql", zap.String("sql", sql))
	// the case will cancel some ddl jobs, so we should ignore the error
	_, err := db.Exec(sql)
	if err != nil {
		if err.Error() == "Error 8214: Cancelled DDL job" {
			log.Info("ddl job cancelled")
		} else {
			Expect(err).ShouldNot(HaveOccurred())
		}
	}
}

func CheckConcurrentRunning(ctx context.Context, db *sql.DB) error {
	log.Info("start check concurrent running")
	sql1 := "admin show ddl jobs where state = 'running' and job_type in ('add index', 'modify column');"
	sql2 := "admin show ddl jobs where state = 'queueing' and job_type in ('add index', 'modify column');"

	for {
		c1 := getDDlInfo(ctx, db, sql1)
		c2 := getDDlInfo(ctx, db, sql2)

		tables := make(map[int]struct{})
		if len(c1) == 0 && len(c2) == 0 {
			break
		}

		for _, ddlInfo := range c1 {
			if _, ok := tables[ddlInfo.tableId]; !ok {
				tables[ddlInfo.tableId] = struct{}{}
			} else {
				return fmt.Errorf("reorg ddl for table %s.%s are runing in concurrent", ddlInfo.dbName, ddlInfo.tableName)
			}
		}

		// when there is a queueing job, the concurrent running job should be larger than 1
		var count int
		if len(c2) > 0 && len(c1) < 2 {
			count++
			log.Info("ddl is not concurrent runing, check it again.")
			if count > 5 {
				return fmt.Errorf("concurrent running job should be larger than 1 while there is a queueing job, corrent runing is %d, queueing is %d", len(c1), len(c2))
			}
		}
	}
	return nil
}