package kafka

import (
	"strconv"
	"strings"
	"time"
)

const (
	DefaultTopicName = "cdc-kafka-test"
	// DefaultProtocol todo: use `canal-json` as the default, after the canal-json fully support in TiCDC
	DefaultProtocol = "open-protocol"
	// DefaultMaxMessageBytes set as `1m`, identical to the default value as broker's `message.max.bytes`
	DefaultMaxMessageBytes = 1024 * 1024
	DefaultPartitionNum    = 3
)

type CMD struct {
	Command string
	Timeout time.Duration
}

func NewCMD(cmd string, timeout time.Duration) *CMD {
	return &CMD{
		Command: cmd,
		Timeout: timeout,
	}
}

type TopicOption struct {
	Topic             string
	Partitions        int
	Config            string
	BootstrapServer   string
	ReplicationFactor int
	Duration          time.Duration

	// Command CommandType
}

func (t *TopicOption) Parse() (commands *CMD, err error) {
	var (
		// At the moment we use zookeeper and broker as image, the worker directory has script to create topic
		// so that we just call `kafka-topics.sh`
		cmd    = "kafka-topics.sh "
		params []string
	)

	params = append(params, "--create")
	params = append(params, "--topic="+t.Topic)
	params = append(params, "--partitions="+strconv.Itoa(t.Partitions))
	params = append(params, "--replication-factor="+strconv.Itoa(t.ReplicationFactor))
	params = append(params, "--config "+t.Config)
	params = append(params, "--bootstrap-server="+t.BootstrapServer)

	cmd = cmd + strings.Join(params, " ")

	timeout := t.Duration + time.Minute*10
	return NewCMD(cmd, timeout), nil
}