package mysql

import (
	"context"
	"database/sql"
	"fmt"
	"reflect"
	"strings"
	"sync"
	"time"

	"github.com/Shopify/sarama"
	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/tidb/model"

	"github.com/pingcap/wormhole/pkg/mq/topic"
	wmysql "github.com/pingcap/wormhole/pkg/mysql"
	"github.com/pingcap/wormhole/pkg/util"
	"github.com/pingcap/wormhole/pkg/whpb"
	"github.com/siddontang/go-mysql/schema"
)

//the following fields identification some type for job
const (
	//JobCreateSchema is the job for create schema
	JobCreateSchema int = iota
	//JobCreateTable is the job for create table
	JobCreateTable
	//JobTableData is the job for table data
	JobTableData
)
const dbTimeout int = 2147483

// Range represents chunk range
type chunkRange struct {
	begin interface{}
	end   interface{}
	// for example:
	// containB and containE is true, means [begin, end]
	// containB is true, containE is false is false, means [begin, end)
	containB bool
	containE bool
	notNil   bool
}

// newChunkRange return a range struct
func newChunkRange(begin, end interface{}, containB, containE bool, notNil bool) chunkRange {
	return chunkRange{
		begin:    begin,
		end:      end,
		containE: containE,
		containB: containB,
		notNil:   notNil,
	}
}

func (d *Dumper) splitRange(chunk *chunkRange, count int64, dbname string, table string, column *schema.TableColumn) ([]chunkRange, error) {
	var chunks []chunkRange

	if count <= 1 {
		chunks = append(chunks, *chunk)
		return chunks, nil
	}

	if !chunk.notNil {
		return nil, errors.Errorf("the chunk is empty: %v", chunk)
	}

	//if chunk.isNumber {
	if reflect.TypeOf(chunk.begin).String() == "int64" {
		min, ok1 := chunk.begin.(int64)
		max, ok2 := chunk.end.(int64)
		if !ok1 || !ok2 {
			return nil, errors.Errorf("can't parse chunk's begin: %v, end: %v", chunk.begin, chunk.end)
		}
		step := (max-min)/count + 1
		cutoff := min
		for cutoff <= max {
			r := newChunkRange(cutoff, cutoff+step, true, false, true)
			chunks = append(chunks, r)
			cutoff += step
		}
		log.Debugf("[dumper] getChunksForTable cut table: cnt=%d min=%v max=%v step=%v chunk=%d",
			count, min, max, step, len(chunks))
	} else if reflect.TypeOf(chunk.begin).String() == "float64" {
		min, ok1 := chunk.begin.(float64)
		max, ok2 := chunk.end.(float64)
		if !ok1 || !ok2 {
			return nil, errors.Errorf("can't parse chunk's begin: %v, end: %v", chunk.begin, chunk.end)
		}
		step := (max-min)/float64(count) + 1
		cutoff := min
		for cutoff <= max {
			r := newChunkRange(cutoff, cutoff+step, true, false, true)
			chunks = append(chunks, r)
			cutoff += step
		}
		log.Debugf("[dumper] getChunksForTable cut table: cnt=%d min=%v max=%v step=%v chunk=%d",
			count, min, max, step, len(chunks))
	} else {
		max, ok1 := chunk.end.(string)
		min, ok2 := chunk.begin.(string)
		if !ok1 || !ok2 {
			return nil, errors.Errorf("can't parse chunk's begin: %v, end: %v", chunk.begin, chunk.end)
		}

		// get random value as split value
		splitValues, err := wmysql.GetRandomValues(d.db, dbname, table, column.Name, count-1, min, max)
		if err != nil {
			return nil, errors.Trace(err)
		}
		var min_tmp, max_tmp string
		var i int64
		for i = 0; i < count; i++ {
			if i == 0 {
				min_tmp = min
				max_tmp = splitValues[i]
			} else if i == int64(len(splitValues)) {
				min_tmp = splitValues[i-1]
				max_tmp = max
			} else {
				min_tmp = splitValues[i-1]
				max_tmp = splitValues[i]
			}
			r := newChunkRange(min_tmp, max_tmp, true, false, true)
			chunks = append(chunks, r)
		}
		log.Debugf("[dumper] getChunksForTable cut table: cnt=%d min=%s max=%s chunk=%d",
			count, min, max, len(chunks))
	}
	chunks[0].containB = chunk.containB
	chunks[len(chunks)-1].containE = chunk.containE
	return chunks, nil
}

