package main

import (
	"bytes"
	"database/sql"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"text/template"
	"time"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	data "github.com/pingcap/schrodinger-test/pkg/ddl_data"
	"github.com/pingcap/schrodinger-test/util"
	"github.com/pingcap/schrodinger/pkg/test_client"
	"golang.org/x/net/context"
)

type replicationMySQL struct {
	hosts   []string
	current int
}

type dbExecutor struct {
	db            *sql.DB
	dbStandby     *sql.DB
	dbName        string
	dbInfo        *data.DBInfo
	standbyDBInfo *data.DBInfo
	tables        []string
}

type caseOptions struct {
	ddlCount          int
	mysqlFlavor       string
	taskMode          string
	enableSharding    bool
	masterSlaveSwitch bool
	cmptDDL           bool
	onlineDDL         bool
	onlineDDLTool     string
	filterDDL         bool
	enableGTID        bool
	failResume        bool
}

// mssOptions means master slave switch options for a dm case
type mssOptions struct {
	leader  bool // whether is leader, only leader needs to update db config
	dbIndex int  // use dbIndex to get db
	abort   bool // perform master-slave-switch or not
}

// DMCase keeps a test case, used for running a DM task
type DMCase struct {
	caseID       int
	master       string
	workers      []string
	mysqlReplica map[int]*replicationMySQL
	caseOpts     *caseOptions

	ctx    context.Context
	cancel context.CancelFunc

	sync.RWMutex
	wg    sync.WaitGroup
	jobWg sync.WaitGroup
	dmlWg sync.WaitGroup

	nemesis *nemesis
	pause   chan struct{}

	sqlExecutors []*executor
	sqlDB        []*sql.DB
	tidb         *sql.DB

	tidbAddr string

	ddlIndex int
	ddls     []string
	ddlCh    chan string
	ddlCli   *data.Client
	pauseDML chan struct{}
	errCh    chan error
}

// NewDMCase returns a new DMCase object
func NewDMCase(ctx context.Context, caseID int, master string, workers []string, mysqls map[int]*replicationMySQL, caseOpts *caseOptions, nemesis *nemesis) (*DMCase, error) {
	cctx, cancel := context.WithCancel(ctx)
	copyMysqls := make(map[int]*replicationMySQL)
	for id, replic := range mysqls {
		copyMysqls[id] = replic.clone()
	}
	return &DMCase{
		ctx:          cctx,
		cancel:       cancel,
		caseID:       caseID,
		master:       master,
		workers:      workers,
		mysqlReplica: copyMysqls,
		caseOpts:     caseOpts,
		nemesis:      nemesis,

		pause:    make(chan struct{}),
		ddlCh:    make(chan string, 1),
		pauseDML: make(chan struct{}),
		errCh:    make(chan error, 2),
	}, nil
}

