package topic

import (
	"fmt"
	"time"

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

const (
	fetchPartitionBackOff = time.Second * 1
)

// MakeDumperTopic creates a dumper topic.
func MakeDumperTopic(schema string, table string, db whpb.DBAccount, p producer.Producer, zkClient *zk.Client) (t *whpb.Topic, err error) {
	return MakeTopic(schema, table, whpb.SourceType_mysql_dump, db, p, zkClient)
}

// MakeSyncerTopic creates a syncer topic
func MakeSyncerTopic(schema string, table string, db whpb.DBAccount, p producer.Producer, zkClient *zk.Client) (t *whpb.Topic, err error) {
	return MakeTopic(schema, table, whpb.SourceType_mysql_binlog, db, p, zkClient)
}

// MakeSpecialTopic creates a special topic when the database has no tables at all. It's specially for drop database in
// the following scenarios:
// 1. the database was just created, and no tables created yet.
// 2. tables may be created before, but all have been dropped.
// Also we can use it in dumper when there are no tables in a database.
func MakeSpecialTopic(schema string, db whpb.DBAccount, p producer.Producer, zkClient *zk.Client, st whpb.SourceType) (t *whpb.Topic, err error) {
	return MakeTopic(schema, "", st, db, p, zkClient)
}

// GetSpecialTopicName returns special topic name.
func GetSpecialTopicName(schema string, host string, port int32, st whpb.SourceType) string {
	return fmt.Sprintf("%s_%d_%s_%s", host, port, schema, st.String())
}

// MakeTopic creates a topic.
func MakeTopic(schema string, table string, st whpb.SourceType, db whpb.DBAccount, p producer.Producer, zkClient *zk.Client) (t *whpb.Topic, err error) {
	t = &whpb.Topic{
		Host:   db.Host,
		Port:   db.Port,
		Schema: schema,
		Table:  table, // TODO: replaced by topic manager.
	}
	err = SetTopicName(t, st)
	if err != nil {
		return nil, errors.Trace(err)
	}

	num, err := getPartitionNum(p, t.GetName(), zkClient)
	if err != nil {
		return nil, errors.Trace(err)
	}

	t.PartitionNum = num
	return t, nil
}

// SetTopicName sets the name of topic.
func SetTopicName(topic *whpb.Topic, source whpb.SourceType) error {
	if topic.GetName() != "" {
		return nil
	}
	if source != whpb.SourceType_mysql_binlog && source != whpb.SourceType_mysql_dump {
		return errors.Errorf("unknown source type %v", source)
	}
	if topic.GetTable() == "" {
		topic.Name = fmt.Sprintf("%s_%d_%s_%s", topic.GetHost(), topic.GetPort(), topic.GetSchema(), source.String())
	} else {
		topic.Name = fmt.Sprintf("%s_%d_%s_%s_%s",
			topic.GetHost(), topic.GetPort(), topic.GetSchema(), topic.GetTable(), source.String())
	}
	return nil
}

// GetTopicNames returns topic list constructed from schemainfo.
func GetTopicNames(info whpb.SchemaInfo, st whpb.SourceType) []string {
	topics := make([]string, 0, 0)
	for _, schema := range info.GetSchemas() {
		for _, table := range schema.GetTables() {
			topics = append(topics,
				fmt.Sprintf("%s_%d_%s_%s_%s", info.GetHost(), info.GetPort(), schema.GetName(), table, st.String()))
		}
	}
	return topics
}

func getPartitionNum(p producer.Producer, topicName string, zkClient *zk.Client) (int32, error) {
	retry := 10
	for i := 0; i < retry; i++ {
		partitions, err := p.(*producer.KafkaProducer).Client.Partitions(topicName)
		if err != nil {
			return 0, errors.Trace(err)
		}
		if len(partitions) == 0 {
			time.Sleep(fetchPartitionBackOff)
			continue
		}
		return int32(len(partitions)), nil
	}

	log.Infof("Topic %s reached max retry using sarama client and partitions still empty. Now trying to fetch partitions by zookeeper client.", topicName)
	partitions, err := zkClient.Partitions(topicName)
	if err != nil {
		return 0, errors.Trace(err)
	}

	return int32(len(partitions)), nil
}