//dumpJob is the struct for job of dump
type dumpJob struct {
	typ    int
	dbName string
	table  string
	column *schema.TableColumn
	where  string
	chunk  chunkRange
	err    chan error
}

func (d *Dumper) processQueue() {
	var err error
	defer d.wg.Done()

	err = wmysql.SetTimeout(d.db, dbTimeout, 0)
	if err != nil {
		log.Errorf("[dumper] processQueue set db timeout fail")
		return
	}
	err = wmysql.SetTransactionIsolation(d.db, wmysql.RepeatableRead)
	if err != nil {
		log.Errorf("[dumper] processQueue set transaction isolation fail")
		return
	}
	err = wmysql.CreateSnapshot(d.db)
	if err != nil {
		log.Errorf("[dumper] processQueue create snapshot fail")
		return
	}

	for {
		log.Debug("[dumper] processQueue wait for new dumpJob")
		select {
		case job := <-d.jobCh:
			log.Debugf("[dumper] processQueue recv a new dumpJob: job=%v", job)
			switch job.typ {
			case JobCreateSchema:
				err = d.doDumpCreateSchema(job)
				if err != nil {
					log.Errorf("[dumper] doDumpCreateSchema fail: %s", errors.ErrorStack(err))
				}
			case JobCreateTable:
				err = d.doDumpCreateTable(job, d.db)
				if err != nil {
					log.Errorf("[dumper] doDumpCreateTable fail: %s", errors.ErrorStack(err))
				}
			case JobTableData:
				err = d.doDumpTableData(job, d.db)
				if err != nil {
					log.Errorf("[dumper] doDumpTableData fail: %s", errors.ErrorStack(err))
				}
			}
			job.err <- err
			d.jobFin++
		case <-d.quit:
			log.Debug("[dumper] processQueue recv quit msg, quit")
			return
		}
	}
}

func (d *Dumper) dumpCreateSchema() error {
	schema := d.mysqlDumpJob.GetSchemas()[0]
	log.Debugf("[dumper] create dumpJob of dump create schema: %s", schema.GetName())
	job := &dumpJob{
		typ:    JobCreateSchema,
		dbName: schema.GetName(),
		err:    make(chan error),
	}
	d.jobCnt++
	d.jobCh <- job
	return <-job.err
}

// NOTE only create database ddl in dumper situation, no alter or drop
func (d *Dumper) doDumpCreateSchema(job *dumpJob) error {
	log.Debugf("[dumper] dump create database %s", job.dbName)
	sql := fmt.Sprintf("CREATE DATABASE IF NOT EXISTS `%s`", job.dbName)
	d.dbDrip = &whpb.Drip{
		Schema: job.dbName,
		Source: whpb.SourceType_mysql_dump,
		Op:     whpb.OPType_DDL,
		Data:   []byte(sql),
		Pos:    &whpb.Position{},
	}

	schema := d.mysqlDumpJob.GetSchemas()[0]
	if len(schema.GetTables()) == 0 {
		t, err := topic.MakeSpecialTopic(schema.GetName(), d.mysqlDumpJob.GetSource(), d.producer, d.zkClient, whpb.SourceType_mysql_dump)
		if err != nil {
			return errors.Trace(err)
		}
		err = d.producer.FanoutProduce([]*whpb.Topic{t}, d.dbDrip)
		if err != nil {
			return errors.Trace(err)
		}
	}

	return nil
}

func (d *Dumper) dumpCreateTable() error {
	log.Debug("[dumper] create dumpJob of dump create table")
	schema := d.mysqlDumpJob.GetSchemas()[0]

	var (
		wg      sync.WaitGroup
		lastErr error
	)
	wg.Add(len(schema.GetTables()))

	for _, table := range schema.GetTables() {
		job := &dumpJob{
			typ:    JobCreateTable,
			dbName: schema.GetName(),
			table:  table,
			err:    make(chan error),
		}

		go func(job *dumpJob) {
			defer wg.Done()
			d.jobCnt++
			d.jobCh <- job
			err := <-job.err
			if err != nil {
				log.Errorf("[dumper] dump create table error: %s", err)
				lastErr = err
			}
		}(job)
	}
	wg.Wait()

	return lastErr
}

