package stargate

import (
	"context"
	"database/sql"
	"time"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/wormhole/pkg/whpb"
)

var (
	maxWaitTime   = 3 * time.Second
	waitTime      = 50 * time.Millisecond
	maxRetryCount = 30
	retryTimeout  = time.Second
)

func (m *MysqlStargate) execute(ctx context.Context, tp *topicPartition, jobCh chan *sqlJob) {
	m.wg.Add(1)
	defer m.wg.Done()

	var (
		pos *whpb.Position
		err error
	)
	key := tp.toString()

	index := 0
	m.RLock()
	db := m.executorDBs[key]
	m.RUnlock()
	batch := int(m.cfg.Batch)
	sqls := make([]string, 0, batch)
	args := make([][]interface{}, 0, batch)
	lastSyncTime := time.Now()
	maxOffset := int64(0)

	executeF := func() {
		err = executeSQL(db, sqls, args)
		m.activeWg.Add(-1 * len(sqls))
		if err != nil {
			// FIXME: not just report error. we should handle it.
			m.reportErr(errors.Trace(err))
			return
		}

		m.markOffsets(tp.topic, tp.partitionID, maxOffset)
		index = 0
		sqls = sqls[0:0]
		args = args[0:0]
		lastSyncTime = time.Now()

		log.Debugf("[stargate] report pos %+v", pos)
		m.status.report(key, pos)
	}

	for {
		select {
		case <-ctx.Done():
			log.Infof("[stargate] executor %s exit, error %v", key, ctx.Err())
			return
		case job, ok := <-jobCh:
			if !ok {
				if len(sqls) > 0 {
					executeF()
				}
				log.Infof("[stargate] executor %s channel closed. exits.", key)
				return
			}

			if job.tp == ddlType {
				// execute dml sqls
				executeF()
			}

			if job.tp == endType {
				log.Infof("[stargate] meet end job type")
				close(jobCh)
				continue
			}

			index++
			pos = job.pos
			sqls = append(sqls, job.sql)
			args = append(args, job.args)
			if job.msg.Offset > maxOffset {
				maxOffset = job.msg.Offset
			}

			if job.tp == ddlType {
				// execute the ddl sql
				executeF()
			}

			if index >= batch {
				executeF()
			}
		default:
			now := time.Now()
			if len(sqls) > 0 && now.Sub(lastSyncTime) >= maxWaitTime {
				log.Debugf("[stargate] now %s, last_sync_time %s", now, lastSyncTime)
				executeF()
			}
			time.Sleep(waitTime)
		}
	}
}

func (m *MysqlStargate) markOffsets(topic string, partition int32, offset int64) {
	if offset <= 0 {
		return
	}

	m.consumer.MarkPartitionOffset(topic, partition, offset, "")
}

func executeSQL(db *sql.DB, sqls []string, args [][]interface{}) error {
	start := time.Now()
	defer func() {
		cost := time.Since(start)
		if cost > time.Second {
			log.Infof("[stargate] execute %d sqls, cost: %s", len(sqls), cost)
		} else {
			log.Debugf("[stargate] execute %d sqls, cost: %s", len(sqls), cost)
		}
	}()

	if len(sqls) == 0 {
		return nil
	}
	var err error
	retryCount := maxRetryCount

	for i := 0; i < retryCount; i++ {
		if i > 0 {
			// only add only one sql in log, because may has too many sqls
			log.Warnf("[stargate] sql stmt_exec retry %d: sql length: %d, first sql: %s, %s", i, len(sqls), sqls[0], args[0])
			time.Sleep(retryTimeout)
		}

		if err = executeSQLImp(db, sqls, args); err != nil {
			// only add only one sql in log, because may has too many sqls
			log.Errorf("[stargate][exec][sql]%s[args]%v[error]%v", sqls[0], args[0], err)
			continue
		}

		return nil
	}
	return errors.Errorf("[stargate] exec sqls failed, the first sql: %s, err:%s", sqls[0], err.Error())
}

func executeSQLImp(db *sql.DB, sqls []string, args [][]interface{}) error {
	txn, err := db.Begin()
	if err != nil {
		log.Errorf("[stargate] exec sqls[%v] begin failed %v", sqls, errors.ErrorStack(err))
		return errors.Trace(err)
	}

	for i := range sqls {
		log.Debugf("[stargate][exec][sql]%s[args]%v", sqls[i], args[i])

		_, err = txn.Exec(sqls[i], args[i]...)
		if err != nil {
			log.Warnf("[stargate][exec][sql]%s[args]%v[error]%v", sqls[i], args[i], err)
			rerr := txn.Rollback()
			if rerr != nil {
				log.Errorf("[stargate][exec][sql]%s[args]%v[error]%v", sqls[i], args[i], rerr)
			}
			return errors.Trace(err)
		}
	}
	err = txn.Commit()
	if err != nil {
		log.Errorf("[stargate]exec sqls[%v] commit failed %v", sqls, errors.ErrorStack(err))
		return errors.Trace(err)
	}
	return nil
}