package common

import (
	"net"
	"strconv"
	"time"

	"github.com/segmentio/kafka-go"
)

type KafkaReadOptionOp func(*kafka.ReaderConfig)

func KafkaGroupIDOp(groupID string) KafkaReadOptionOp {
	return func(config *kafka.ReaderConfig) {
		if groupID != "" {
			config.GroupID = groupID
		}
	}
}

func KafkaTopicOp(topic string) KafkaReadOptionOp {
	return func(config *kafka.ReaderConfig) {
		if topic != "" {
			config.Topic = topic
		}
	}
}

func GenerateKafkaReader(brokers []string, opts ...KafkaReadOptionOp) *kafka.Reader {
	config := NewKafkaReaderConfig(brokers, opts)
	return kafka.NewReader(*config)
}

func NewKafkaReaderConfig(brokers []string, opts []KafkaReadOptionOp) *kafka.ReaderConfig {
	config := &kafka.ReaderConfig{
		Brokers:  brokers,
		GroupID:  "consumer-group-id",
		Topic:    "topic-A",
		MinBytes: 10e3, // 10KB, default
		MaxBytes: 10e6, // 10MB, default
		MaxWait:  1 * time.Second,
	}
	for _, opt := range opts {
		opt(config)
	}
	return config
}

func GenerateKafkaControllerConn(broker string) (*kafka.Conn, error) {
	conn, err := kafka.Dial("tcp", broker)
	if err != nil {
		panic(err.Error())
	}

	controller, err := conn.Controller()
	if err != nil {
		panic(err.Error())
	}
	return kafka.Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port)))
}