func (d *Dumper) doDumpCreateTable(job *dumpJob, db *sql.DB) error {
	log.Debugf("[dumper] dump create table %s.%s", job.dbName, job.table)
	query := fmt.Sprintf("SHOW CREATE TABLE `%s`.`%s`", job.dbName, job.table)
	row := db.QueryRow(query)

	var tbl, createTable string
	err := row.Scan(&tbl, &createTable)
	if err != nil {
		return errors.Trace(err)
	}
	createTable = strings.Replace(createTable, "CREATE TABLE", "CREATE TABLE IF NOT EXISTS", 1)
	createTable = fmt.Sprintf("USE `%s`; %s", job.dbName, createTable)

	t, err := topic.MakeDumperTopic(job.dbName, job.table, d.mysqlDumpJob.GetSource(), d.producer, d.zkClient)
	if err != nil {
		return errors.Trace(err)
	}

	d.dbDrip.Table = job.table
	err = d.producer.FanoutProduce([]*whpb.Topic{t}, d.dbDrip)
	if err != nil {
		return errors.Trace(err)
	}

	fenceID, err := d.ec.CreateFence(context.Background(), []*whpb.Topic{t})
	if err != nil {
		return errors.Annotate(err, "[dumper] create fence for table ddl fail")
	}

	drip := &whpb.Drip{
		Schema:     job.dbName,
		Table:      job.table,
		Source:     whpb.SourceType_mysql_dump,
		Op:         whpb.OPType_DDL,
		Data:       []byte(createTable),
		FenceId:    fenceID,
		Pos:        &whpb.Position{},
		CreateTime: time.Now().Unix(),
	}

	err = d.producer.FanoutProduce([]*whpb.Topic{t}, drip)
	return errors.Trace(err)
}

func (d *Dumper) dumpTableData(dbname string, tables []string) error {
	jobBucket := make([][]*dumpJob, len(tables))
	errCh := make(chan error)
	var jobCnt int
	for i, table := range tables {
		column, err := d.findSuitableField(dbname, table)
		if err != nil {
			return errors.Trace(err)
		}

		chunks, err := d.getChunksForTable(dbname, table, column)
		if err != nil {
			return errors.Trace(err)
		}

		jobCnt += len(chunks)
		var (
			chunk chunkRange
			where string
		)
		for {
			length := len(chunks)
			if length == 0 {
				break
			}
			if length%2 == 0 {
				chunk = chunks[0]
				chunks = chunks[1:]
			} else {
				chunk = chunks[length-1]
				chunks = chunks[:length-1]
			}
			if chunk.notNil {
				gt := ">"
				lt := "<"
				if chunk.containB {
					gt = ">="
				}
				if chunk.containE {
					lt = "<="
				}
				if reflect.TypeOf(chunk.begin).String() == "int64" {
					where = fmt.Sprintf("(`%s` %s %d AND `%s` %s %d)", column.Name, gt, chunk.begin, column.Name, lt, chunk.end)
				} else if reflect.TypeOf(chunk.begin).String() == "float64" {
					where = fmt.Sprintf("(`%s` %s %f AND `%s` %s %f)", column.Name, gt, chunk.begin, column.Name, lt, chunk.end)
				} else {
					where = fmt.Sprintf("(`%s` %s \"%v\" AND `%s` %s \"%v\")", column.Name, gt, chunk.begin, column.Name, lt, chunk.end)
				}
			} else {
				where = ""
			}

			jobBucket[i] = append(jobBucket[i], &dumpJob{
				typ:    JobTableData,
				dbName: dbname,
				table:  table,
				column: column,
				where:  where,
				chunk:  chunk,
				err:    errCh,
			})
		}
	}
	d.jobCnt += uint64(jobCnt)

	go func() {
		bucketCnt := len(jobBucket)
		for i, idx := 0, 0; i < int(jobCnt); idx++ {
			idx = idx % bucketCnt
			if len(jobBucket[idx]) < 1 {
				continue
			}
			d.jobCh <- jobBucket[idx][0]
			jobBucket[idx] = jobBucket[idx][1:]
			i++
		}
	}()

	var lastErr error
	for i := 0; i < int(jobCnt); i++ {
		err := <-errCh
		if err != nil {
			log.Errorf("[dumper] get error: %s", err)
			lastErr = err
		}
	}

	return lastErr
}

