package mysql

import (
	"database/sql"
	"fmt"
	"sync"
	"sync/atomic"
	"time"

	"github.com/Shopify/sarama"
	"github.com/coreos/etcd/clientv3"
	"github.com/gogo/protobuf/proto"
	"github.com/juju/errors"
	"github.com/juju/utils/set"
	"github.com/ngaut/log"
	"github.com/pingcap/wormhole/config"
	"github.com/pingcap/wormhole/pkg/etcdutil"
	"github.com/pingcap/wormhole/pkg/ether-client"
	"github.com/pingcap/wormhole/pkg/mq/producer"
	"github.com/pingcap/wormhole/pkg/mq/zk"
	wmysql "github.com/pingcap/wormhole/pkg/mysql"
	"github.com/pingcap/wormhole/pkg/whpb"
	"golang.org/x/net/context"
)

//Dumper is a collections for dumping
type Dumper struct {
	cfg          *config.Config
	job          *whpb.Job
	mysqlDumpJob *whpb.MysqlDumpJob

	jobCh    chan *dumpJob
	db       *sql.DB
	ec       ether.Client
	producer producer.Producer
	dbDrip   *whpb.Drip

	jobCnt   uint64
	jobFin   uint64
	lastTime int64

	closed int64
	quit   chan struct{}  // worker routine
	wg     sync.WaitGroup // ditto.

	zkClient *zk.Client
}

// NewDumper creates a new Dumper.
func NewDumper(job *whpb.Job, ec ether.Client, zkClient *zk.Client, cfg *config.Config) (*Dumper, error) {
	d := &Dumper{
		cfg:      cfg,
		job:      job,
		jobCh:    make(chan *dumpJob, cfg.RoutineCnt),
		ec:       ec,
		lastTime: time.Now().Unix(),
		quit:     make(chan struct{}),
	}

	d.zkClient = zkClient
	d.mysqlDumpJob = &whpb.MysqlDumpJob{}
	err := proto.Unmarshal(d.job.Info, d.mysqlDumpJob)
	if err != nil {
		return nil, errors.Annotate(err, "[dumper] unmarshal to MysqlDumpJob fail")
	}

	schema := d.mysqlDumpJob.GetSchemas()[0]
	log.Infof("[dumper] dump schema %s, tables %+v", schema, schema.GetTables())

	d.db, err = wmysql.CreateDB(d.mysqlDumpJob.GetSource())
	if err != nil {
		return nil, errors.Annotate(err, "[dumper] connect to db fail")
	}
	log.Debugf("[dumper] connect to db success")

	err = d.genProducer()
	if err != nil {
		return nil, errors.Annotate(err, "[dumper] generate kafka producer fail")
	}
	return d, nil
}

// Close implements Processor interface.
func (d *Dumper) Close() {
	if !atomic.CompareAndSwapInt64(&d.closed, 0, 1) {
		// dumper is already closed
		return
	}

	close(d.quit)
	d.wg.Wait()

	d.producer.Close()
	d.db.Close()
	log.Infof("[dumper] dumper closed")
}

