package mysql

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

	"github.com/Shopify/sarama"
	"github.com/coreos/etcd/clientv3"
	"github.com/gogo/protobuf/proto"
	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/tidb/ast"
	uuid "github.com/satori/go.uuid"
	"github.com/siddontang/go-mysql/mysql"
	"github.com/siddontang/go-mysql/replication"
	"github.com/siddontang/go-mysql/schema"
	"github.com/siddontang/go/sync2"
	"golang.org/x/net/context"

	"github.com/pingcap/wormhole/config"
	"github.com/pingcap/wormhole/pkg/etcdutil"
	ether "github.com/pingcap/wormhole/pkg/ether-client"
	"github.com/pingcap/wormhole/pkg/mq/producer"
	"github.com/pingcap/wormhole/pkg/mq/topic"
	"github.com/pingcap/wormhole/pkg/mq/zk"
	pkgmysql "github.com/pingcap/wormhole/pkg/mysql"
	"github.com/pingcap/wormhole/pkg/mysql/ddl"
	"github.com/pingcap/wormhole/pkg/util"
	"github.com/pingcap/wormhole/pkg/whpb"
)

const (
	maxRetryCount       = 100
	retryTimeout        = 3 * time.Second
	eventTimeout        = 1 * time.Hour
	statusTime          = 30 * time.Second
	prepareEventTimeout = 1 * time.Minute
)

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

	cfg *config.Config

	syncer *replication.BinlogSyncer

	wg    sync.WaitGroup
	jobWg sync.WaitGroup

	db *sql.DB

	closed sync2.AtomicBool

	start    time.Time
	lastTime time.Time

	lastCount sync2.AtomicInt64
	count     sync2.AtomicInt64

	ctx    context.Context
	cancel context.CancelFunc

	lackOfReplClientPrivilege bool

	lastGS         pkgmysql.GTIDSet
	lastPos        mysql.Position
	producer       producer.Producer
	ec             ether.Client
	etcdClient     *etcdutil.Client
	zkClient       *zk.Client
	job            *whpb.Job
	mysqlBinlogJob *whpb.MysqlBinlogJob
	ddl            ddl.DDL
}

// NewSyncer creates a new Syncer.
func NewSyncer(job *whpb.Job, ec ether.Client, zkClient *zk.Client, cfg *config.Config) (s *Syncer, err error) {
	s = &Syncer{
		cfg:      cfg,
		job:      job,
		lastTime: time.Now(),
		ec:       ec,
		zkClient: zkClient,
	}
	// some easy init jobs are put here.
	s.ctx, s.cancel = context.WithCancel(context.Background())

	err = s.init()
	return s, errors.Trace(err)
}

func (s *Syncer) init() (err error) {
	// some complicated init jobs are put here.

	// mysqljob
	mysqlJob := &whpb.MysqlBinlogJob{}
	err = proto.Unmarshal(s.job.Info, mysqlJob)
	if err != nil {
		return errors.Trace(err)
	}
	s.mysqlBinlogJob = mysqlJob

	log.Infof("[syncer] recv job %+v", mysqlJob)

	// etcdutil.Client
	etcd, err := clientv3.New(clientv3.Config{
		Endpoints: s.cfg.EtherEndpoints,
	})
	if err != nil {
		return errors.Trace(err)
	}
	rootpath := s.ec.GetStorageRootPath()
	etcdClient := etcdutil.NewClient(etcd, rootpath)
	s.etcdClient = etcdClient

	// ddl
	s.ddl = ddl.NewDDL(s.etcdClient, mysqlJob.GetSource())

	// producer
	saramaCfg := sarama.NewConfig()
	// for SyncProducer, it must set true.
	saramaCfg.Producer.Return.Successes = true
	s.producer, err = producer.NewProducer(s.cfg.KafkaBrokers, saramaCfg, etcdClient, producer.EnableSavepoint())
	if err != nil {
		return errors.Trace(err)
	}

	// db
	s.db, err = pkgmysql.CreateDB(s.mysqlBinlogJob.GetSource())
	if err != nil {
		return errors.Trace(err)
	}
	return nil
}

