// Copyright 2016 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package main

import (
	"context"
	"database/sql"
	"fmt"
	"runtime/debug"
	"strings"
	"sync"
	"time"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/parser"
	"github.com/pingcap/tidb-enterprise-tools/pkg/filter"
	route "github.com/pingcap/tidb-enterprise-tools/pkg/tableroute"
	"github.com/pingcap/tidb-enterprise-tools/pkg/utils"
	"github.com/siddontang/go-mysql/mysql"
	"github.com/siddontang/go-mysql/replication"
	"github.com/siddontang/go/sync2"
)

var (
	maxRetryCount    = 100
	maxDDLRetryCount = 3

	retryTimeout = 3 * time.Second
	waitTime     = 10 * time.Millisecond
	eventTimeout = 1 * time.Hour
	statusTime   = 30 * time.Second

	slaveReadTimeout      = 1 * time.Minute  // slave read binlog data timeout, ref: https://dev.mysql.com/doc/refman/8.0/en/replication-options-slave.html#sysvar_slave_net_timeout
	masterHeartbeatPeriod = 30 * time.Second // master server send heartbeat period: ref: `MASTER_HEARTBEAT_PERIOD` in https://dev.mysql.com/doc/refman/8.0/en/change-master-to.html

	adminQueueName = "admin queue"
	// put queues into bucket to refine display
	// e.g. we have 32 workers, it's bad to display them all
	// so put them into bucket and display them
	defaultBucketQueueCount = 8
	queueBucketMapping      []string
)

// safeMode makes syncer reentrant.
// we make each operator reentrant to make syncer reentrant.
// `replace` and `delete` are naturally reentrant.
// use `delete`+`replace` to represent `update` can make `update`  reentrant.
// but there are no ways to make `update` idempotent,
// if we start syncer at an early position, database must bear a period of inconsistent state,
// it's eventual consistency.
var safeMode sync2.AtomicBool

// Syncer can sync your MySQL data to another MySQL database.
type Syncer struct {
	sync.Mutex

	cfg *Config

	meta Meta

	syncer *replication.BinlogSyncer

	wg    sync.WaitGroup
	jobWg sync.WaitGroup

	tables map[string]*table

	pt *PersistentTable

	fromDB *sql.DB
	toDBs  []*sql.DB
	ddlDB  *sql.DB

	jobs []chan *job
	c    *causality

	tableRouter route.TableRouter

	closed sync2.AtomicBool

	start    time.Time
	lastTime time.Time
	timezone *time.Location

	lastFinishedCount sync2.AtomicInt64
	totalCount        sync2.AtomicInt64
	finishedCount     sync2.AtomicInt64

	ctx    context.Context
	cancel context.CancelFunc

	done chan struct{}

	filter       *filter.Filter
	skipDMLRules *SkipDMLRules

	lackOfReplClientPrivilege bool
	lastSlaveConnectionID     uint32
}

// NewSyncer creates a new Syncer.
func NewSyncer(cfg *Config) *Syncer {
	syncer := new(Syncer)
	syncer.cfg = cfg
	syncer.meta = NewLocalMeta(cfg.Meta, cfg.Flavor)
	syncer.closed.Set(false)
	syncer.lastFinishedCount.Set(0)
	syncer.finishedCount.Set(0)
	syncer.totalCount.Set(0)
	syncer.jobs = newJobChans(cfg.WorkerCount+1, cfg.ExecutionQueueLength)
	syncer.tables = make(map[string]*table)
	if len(cfg.PersistentTableDir) > 0 {
		// if set PersistentTableDir, we enable PersistentTable
		syncer.pt = newPersistentTable(cfg.PersistentTableDir)
	}
	syncer.c = newCausality()
	syncer.ctx, syncer.cancel = context.WithCancel(context.Background())
	syncer.done = make(chan struct{})
	rules := &filter.Rules{
		DoDBs:        cfg.DoDBs,
		DoTables:     cfg.DoTables,
		IgnoreDBs:    cfg.IgnoreDBs,
		IgnoreTables: cfg.IgnoreTables,
	}
	syncer.filter = filter.New(rules)
	return syncer
}

func newJobChans(count int, length int) []chan *job {
	jobs := make([]chan *job, 0, count)
	for i := 0; i < count; i++ {
		jobs = append(jobs, make(chan *job, length))
	}

	return jobs
}

func closeJobChans(jobs []chan *job) {
	for _, ch := range jobs {
		close(ch)
	}
}