// Process implements Processor interface.
// cut mysql dump job and dispatch
func (d *Dumper) Process() error {
	defer d.Close()

	err := wmysql.CheckBinlog(d.db)
	if err != nil {
		return errors.Trace(err)
	}

	err = d.reportRunningStatus()
	if err != nil {
		return errors.Trace(err)
	}

	// fixme: compatible with mutiple schemas
	if !d.mysqlDumpJob.GetSchemas()[0].GetNoLocks() {
		err = wmysql.LockTables(d.db)
		if err != nil {
			return errors.Trace(err)
		}
		log.Debug("[dumper] flush tables with read lock")
	}

	err = wmysql.CreateSnapshot(d.db)
	if err != nil {
		return errors.Trace(err)
	}
	log.Debug("[dumper] create snapshot")

	pos, gtid, err := wmysql.GetMasterStatus(d.db)
	if err != nil {
		return errors.Trace(err)
	}
	d.job.GetStatus().Pos = &whpb.Position{
		MysqlBinlogFile: pos.Name,
		MysqlBinlogPos:  uint64(pos.Pos),
		MysqlBinlogGtid: gtid.String(),
	}

	log.Debugf("[dumper] start %d goroutine of processQueue", d.cfg.RoutineCnt)
	for i := 0; i < d.cfg.RoutineCnt; i++ {
		d.wg.Add(1)
		go d.processQueue()
	}

	err = d.dumpCreateSchema()
	if err != nil {
		return errors.Trace(err)
	}

	err = d.dumpCreateTable()
	if err != nil {
		return errors.Trace(err)
	}

	if len(d.mysqlDumpJob.GetSchemas()) == 0 {
		return errors.New("dump job don't get schema")
	}
	// now dumper job only contains one schema
	schema := d.mysqlDumpJob.GetSchemas()[0]

	innoDBTables, nonInnoDBTables, err := d.markOffInnoDB(schema)
	if err != nil {
		return errors.Trace(err)
	}
	log.Infof("[dumper] markoff innodb=%v and nonInnodb=%v", innoDBTables, nonInnoDBTables)

	err = d.dumpTableData(schema.GetName(), nonInnoDBTables)
	if err != nil {
		return errors.Trace(err)
	}
	log.Info("[dumper] finish dump non innodb table data")

	if !d.mysqlDumpJob.GetSchemas()[0].GetNoLocks() {
		err = wmysql.UnlockTables(d.db)
		if err != nil {
			return errors.Trace(err)
		}
	}

	err = d.dumpTableData(schema.GetName(), innoDBTables)
	if err != nil {
		return errors.Trace(err)
	}
	log.Info("[dumper] finish dump innodb table data")

	err = d.sendJobEnd()
	if err != nil {
		return errors.Trace(err)
	}
	return nil
}

func (d *Dumper) reportRunningStatus() error {
	d.job.Status.Stage = whpb.JobStage_Running
	cloneJob := proto.Clone(d.job).(*whpb.Job)
	err := d.ec.SendHeartbeat(context.Background(), []*whpb.JobStatus{cloneJob.Status})
	return errors.Trace(err)
}

// GetStatus implements Processor interface.
func (d *Dumper) GetStatus() *whpb.JobStatus {
	d.job.Status.Total = d.jobCnt
	if d.job.GetStatus().GetStage() == whpb.JobStage_Resolved {
		d.job.Status.Percent = 100
		d.job.Status.Qps = 0
	} else {
		if d.jobCnt != 0 {
			d.job.Status.Percent = d.jobFin * 100 / d.jobCnt
		} else {
			d.job.Status.Percent = 0
		}
		duration := uint64(time.Now().Unix() - d.lastTime)
		if duration != 0 {
			d.job.Status.Qps = d.jobFin * d.cfg.ChunkSize / duration
		}
	}
	// return a clone to prevent modification.
	return proto.Clone(d.job).(*whpb.Job).GetStatus()
}

func (d *Dumper) genProducer() error {
	saramaCfg := sarama.NewConfig()
	saramaCfg.Producer.Return.Successes = true
	saramaCfg.Producer.MaxMessageBytes = 10 * 1024 * 1024

	etcd, err := clientv3.New(clientv3.Config{
		Endpoints: d.cfg.EtherEndpoints,
	})
	if err != nil {
		return errors.Trace(err)
	}

	rootpath := d.ec.GetStorageRootPath()
	etcdClient := etcdutil.NewClient(etcd, rootpath)

	d.producer, err = producer.NewProducer(d.cfg.KafkaBrokers, saramaCfg, etcdClient)
	if err != nil {
		return errors.Trace(err)
	}
	return nil
}

func (d *Dumper) markOffInnoDB(schema *whpb.Schema) ([]string, []string, error) {
	dbname := schema.GetName()
	sql := fmt.Sprintf("SHOW TABLE STATUS FROM %s", dbname)
	log.Debugf("[dumper] markOffInnoDB sql=%v", sql)
	rows, err := d.db.Query(sql)
	if err != nil {
		return nil, nil, errors.Trace(err)
	}
	defer rows.Close()

	innoDBTable := set.NewStrings()
	nonInnoDBTable := set.NewStrings()
	for rows.Next() {
		colVals, err := wmysql.ScanRow(rows)
		if err != nil {
			return nil, nil, errors.Trace(err)
		}
		table := string(colVals["Name"])
		engine := string(colVals["Engine"])
		if engine == "InnoDB" {
			innoDBTable.Add(table)
		} else {
			nonInnoDBTable.Add(table)
		}
	}

	allTable := set.NewStrings()
	tables := schema.GetTables()
	for _, table := range tables {
		allTable.Add(table)
	}
	return innoDBTable.Intersection(allTable).Values(),
		nonInnoDBTable.Intersection(allTable).Values(), nil
}