// Init inits DMCase
func (dmc *DMCase) Init(cli *client.Client, dmctl *Dmctl) error {
	var err error

	tidbIP := strings.Split(cli.TiDBService(), ":")[0]
	mysqls := make([]string, 0, len(dmc.mysqlReplica))
	dbInfo := make([]*data.DBInfo, 0, len(dmc.mysqlReplica))
	for i := 0; i < len(dmc.mysqlReplica); i++ {
		instID := i + 1
		mysql, ok := dmc.mysqlReplica[instID]
		if !ok {
			return errors.NotFoundf("instance %d in case %d mysqlReplica", instID, dmc.caseID)
		}
		mysqls = append(mysqls, mysql.hosts[0])
		dbInfo = append(dbInfo, data.NewDBInfo("root", "", mysql.hosts[0], 3306))
	}

	err = dmc.createTaskConfig(tidbIP, mysqls)
	if err != nil {
		return errors.Annotate(err, "[dm] create task config with error")
	}

	dbs, err := dmc.initialDBs(tidbIP, mysqls)
	if err != nil {
		return errors.Annotate(err, "[dm] init dbs with error")
	}
	dmc.sqlDB = dbs[1:]
	dmc.tidb = dbs[0]
	dmc.tidbAddr = tidbIP

	var dbStandby []*sql.DB
	var standbyDBInfo []*data.DBInfo
	if dmc.caseOpts.masterSlaveSwitch {
		standbys := make([]string, 0, len(dmc.mysqlReplica))
		standbyDBInfo = make([]*data.DBInfo, 0, len(dmc.mysqlReplica))
		for i := 0; i < len(dmc.mysqlReplica); i++ {
			instID := i + 1
			mysql, ok := dmc.mysqlReplica[instID]
			if !ok {
				return errors.NotFoundf("instance %d in case %d mysqlReplica", instID, dmc.caseID)
			}
			standbys = append(standbys, mysql.hosts[1])
			standbyDBInfo = append(standbyDBInfo, data.NewDBInfo("root", "", mysql.hosts[1], 3306))
		}
		dbStandby, err = initialMySQLDBs(standbys, dmc.sourceDB())
		if err != nil {
			return errors.Annotatef(err, "[dm] init standby dbs with error")
		}
	} else {
		dbStandby = make([]*sql.DB, len(dmc.mysqlReplica))
		standbyDBInfo = make([]*data.DBInfo, len(dmc.mysqlReplica))
	}

	for idx, db := range dbs[1:] {
		instID := idx + 1
		dmc.addExecutor(&dbExecutor{
			db:            db,
			dbStandby:     dbStandby[idx],
			dbName:        dmc.sourceDB(),
			dbInfo:        dbInfo[idx],
			standbyDBInfo: standbyDBInfo[idx],
			tables:        getSubTaskTables(dmc.caseID, instID),
		}, &executorOpts{
			onlineDDL:     dmc.caseOpts.onlineDDL,
			onlineDDLTool: dmc.caseOpts.onlineDDLTool,
			filterDDL:     dmc.caseOpts.filterDDL,
		})
	}

	ancestorCli := data.NewClient(dmc.ctx)
	ddlTypes := []int{data.AddColumn, data.DeletColumn}
	if dmc.caseOpts.cmptDDL {
		ddlTypes = append(ddlTypes, []int{data.Flush, data.OptimizeTable}...)
	}
	if dmc.caseOpts.failResume {
		ddlTypes = append(ddlTypes, []int{data.AddAndAlterColumn, data.AddWithIndexAndDeleteColumn}...)
	}
	indexTypes := []int{data.NormalIndex}
	ddls := ancestorCli.CreateRandomDDLData(dmc.caseOpts.ddlCount, ddlTypes, indexTypes)
	dmc.ddls = ddls
	dmc.ddlCli = ancestorCli

	dmc.nemesis.Lock()
	dmc.nemesis.configs[dmc.caseID] = dmc.taskConfigName()
	dmc.nemesis.Unlock()

	dmc.nemesis.dmlMu.Lock()
	dmc.nemesis.dmlCh[dmc.caseID] = make(chan bool, 1)
	dmc.nemesis.dmlStat[dmc.caseID] = 0
	dmc.nemesis.dmlMu.Unlock()

	dmc.nemesis.checkMu.Lock()
	dmc.nemesis.checkCh[dmc.caseID] = make(chan bool, 1)
	dmc.nemesis.checkStat[dmc.caseID] = 0
	dmc.nemesis.checkMu.Unlock()

	dmc.nemesis.emptyMu.Lock()
	dmc.nemesis.emptyCh[dmc.caseID] = make(chan bool, 1)
	dmc.nemesis.emptyMu.Unlock()

	dmc.nemesis.mssG.precheck.Lock()
	dmc.nemesis.mssG.precheck.ch[dmc.caseID] = make(chan int, 1)
	dmc.nemesis.mssG.precheck.Unlock()

	dmc.nemesis.mssG.ready.Lock()
	dmc.nemesis.mssG.ready.ch[dmc.caseID] = make(chan bool, 1)
	dmc.nemesis.mssG.ready.Unlock()

	dmc.nemesis.mssG.switch1.Lock()
	dmc.nemesis.mssG.switch1.ch[dmc.caseID] = make(chan *mssOptions, 1)
	dmc.nemesis.mssG.switch1.Unlock()

	dmc.nemesis.mssG.switch2.Lock()
	dmc.nemesis.mssG.switch2.ch[dmc.caseID] = make(chan bool, 1)
	dmc.nemesis.mssG.switch2.Unlock()

	dmc.nemesis.mssG.done.Lock()
	dmc.nemesis.mssG.done.ch[dmc.caseID] = make(chan bool, 1)
	dmc.nemesis.mssG.done.Unlock()

	dmc.nemesis.queryG.request.Lock()
	dmc.nemesis.queryG.request.ch[dmc.caseID] = make(chan *Message, 1)
	dmc.nemesis.queryG.request.Unlock()

	dmc.nemesis.queryG.response.Lock()
	dmc.nemesis.queryG.response.ch[dmc.caseID] = make(chan *MessageResponse, 1)
	dmc.nemesis.queryG.response.Unlock()

	return nil
}

// Run starts DMCase test
func (dmc *DMCase) Run(wg, startupWg *sync.WaitGroup, fullDone *int32) {
	defer wg.Done()
	switch dmc.caseOpts.taskMode {
	case AllMode:
		dmc.testAll(startupWg, fullDone)
	case FullMode:
		dmc.testFull(startupWg)
	}

	for {
		select {
		case <-dmc.ctx.Done():
			client.Infof("case %d all test passed", dmc.caseID)
			return
		case err := <-dmc.errCh:
			dmc.nemesis.errCh <- errors.Annotatef(err, "case %d error", dmc.caseID)
			return
		}
	}
}

func (dmc *DMCase) testAll(startupWg *sync.WaitGroup, fullDone *int32) {
	go dmc.executeDDLs()
	err := dmc.nemesis.dmctl.StartTask(dmc.taskConfigName())
	if err != nil {
		dmc.errCh <- errors.Trace(err)
	}
	go dmc.startTest(startupWg, fullDone)
}

func (dmc *DMCase) testFull(startupWg *sync.WaitGroup) {
	go dmc.executeDDLs()
	go dmc.startFullTest(startupWg)
}