func (s *Syncer) initFromDataSource(cfg replication.BinlogSyncerConfig) (pos *whpb.Position, err error) {
	// show master status, get pos 1
	pos1, gs1, err := s.getMasterStatus()
	if err != nil {
		return nil, errors.Trace(err)
	}
	log.Debugf("[syncer] pos1 %+v, gs1 %+v", pos1, gs1)

	var (
		tableNames []string
		tables     []*schema.Table
	)
	schemaNames, err := s.showSchemas()
	if err != nil {
		return nil, errors.Trace(err)
	}

	for _, schemaName := range schemaNames {
		tableNames, err = pkgmysql.ShowTables(s.db, schemaName)
		if err != nil {
			return nil, errors.Trace(err)
		}

		// show create tables
		tables, err = s.getTables(schemaName, tableNames)
		if err != nil {
			return nil, errors.Trace(err)
		}

		if len(tables) == 0 {
			log.Infof("[syncer] schema %s has no tables", schemaName)
			s.ddl.SetSchema(schemaName)
			continue
		}

		log.Infof("[syncer] schema %s tables %+v", schemaName, tables)
		for _, table := range tables {
			tableInfo := pkgmysql.ToTableInfo(table)
			s.ddl.SetTable(schemaName, tableInfo)
		}
	}

	// show master status, get pos 2 (now we have got base schema info)
	pos2, gs2, err := s.getMasterStatus()
	if err != nil {
		return nil, errors.Trace(err)
	}
	log.Debugf("pos2 %+v, gs2 %+v", pos2, gs2)

	// pos1 == pos2  we can say there is no scheam change during fetching tables.
	if pos1.Compare(pos2) == 0 {
		log.Debugf("[syncer] no schema changes during initialization.")
		pos = &whpb.Position{
			MysqlBinlogFile: pos2.Name,
			MysqlBinlogPos:  uint64(pos2.Pos),
			MysqlBinlogGtid: gs2.String(),
		}
		log.Debugf("[syncer] pos %+v", pos)
		return pos, nil
	}

	// temporary streamer for only prepare
	streamer, _, err := s.getStreamer(pos1, gs1)
	if err != nil {
		return nil, errors.Trace(err)
	}
	// pull binlog from pos1
	// FIXME: redundant code compare to func Process()

	currentPos := pos1

FindDDL:
	for {
		ctx, cancel := context.WithTimeout(s.ctx, prepareEventTimeout)
		e, err := streamer.GetEvent(ctx)
		cancel()
		if err == context.Canceled {
			log.Infof("[syncer] ready to quit! [%v]", s.lastPos)
			return nil, nil
		} else if err == context.DeadlineExceeded {
			log.Info("[syncer] deadline exceeded.")
			continue
		}

		// ignore dmls and apply ddls to schema info
		switch ev := e.Event.(type) {
		case *replication.RotateEvent:
			binlogEventsTotal.WithLabelValues("rotate").Inc()
			currentPos = mysql.Position{
				Name: string(ev.NextLogName),
				Pos:  uint32(ev.Position),
			}

		case *replication.QueryEvent:
			currentPos.Pos = e.Header.LogPos

			sql := strings.TrimSpace(string(ev.Query))
			ddls, err := resolveDDLSQL(sql, string(ev.Schema))
			if err != nil {
				if skipQueryEvent(sql) {
					log.Warnf("[skip query-sql]%s  [schema]:%s", sql, ev.Schema)
					continue
				}
				return nil, errors.Errorf("[syncer]parse query event failed: %v", err)
			}

			for _, ddl := range ddls {
				err = s.ddl.Execute(ddl.stmt)
				if err != nil {
					if s.checkNoPrivilegedDBError(err) {
						continue
					}
					return nil, errors.Trace(err)
				}
			}
		case *replication.XIDEvent:
			currentPos.Pos = e.Header.LogPos

		default:
			// if cur_pos > pos2, finish initialization. then produce drips(go to Process).
			if currentPos.Compare(pos2) == 1 {
				break FindDDL
			}

		}
	}

	// reset s.syncer, for later use
	s.syncer.Close()
	s.syncer = replication.NewBinlogSyncer(cfg)

	pos = &whpb.Position{
		MysqlBinlogFile: pos2.Name,
		MysqlBinlogPos:  uint64(pos2.Pos),
		MysqlBinlogGtid: gs2.String(),
	}

	return pos, nil
}

func (s *Syncer) getTables(schemaName string, tableNames []string) (tables []*schema.Table, err error) {
	tables = make([]*schema.Table, 0, len(tableNames))
	for _, tableName := range tableNames {
		table, err := pkgmysql.GetSchemaTable(s.db, schemaName, tableName)
		if err != nil {
			return nil, errors.Trace(errors.Annotatef(err, "get table error. schema %s, table %s", schemaName, tableName))
		}
		tables = append(tables, table)
	}

	return tables, nil
}