func (d *Dumper) doDumpTableData(job *dumpJob, db *sql.DB) error {
	log.Debugf("[dumper] do job of dump table data table=%s.%s, where=%s", job.dbName, job.table, job.where)
	// construct dml data
	t, err := topic.MakeDumperTopic(job.dbName, job.table, d.mysqlDumpJob.GetSource(), d.producer, d.zkClient)
	if err != nil {
		return errors.Trace(err)
	}

	table, err := wmysql.GetSchemaTable(d.db, job.dbName, job.table)
	if err != nil {
		return errors.Trace(err)
	}

	tableInfo := wmysql.ToTableInfo(table)
	dmlData := &whpb.DMLData{
		Schema:  job.dbName,
		Table:   job.table,
		Tp:      whpb.EventType_Insert,
		Columns: wmysql.ToColumns(tableInfo.Columns),
	}

	drip, err := d.genDumpTableData(job, dmlData, t, tableInfo)
	if err != nil {
		return errors.Trace(err)
	}
	if drip == nil {
		return nil
	}
	
	drips := []*whpb.Drip{drip}
	dmlDatas := []*whpb.DMLData{dmlData}
	var lastErr error

	for i := 0; i < len(drips); i++ {
		drip := drips[i]
		_, _, err = d.producer.Produce(t, drip)
		// split job when meet message too large error
		if util.OriginError(err) == sarama.ErrMessageTooLarge || util.OriginError(err) == sarama.ErrMessageSizeTooLarge {
			log.Debugf("[dumper] meet err of ErrMessageTooLarge or ErrMessageSizeTooLarge, split drip data")
			splitDrips, splitDmlDatas, err2 := splitDrip(drips[i], dmlDatas[i], 2)
			if err2 != nil {
				log.Warnf("[dumper] fail to split drip: job=%v", drips[i])
				return errors.Trace(err2)
			}
			if len(splitDrips) < 2 || len(splitDmlDatas) < 2 {
				return errors.Errorf("split drip data failed!")
			}
			drips = append(drips, splitDrips[0])
			dmlDatas = append(dmlDatas, splitDmlDatas[0])
			drips = append(drips, splitDrips[1])
			dmlDatas = append(dmlDatas, splitDmlDatas[1])

		} else if err != nil {
			log.Errorf("[dumper] dump table data error: %s", err)
			lastErr = err
		}
	}

	return errors.Trace(lastErr)
}

func (d *Dumper) genDumpTableData(job *dumpJob, dmlData *whpb.DMLData, topic *whpb.Topic, tableInfo *model.TableInfo) (*whpb.Drip, error) {
	var query string
	if job.where != "" {
		query = fmt.Sprintf("SELECT %s * FROM `%s`.`%s` WHERE %s",
			"/*!40001 SQL_NO_CACHE */", job.dbName, job.table, job.where)
	} else {
		query = fmt.Sprintf("SELECT %s * FROM `%s`.`%s`",
			"/*!40001 SQL_NO_CACHE */", job.dbName, job.table)
	}
	log.Debugf("[dumper] dump table data query: sql=%s", query)

	rows, err := d.db.Query(query)
	if err != nil {
		return nil, errors.Trace(err)
	}

	defer rows.Close()
	rowsData, err := wmysql.ScanRowsToInterfaces(rows)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if len(rowsData) == 0 {
		return nil, nil
	}
	log.Debugf("[dumper] dump table data query: rows=%d", len(rowsData))

	dmlData.Rows = wmysql.ToRowData(rowsData, tableInfo.Columns)
	data, err := dmlData.Marshal()
	if err != nil {
		return nil, errors.Trace(err)
	}

	drip := &whpb.Drip{
		Schema: job.dbName,
		Table:  job.table,
		Source: whpb.SourceType_mysql_dump,
		Op:     whpb.OPType_DML,
		Data:   data,
		Pos:    &whpb.Position{},
	}
	return drip, nil
}

func splitDrip(drip *whpb.Drip, dmlData *whpb.DMLData, count int) ([]*whpb.Drip, []*whpb.DMLData, error) {
	if len(dmlData.Rows) < count {
		return nil, nil, errors.Errorf("dmlData: %v can't be split", dmlData)
	}

	drips := make([]*whpb.Drip, 0, 2)
	dmlDatas := make([]*whpb.DMLData, 0, 2)
	var min, max int
	for i := 0; i < count; i++ {
		min = max
		if i == count-1 {
			max = len(dmlData.Rows)
		} else {
			max = len(dmlData.Rows) / count * (i + 1)
		}
		tmpDmlData := &whpb.DMLData{
			Schema:  dmlData.Schema,
			Table:   dmlData.Table,
			Tp:      dmlData.Tp,
			Columns: dmlData.Columns,
		}
		tmpDmlData.Rows = dmlData.Rows[min:max]
		data, err := tmpDmlData.Marshal()
		if err != nil {
			return nil, nil, errors.Trace(err)
		}

		tmpDrip := &whpb.Drip{
			Schema: drip.Schema,
			Table:  drip.Table,
			Source: drip.Source,
			Op:     drip.Op,
			Data:   data,
			Pos:    drip.Pos,
		}
		drips = append(drips, tmpDrip)
		dmlDatas = append(dmlDatas, tmpDmlData)
	}

	return drips, dmlDatas, nil
}