func renderConfig(name, tpl string, data map[string]interface{}) (string, error) {
	t := template.Must(template.New(name).Parse(tpl))
	var buf bytes.Buffer
	err := t.Execute(&buf, data)
	if err != nil {
		return "", errors.Trace(err)
	}
	return buf.String(), nil
}

func (dmc *DMCase) createTaskConfig(tidb string, mysqls []string) error {
	var (
		taskCfg string
		cfg     string
		err     error
	)

	if dmc.caseOpts.enableSharding {
		subtaskCfgs := make([]string, 0, len(mysqls))
		columnMappings := make([]string, 0, len(mysqls))
		for i, mysql := range mysqls {
			instID := i + 1
			data := map[string]interface{}{
				"InstanceID": instID,
				"RuleID1":    1,
				"RuleID2":    2,
			}
			cfg, err = renderConfig("columnMappingTpl", columnMappingTpl, data)
			if err != nil {
				return errors.Trace(err)
			}
			columnMappings = append(columnMappings, cfg)
			rules := fmt.Sprintf(`"instance-%[1]d-1", "instance-%[1]d-2"`, instID)
			data = map[string]interface{}{
				"SourceHost":         mysql,
				"ColumnMappingRules": rules,
				"SchemaIndex":        dmc.caseID,
			}
			cfg, err = renderConfig("subTaskCfg", subTaskCfg, data)
			if err != nil {
				return errors.Trace(err)
			}
			subtaskCfgs = append(subtaskCfgs, cfg)
		}
		data := map[string]interface{}{
			"TaskName":           dmc.taskName(),
			"TaskMode":           dmc.caseOpts.taskMode,
			"RemoveMeta":         dmc.caseOpts.taskMode != "all",
			"Host":               tidb,
			"MySQLInstances":     strings.Join(subtaskCfgs, "\n"),
			"SchemaIndex":        dmc.caseID,
			"ColumnMappingRules": strings.Join(columnMappings, "\n"),
			"OnlineDDLScheme":    dmc.caseOpts.onlineDDLTool,
		}
		cfg, err = renderConfig("baseTaskCfg", baseTaskCfg, data)
		if err != nil {
			return errors.Trace(err)
		}
		taskCfg = cfg
	} else {
		subtaskCfgs := make([]string, 0, len(mysqls))
		for i, mysql := range mysqls {
			// use different black-white-list
			var bwl string
			if i == 0 {
				bwl = "instance"
			} else {
				bwl = "instance2"
			}
			data := map[string]interface{}{
				"SourceHost":     mysql,
				"SchemaIndex":    dmc.caseID,
				"BlackWhiteList": bwl,
			}
			cfg, err = renderConfig("subTaskCfgNonSharding", subTaskCfgNonSharding, data)
			if err != nil {
				return errors.Trace(err)
			}
			subtaskCfgs = append(subtaskCfgs, cfg)
		}

		data := map[string]interface{}{
			"TaskName":        dmc.taskName(),
			"TaskMode":        dmc.caseOpts.taskMode,
			"RemoveMeta":      dmc.caseOpts.taskMode != "all",
			"Host":            tidb,
			"SchemaIndex":     dmc.caseID,
			"MySQLInstances":  strings.Join(subtaskCfgs, "\n"),
			"OnlineDDLScheme": dmc.caseOpts.onlineDDLTool,
		}
		cfg, err = renderConfig("baseTaskCfgNonSharding", baseTaskCfgNonSharding, data)
		if err != nil {
			return errors.Trace(err)
		}
		taskCfg = cfg
	}
	return errors.Trace(writeTaskConfigFile(taskCfg, dmc.taskConfigName()))
}

func (dmc *DMCase) initialDBs(tidb string, mysqls []string) ([]*sql.DB, error) {
	var err error
	dbs := make([]*sql.DB, len(mysqls)+1)

	defer func() {
		if err != nil {
			for _, db := range dbs {
				if db != nil {
					db.Close()
				}
			}
		}
	}()

	dbs[0], err = util.OpenDB(fmt.Sprintf("root:@tcp(%s:4000)/", tidb), 20)
	if err != nil {
		return nil, errors.Trace(err)
	}
	for idx, mysql := range mysqls {
		dbs[idx+1], err = util.OpenDB(fmt.Sprintf("root:@tcp(%s:3306)/", mysql), 20)
		if err != nil {
			return nil, errors.Trace(err)
		}
	}

	err = initialTaskTables(dmc.caseID, dbs[1:], dmc.sourceDB())
	if err != nil {
		return nil, errors.Trace(err)
	}

	err = createDB(dbs[0], dmc.targetDB())
	if err != nil {
		return nil, errors.Trace(err)
	}

	_, err = dbs[0].Exec("SET @@GLOBAL.SQL_MODE = '';")
	if err != nil {
		return nil, errors.Trace(err)
	}

	return dbs, nil
}

func (dmc *DMCase) addExecutor(dbc *dbExecutor, opts *executorOpts) {
	executor := newExecutor(
		dmc.ctx,
		dbc.db,
		dbc.dbStandby,
		dbc.dbName,
		dbc.dbInfo,
		dbc.standbyDBInfo,
		dbc.tables,
		&dmc.jobWg,
		opts)
	dmc.sqlExecutors = append(dmc.sqlExecutors, executor)
}