func (s *Syncer) prepare() (steamer *replication.BinlogStreamer, cfg replication.BinlogSyncerConfig, err error) {
	serverid := uint32(s.mysqlBinlogJob.GetServerId())
	if serverid == 0 {
		log.Warn("server id is 0, use the config file's MySQLServerID")
		serverid = uint32(s.cfg.MySQLServerID)
	}
	// syncer
	cfg = replication.BinlogSyncerConfig{
		ServerID: serverid,
		Flavor:   "mysql",
		Host:     s.mysqlBinlogJob.GetSource().Host,
		Port:     uint16(s.mysqlBinlogJob.GetSource().Port),
		User:     s.mysqlBinlogJob.GetSource().Username,
		Password: s.mysqlBinlogJob.GetSource().Password,
	}
	s.syncer = replication.NewBinlogSyncer(cfg)

	err = pkgmysql.CheckBinlog(s.db)
	if err != nil {
		return nil, cfg, errors.Trace(err)
	}
	go s.printStatus()

	var pos *whpb.Position
	if s.job.Status.GetPos() != nil {
		pos = s.job.Status.Pos
	}

	// initialize schema informations
	err = s.ddl.Init()
	if err != nil {
		if util.OriginError(err) != ddl.ErrNotInitialized {
			return nil, cfg, errors.Trace(err)
		}
		log.Debugf("[syncer] init from data source %+v", cfg)
		// initial from data source
		pos, err = s.initFromDataSource(cfg)
		if err != nil {
			return nil, cfg, errors.Trace(err)
		}
	}

	log.Infof("[syncer] job status %+v,", s.job.Status)
	// initialize schema meta
	if s.job.Status.GetStage() != whpb.JobStage_Running {
		summary := s.ddl.GetSchemaSummary()
		schemas := make([]*whpb.Schema, 0, len(summary))
		for schema, tables := range summary {
			schema := &whpb.Schema{
				Name:   schema,
				Tables: tables,
			}
			schemas = append(schemas, schema)
		}

		meta := &whpb.SchemaInfo{
			Host:    s.mysqlBinlogJob.GetSource().Host,
			Port:    s.mysqlBinlogJob.GetSource().Port,
			Schemas: schemas,
		}
		log.Debugf("[syncer] save schema meta %+v, schemas %+v", meta, schemas)
		err = s.etcdClient.SaveSchemaMeta(s.mysqlBinlogJob.GetSource().Host, s.mysqlBinlogJob.GetSource().Port, meta)
		if err != nil {
			return nil, cfg, errors.Trace(err)
		}

		s.job.Status.Pos = pos
		err = s.reportRunningStatus()
		if err != nil {
			return nil, cfg, errors.Trace(err)
		}
		log.Infof("[syncer] datasource %+v has initialized", s.mysqlBinlogJob.GetSource())
	}

	s.lastPos = mysql.Position{
		Name: pos.GetMysqlBinlogFile(),
		Pos:  uint32(pos.GetMysqlBinlogPos()),
	}
	s.lastGS, err = pkgmysql.ParseGTIDSet(pos.GetMysqlBinlogGtid())
	if err != nil {
		return nil, cfg, errors.Trace(err)
	}
	log.Debugf("[syncer] last_pos %+v, last_gtid_set %+v", s.lastPos, s.lastGS)

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

	return streamer, cfg, nil
}

func (s *Syncer) reportRunningStatus() error {
	// we want to reuse the connection, so we ignore `cancel`` here.
	s.job.Status.Stage = whpb.JobStage_Running
	cloneJob := proto.Clone(s.job).(*whpb.Job)
	err := s.ec.SendHeartbeat(s.ctx, []*whpb.JobStatus{cloneJob.Status})
	return errors.Trace(err)
}

// GetStatus implements Processor interface.
func (s *Syncer) GetStatus() *whpb.JobStatus {
	if s.job.Status == nil {
		s.job.Status = &whpb.JobStatus{}
	}
	s.job.Status.Pos = &whpb.Position{
		MysqlBinlogFile: s.lastPos.Name,
		MysqlBinlogPos:  uint64(s.lastPos.Pos),
		MysqlBinlogGtid: s.lastGS.String(),
	}
	return proto.Clone(s.job).(*whpb.Job).Status
}

