package stargate

import (
	"strings"
	"database/sql"
	"fmt"
	"runtime/debug"
	"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/siddontang/go/sync2"
	"golang.org/x/net/context"

	"github.com/pingcap/wormhole/config"
	"github.com/pingcap/wormhole/pkg/etcdutil"
	"github.com/pingcap/wormhole/pkg/ether-client"
	"github.com/pingcap/wormhole/pkg/mq/consumer"
	"github.com/pingcap/wormhole/pkg/mq/topic"
	"github.com/pingcap/wormhole/pkg/mq/zk"
	"github.com/pingcap/wormhole/pkg/mysql"
	"github.com/pingcap/wormhole/pkg/whpb"
)

const (
	defaultChannelSize = 1024
	dmlType = "dml"
	ddlType = "ddl"
	endType = "end"
)

// MysqlStargate is mysql stargate
type MysqlStargate struct {
	sync.RWMutex
	errCh chan error

	// target database
	db           whpb.DBAccount
	topics       []string
	kafkaAddr    []string
	job          *whpb.Job
	mysqlSyncJob *whpb.MysqlSyncJob
	etcdClient   *etcdutil.Client
	ec           ether.Client
	zkClient     *zk.Client
	consumer     *consumer.Consumer

	lastTime time.Time

	translators map[string]chan *sarama.ConsumerMessage
	executors   map[string]chan *sqlJob
	executorDBs map[string]*sql.DB
	status      *status

	lastCount sync2.AtomicInt64
	count     sync2.AtomicInt64
	closed    sync2.AtomicInt64
	finished  sync2.AtomicBool

	wg       sync.WaitGroup
	activeWg sync.WaitGroup

	cfg    *config.Config
	ctx    context.Context
	cancel context.CancelFunc

	produced int64
	consumed int64

	consumeMap map[string]map[int32]bool
	//Rules         []*whpb.RouteRule  //TODO
}

// New returns MysqlStargate
func New(job *whpb.Job, etherClient ether.Client, zkClient *zk.Client, cfg *config.Config) (*MysqlStargate, error) {
	mysqlSyncJob := &whpb.MysqlSyncJob{}
	err := mysqlSyncJob.Unmarshal(job.GetInfo())
	if err != nil {
		return nil, errors.Trace(err)
	}

	log.Infof("[stargate] stargate recv job %+v", mysqlSyncJob)

	//TODO: init it from handler. share with syncer.
	etcd, err := clientv3.New(clientv3.Config{
		Endpoints: cfg.EtherEndpoints,
	})
	if err != nil {
		return nil, errors.Trace(err)
	}
	rootpath := etherClient.GetStorageRootPath()
	etcdClient := etcdutil.NewClient(etcd, rootpath)

	ctx, cancel := context.WithCancel(context.Background())
	ms := &MysqlStargate{
		db:           mysqlSyncJob.GetTarget(),
		kafkaAddr:    cfg.KafkaBrokers,
		job:          job,
		mysqlSyncJob: mysqlSyncJob,
		translators:  make(map[string]chan *sarama.ConsumerMessage),
		executors:    make(map[string]chan *sqlJob),
		executorDBs:  make(map[string]*sql.DB),
		status:       newStatus(),
		ctx:          ctx,
		cancel:       cancel,
		errCh:        make(chan error),
		ec:           etherClient,
		etcdClient:   etcdClient,
		zkClient:     zkClient,
		cfg:          cfg,
		consumeMap:     make(map[string]map[int32]bool),
	}

	topics := topic.GetTopicNames(mysqlSyncJob.GetSource(), job.GetSource())
	for _, schema := range mysqlSyncJob.Source.GetSchemas() {
		t := topic.GetSpecialTopicName(schema.GetName(), mysqlSyncJob.Source.GetHost(), mysqlSyncJob.Source.GetPort(), job.GetSource())
		topics = append(topics, t)
	}

	ms.lastTime = time.Now()
	ms.topics = topics
	log.Infof("[stargate] topics %+v", topics)

	// initialize consumer
	if err = ms.initConsumer(); err != nil {
		return nil, errors.Trace(err)
	}
	return ms, nil
}

