package stargate

import (
	"context"

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

func (m *MysqlStargate) translate(ctx context.Context, tp *topicPartition, msgCh chan *sarama.ConsumerMessage, jobCh chan *sqlJob) {
	m.wg.Add(1)
	defer m.wg.Done()
	key := tp.toString()

	for {
		select {
		case <-ctx.Done():
			log.Infof("[stargate] translator %s exit, error %v", key, ctx.Err())
			return
		case msg, ok := <-msgCh:
			if !ok {
				job := m.newJob("", nil, nil, nil, endType)
 				jobCh <- job
				log.Infof("[stargate] translator %s channel closed. exits", key)
				return
			}
			drip := &whpb.Drip{}
			err := drip.Unmarshal(msg.Value)
			if err != nil {
				m.reportErr(errors.Trace(err))
				return
			}

			var (
				sqls []string
				args [][]interface{}
				pos  = drip.Pos
				sqlType string
			)
			if drip.Op == whpb.OPType_DDL {
				sqlType = ddlType
				sqls = []string{string(drip.GetData())}
				log.Debugf("[translator] get ddl drip, data=%+v", sqls)
				args = append(args, nil)
			} else if drip.Op == whpb.OPType_DML {
				sqlType = dmlType
				dmlData := &whpb.DMLData{}
				err = dmlData.Unmarshal(drip.GetData())
				if err != nil {
					m.reportErr(errors.Trace(err))
					return
				}
				log.Debugf("[translator] get dml drip, data=%+v", dmlData)

				switch dmlData.Tp {
				case whpb.EventType_Insert:
					sqls, args, err = genInsertSQLs(dmlData)
				case whpb.EventType_Update:
					sqls, args, err = genUpdateSQLs(dmlData)
				case whpb.EventType_Delete:
					sqls, args, err = genDeleteSQLs(dmlData)
				}
			} else if drip.Op == whpb.OPType_END {
				log.Infof("[stargate] meet end drip %+v. close the channel for key %s", drip, key)
				m.RLock()
				msgCh := m.translators[key]
				m.RUnlock()
				close(msgCh)
			} else {
				log.Debugf("[stargate] ignore drip %+v", drip)
			}
			m.activeWg.Add(len(sqls))
			m.activeWg.Done()

			if err != nil {
				err1 := errors.Errorf("[stargate] gen dml sqls failed: %v, schema: %s, table: %s", err, drip.Schema, drip.Table)
				m.reportErr(errors.Trace(err1))
				return
			}

			for i := range sqls {
				job := m.newJob(sqls[i], args[i], pos, msg, sqlType)
				jobCh <- job
			}

			m.count.Add(int64(len(sqls)))
		}
	}
}