// Process implements Processor.
func (s *Syncer) Process() error {
	var (
		err     error
		uuidSet *mysql.UUIDSet
	)

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

	defer func() {
		if err1 := recover(); err1 != nil {
			log.Errorf("[syncer] panic. err: %s, stack: %s", err1, debug.Stack())
			err = errors.Errorf("panic error: %v", err1)
		}
		s.Close()
	}()

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

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

		if err != nil {
			log.Errorf("[syncer] get binlog error %v", err)
			return errors.Trace(err)
		}
		// get binlog event, reset tryReSync, so we can re-sync binlog while syncer meets errors next time
		binlogPos.WithLabelValues("syncer").Set(float64(e.Header.LogPos))
		binlogFile.WithLabelValues("syncer").Set(getBinlogIndex(s.lastPos.Name))

		switch ev := e.Event.(type) {
		case *replication.RotateEvent:
			binlogEventsTotal.WithLabelValues("rotate").Inc()
			currentPos := mysql.Position{
				Name: string(ev.NextLogName),
				Pos:  uint32(ev.Position),
			}
			if currentPos.Name > s.lastPos.Name {
				s.lastPos = currentPos
			}
			log.Infof("[syncer] rotate binlog to %v", s.lastPos)

		case *replication.RowsEvent:
			// binlogEventsTotal.WithLabelValues("type", "rows").Add(1)
			schemaName, tableName := string(ev.Table.Schema), string(ev.Table.Table)
			// DON'T skip row event, let it done in stargate.

			log.Debugf("[syncer] schema %s, table %s rows data %+v", schemaName, tableName, ev.Rows)

			if isIgnoredSchema(schemaName) {
				log.Infof("[syncer] ignore schema %s", schemaName)
				continue
			}

			switch e.Header.EventType {
			case replication.WRITE_ROWS_EVENTv0, replication.WRITE_ROWS_EVENTv1, replication.WRITE_ROWS_EVENTv2:
				binlogEventsTotal.WithLabelValues("write_rows").Inc()
				s.count.Add(int64(len(ev.Rows)))
				err = s.produceDMLData(schemaName, tableName, whpb.EventType_Insert, ev.Rows)
				if err != nil {
					return errors.Trace(err)
				}

			case replication.UPDATE_ROWS_EVENTv0, replication.UPDATE_ROWS_EVENTv1, replication.UPDATE_ROWS_EVENTv2:
				binlogEventsTotal.WithLabelValues("update_rows").Inc()
				s.count.Add(int64(len(ev.Rows) / 2))
				err = s.produceDMLData(schemaName, tableName, whpb.EventType_Update, ev.Rows)
				if err != nil {
					return errors.Trace(err)
				}

			case replication.DELETE_ROWS_EVENTv0, replication.DELETE_ROWS_EVENTv1, replication.DELETE_ROWS_EVENTv2:
				binlogEventsTotal.WithLabelValues("delete_rows").Inc()
				s.count.Add(int64(len(ev.Rows)))
				err = s.produceDMLData(schemaName, tableName, whpb.EventType_Delete, ev.Rows)
				if err != nil {
					return errors.Trace(err)
				}
			}

		case *replication.QueryEvent:
			binlogEventsTotal.WithLabelValues("query").Inc()

			sql := strings.TrimSpace(string(ev.Query))
			nextPos := mysql.Position{
				Name: s.lastPos.Name,
				Pos:  e.Header.LogPos,
			}

			ddls, err := resolveDDLSQL(sql, string(ev.Schema))
			if err != nil {
				if skipQueryEvent(sql) {
					binlogSkippedEventsTotal.WithLabelValues("query").Inc()
					log.Warnf("[skip query-sql]%s  [schema]:%s", sql, ev.Schema)
					continue
				}
				log.Infof("[syncer][query]%s [pos]%v [next pos]%v [gtid_next]%v", sql, s.lastPos, nextPos, uuidSet)
				return errors.Errorf("[syncer] parse query event failed: %v", err)
			}

			if len(ddls) == 0 {
				continue
			}
			s.lastGS.AddSet(uuidSet)
			log.Infof("[syncer][query]%s [pos]%v [next pos]%v [gtid]%v", sql, s.lastPos, nextPos, s.lastGS)
			s.lastPos = nextPos

			for _, ddl := range ddls {
				err = s.produceDDLData(ddl)
				if err != nil {
					return errors.Trace(err)
				}
			}

		case *replication.XIDEvent:
			s.lastPos.Pos = e.Header.LogPos
			s.lastGS.AddSet(uuidSet)
		case *replication.GTIDEvent:
			// only support mysql gtid, not for mariadb
			s.lastPos.Pos = e.Header.LogPos
			u, err := uuid.FromBytes(ev.SID)
			if err != nil {
				return errors.Trace(err)
			}
			gtid := fmt.Sprintf("%s:1-%d", u.String(), ev.GNO)
			uuidSet, err = mysql.ParseUUIDSet(gtid)
			if err != nil {
				return errors.Trace(err)
			}
			log.Debugf("[syncer] gtid information: binlog %v, gtid %s", s.lastPos, gtid)
			binlogGTID.WithLabelValues("syncer").Set(float64(ev.GNO))
		}
	}
}