func (dmc *DMCase) startFullTest(startupWg *sync.WaitGroup) {
	var index int64

	dmc.nemesis.dmlMu.Lock()
	dmlCh := dmc.nemesis.dmlCh[dmc.caseID]
	dmc.nemesis.dmlMu.Unlock()

	dmc.nemesis.checkMu.Lock()
	checkCh := dmc.nemesis.checkCh[dmc.caseID]
	dmc.nemesis.checkMu.Unlock()

	dmc.nemesis.emptyMu.Lock()
	emptyCh := dmc.nemesis.emptyCh[dmc.caseID]
	dmc.nemesis.emptyMu.Unlock()

	startupWg.Done()

	for {
		select {
		case <-dmc.ctx.Done():
			return
		default:
			done := dmc.produceDDL()
			if done {
				client.Info("produce done before wait..")
				dmc.jobWg.Wait()
				client.Info("produce done after wait..")
				dmc.cancel()
				return
			}

			select {

			case <-dmlCh:
				dmc.jobWg.Wait()
				client.Infof("start case-%d dml", dmc.caseID)
				dmc.dmlWg.Add(1)
				go dmc.startDML()
				client.Infof("stop case-%d dml", dmc.caseID)
				dmc.stopDML()
				dmc.nemesis.dmlMu.Lock()
				dmc.nemesis.dmlStat[dmc.caseID] = 0
				dmc.nemesis.dmlMu.Unlock()

			case <-checkCh:
				var err error

				resetCheckStat := func() {
					dmc.nemesis.checkMu.Lock()
					dmc.nemesis.checkStat[dmc.caseID] = 0
					dmc.nemesis.checkMu.Unlock()
				}
				dmc.jobWg.Wait()
				dmc.dmlWg.Wait()

				atomic.AddInt64(&index, 1)
				client.Infof("[full testing] start %d round", index)

				dmc.dropschema()
				err = dmc.nemesis.dmctl.StartTask(dmc.taskConfigName())
				if err != nil {
					dmc.errCh <- err
					resetCheckStat()
					return
				}

				for i := 0; i < 20; i++ {
					err = dmc.checkData()
					if err == nil {
						break
					} else {
						client.Warningf("fail checking %d-th, retry later %v", i, err)
						time.Sleep(time.Minute)
					}
				}
				if err != nil {
					dmc.errCh <- err
					resetCheckStat()
					return
				}

				err = dmc.nemesis.dmctl.stopTask(dmc.taskName())
				if err != nil {
					dmc.errCh <- err
					resetCheckStat()
					return
				}
				resetCheckStat()

			case <-emptyCh:
				continue
			}

		}
	}
}