// GetStatus implements the Processor interface.
func (m *MysqlStargate) GetStatus() *whpb.JobStatus {
	pos := m.status.statistics()
	var percent, total uint64
	// only dumpjob can have percent and total.
	if m.job.GetSource() == whpb.SourceType_mysql_dump {
		percent, total = m.calcPercent()
	}

	m.RLock()
	j := proto.Clone(m.job).(*whpb.Job)
	j.Status.Pos = pos
	j.Status.Percent = percent
	j.Status.Total = total
	m.RUnlock()

	return j.Status
}

// In ideal conditions, the stargate-dump progress should always equal to portal-dump progress.
// but in reality, there is delay in kafka consumer, so the message consumed always less than or equal
// to produced, thats consumed <= produced.
// So percent_portal_dump * (consumed/produced) * 100% = percent_stargate_dump.
// But for convenient, we only calculate consumed/produced, and the front-end does the rest of it.
func (m *MysqlStargate) calcPercent() (percent uint64, total uint64) {
	var (
		produced int64
		consumed int64
	)
	// produced
	for _, topic := range m.topics {
		partitions, err := m.etcdClient.LoadProducerPartitions(topic)
		if err != nil {
			m.reportErr(errors.Trace(err))
			return 0, 0
		}
		for _, p := range partitions {
			produced += p.Pos.GetKafkaOffset()
		}
	}

	// consumed
	snap := m.consumer.Snapshot()
	for _, partitions := range snap {
		for _, offset := range partitions {
			consumed += offset
		}
	}

	if m.produced < produced {
		m.produced = produced
	}
	// denominator can't be zero.
	if m.produced == 0 {
		return 0, 0
	}

	// consumed is 0 means can't get snapshot from etcd, the stargate job is finished
	if consumed == 0 {
		m.consumed = m.produced
	} else if m.consumed < consumed {
		m.consumed = consumed
	}

	percent = uint64(m.consumed * 100 / m.produced)
	log.Debugf("consumed %d, produced %d, percent %d, snap %+v", m.consumed, m.produced, percent, snap)
	if percent > 100 {
		log.Warnf("percent %d greater than 100", percent)
		percent = 100
	}

	return percent, uint64(m.produced)
}

func (m *MysqlStargate) getQPS() int64 {
	now := time.Now()
	seconds := now.Unix() - m.lastTime.Unix()
	last := m.lastCount.Get()
	total := m.count.Get()

	tps := int64(0)
	if seconds > 0 {
		tps = (total - last) / seconds
	}
	log.Debugf("[stargate] total events = %d, last = %d, seconds= %d, recent tps = %d", total, last, seconds, tps)

	m.lastTime = time.Now()
	m.lastCount.Set(total)
	return tps
}

