package stargate

import (
	"fmt"
	"regexp"
	"strconv"
	"strings"
	"time"

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

const (
	identifier = "[a-zA-Z0-9$_]+"
)

func (m *MysqlStargate) initConsumer() error {
	cfg := consumer.NewConfig()
	cfg.Consumer.Return.Errors = true
	// if no offset was previously committed, comsumes from oldest offset.
	cfg.Consumer.Offsets.Initial = sarama.OffsetOldest
	cfg.Metadata.RefreshFrequency = time.Second * 5

	if m.job.GetSource() == whpb.SourceType_mysql_binlog {
		// potential topics listening.
		cfg.Group.Topics.Whitelist = m.genTopicWhiteList()
		log.Debugf("[stargate] topics whitelist pattern %s", cfg.Group.Topics.Whitelist.String())
	}

	jobIDStr := strconv.FormatUint(m.job.GetId(), 10)

	// Init consumer
	var err error
	m.consumer, err = consumer.NewConsumer(m.kafkaAddr, jobIDStr, m.topics, cfg, m.etcdClient, m.ec, m.zkClient)
	return errors.Trace(err)
}

// TODO: more filter.
func (m *MysqlStargate) genTopicWhiteList() *regexp.Regexp {
	whiteList := make([]string, 0, 0)
	host := m.mysqlSyncJob.Source.GetHost()
	port := m.mysqlSyncJob.Source.GetPort()
	// now, we only support known database and known tables consuming.
	// that is, databases that created during wormhole running would not be consumed.
	for _, schema := range m.mysqlSyncJob.Source.GetSchemas() {
		whiteList = append(whiteList, fmt.Sprintf("%s_%d_%s_%s_mysql_binlog", host, port, schema.Name, identifier))
	}

	return regexp.MustCompile("(?i)" + strings.Join(whiteList, "|"))
}