// starts start all check and nemesis
func (dmc *DMCase) startTest(startupWg *sync.WaitGroup, fullDone *int32) {
	dmc.nemesis.dmlMu.Lock()
	dmlCh := dmc.nemesis.dmlCh[dmc.caseID]
	dmc.nemesis.dmlMu.Unlock()

	dmc.nemesis.checkMu.Lock()
	checkCh := dmc.nemesis.checkCh[dmc.caseID]
	dmc.nemesis.checkMu.Unlock()

	dmc.nemesis.emptyMu.Lock()
	emptyCh := dmc.nemesis.emptyCh[dmc.caseID]
	dmc.nemesis.emptyMu.Unlock()

	dmc.nemesis.mssG.precheck.Lock()
	mssPreCh := dmc.nemesis.mssG.precheck.ch[dmc.caseID]
	dmc.nemesis.mssG.precheck.Unlock()

	dmc.nemesis.mssG.ready.Lock()
	mssReadyCh := dmc.nemesis.mssG.ready.ch[dmc.caseID]
	dmc.nemesis.mssG.ready.Unlock()

	dmc.nemesis.mssG.switch1.Lock()
	mssSwitch1Ch := dmc.nemesis.mssG.switch1.ch[dmc.caseID]
	dmc.nemesis.mssG.switch1.Unlock()

	dmc.nemesis.mssG.switch2.Lock()
	mssSwitch2Ch := dmc.nemesis.mssG.switch2.ch[dmc.caseID]
	dmc.nemesis.mssG.switch2.Unlock()

	dmc.nemesis.mssG.done.Lock()
	mssDoneCh := dmc.nemesis.mssG.done.ch[dmc.caseID]
	dmc.nemesis.mssG.done.Unlock()

	dmc.nemesis.queryG.request.Lock()
	requestCh := dmc.nemesis.queryG.request.ch[dmc.caseID]
	dmc.nemesis.queryG.request.Unlock()

	dmc.nemesis.queryG.response.Lock()
	responseCh := dmc.nemesis.queryG.response.ch[dmc.caseID]
	dmc.nemesis.queryG.response.Unlock()

	startupWg.Done()

	// wait for all subtasks enter sync stage
	for atomic.LoadInt32(fullDone) == 0 {
		client.Debugf("case-%d wait for full stage done", dmc.caseID)
		time.Sleep(10 * time.Second)
	}

	for {
		select {
		case <-dmc.ctx.Done():
			return
		default:
			done := dmc.produceDDL()
			if done {
				client.Infof("case-%d produce done before wait..", dmc.caseID)
				dmc.jobWg.Wait()
				client.Infof("case-%d produce done after wait..", dmc.caseID)
				dmc.cancel()
				return
			}

			select {

			case <-dmlCh:
				dmc.jobWg.Wait()
				client.Infof("start case: %d dml", dmc.caseID)
				dmc.dmlWg.Add(1)
				go dmc.startDML()
				client.Infof("stop case: %d dml", dmc.caseID)
				dmc.stopDML()
				dmc.nemesis.dmlMu.Lock()
				dmc.nemesis.dmlStat[dmc.caseID] = 0
				dmc.nemesis.dmlMu.Unlock()

			case <-checkCh:
				resetCheckStat := func() {
					dmc.nemesis.checkMu.Lock()
					dmc.nemesis.checkStat[dmc.caseID] = 0
					dmc.nemesis.checkMu.Unlock()
				}
				dmc.jobWg.Wait()

				var checkError error
				for i := 0; i < 10; i++ {
					checkError = dmc.checkData()
					if checkError == nil {
						break
					} else {
						client.Warningf("case-%d fail checking %d-th, retry later %v", dmc.caseID, i, checkError)
						time.Sleep(time.Second * 30)
					}
				}
				if checkError != nil {
					dmc.errCh <- errors.Annotatef(checkError, "case %d error", dmc.caseID)
					resetCheckStat()
					return
				}
				resetCheckStat()

			case msg := <-requestCh:
				resp, err := dmc.Check(msg)
				if err != nil {
					dmc.errCh <- errors.Trace(err)
					return
				}
				responseCh <- resp

			case dbIndex := <-mssPreCh:
				dmc.jobWg.Wait()
				dmc.dmlWg.Wait()

				resolved, err := dmc.waitUnresolvedGroup()
				if err != nil {
					dmc.errCh <- errors.Trace(err)
					return
				}

				checkpointSaved, err := dmc.checkSyncerCheckpoint(dbIndex)
				if err != nil {
					dmc.errCh <- errors.Trace(err)
					return
				}

				if !resolved {
					client.Infof("case-%d has unresolved sharding group", dmc.caseID)
				}
				if !checkpointSaved {
					client.Infof("case-%d syncer checkpoint not saved", dmc.caseID)
				}
				if resolved && checkpointSaved {
					mssReadyCh <- true
				} else {
					mssReadyCh <- false
				}
				client.Debugf("case-%d sent ready response", dmc.caseID)

				mssOpts := <-mssSwitch1Ch
				if mssOpts.abort {
					continue
				}
				err2 := dmc.upstreamMasterSlaveSwitch(mssOpts)
				if err2 != nil {
					dmc.errCh <- errors.Trace(err2)
					return
				}
				mssSwitch2Ch <- true
				<-mssDoneCh

			case <-emptyCh:
				continue
			}
		}
	}
}

func (dmc *DMCase) produceDDL() bool {
	if dmc.ddlIndex >= len(dmc.ddls) {
		return true
	}

	dmc.jobWg.Add(len(dmc.sqlExecutors))
	client.Debugf("case-%d produce ddl...round: %d", dmc.caseID, dmc.ddlIndex)
	dmc.ddlCh <- dmc.ddls[dmc.ddlIndex]
	client.Debugf("case-%d produce ddl %s success, round: %d", dmc.caseID, dmc.ddls[dmc.ddlIndex], dmc.ddlIndex)
	dmc.ddlIndex++
	return false
}

func (dmc *DMCase) executeDDLs() {
	for {
		select {
		case ddl := <-dmc.ddlCh:
			client.Infof("execute ddl %s", ddl)
			dmc.executeDDL(ddl)
		}
	}
}

func (dmc *DMCase) executeDDL(ddl string) {
	for _, e := range dmc.sqlExecutors {
		e.addJob(ddl)
	}
}

// execute dmls
func (dmc *DMCase) startDML() {
	defer dmc.dmlWg.Done()
	for {
		select {
		case <-dmc.ctx.Done():
			return
		case <-dmc.pauseDML:
			return
		default:
		}

		client.Info("execute dml round")
		for _, e := range dmc.sqlExecutors {
			err := e.startDML()
			if err != nil {
				dmc.errCh <- err
				return
			}
		}
	}
}

func (dmc *DMCase) stopDML() {
	dmc.pauseDML <- struct{}{}
}