// Start starts syncer.
func (s *Syncer) Start() error {
	defer func() {
		close(s.done)
	}()
	err := s.meta.Load()
	if err != nil {
		return errors.Trace(err)
	}

	if s.pt != nil {
		err = s.pt.Load()
		if err != nil {
			return errors.Trace(err)
		}
		log.Infof("[syncer] load persistent tables (%d) from %s successfully", len(s.pt.Tables), s.cfg.PersistentTableDir)
	}

	cfg := replication.BinlogSyncerConfig{
		ServerID:                uint32(s.cfg.ServerID),
		Flavor:                  s.cfg.Flavor,
		Host:                    s.cfg.From.Host,
		Port:                    uint16(s.cfg.From.Port),
		User:                    s.cfg.From.User,
		Password:                s.cfg.From.Password,
		UseDecimal:              true,
		ReadTimeout:             slaveReadTimeout,
		HeartbeatPeriod:         masterHeartbeatPeriod,
		TimestampStringLocation: s.cfg.Timezone,
	}
	err = s.init(cfg)
	if err != nil {
		return errors.Trace(err)
	}

	err = s.run(cfg)
	if err != nil {
		return errors.Trace(err)
	}

	return nil
}

func (s *Syncer) getServerUUID() (string, error) {
	return getServerUUID(s.fromDB)
}

func (s *Syncer) getMasterStatus() (mysql.Position, GTIDSet, error) {
	return getMasterStatus(s.fromDB, s.cfg.Flavor)
}

func (s *Syncer) clearTable(schemaName, tableName string) {
	key := buildTableName(schemaName, tableName)
	delete(s.tables, key)
}

func (s *Syncer) putTableIntoHistory(t *filter.Table) error {
	if len(t.Name) == 0 || s.pt == nil {
		return nil
	}

	key := buildTableName(t.Schema, t.Name)

	tbl, err := s.getTable(t.Schema, t.Name)
	if err != nil && isErrTableNotExists(err) {
		// triggered by CREATE TABLE stmt, but not created yet
		log.Warnf("[syncer] table %s not exists, no table info can be put into history", key)
		return nil
	}
	if err != nil || tbl == nil {
		return errors.Trace(err)
	}

	var previousTable *table
	length := len(s.pt.Tables[key])
	if length > 0 {
		previousTable = s.pt.Tables[key][length-1]
	}

	if compareTableColumns(tbl, previousTable) {
		s.pt.Tables[key][length-1] = tbl
		log.Infof("[syncer] update history table %s:%d with column length %d", key, length-1, len(tbl.Columns))
	} else {
		s.pt.Tables[key] = append(s.pt.Tables[key], tbl)
		log.Infof("[syncer] append history table %s:%d with column length %d", key, len(s.pt.Tables[key])-1, len(tbl.Columns))
	}

	return nil
}

// in sharding table case, try compatible with
// newest and previous version table
// limits: don't support to adjust the order of the columns
func (s *Syncer) chooseTable(t *table, rows [][]interface{}) *table {
	if len(rows) == 0 || s.pt == nil {
		return t
	}

	colLen := len(rows[0])
	if colLen == len(t.Columns) {
		return t //  in most cases should hit this
	}

	key := buildTableName(t.Schema, t.Name)
	historyTables, ok := s.pt.Tables[key]
	if ok {
		// maybe iterate in reverse order is better
		for i := len(historyTables) - 1; i >= 0; i-- {
			historyTable := historyTables[i]
			if colLen == len(historyTable.Columns) {
				log.Debugf("[syncer] choose a history table %s:%d with column length %d", key, i, colLen)
				return historyTable
			}
		}
	}

	return t
}

func (s *Syncer) persistentTable() error {
	if s.pt == nil {
		return nil
	}

	for key, tbl := range s.tables {
		var previousTable *table
		length := len(s.pt.Tables[key])
		if length > 0 {
			previousTable = s.pt.Tables[key][length-1]
		}

		if compareTableColumns(tbl, previousTable) {
			s.pt.Tables[key][length-1] = tbl
		} else {
			s.pt.Tables[key] = append(s.pt.Tables[key], tbl)
		}
	}

	err := s.pt.Save()
	if err != nil {
		return errors.Trace(err)
	}

	log.Infof("[syncer] save persistent tables (%d) to %s successfully", len(s.pt.Tables), s.cfg.PersistentTableDir)
	return nil
}

func (s *Syncer) getTableFromDB(db *sql.DB, schema string, name string) (*table, error) {
	table := &table{}
	table.Schema = schema
	table.Name = name
	table.IndexColumns = make(map[string][]*column)

	err := getTableColumns(s.ctx, db, table, s.cfg.MaxRetry)
	if err != nil {
		return nil, errors.Trace(err)
	}

	err = getTableIndex(s.ctx, db, table, s.cfg.MaxRetry)
	if err != nil {
		return nil, errors.Trace(err)
	}

	if len(table.Columns) == 0 {
		return nil, errors.Errorf("invalid table %s", buildTableName(schema, name))
	}

	return table, nil
}

func (s *Syncer) getTable(schema string, table string) (*table, error) {
	key := buildTableName(schema, table)

	value, ok := s.tables[key]
	if ok {
		return value, nil
	}

	db := s.toDBs[len(s.toDBs)-1]
	t, err := s.getTableFromDB(db, schema, table)
	if err != nil {
		return nil, errors.Trace(err)
	}

	s.tables[key] = t
	return t, nil
}

