// to create topics when auto.create.topics.enable='false'
package service

import (
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"net"
	"strconv"
	"strings"
	"time"

	"github.com/pingcap/log"
	kafka "github.com/segmentio/kafka-go"
	"github.com/segmentio/kafka-go/sasl/scram"
)

type KafkaTopicParam struct {
	TopicName         string
	NumPartitions     int
	ReplicationFactor int
}

func (svc *apiService) CreateKafkaTopic(nParam KafkaTopicParam) (string, error) {
	log.Debug(fmt.Sprintf("Create Kafka Topic, %v, %v", nParam, svc.Kafkaconf))
	if err := svc.Kafkaconf.Validate(); err != nil {
		return "", err
	}
	mechanism, err := scram.Mechanism(scram.SHA512, svc.Kafkaconf.User, svc.Kafkaconf.Passwd)
	if err != nil {
		log.Error(err.Error())
		return "", fmt.Errorf("scram failed, %s", err)
	}
	dialer := &kafka.Dialer{
		Timeout:       10 * time.Second,
		DualStack:     true,
		SASLMechanism: mechanism,
		TLS:           &tls.Config{InsecureSkipVerify: true},
	}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	var conn *kafka.Conn
	for _, endpoint := range strings.Split(svc.Kafkaconf.BrokerList, ",") {
		conn, err = dialer.DialContext(ctx, "tcp", endpoint)
		if err != nil {
			log.Error(fmt.Sprintf("dail failed, %s", err.Error()))
			return "", err
		} else {
			break
		}
	}
	if conn == nil {
		return "", errors.New("kafka dailer failed")
	}
	defer conn.Close()

	controller, err := conn.Controller()
	if err != nil {
		log.Error(fmt.Sprintf("controller failed, %s", err.Error()))
		return "", err
	}
	// var controllerConn *kafka.Conn
	controllerConn, err := dialer.Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port)))
	if err != nil {
		log.Error(fmt.Sprintln("Join controller failed", err.Error()))
		return "", err
	}
	defer controllerConn.Close()

	topicConfigs := FormConf([]KafkaTopicParam{nParam})
	err = controllerConn.CreateTopics(topicConfigs...)
	if err != nil {
		log.Error(fmt.Sprintf("create topic failed, %s", err.Error()))
		return "", err
	}
	return fmt.Sprintf("%s/%s", svc.Kafkaconf.BrokerList, nParam.TopicName), nil
}

func (svc *apiService) DeleteKafkaTopic(topic string) error {
	if err := svc.Kafkaconf.Validate(); err != nil {
		return err
	}
	mechanism, err := scram.Mechanism(scram.SHA512, svc.Kafkaconf.User, svc.Kafkaconf.Passwd)
	if err != nil {
		log.Error(err.Error())
		return fmt.Errorf("scram failed, %s", err)
	}
	dialer := &kafka.Dialer{
		Timeout:       10 * time.Second,
		DualStack:     true,
		SASLMechanism: mechanism,
		TLS:           &tls.Config{InsecureSkipVerify: true},
	}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	var conn *kafka.Conn
	for _, endpoint := range strings.Split(svc.Kafkaconf.BrokerList, ",") {
		conn, err = dialer.DialContext(ctx, "tcp", endpoint)
		if err != nil {
			log.Error(fmt.Sprintf("dail failed, %s", err.Error()))
			return err
		} else {
			break
		}
	}
	if conn == nil {
		return errors.New("kafka dailer failed")
	}
	defer conn.Close()

	controller, err := conn.Controller()
	if err != nil {
		log.Error(fmt.Sprintf("controller failed, %s", err.Error()))
		return err
	}
	// var controllerConn *kafka.Conn
	controllerConn, err := dialer.Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port)))
	if err != nil {
		log.Error(fmt.Sprintln("Join controller failed", err.Error()))
		return err
	}
	defer controllerConn.Close()

	err = controllerConn.DeleteTopics(topic)
	if err != nil {
		log.Error(fmt.Sprintf("create topic failed, %s", err.Error()))
		return err
	}
	return nil
}

func FormConf(params []KafkaTopicParam) []kafka.TopicConfig {
	nTopicConfigs := make([]kafka.TopicConfig, 0)
	for _, param := range params {
		nTopicConfigs = append(nTopicConfigs, kafka.TopicConfig{
			Topic:             param.TopicName,
			NumPartitions:     param.NumPartitions,
			ReplicationFactor: param.ReplicationFactor,
		})
	}
	return nTopicConfigs
}