// upstreamMasterSlaveSwitch reacts to upstream MySQL master slave ip switch
func (dmc *DMCase) upstreamMasterSlaveSwitch(opts *mssOptions) error {
	var err error

	dbIndex := opts.dbIndex
	workerHost := dmc.nemesis.workers[dbIndex]
	instanceID := dbIndex + 1
	mysqlReplica, ok := dmc.mysqlReplica[instanceID]
	if !ok {
		return errors.NotFoundf("mysql replica:%d", instanceID)
	}
	oldMySQL, newMySQL := mysqlReplica.rotateReplicationHost()
	client.Infof("case %d switch db connection to upstream from %s to %s", dmc.caseID, oldMySQL, newMySQL)

	if opts.leader {
		err = updateDmWorkerConfig(workerHost, oldMySQL, newMySQL)
		if err != nil {
			return errors.Trace(err)
		}
	}

	// no need to update dm task config now, as only source-id used in task config
	// caseHost := dmc.nemesis.caseHost
	// err = updateDmTaskConfig(caseHost, dmc.taskConfigName(), oldMySQL, newMySQL)
	// if err != nil {
	// 	return errors.Trace(err)
	// }

	// In MySQL master-master architecture, the two MySQL instances have
	// different master binlog positon. So we exchange db and dbStandby
	// to execute SQL in MySQL with new master ip.
	// dbMaster, dbStandby, err := dmc.sqlExecutors[dbIndex].swapDBStandby()
	_, _, err = dmc.sqlExecutors[dbIndex].swapDBStandby()
	if err != nil {
		return errors.Trace(err)
	}

	return nil
}

func (dmc *DMCase) checkData() error {
	msg, err := dmc.nemesis.dmctl.queryStatus()
	if err != nil {
		client.Errorf("get query status failed %+v", err)
		return errors.Trace(err)
	}

	client.Infof("check data of case-%d, status %s", dmc.caseID, msg)

	if !dmc.caseOpts.enableSharding {
		return errors.Trace(dmc.checkDataNonSharding())
	}
	return errors.Trace(dmc.checkDataSharding())
}

func (dmc *DMCase) checkDataNonSharding() error {
	var results, checkResults []*checkResult
	var err error

	instID := 1
	for _, table := range filterIgnoreTables(getSubTaskTables(dmc.caseID, instID)) {
		ss, err := getTableCrc32(dmc.sqlDB[0], &checkConfig{
			calculateMore: true,
			instanceID:    int64(instID),
		}, dmc.sourceDB(), table, []string{"id"})
		if err != nil {
			return errors.Trace(err)
		}
		results = append(results, ss...)
	}

	for _, table := range filterIgnoreTables(getSubTaskTables(dmc.caseID, instID)) {
		ss, err := getTableCrc32(dmc.tidb, &checkConfig{
			calculateMore: true,
			instanceID:    int64(instID),
		}, dmc.targetDB(), table, []string{"id"})
		if err != nil {
			return errors.Trace(err)
		}
		checkResults = append(checkResults, ss...)
	}

	mappings := make(map[int64]int64)
	for _, res := range checkResults {
		mappings[res.mappingID] = res.crc32
	}

	err = dmc.checkColumnMappingEquals(mappings, results, true)
	if err != nil {
		return errors.Trace(err)
	}

	client.Infof("case-%d check pass, total object: %d", dmc.caseID, len(results))
	return nil
}

func (dmc *DMCase) checkDataSharding() error {
	var checkCount int
	var results []*checkResult
	for idx, db := range dmc.sqlDB {
		instID := idx + 1
		for _, table := range filterIgnoreTables(getSubTaskTables(dmc.caseID, instID)) {
			ss, err := getTableCrc32(db, &checkConfig{
				calculateMore: true,
				instanceID:    int64(instID),
			}, dmc.sourceDB(), table, []string{"id"})
			if err != nil {
				return errors.Trace(err)
			}
			results = append(results, ss...)
		}
	}
	if len(results) > 0 {
		tidbResults, err := getTableCrc32(dmc.tidb, &checkConfig{}, dmc.targetDB(), "t_target", []string{"id"})
		if err != nil {
			return errors.Trace(err)
		}

		mappings := make(map[int64]int64)
		for _, res := range tidbResults {
			mappings[res.originID] = res.crc32
		}

		err = dmc.checkColumnMappingEquals(mappings, results, true)
		if err != nil {
			return errors.Trace(err)
		}
		checkCount += len(results)
	}

	// check filter table if enabled
	if dmc.caseOpts.filterDDL {
		var mysqlResults, tidbResults []*checkResult

		for idx, db := range dmc.sqlDB {
			instID := idx + 1
			for _, table := range filterFilterTables(getSubTaskTables(dmc.caseID, instID)) {
				ss, err := getTableCrc32(db, &checkConfig{
					calculateMore: true,
					instanceID:    int64(instID),
				}, dmc.sourceDB(), table, []string{"id"})
				if err != nil {
					return errors.Trace(err)
				}
				mysqlResults = append(mysqlResults, ss...)
			}
		}

		if len(mysqlResults) > 0 {
			for _, table := range filterTables {
				ss, err := getTableCrc32(dmc.tidb, &checkConfig{
					calculateMore: false,
				}, dmc.targetDB(), table, []string{"id"})
				if err != nil {
					return errors.Trace(err)
				}
				tidbResults = append(tidbResults, ss...)
			}

			mappings := make(map[int64]int64)
			for _, res := range tidbResults {
				mappings[res.originID] = res.crc32
			}

			err := dmc.checkColumnMappingEquals(mappings, mysqlResults, false)
			if err != nil {
				return errors.Trace(err)
			}
			checkCount += len(mysqlResults)
		}
	}

	client.Infof("case-%d check pass, total object: %d", dmc.caseID, checkCount)
	return nil
}