func (d *Dumper) getChunksForTable(dbname string, table string, column *schema.TableColumn) ([]chunkRange, error) {
	noChunks := []chunkRange{{}}
	if column == nil {
		log.Warnf("[dumper] No suitable index found for %s.%s", dbname, table)
		return noChunks, nil
	}

	field := column.Name

	// fetch min, max
	query := fmt.Sprintf("SELECT %s MIN(`%s`) as MIN, MAX(`%s`) as MAX FROM `%s`.`%s`",
		"/*!40001 SQL_NO_CACHE */", field, field, dbname, table)
	log.Debugf("[dumper] getChunksForTable query sql: %s", query)

	// get the chunk count
	cnt, err := wmysql.EstimateCount(d.db, dbname, table, field)
	if err != nil {
		return nil, errors.Trace(err)
	}

	chunkCnt := cnt / int64(d.cfg.ChunkSize)

	//var chunks []chunkRange
	var chunk chunkRange
	if column.Type == schema.TYPE_NUMBER {
		var min, max sql.NullInt64
		err := d.db.QueryRow(query).Scan(&min, &max)
		if err != nil {
			return nil, errors.Trace(err)
		}
		if !min.Valid {
			// min is NULL, means that no table data.
			return []chunkRange{}, nil
		}
		chunk = newChunkRange(min.Int64, max.Int64+1, true, false, true)
	} else if column.Type == schema.TYPE_FLOAT {
		var min, max sql.NullFloat64
		err := d.db.QueryRow(query).Scan(&min, &max)
		if err != nil {
			return nil, errors.Trace(err)
		}
		if !min.Valid {
			// min is NULL, means that no table data.
			return []chunkRange{}, nil
		}
		chunk = newChunkRange(min.Float64-0.1, max.Float64+0.1, false, false, true)
	} else {
		var min, max string
		err := d.db.QueryRow(query).Scan(&min, &max)
		if err != nil {
			return nil, errors.Trace(err)
		}
		if min == "" || max == "" {
			// min is NULL, means that no table data.
			return []chunkRange{}, nil
		}
		chunk = newChunkRange(min, max, true, true, true)
	}
	return d.splitRange(&chunk, chunkCnt, dbname, table, column)
}

func (d *Dumper) sendJobEnd() error {
	log.Info("[dumper] send dump job end drip")
	schema := d.mysqlDumpJob.GetSchemas()[0]

	drip := &whpb.Drip{
		Schema: schema.GetName(),
		Source: whpb.SourceType_mysql_dump,
		Op:     whpb.OPType_END,
		Data:   []byte{},
		Pos:    &whpb.Position{},
	}

	topics := make([]*whpb.Topic, 0, 8)
	if len(schema.GetTables()) == 0 {
		t, err := topic.MakeSpecialTopic(schema.GetName(), d.mysqlDumpJob.GetSource(), d.producer, d.zkClient, whpb.SourceType_mysql_dump)
		if err != nil {
			return errors.Trace(err)
		}
		topics = append(topics, t)
	}

	for _, table := range schema.GetTables() {
		t, err := topic.MakeDumperTopic(schema.GetName(), table, d.mysqlDumpJob.GetSource(), d.producer, d.zkClient)
		if err != nil {
			return errors.Trace(err)
		}
		topics = append(topics, t)
	}

	err := d.producer.FanoutProduce(topics, drip)
	return errors.Trace(err)
}

func (d *Dumper) findSuitableField(dbname string, table string) (*schema.TableColumn, error) {
	// first select the index with number type
	column, err := wmysql.FindSuitableIndex(d.db, dbname, table, true)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if column != nil {
		return column, nil
	}
	log.Infof("%s.%s don't have index, will use a number column as split field", dbname, table)

	// select a number column
	column, err = wmysql.FindNumberColumn(d.db, dbname, table)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if column != nil {
		return column, nil
	}
	log.Infof("%s.%s don't have number columns, will use the first column as split field", dbname, table)

	// use the first column
	column, err = wmysql.GetFirstColumn(d.db, dbname, table)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if column != nil {
		return column, nil
	}

	return nil, errors.Errorf("no column find in table %s.%s", dbname, table)
}