package kafka

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

	"github.com/pingcap/test-infra/caselib/pkg/model/workload"
)

type ConsumerOption struct {
	brokerAddr string
	targetAddr string

	Topic           string
	Protocol        string
	MaxMessageBytes int
	PartitionNum    int
	GroupID         string
}

func (t *ConsumerOption) Parse() (commands *workload.CMD, err error) {
	var (
		cmd = "nohup /cdc_kafka_consumer %s &"
	)

	// build broker uri first.
	topic := DefaultTopicName
	if t.Topic != "" {
		topic = t.Topic
	}
	brokerURI := t.brokerAddr + "/" + topic

	params := make([]string, 0)
	protocol := DefaultProtocol
	if t.Protocol != "" {
		protocol = t.Protocol
	}
	params = append(params, "protocol="+protocol)

	maxMessageBytes := DefaultMaxMessageBytes
	if t.MaxMessageBytes != 0 {
		maxMessageBytes = t.MaxMessageBytes
	}
	params = append(params, "max-message-bytes="+strconv.Itoa(maxMessageBytes))

	// partitionNum := DefaultPartitionNum
	// if t.PartitionNum != 0 {
	// 	partitionNum = t.PartitionNum
	// }
	// params = append(params, "partition-num="+strconv.Itoa(partitionNum))
	if len(params) != 0 {
		brokerURI += "?"
		brokerURI += strings.Join(params, "&")
	}
	params = params[:0]
	params = append(params, "--upstream-uri="+strconv.Quote(brokerURI))
	params = append(params, "--downstream-uri="+strconv.Quote(t.targetAddr))

	// nohup cdc_kafka_consumer --upstream-uri=kafka://downstream-kafka.testbed-lingjin-pmj7b:9092/cdc-event-canal-json-1?protocol=canal-json&max-message-bytes=1048576&partition-num=3 --downstream-uri=mysql://root:@target.testbed-lingjin-pmj7b:3306 &
	cmd = fmt.Sprintf(cmd, strings.Join(params, " "))
	timeout := 10 * time.Second

	return workload.NewCMD(cmd, timeout), nil
}

func (t *ConsumerOption) WithBrokerAddr(addr string) *ConsumerOption {
	t.brokerAddr = addr
	return t
}

func (t *ConsumerOption) WithTargetAddr(addr string) *ConsumerOption {
	t.targetAddr = addr
	return t
}