func (dmc *DMCase) checkColumnMappingEquals(targets map[int64]int64, source []*checkResult, checkMoreData bool) error {
	for _, obj := range source {
		mysqlReplica, ok := dmc.mysqlReplica[int(obj.instanceID)]
		if !ok {
			return errors.NotFoundf("mysql replica:%d", obj.instanceID)
		}
		mappingID := dmc.genTargetID(obj, true)
		targetID := dmc.genTargetID(obj, false)
		crc32, ok := targets[mappingID]
		tidbTable := dmc.routeTableName(obj.tableName)
		if !ok {
			return errors.NotFoundf("upstream mysql:%s schema: %s table: %s id:%d in tidb:%s schema: %s table: %s id:%d",
				mysqlReplica.hosts[mysqlReplica.current], obj.schemaName, obj.tableName, obj.originID, dmc.tidbAddr, dmc.targetDB(), tidbTable, targetID)
		}
		if crc32 != obj.crc32 {
			mysqlObj, tidbObj, err := dmc.getRow(obj, dmc.sourceDB(), dmc.targetDB(), targetID)
			if err != nil {
				client.Errorf("get row failed %v", errors.ErrorStack(err))
				return errors.Errorf("upstream mysql:%s schema: %s table: %s id:%d  crc32 %d not equal to tidb: %s, schema: %s table: %s id:%d crc32: %d",
					mysqlReplica.hosts[mysqlReplica.current], obj.schemaName, obj.tableName, obj.originID, obj.crc32, dmc.tidbAddr, dmc.targetDB(), tidbTable, targetID, crc32)
			}

			// some timestamp special check
			skipError := true
			if len(mysqlObj) == len(tidbObj) {
				for columnName, value := range mysqlObj {
					if columnName == "id" {
						continue
					}
					if tidbValue, ok := tidbObj[columnName]; ok {
						if value == tidbValue {
							continue
						}
						info, err := util.GetColumnInfo(dmc.sqlDB[obj.instanceID-1], obj.schemaName, obj.tableName, columnName)
						if err != nil {
							return errors.Trace(err)
						}
						columnType := strings.ToUpper(info["Type"])
						client.Debugf("check data inconsistent because of different value: mysql:%v tidb:%v, columnType: %s", value, tidbValue, columnType)
						// MySQL 5.6 and MySQL 5.7 have different conversion in some DST gap,
						// such as unix_timestamp for "1990-04-15 01:30:12 CST" are different.
						// So we query unix_timestamp directly from upstream MySQL and test
						// if it equals to the unix_timestamp converted from downstream TiDB
						// with 8 hours shift, because upstream is CST-8, downstream is UTC.
						if columnType == "TIMESTAMP" || columnType == "DATETIME" {
							upstreamDB := dmc.sqlDB[obj.instanceID-1]
							ts1, err := dmc.getRowTimestamp(upstreamDB, dmc.sourceDB(), obj.tableName, columnName, obj.originID)
							if err != nil {
								return errors.Trace(err)
							}

							ts2, err := dmc.getRowTimestamp(dmc.tidb, dmc.targetDB(), dmc.routeTableName(obj.tableName), columnName, targetID)
							if err != nil {
								return errors.Trace(err)
							}

							client.Debugf("fixed dst ts1: %d ts2: %d value1: %s, value2: %s, offset: %d", ts1, ts2, value, tidbValue, ts2-ts1)
							if ts2-ts1 == 28800 || ts2-ts1 == 32400 {
								continue
							}
						}
						skipError = false
						break
					} else {
						skipError = false
						break
					}
				}
			} else {
				skipError = false
			}

			if !skipError {
				return errors.Errorf("upstream mysql:%s schema: %s table: %s id:%d  crc32 %d "+
					"not equal to tidb: %s, schema: %s table: %s id:%d crc32: %d, additional message mysql: %v  tidb:%v",
					mysqlReplica.hosts[mysqlReplica.current], obj.schemaName, obj.tableName, obj.originID, obj.crc32,
					dmc.tidbAddr, dmc.targetDB(), tidbTable, targetID, crc32, mysqlObj, tidbObj)
			}
		}
		delete(targets, mappingID)
	}
	if checkMoreData {
		if len(targets) > 0 {
			return errors.Errorf("there are data still exist in tidb %s %+v", dmc.tidbAddr, targets)
		}
	} else {
		if len(targets) == 0 {
			log.Warning("has no more data in target table in checkMoreData mode")
		}
	}

	return nil
}

func (dmc *DMCase) getRowTimestamp(db *sql.DB, schemaName, tableName, columnName string, id int64) (int64, error) {
	stmt := fmt.Sprintf("select unix_timestamp(%s) from `%s`.`%s` where id = ?", columnName, schemaName, tableName)
	var ts sql.NullInt64
	err := db.QueryRowContext(dmc.ctx, stmt, id).Scan(&ts)
	if err != nil {
		return 0, errors.Trace(err)
	}
	if !ts.Valid {
		return 0, errors.NotFoundf("table `%s`.`%s`", schemaName, tableName)
	}
	return ts.Int64, nil
}