func (s *Syncer) addCount(isFinished bool, bucket string, tp opType, n int64) {
	jobM := addedJobsTotal
	if isFinished {
		jobM = finishedJobsTotal
	}

	switch tp {
	case insert:
		jobM.WithLabelValues("insert", bucket).Add(float64(n))
	case update:
		jobM.WithLabelValues("update", bucket).Add(float64(n))
	case del:
		jobM.WithLabelValues("del", bucket).Add(float64(n))
	case ddl:
		jobM.WithLabelValues("ddl", bucket).Add(float64(n))
	case xid:
		// ignore xid jobs
	case flush:
		jobM.WithLabelValues("flush", bucket).Add(float64(n))
	default:
		log.Warnf("unknown optype %v", tp)
	}

	if isFinished {
		s.finishedCount.Add(n)
	} else {
		s.totalCount.Add(n)
	}
}

func (s *Syncer) checkWait(job *job) bool {
	if job.tp == ddl {
		return true
	}

	if s.meta.Check() {
		return true
	}

	return false
}

func (s *Syncer) addJob(job *job) error {
	switch job.tp {
	case xid:
		s.meta.Save(job.pos, job.gtidSet)
		return nil
	case ddl:
		// wait 3 seconds? refine it later
		// while meet ddl, we should wait all dmls finished firstly
		s.jobWg.Wait()
		if s.cfg.StopOnDDL {
			// flush current position.
			s.meta.Flush()
			log.Fatalf("stop-on-ddl:%v, ddl should be executed manually. ddl: %s, next-pos: %v", s.cfg.StopOnDDL, job.sql, job.pos)
		}
	case flush:
		addedJobsTotal.WithLabelValues("flush", adminQueueName).Inc()

		// ugly code addJob and sync, refine it later
		s.jobWg.Add(s.cfg.WorkerCount)
		for i := 0; i < s.cfg.WorkerCount; i++ {
			s.jobs[i] <- job
		}
		s.jobWg.Wait()
		finishedJobsTotal.WithLabelValues("flush", adminQueueName).Inc()
		err := s.meta.Flush()
		return errors.Trace(err)
	}

	if len(job.sql) > 0 {
		s.jobWg.Add(1)
		if job.tp == ddl {
			addedJobsTotal.WithLabelValues("ddl", adminQueueName).Inc()
			s.jobs[s.cfg.WorkerCount] <- job
		} else {
			idx := int(utils.GenHashKey(job.key)) % s.cfg.WorkerCount
			s.addCount(false, queueBucketMapping[idx], job.tp, 1)
			s.jobs[idx] <- job
		}
	}

	wait := s.checkWait(job)
	if wait {
		s.jobWg.Wait()
		s.c.reset()
	}

	if job.tp == ddl && job.isLast {
		// only save pos for DDL and XID (see above)
		// and only when the ddl is the last ddl in a binlog event will save pos
		s.meta.Save(job.pos, job.gtidSet)
	}
	if wait {
		err := s.meta.Flush()
		if err != nil {
			return errors.Trace(err)
		}
	}

	return nil
}

func (s *Syncer) sync(ctx context.Context, queueBucket string, db *sql.DB, jobChan chan *job) {
	defer s.wg.Done()

	idx := 0
	count := s.cfg.Batch
	sqls := make([]string, 0, count)
	args := make([][]interface{}, 0, count)
	tpCnt := make(map[opType]int64)

	clearF := func() {
		for i := 0; i < idx; i++ {
			s.jobWg.Done()
		}

		idx = 0
		sqls = sqls[0:0]
		args = args[0:0]
		for tpName, v := range tpCnt {
			s.addCount(true, queueBucket, tpName, v)
			tpCnt[tpName] = 0
		}
	}

	var err error
	for {
		select {
		case job, ok := <-jobChan:
			if !ok {
				return
			}
			idx++

			if job.tp == ddl {
				err = executeSQL(s.ctx, db, sqls, args, s.cfg.MaxRetry)
				if err != nil {
					log.Fatalf(errors.ErrorStack(err))
				}

				// executes DDL SQL
				err = executeSQL(s.ctx, db, []string{job.sql}, [][]interface{}{job.args}, maxDDLRetryCount)
				if err != nil {
					if !ignoreDDLError(err) {
						log.Fatalf("[error query event][error type]%s [sql]%s %s [error message]%v", ErrExecution, job.sql, job.annotation, errors.ErrorStack(err))
					} else {
						log.Warnf("[ignore ddl error][sql]%s[args]%v[error]%v", job.sql, job.args, err)
					}
				}

				tpCnt[job.tp]++
				clearF()

			} else if job.tp != flush {
				sqls = append(sqls, job.sql)
				args = append(args, job.args)
				tpCnt[job.tp]++
			}

			if idx >= count || job.tp == flush {
				err = executeSQL(s.ctx, db, sqls, args, s.cfg.MaxRetry)
				if err != nil {
					log.Fatalf("[error rows event] %v", errors.ErrorStack(err))
				}
				clearF()
			}

		default:
			if len(sqls) > 0 {
				err = executeSQL(s.ctx, db, sqls, args, s.cfg.MaxRetry)
				if err != nil {
					log.Fatalf("[error rows event] %v", errors.ErrorStack(err))
				}
				clearF()
			} else {
				time.Sleep(waitTime)
			}
		}
	}
}