// produce DML data to kafka.
func (s *Syncer) produceDMLData(schemaName, tableName string, et whpb.EventType, rowsData [][]interface{}) error {
	tableInfo, err := s.ddl.Snapshot(schemaName, tableName)
	if err != nil {
		if s.checkNoPrivilegedDBError(err) {
			return nil
		}
		return errors.Trace(err)
	}

	rows := pkgmysql.ToRowData(rowsData, tableInfo.Columns)
	metaCols := pkgmysql.ToColumns(tableInfo.Columns)
	dmlData := &whpb.DMLData{
		Schema:  schemaName,
		Table:   tableName,
		Tp:      et,
		Rows:    rows,
		Columns: metaCols,
	}
	data, err := proto.Marshal(dmlData)
	if err != nil {
		return errors.Trace(err)
	}

	t, err := topic.MakeSyncerTopic(schemaName, tableName, s.mysqlBinlogJob.GetSource(), s.producer, s.zkClient)
	if err != nil {
		return errors.Trace(err)
	}

	dmlDrip := s.dmlDrip(schemaName, tableName, data)
	log.Debugf("[syncer] dml drip %+v", dmlDrip)
	_, _, err = s.producer.Produce(t, dmlDrip)
	return errors.Trace(err)
}

// Sometimes, `s.mysqlBinlogJob.GetSource().Username` doesn't have privilege on some schemas.
// So we should ignore these schemas.
func (s *Syncer) checkNoPrivilegedDBError(err error) bool {
	oerr, ok := util.OriginError(err).(*ddl.DBNotExistsError)
	if !ok {
		return false
	}

	schemaNames, err := s.showSchemas()
	if err != nil {
		log.Errorf("[syncer] show schemas error %s", err)
		return false
	}
	for _, schema := range schemaNames {
		if strings.ToLower(oerr.Name()) == strings.ToLower(schema) {
			// if found, it means DBNotExistsError is unexpected.
			return false
		}
	}
	log.Warnf("[syncer] skip schema %s because user %s doesn't has privilege on it", oerr.Name(), s.mysqlBinlogJob.GetSource().Username)
	return true
}