func (dmc *DMCase) getRow(obj *checkResult, mysqlSchema, tidbSchema string, tidbID int64) (map[string]string, map[string]string, error) {
	mysqlObj, err := util.GetRow(dmc.sqlDB[obj.instanceID-1], mysqlSchema, obj.tableName, obj.originID)
	if err != nil {
		return nil, nil, errors.Trace(err)
	}

	tidbTable := dmc.routeTableName(obj.tableName)
	tidbObj, err := util.GetRow(dmc.tidb, tidbSchema, tidbTable, tidbID)
	if err != nil {
		return nil, nil, errors.Trace(err)
	}

	return mysqlObj, tidbObj, nil
}

// genTargetID generates partitionID in sharding mode, generates mappingID in non sharding mode
func (dmc *DMCase) genTargetID(info *checkResult, nonShardingMapping bool) int64 {
	if dmc.caseOpts.enableSharding {
		return info.originID | (info.instanceID << 59) | (info.schemaID << 52) | (info.tableID << 44)
	}
	if nonShardingMapping {
		return info.mappingID
	}
	return info.originID
}

func (dmc *DMCase) routeTableName(tableName string) string {
	// hard code, assume table t_* is in sharding mode is mapping to t_target,
	// other tables route directly
	tidbTable := tableName
	if dmc.caseOpts.enableSharding && strings.HasPrefix(tableName, "t_") {
		tidbTable = "t_target"
	}
	return tidbTable
}

func (dmc *DMCase) sourceDB() string {
	return fmt.Sprintf("test_%d", dmc.caseID)
}

func (dmc *DMCase) targetDB() string {
	return fmt.Sprintf("test_target_%d", dmc.caseID)
}

func (dmc *DMCase) instanceID(idx int) int {
	return idx + 1
}

func (dmc *DMCase) taskName() string {
	return genTaskName(dmc.caseID)
}

func (dmc *DMCase) taskConfigName() string {
	return fmt.Sprintf("%s.yaml", dmc.taskName())
}

// rotateReplicationHost returns current host and next host
func (replic *replicationMySQL) rotateReplicationHost() (string, string) {
	current := replic.hosts[replic.current]
	replic.current++
	if replic.current >= len(replic.hosts) {
		replic.current = 0
	}
	return current, replic.hosts[replic.current]
}

func (dmc *DMCase) dropschema() error {
	var err error
	for i := 0; i < 100; i++ {
		stmt := fmt.Sprintf("drop schema if exists %s", dmc.targetDB())
		if _, err = dmc.tidb.Exec(stmt); err == nil {
			return nil
		}

		client.Errorf("drop schema %s error %v", dmc.targetDB(), err)
		time.Sleep(1 * time.Second)
	}

	return errors.Trace(err)
}

// waitUnresolvedGroup waits for sharding group resolved in certain times.
// if wait exceeds certain times returns false.
func (dmc *DMCase) waitUnresolvedGroup() (bool, error) {
	retry := 30
	for retry > 0 {
		unresolved, err := dmc.nemesis.dmctl.existsUnresolvedGroups(dmc.taskName())
		if err != nil {
			return false, errors.Trace(err)
		}
		if unresolved {
			client.Warnf("case-%d detect unresovled groups, wait for sometime", dmc.caseID)
			time.Sleep(time.Second * 10)
		} else {
			return true, nil
		}
		retry--
	}
	client.Warnf("case-%d wait unresolved group too long", dmc.caseID)
	return false, nil
}

// checkSyncerCheckpoint checks whether syncer checkpoint already saved
// if syncer checkpoint is not saved, after master slave switch it will
// read mysql position from dumped_meta file, which is a wrong position.
func (dmc *DMCase) checkSyncerCheckpoint(dbIndex int) (bool, error) {
	// non sharding task only use first db and DM-worker
	if !dmc.caseOpts.enableSharding && dbIndex > 0 {
		return true, nil
	}
	instID := dbIndex + 1
	id := fmt.Sprintf("%s:3306", dmc.mysqlReplica[instID].hosts[0])
	name, pos, err := getSyncerCheckpoint(dmc.tidb, id, dmc.taskName())
	if err != nil {
		return false, errors.Trace(err)
	}
	return (name != "" && pos > 0), nil
}

// Check processes check request
func (dmc *DMCase) Check(msg *Message) (*MessageResponse, error) {
	switch msg.t {
	case msgSyncCheckpoint:
		resp := &MessageResponse{
			caseID:         dmc.caseID,
			syncCheckpoint: make(map[int]bool),
		}
		for dbIndex := range dmc.sqlDB {
			ready, err := dmc.checkSyncerCheckpoint(dbIndex)
			if err != nil {
				return nil, errors.Trace(err)
			}
			resp.syncCheckpoint[dbIndex] = ready
		}
		return resp, nil
	case msgUnresolvedGroup:
		ready, err := dmc.waitUnresolvedGroup()
		if err != nil {
			return nil, errors.Trace(err)
		}
		return &MessageResponse{caseID: dmc.caseID, ddlGroupReady: ready}, nil
	default:
		return nil, errors.Errorf("unknown message type: %d", msg.t)
	}
}

// clone returns a new replicationMySQL object copy
func (replic *replicationMySQL) clone() *replicationMySQL {
	return &replicationMySQL{
		hosts:   replic.hosts[:],
		current: replic.current,
	}
}