func (s *Syncer) init(cfg replication.BinlogSyncerConfig) error {
	s.start = time.Now()
	s.lastTime = s.start

	s.syncer = replication.NewBinlogSyncer(cfg)

	err := s.createDBs()
	if err != nil {
		return errors.Trace(err)
	}

	err = checkBinlogFormat(s.fromDB)
	if err != nil {
		return errors.Trace(err)
	}

	// support regex
	s.genSkipDMLRules()
	err = s.genRouter()
	return errors.Trace(err)
}

func (s *Syncer) run(cfg replication.BinlogSyncerConfig) (err error) {
	parser2, err := getParser(s.fromDB, s.cfg.EnableANSIQuotes)
	if err != nil {
		return errors.Trace(err)
	}

	streamer, isGTIDMode, err := s.getBinlogStreamer()
	if err != nil {
		return errors.Trace(err)
	}

	s.wg.Add(s.cfg.WorkerCount + 1)
	for i := 0; i < s.cfg.WorkerCount; i++ {
		name := queueBucketName(i)
		queueBucketMapping = append(queueBucketMapping, name)
		go func(i int, n string) {
			ctx, cancel := context.WithCancel(s.ctx)
			s.sync(ctx, n, s.toDBs[i], s.jobs[i])
			cancel()
		}(i, name)
	}
	go func() {
		ctx, cancel := context.WithCancel(s.ctx)
		s.sync(ctx, adminQueueName, s.ddlDB, s.jobs[s.cfg.WorkerCount])
		cancel()
	}()
	go func() {
		ctx, cancel := context.WithCancel(s.ctx)
		s.printStatus(ctx)
		cancel()
	}()

	defer func() {
		if err1 := recover(); err1 != nil {
			log.Errorf("panic. err: %s, stack: %s", err1, debug.Stack())
			err = errors.Errorf("panic error: %v", err1)
		}
		if err1 := s.persistentTable(); err1 != nil {
			log.Errorf("fail to persistent history table error: %v", err1)
		}
		if err1 := s.flushJobs(); err1 != nil {
			log.Errorf("fail to finish all jobs error: %v", err1)
		}
	}()

	pos := s.meta.Pos()
	gs, err := s.meta.GTID()
	if err != nil {
		return errors.Trace(err)
	}

	s.enableSafeModeInitializationPhase()

	tryReSync := true
	for {
		ctx, cancel := context.WithTimeout(s.ctx, eventTimeout)
		e, err := streamer.GetEvent(ctx)
		cancel()

		startTime := time.Now()

		if err == context.Canceled {
			log.Infof("ready to quit! [%v]", pos)
			return nil
		} else if err == context.DeadlineExceeded {
			log.Info("deadline exceeded.")
			if s.needResync() {
				log.Info("timeout, resync")
				streamer, isGTIDMode, err = s.reopenWithRetry(cfg)
				if err != nil {
					return errors.Trace(err)
				}
			}
			continue
		}

		if err != nil {
			log.Errorf("get binlog error %v", err)
			// try to re-sync in gtid mode
			if tryReSync && isGTIDMode && isBinlogPurgedError(err) && s.cfg.AutoFixGTID {
				time.Sleep(retryTimeout)
				streamer, isGTIDMode, err = s.reSyncBinlog(cfg)
				if err != nil {
					return errors.Trace(err)
				}
				tryReSync = false
				continue
			}

			return errors.Trace(err)
		}
		// get binlog event, reset tryReSync, so we can re-sync binlog while syncer meets errors next time
		tryReSync = true
		binlogPos.WithLabelValues("syncer").Set(float64(e.Header.LogPos))
		binlogFile.WithLabelValues("syncer").Set(getBinlogIndex(s.meta.Pos().Name))

		switch ev := e.Event.(type) {
		case *replication.RotateEvent:
			currentPos := mysql.Position{
				Name: string(ev.NextLogName),
				Pos:  uint32(ev.Position),
			}

			if currentPos.Name > pos.Name {
				// only update pos when rotating to a NEW binlog file
				pos = currentPos
			}
			// ignore fake binlog
			if e.Header.LogPos != 0 {
				log.Infof("rotate binlog to %v", currentPos)
			}
		case *replication.RowsEvent:
			// binlogEventsTotal.WithLabelValues("type", "rows").Add(1)
			schemaName, tableName := s.renameShardingSchema(string(ev.Table.Schema), string(ev.Table.Table))
			log.Debugf("source-db:%s table:%s; target-db:%s table:%s, RowsEvent data: %v", ev.Table.Schema, ev.Table.Table, schemaName, tableName, ev.Rows)

			table := &table{}
			if s.skipRowEvent(schemaName, tableName, e.Header.EventType) {
				binlogSkippedEventsTotal.WithLabelValues("rows").Inc()
				if err = s.recordSkipSQLsPos(insert, pos, gs, "", false); err != nil {
					return errors.Trace(err)
				}

				continue
			}
			table, err = s.getTable(schemaName, tableName)
			if err != nil {
				return errors.Trace(err)
			}

			var (
				sqls []string
				keys [][]string
				args [][]interface{}
			)
			switch e.Header.EventType {
			case replication.WRITE_ROWS_EVENTv0, replication.WRITE_ROWS_EVENTv1, replication.WRITE_ROWS_EVENTv2:
				table = s.chooseTable(table, ev.Rows)
				sqls, keys, args, err = genInsertSQLs(table.Schema, table.Name, ev.Rows, table.Columns, table.IndexColumns)
				if err != nil {
					return errors.Errorf("gen insert sqls failed: %v, schema: %s, table: %s", err, table.Schema, table.Name)
				}
				binlogEvents.WithLabelValues("write_rows").Observe(time.Since(startTime).Seconds())

				for i := range sqls {
					log.Debugf("[write rows] %s\n args: %v\n, keys %v", sqls[i], args[i], keys[i])
					err = s.commitJob(insert, sqls[i], args[i], keys[i], true, pos, gs, i == (len(sqls)-1))
					if err != nil {
						return errors.Trace(err)
					}
				}

			case replication.UPDATE_ROWS_EVENTv0, replication.UPDATE_ROWS_EVENTv1, replication.UPDATE_ROWS_EVENTv2:
				table = s.chooseTable(table, ev.Rows)
				sqls, keys, args, err = genUpdateSQLs(table.Schema, table.Name, ev.Rows, table.Columns, table.IndexColumns)
				if err != nil {
					return errors.Errorf("gen update sqls failed: %v, schema: %s, table: %s", err, table.Schema, table.Name)
				}
				binlogEvents.WithLabelValues("update_rows").Observe(time.Since(startTime).Seconds())

				for i := range sqls {
					log.Debugf("[update rows] %s\n args: %v\n, keys %v", sqls[i], args[i], keys[i])
					err = s.commitJob(update, sqls[i], args[i], keys[i], true, pos, gs, i == (len(sqls)-1))
					if err != nil {
						return errors.Trace(err)
					}
				}
			case replication.DELETE_ROWS_EVENTv0, replication.DELETE_ROWS_EVENTv1, replication.DELETE_ROWS_EVENTv2:
				table = s.chooseTable(table, ev.Rows)
				sqls, keys, args, err = genDeleteSQLs(table.Schema, table.Name, ev.Rows, table.Columns, table.IndexColumns)
				if err != nil {
					return errors.Errorf("gen delete sqls failed: %v, schema: %s, table: %s", err, table.Schema, table.Name)
				}
				binlogEvents.WithLabelValues("delete_rows").Observe(time.Since(startTime).Seconds())

				for i := range sqls {
					log.Debugf("[delete rows] %s\n args: %v\n, keys %v", sqls[i], args[i], keys[i])
					err = s.commitJob(del, sqls[i], args[i], keys[i], true, pos, gs, i == (len(sqls)-1))
					if err != nil {
						return errors.Trace(err)
					}
				}
			}

		case *replication.QueryEvent:
			sql := strings.TrimSpace(string(ev.Query))
			nextPos := mysql.Position{
				Name: pos.Name,
				Pos:  e.Header.LogPos,
			}

			eventAnnotation := fmt.Sprintf("[current pos]%v [next pos]%v [current gtid set]%v [next gtid set]%v", pos, nextPos, gs, ev.GSet)

			sqls, err := resolveDDLSQL(sql, parser2)
			if err != nil {
				if s.skipQueryEvent(sql) {
					binlogSkippedEventsTotal.WithLabelValues("query").Inc()
					log.Warnf("[skip query event][schema]%s [sql]%s %s", ev.Schema, sql, eventAnnotation)
					continue
				}
				log.Errorf("[error query event][schema]%s %s %s", ev.Schema, err.Error(), eventAnnotation)
				return errors.Trace(err)
			}

			if len(sqls) > 0 {
				// only log real ddl, ignore BEGIN
				log.Infof("[query]%s %s", sql, eventAnnotation)
				gs.set(ev.GSet)
				pos = nextPos

				binlogEvents.WithLabelValues("query").Observe(time.Since(startTime).Seconds())
			}

			for i, sql := range sqls {
				// refine skip later
				if s.skipQueryEvent(sql) {
					binlogSkippedEventsTotal.WithLabelValues("query").Inc()
					log.Warnf("[skip query event][schema]%s [sql]%s %s", ev.Schema, sql, eventAnnotation)
					continue
				}

				tableNames, err := s.fetchDDLTableNames(sql, string(ev.Schema), parser2)
				if err != nil {
					return errors.Trace(err)
				}

				sqlAnnotation := fmt.Sprintf("[schema]%s [table]%s %s", tableNames[0][0].Schema, tableNames[0][0].Name, eventAnnotation)
				if s.skipQueryDDL(sql, tableNames[1]) {
					binlogSkippedEventsTotal.WithLabelValues("query_ddl").Inc()
					if err = s.recordSkipSQLsPos(ddl, pos, gs, sqlAnnotation, i == (len(sqls)-1)); err != nil {
						return errors.Trace(err)
					}

					log.Warnf("[skip query event][sql]%s %s", sql, sqlAnnotation)
					continue
				}

				err = s.putTableIntoHistory(tableNames[1][0])
				if err != nil {
					return errors.Trace(err)
				}

				sql, err = genDDLSQL(sql, tableNames[0], tableNames[1], parser2)
				if err != nil {
					return errors.Trace(err)
				}

				log.Infof("[ddl][schema]%s [start]%s", string(ev.Schema), sql)

				job := newJob(ddl, sql, nil, "", false, pos, gs, sqlAnnotation, i == (len(sqls)-1))
				err = s.addJob(job)
				if err != nil {
					return errors.Trace(err)
				}

				log.Infof("[ddl][end]%s", sql)

				s.clearTable(tableNames[1][0].Schema, tableNames[1][0].Name)
			}
		case *replication.XIDEvent:
			pos.Pos = e.Header.LogPos
			gs.set(ev.GSet)

			log.Debugf("[XID event][pos]%v [gtid set]%v", pos, gs)

			job := newXIDJob(pos, gs)
			s.addJob(job)
		}
	}
}