// Process implements the Processor interface.
func (m *MysqlStargate) Process() (err error) {
	start := time.Now()
	saveStatusF := func() {
		qps := m.getQPS()
		count := m.count.Get()

		m.job.Status.Qps = uint64(qps)
		m.lastCount.Set(count)
	}

	defer func() {
		cost := time.Since(start)
		log.Infof("[stargate] Process cost: %s", cost)

		saveStatusF()

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

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

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

	go m.catchErr(m.ctx)
	go m.waitFinished()

	for _, topic := range m.topics {
		// topic like 192.168.1.29_3307_test_NTEST_mysql_dump
		item := strings.Split(topic, "_")
		if len(item) == 5 {
			// is a special topic for database
			continue
		}
		m.consumeMap[topic] = make(map[int32]bool)
	}

	for {
		select {
		case <-m.ctx.Done():
			log.Infof("[stargate] job %d's process exit!", m.job.GetId())
			return m.ctx.Err()
		case msg, ok := <-m.consumer.Messages():
			if !ok {
				return errors.New("[stargate] consumer group closed")
			}

			if msg.Sync {
				m.activeWg.Wait()
			}
			err := m.process(msg.Message)
			if msg.Sync {
				m.activeWg.Wait()
				msg.Done <- struct{}{}
			}
			if err != nil {
				return errors.Trace(err)
			}

		case err := <-m.consumer.Errors():
			if errors.Cause(err) == sarama.ErrNoError {
				continue
			}
			m.reportErr(errors.Trace(err))
		case <-timer.C:
			saveStatusF()

			if !m.finished.Get() {
				continue
			}

			log.Infof("[stargate] process end")
			return nil
		}
	}
}

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

func (m *MysqlStargate) process(msg *sarama.ConsumerMessage) error {
	msgCh, err := m.handleMessage(msg.Topic, msg.Partition)
	if err != nil {
		return errors.Trace(err)
	}

	m.activeWg.Add(1)
	msgCh <- msg

	return nil
}

func (m *MysqlStargate) handleMessage(topic string, partition int32) (chan *sarama.ConsumerMessage, error) {
	err := m.registerConsumer(topic, partition)
	if err != nil {
		return nil, errors.Trace(err)
	}

	tp := &topicPartition{
		topic:       topic,
		partitionID: partition,
	}
	key := tp.toString()

	m.Lock()
	defer m.Unlock()

	msgCh, ok := m.translators[key]
	if !ok {
		msgCh = make(chan *sarama.ConsumerMessage, defaultChannelSize)
		jobCh := make(chan *sqlJob, defaultChannelSize)
		db, err := mysql.CreateDB(m.db)
		if err != nil {
			return nil, errors.Trace(err)
		}
		m.executors[key] = jobCh
		m.translators[key] = msgCh
		m.executorDBs[key] = db
		go m.translate(m.ctx, tp, msgCh, jobCh)
		go m.execute(m.ctx, tp, jobCh)
		log.Debugf("run translate and execute task for topic: %s, partition: %d", topic, partition)
	}

	return msgCh, nil
}

// Close implements the Processor interface.
func (m *MysqlStargate) Close() {
	log.Info("[stargate] going to close")
	m.Lock()
	defer m.Unlock()

	if !m.closed.CompareAndSwap(0, 1) {
		log.Info("[stargate] stargate has already been closed.")
		return
	}

	m.cancel()
	m.wg.Wait()

	for _, db := range m.executorDBs {
		db.Close()
	}

	if m.consumer != nil {
		m.consumer.Close()
	}
	log.Infof("[stargate] closed. job %+v", m.job)
}

func (m *MysqlStargate) waitFinished() {
	// only after running can we begin waiting.
	ticker := time.NewTicker(time.Millisecond * 50)
	defer ticker.Stop()

waitRunning:
	for {
		select {
		case <-ticker.C:
			if m.checkConsumer() {
				break waitRunning
			}
		}
	}

	log.Debugf("[stargate] processor running and wait for it finished")
	m.wg.Wait()
	m.finished.Set(true)
	log.Info("[stargate] wait finished thread exits.")
}

func (m *MysqlStargate) catchErr(ctx context.Context) {
	cctx, _ := context.WithCancel(ctx)
	select {
	case <-cctx.Done():
		log.Info("[stargate] catch error thread exit")
		return
	case err := <-m.errCh:
		log.Errorf("[stargate] catch error %v", errors.ErrorStack(err))
		m.Lock()
		m.job.Status.State = whpb.JobState_Abnormal
		m.job.Status.Message = err.Error()
		m.Unlock()
	}
}

func (m *MysqlStargate) reportErr(err error) {
	m.errCh <- err
}

func (m *MysqlStargate) checkConsumer() bool {
	m.RLock()
	defer m.RUnlock()

	log.Debugf("m.consumeMap: %v", m.consumeMap)
	for _, partitionMap := range m.consumeMap {
		if len(partitionMap) == 0 {
			return false
		}
		for _, consumed := range partitionMap {
			if consumed == false {
				return false
			}
		}
	}
	log.Debug("checkConsumer: all topic's partition is being consumed!")
	return true
}

func (m *MysqlStargate) registerConsumer(topic string, partition int32) error {
	m.Lock()
	defer m.Unlock()

	partitionMap, ok := m.consumeMap[topic]
	if !ok {
		m.consumeMap[topic] = make(map[int32]bool)
	}

	if len(partitionMap) == 0 {
		partitions, err := m.zkClient.Partitions(topic)
		if err != nil {
			log.Errorf("get partition of topic: %s failed!", topic)
			return errors.Trace(err)
		}
		for _, p := range partitions {
			partitionMap[p] = false
		}
	}
	partitionMap[partition] = true
	
	return nil
}

type topicPartition struct {
	topic       string
	partitionID int32
}

func (t *topicPartition) toString() string {
	return fmt.Sprintf("topic_%s_partition_%d", t.topic, t.partitionID)
}