// produce ddl data to message queue.
// 1. execute sql in memory ddl
// 2. make topic
// 3. create fence
// 4. make drip
// 5. fanout producer(topic, drip)
func (s *Syncer) produceDDLData(ddlNode *ddlNode) (err error) {
	s.count.Add(1)

	err = s.ddl.Execute(ddlNode.stmt)
	if err != nil {
		if s.checkNoPrivilegedDBError(err) {
			return nil
		}
		return errors.Trace(err)
	}

	var (
		dripSchema string
		dripTable  string
	)
	topics := make([]*whpb.Topic, 0, 0)

	switch v := ddlNode.stmt.(type) {
	case *ast.CreateDatabaseStmt:
		return nil
	case *ast.DropDatabaseStmt:
		tables := s.ddl.GetTables(v.Name)
		for _, table := range tables {
			//fanout drips to all tables that belong to it.
			t, err := topic.MakeSyncerTopic(v.Name, table, s.mysqlBinlogJob.GetSource(), s.producer, s.zkClient)
			if err != nil {
				return errors.Trace(err)
			}
			topics = append(topics, t)
		}
		if len(tables) == 0 {
			t, err := topic.MakeSpecialTopic(v.Name, s.mysqlBinlogJob.GetSource(), s.producer, s.zkClient, whpb.SourceType_mysql_binlog)
			if err != nil {
				return errors.Trace(err)
			}
			topics = append(topics, t)
		}
		dripSchema = v.Name

	case *ast.CreateTableStmt:
		t, err := topic.MakeSyncerTopic(ddlNode.table.Schema, ddlNode.table.Name, s.mysqlBinlogJob.GetSource(), s.producer, s.zkClient)
		if err != nil {
			return errors.Trace(err)
		}
		topics = append(topics, t)
		sql := createDatabaseStmtToSQL(&ast.CreateDatabaseStmt{Name: ddlNode.table.Schema})
		drip := s.ddlDrip(ddlNode.table.Schema, ddlNode.table.Name, sql, 0)
		err = s.producer.FanoutProduce(topics, drip)
		if err != nil {
			return errors.Trace(err)
		}

		dripSchema = ddlNode.table.Schema
		dripTable = ddlNode.table.Name

	case *ast.AlterTableStmt:
		// if rename table, send ddl to new topic(new table)
		// both old table and new table should have same fence.
		t, err := topic.MakeSyncerTopic(ddlNode.table.Schema, ddlNode.table.Name, s.mysqlBinlogJob.GetSource(), s.producer, s.zkClient)
		if err != nil {
			return errors.Trace(err)
		}
		topics = append(topics, t)

		if ddlNode.newTable.Name != "" {
			nt, err := topic.MakeSyncerTopic(ddlNode.newTable.Schema, ddlNode.newTable.Name, s.mysqlBinlogJob.GetSource(), s.producer, s.zkClient)
			if err != nil {
				return errors.Trace(err)
			}
			topics = append(topics, nt)
		}
		dripSchema = ddlNode.table.Schema
		dripTable = ddlNode.table.Name

	case *ast.RenameTableStmt:
		// both old table and new table should have same fence.
		ot, err := topic.MakeSyncerTopic(ddlNode.table.Schema, ddlNode.table.Name, s.mysqlBinlogJob.GetSource(), s.producer, s.zkClient)
		if err != nil {
			return errors.Trace(err)
		}
		nt, err := topic.MakeSyncerTopic(ddlNode.newTable.Schema, ddlNode.newTable.Name, s.mysqlBinlogJob.GetSource(), s.producer, s.zkClient)
		if err != nil {
			return errors.Trace(err)
		}
		topics = append(topics, ot, nt)
		dripSchema = ddlNode.table.Schema
		dripTable = ddlNode.table.Name

	case *ast.DropTableStmt, *ast.TruncateTableStmt, *ast.DropIndexStmt, *ast.CreateIndexStmt:
		t, err := topic.MakeSyncerTopic(ddlNode.table.Schema, ddlNode.table.Name, s.mysqlBinlogJob.GetSource(), s.producer, s.zkClient)
		if err != nil {
			return errors.Trace(err)
		}
		topics = append(topics, t)
		dripSchema = ddlNode.table.Schema
		dripTable = ddlNode.table.Name

	default:
		// according to tidb source code, there are only 9 statements(the above statements) that implements DDLNode interface.
		panic("unreachable code")
	}

	return errors.Trace(s.fanoutProduceDDL(topics, dripSchema, dripTable, ddlNode.sql))
}

func (s *Syncer) fanoutProduceDDL(topics []*whpb.Topic, dripSchema, dripTable, query string) error {
	fenceID, err := s.ec.CreateFence(s.ctx, topics)
	if err != nil {
		return errors.Trace(err)
	}
	drip := s.ddlDrip(dripSchema, dripTable, query, fenceID)
	err = s.producer.FanoutProduce(topics, drip)
	return errors.Trace(err)
}

func (s *Syncer) dmlDrip(schema string, table string, data []byte) *whpb.Drip {
	return &whpb.Drip{
		Schema: schema,
		Table:  table,
		Source: whpb.SourceType_mysql_binlog,
		Op:     whpb.OPType_DML,
		Data:   data,
		Pos: &whpb.Position{
			MysqlBinlogFile: s.lastPos.Name,
			MysqlBinlogPos:  uint64(s.lastPos.Pos),
			MysqlBinlogGtid: s.lastGS.String(),
		},
		CreateTime: time.Now().Unix(),
	}
}