func (s *Syncer) commitJob(tp opType, sql string, args []interface{}, keys []string, retry bool, pos mysql.Position, gs GTIDSet, isLast bool) error {
	key, err := s.resolveCasuality(keys)
	if err != nil {
		return errors.Errorf("resolve karam error %v", err)
	}
	job := newJob(tp, sql, args, key, retry, pos, gs, "", isLast)
	err = s.addJob(job)
	return errors.Trace(err)
}

func (s *Syncer) resolveCasuality(keys []string) (string, error) {
	if s.cfg.DisableCausality {
		if len(keys) > 0 {
			return keys[0], nil
		}
		return "", nil
	}
	if s.c.detectConflict(keys) {
		log.Debug("[causality] meet causality key, will generate a flush job and wait all sqls executed")
		if err := s.flushJobs(); err != nil {
			return "", errors.Trace(err)
		}
		s.c.reset()
	}
	if err := s.c.add(keys); err != nil {
		return "", errors.Trace(err)
	}
	var key string
	if len(keys) > 0 {
		key = keys[0]
	}
	return s.c.get(key), nil
}

func (s *Syncer) genRouter() error {
	s.tableRouter = route.NewTrieRouter()
	for _, rule := range s.cfg.RouteRules {
		err := s.tableRouter.Insert(rule.PatternSchema, rule.PatternTable, rule.TargetSchema, rule.TargertTable)
		if err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

func (s *Syncer) genSkipDMLRules() {
	s.skipDMLRules = &SkipDMLRules{
		skipByDML:    make(map[dmlType]struct{}),
		skipBySchema: make(map[string]map[dmlType]struct{}),
		skipByTable:  make(map[string]map[string]map[dmlType]struct{}),
	}

	// for backward compatibility
	for _, dml := range s.cfg.SkipEvents {
		dt := toDmlType(dml)
		if dt == dmlInvalid {
			continue
		}
		s.skipDMLRules.skipByDML[dt] = struct{}{}
	}

	for _, skipDML := range s.cfg.SkipDMLs {
		dt := toDmlType(skipDML.Type)
		if dt == dmlInvalid {
			continue
		}
		if skipDML.Schema == "" && skipDML.Table == "" {
			s.skipDMLRules.skipByDML[dt] = struct{}{}

		} else if skipDML.Schema == "" && skipDML.Table != "" {
			// it shouldn't have such rule since wo do precheck in config `adjust` function,
			// but check it case we really have one.
			log.Warnf("[syncer] invalid skip dml rule %+v and ignore it", skipDML)
			continue

		} else if skipDML.Schema != "" && skipDML.Table == "" {
			if _, ok := s.skipDMLRules.skipBySchema[skipDML.Schema]; !ok {
				s.skipDMLRules.skipBySchema[skipDML.Schema] = make(map[dmlType]struct{})
			}
			s.skipDMLRules.skipBySchema[skipDML.Schema][dt] = struct{}{}

		} else {
			if _, ok := s.skipDMLRules.skipByTable[skipDML.Schema]; !ok {
				s.skipDMLRules.skipByTable[skipDML.Schema] = make(map[string]map[dmlType]struct{})
			}
			if _, ok := s.skipDMLRules.skipByTable[skipDML.Schema][skipDML.Table]; !ok {
				s.skipDMLRules.skipByTable[skipDML.Schema][skipDML.Table] = make(map[dmlType]struct{})
			}
			s.skipDMLRules.skipByTable[skipDML.Schema][skipDML.Table][dt] = struct{}{}
		}
	}
}

func (s *Syncer) printStatus(ctx context.Context) {
	s.wg.Add(1)
	defer s.wg.Done()

	timer := time.NewTicker(statusTime)
	defer timer.Stop()

	var (
		err                 error
		latestMasterPos     mysql.Position
		latestmasterGTIDSet GTIDSet
	)

	for {
		select {
		case <-ctx.Done():
			log.Infof("print status exits, err:%s", ctx.Err())
			return
		case <-timer.C:
			now := time.Now()
			seconds := float64(now.Unix() - s.lastTime.Unix())
			lastFinished := s.lastFinishedCount.Get()
			finished := s.finishedCount.Get()
			total := s.totalCount.Get()

			tps := float64(0)
			if seconds > 0 {
				tps = float64(finished-lastFinished) / seconds
			}

			if !s.lackOfReplClientPrivilege {
				latestMasterPos, latestmasterGTIDSet, err = s.getMasterStatus()
				if err != nil {
					if isAccessDeniedError(err) {
						s.lackOfReplClientPrivilege = true
					}
					log.Errorf("[syncer] get master status error %s", err)
				} else {
					binlogPos.WithLabelValues("master").Set(float64(latestMasterPos.Pos))
					binlogFile.WithLabelValues("master").Set(getBinlogIndex(latestMasterPos.Name))
				}
			}

			log.Infof("[syncer]total events = %d,  tps = %f, master-binlog = %v, master-binlog-gtid=%v, %s",
				total, tps, latestMasterPos, latestmasterGTIDSet, s.meta)

			s.lastFinishedCount.Set(finished)
			s.lastTime = time.Now()
		}
	}
}

func (s *Syncer) getBinlogStreamer() (*replication.BinlogStreamer, bool, error) {
	defer func() {
		s.lastSlaveConnectionID = s.syncer.LastConnectionID()
		log.Infof("[syncer] last slave connection id %d", s.lastSlaveConnectionID)
	}()
	if s.cfg.EnableGTID {
		gs, err := s.meta.GTID()
		if err != nil {
			return nil, false, errors.Trace(err)
		}

		streamer, err := s.syncer.StartSyncGTID(gs.origin())
		if err != nil {
			log.Errorf("start sync in gtid mode error %v", err)
			return s.startSyncByPosition()
		}

		log.Infof("begin to sync binlog from GTID set %s", gs)
		return streamer, true, errors.Trace(err)
	}

	return s.startSyncByPosition()
}

func (s *Syncer) createDBs() error {
	var err error
	s.fromDB, err = createDB(s.cfg.From, s.cfg.MaxDMLConnectionTimeout)
	if err != nil {
		return errors.Trace(err)
	}

	s.toDBs = make([]*sql.DB, 0, s.cfg.WorkerCount)
	s.toDBs, err = createDBs(s.cfg.To, s.cfg.WorkerCount, s.cfg.MaxDMLConnectionTimeout)
	if err != nil {
		return errors.Trace(err)
	}
	// db for ddl
	s.ddlDB, err = createDB(s.cfg.To, s.cfg.MaxDDLConnectionTimeout)
	if err != nil {
		return errors.Trace(err)
	}

	return nil
}

// record skip ddl/dml sqls' position
// make newJob's sql argument empty to distinguish normal sql and skips sql
func (s *Syncer) recordSkipSQLsPos(op opType, pos mysql.Position, gtidSet GTIDSet, annotation string, isLast bool) error {
	job := newJob(op, "", nil, "", false, pos, gtidSet, annotation, isLast)
	err := s.addJob(job)
	return errors.Trace(err)
}

func (s *Syncer) flushJobs() error {
	log.Infof("flush all jobs meta = %v", s.meta)
	job := newFlushJob()
	err := s.addJob(job)
	return errors.Trace(err)
}

func (s *Syncer) reSyncBinlog(cfg replication.BinlogSyncerConfig) (*replication.BinlogStreamer, bool, error) {
	err := s.retrySyncGTIDs()
	if err != nil {
		return nil, false, errors.Trace(err)
	}
	// close still running sync
	return s.reopenWithRetry(cfg)
}

func (s *Syncer) reopenWithRetry(cfg replication.BinlogSyncerConfig) (streamer *replication.BinlogStreamer, isGTIDMode bool, err error) {
	for i := 0; i < maxRetryCount; i++ {
		streamer, isGTIDMode, err = s.reopen(cfg)
		if err == nil {
			return
		}
		if needRetryReplicate(err) {
			log.Infof("[syncer] retry open binlog streamer %v", err)
			time.Sleep(retryTimeout)
			continue
		}
		break
	}
	return nil, false, errors.Trace(err)
}

func (s *Syncer) reopen(cfg replication.BinlogSyncerConfig) (*replication.BinlogStreamer, bool, error) {
	s.syncer.Close()
	log.Infof("[syncer] killing connectionid %d", s.lastSlaveConnectionID)
	// kill last slave connection by `KILL [id]`
	if s.lastSlaveConnectionID > 0 {
		err := killConn(s.fromDB, s.lastSlaveConnectionID)
		if err != nil {
			if !isNoSuchThreadError(err) {
				return nil, false, errors.Trace(err)
			}
		}
	}

	s.syncer = replication.NewBinlogSyncer(cfg)
	return s.getBinlogStreamer()
}

func (s *Syncer) startSyncByPosition() (*replication.BinlogStreamer, bool, error) {
	pos := s.meta.Pos()
	streamer, err := s.syncer.StartSync(pos)
	log.Infof("begin to sync binlog from position %s", pos)
	return streamer, false, errors.Trace(err)
}

// the result contains [source TableNames, target TableNames]
// the detail of TableNames refs `parserDDLTableNames()`
func (s *Syncer) fetchDDLTableNames(sql string, schema string, parser *parser.Parser) ([][]*filter.Table, error) {
	tableNames, err := parserDDLTableNames(sql, parser)
	if err != nil {
		return nil, errors.Trace(err)
	}

	var targetTableNames []*filter.Table
	for i := range tableNames {
		if tableNames[i].Schema == "" {
			tableNames[i].Schema = schema
		}
		schema, table := s.renameShardingSchema(tableNames[i].Schema, tableNames[i].Name)
		tableName := &filter.Table{
			Schema: schema,
			Name:   table,
		}
		targetTableNames = append(targetTableNames, tableName)
	}

	return [][]*filter.Table{tableNames, targetTableNames}, nil
}

func (s *Syncer) renameShardingSchema(schema, table string) (string, string) {
	if schema == "" {
		return schema, table
	}
	schemaL := strings.ToLower(schema)
	tableL := strings.ToLower(table)
	targetSchema, targetTable := s.tableRouter.Match(schemaL, tableL)
	if targetSchema == "" {
		return schema, table
	}
	if targetTable == "" {
		targetTable = table
	}

	return targetSchema, targetTable
}

func (s *Syncer) isClosed() bool {
	return s.closed.Get()
}

// Close closes syncer.
func (s *Syncer) Close() {
	s.Lock()
	defer s.Unlock()

	if s.isClosed() {
		return
	}

	s.cancel()
	<-s.done

	closeJobChans(s.jobs)

	s.wg.Wait()

	if s.lastSlaveConnectionID > 0 {
		err := killConn(s.fromDB, s.lastSlaveConnectionID)
		if err != nil {
			log.Errorf("[syncer] kill last connection %d err %v", s.lastSlaveConnectionID, err)
		}
	}

	closeDBs(s.fromDB)
	closeDBs(s.toDBs...)
	closeDBs(s.ddlDB)

	if s.syncer != nil {
		s.syncer.Close()
		s.syncer = nil
	}

	s.closed.Set(true)
}

func (s *Syncer) needResync() bool {
	if s.lackOfReplClientPrivilege {
		return false
	}
	masterPos, _, err := s.getMasterStatus()
	if err != nil {
		log.Errorf("get master status err:%s", err)
		return false
	}

	// Why 190 ?
	// +------------------+-----+----------------+-----------+-------------+-------------------------------------------------------------------+
	// | Log_name         | Pos | Event_type     | Server_id | End_log_pos | Info                                                              |
	// +------------------+-----+----------------+-----------+-------------+-------------------------------------------------------------------+
	// | mysql-bin.000002 |   4 | Format_desc    |         1 |         123 | Server ver: 5.7.18-log, Binlog ver: 4                             |
	// | mysql-bin.000002 | 123 | Previous_gtids |         1 |         194 | 00020393-1111-1111-1111-111111111111:1-7
	//
	// Currently, syncer doesn't handle Format_desc and Previous_gtids events. When binlog rotate to new file with only two events like above,
	// syncer won't save pos to 194. Actually it save pos 4 to meta file. So We got a experience value of 194 - 4 = 190.
	// If (mpos.Pos - spos.Pos) > 190, we could say that syncer is not up-to-date.
	return compareBinlogPos(masterPos, s.meta.Pos(), 190) == 1
}

func (s *Syncer) enableSafeModeInitializationPhase() {
	safeMode.Set(true)

	go func() {
		ctx, cancel := context.WithCancel(s.ctx)
		defer func() {
			cancel()
			safeMode.Set(s.cfg.SafeMode)
		}()

		select {
		case <-ctx.Done():
		case <-time.After(5 * time.Minute):
		}
	}()
}