func (s *Syncer) ddlDrip(schema string, table string, query string, fenceID uint64) *whpb.Drip {
	return &whpb.Drip{
		Schema:  schema,
		Table:   table,
		Source:  whpb.SourceType_mysql_binlog,
		Op:      whpb.OPType_DDL,
		Data:    []byte(query),
		FenceId: fenceID,
		Pos: &whpb.Position{
			MysqlBinlogFile: s.lastPos.Name,
			MysqlBinlogPos:  uint64(s.lastPos.Pos),
			MysqlBinlogGtid: s.lastGS.String(),
		},
		CreateTime: time.Now().Unix(),
	}
}

func (s *Syncer) getServerUUID() (string, error) {
	return pkgmysql.GetServerUUID(s.db)
}

func (s *Syncer) getMasterStatus() (mysql.Position, pkgmysql.GTIDSet, error) {
	return pkgmysql.GetMasterStatus(s.db)
}

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

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

	var (
		err                 error
		latestMasterPos     mysql.Position
		latestMasterGTIDSet pkgmysql.GTIDSet
	)

	for {
		select {
		case <-s.ctx.Done():
			return
		case <-timer.C:
			now := time.Now()
			seconds := now.Unix() - s.lastTime.Unix()
			last := s.lastCount.Get()
			total := s.count.Get()

			tps := int64(0)
			if seconds > 0 {
				tps = (total - last) / seconds
			}
			s.job.Status.Qps = uint64(tps)

			s.lastCount.Set(total)
			s.lastTime = time.Now()

			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))
					masterGTIDGauge(latestMasterGTIDSet, s.db)
				}
			}
		}
	}
}

func (s *Syncer) isGTIDEnabled() bool {
	mode, err := pkgmysql.IsGTIDModeEnabled(s.db)
	if err != nil {
		log.Errorf("[syncer] check mysql gtid mode err %s", errors.ErrorStack(err))
		return false
	}
	return mode
}

func (s *Syncer) getBinlogStreamer() (*replication.BinlogStreamer, bool, error) {
	return s.getStreamer(s.lastPos, s.lastGS)
}

func (s *Syncer) getStreamer(pos mysql.Position, gs pkgmysql.GTIDSet) (*replication.BinlogStreamer, bool, error) {
	if s.isGTIDEnabled() {
		streamer, err := s.syncer.StartSyncGTID(gs)
		if err != nil {
			log.Errorf("[syncer] start sync in gtid mode error %v", err)
			return s.startSyncByPosition(pos)
		}

		return streamer, true, errors.Trace(err)
	}

	return s.startSyncByPosition(pos)
}

// 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.reopen(cfg)
// }

func (s *Syncer) reopen(cfg replication.BinlogSyncerConfig) (*replication.BinlogStreamer, bool, error) {
	s.syncer.Close()
	s.syncer = replication.NewBinlogSyncer(cfg)
	return s.getBinlogStreamer()
}

func (s *Syncer) startSyncByPosition(pos mysql.Position) (*replication.BinlogStreamer, bool, error) {
	streamer, err := s.syncer.StartSync(pos)
	return streamer, false, errors.Trace(err)
}

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

// Close implements Processor interface.
func (s *Syncer) Close() {
	s.Lock()
	defer s.Unlock()

	if s.isClosed() {
		return
	}

	s.cancel()
	s.wg.Wait()

	if err := s.producer.Close(); err != nil {
		log.Errorf("[syncer] close producer error:%s", err)
	}
	if err := s.db.Close(); err != nil {
		log.Errorf("[syncer] close db error:%s", err)
	}

	if s.syncer != nil {
		s.syncer.Close()
		s.syncer = nil
	}
	log.Infof("[syncer] syncer closed with job %+v", s.job)
	s.closed.Set(true)
}

func (s *Syncer) needResync() bool {
	if s.lackOfReplClientPrivilege {
		return false
	}
	masterPos, _, err := s.getMasterStatus()
	if err != nil {
		log.Errorf("[syncer] 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 util.CompareBinlogPosWithDeviation(masterPos, s.lastPos, 190) == 1
}

func (s *Syncer) showSchemas() (schemas []string, err error) {
	originSchemas, err := pkgmysql.ShowSchemas(s.db)
	if err != nil {
		return nil, errors.Trace(err)
	}
	schemas = make([]string, 0, len(originSchemas))
	for _, schema := range originSchemas {
		if !isIgnoredSchema(schema) {
			schemas = append(schemas, schema)
		}
	}

	return schemas, nil
}