package central

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

	"github.com/pingcap/log"
	"github.com/segmentio/kafka-go"
	"github.com/segmentio/kafka-go/sasl"
	"github.com/segmentio/kafka-go/sasl/plain"
	"github.com/segmentio/kafka-go/sasl/scram"
	"go.uber.org/zap"
)

const (
	minInsyncReplicasConfigName = "min.insync.replicas"
	defaultTimeout              = 4 * time.Second

	// always skip verify hostname or cert, since we don't know the network type here
	skipInsecurityVerify = true
)

/*
There is two validation logic here from different calls (FIXME: we should separate them)
1. test reachability to every Kafka brokers when configuring destination for Kafka changefeed
2. validate the replication configuration for changefeed after reachability test pass.
*/
func checkKafkaAccessV2(ctx context.Context, model *connectionModel) error {
	isKafkaBrokersReachabilityTest := model.KafkaItem.ReplicationFactor == 0 && model.KafkaItem.SchemaRegistryEndpoints == ""
	if isKafkaBrokersReachabilityTest {
		return testAllBrokersReachability(model)
	} else {
		return validateReplicationConfiguration(ctx, model)
	}
}

func testAllBrokersReachability(model *connectionModel) error {
	kafkaConn, _, err := createKafkaConnection(model)
	if err != nil {
		return err
	}
	brokers, err := kafkaConn.Brokers()
	if err != nil {
		log.Warn("get conn brokers error", zap.Any("error", err.Error()), zap.Any("connectionModel", model.ToLoggingString()))
		return fmt.Errorf("get conn brokers error %s", err.Error())
	}

	if len(brokers) == 0 {
		log.Warn("no brokers found", zap.Any("connectionModel", model.ToLoggingString()))
		return fmt.Errorf("no brokers found")
	}

	isAllConnectable := true
	errorStrings := []string{
		"The following Kafka brokers can not be connected:",
	}
	for _, v := range brokers {
		err = testSingleBrokerReachability(v.Host, v.Port)
		if err != nil {
			isAllConnectable = false
			errorStrings = append(errorStrings, fmt.Sprintf("broker: %s:%d, connection error: %s", v.Host, v.Port, err.Error()))
			log.Warn("can not connect to broker",
				zap.Any("host", v.Host),
				zap.Any("port", v.Port),
				zap.Any("error", err.Error()),
				zap.Any("connectionModel", model.ToLoggingString()))
			continue
		}
		log.Info("successfully connect to broker",
			zap.Any("host", v.Host),
			zap.Any("port", v.Port),
			zap.Any("connectionModel", model.ToLoggingString()))
	}

	if !isAllConnectable {
		log.Warn("not all brokers can be connected",
			zap.Any("connectionModel", model.ToLoggingString()))
		return errors.New(strings.Join(errorStrings, "\n"))
	}
	log.Info("all brokers can be connected",
		zap.Any("connectionModel", model.ToLoggingString()))

	return nil
}

func testSingleBrokerReachability(host string, port int) error {
	address := net.JoinHostPort(host, strconv.Itoa(port))
	_, err := net.DialTimeout("tcp", address, defaultTimeout)
	return err
}

func validateReplicationConfiguration(ctx context.Context, model *connectionModel) error {
	err := validateReplicationFactorIfNeeded(ctx, model)
	if err != nil {
		return err
	}

	err = validateSchemaRegistryEndpointIfNeeded(model.KafkaItem.SchemaRegistryEndpoints)
	if err != nil {
		return err
	}

	return nil
}

func validateReplicationFactorIfNeeded(ctx context.Context, model *connectionModel) error {
	if model.KafkaItem.ReplicationFactor == 0 {
		return nil
	}
	kafkaConn, theConnectedEndpoint, err := createKafkaConnection(model)
	if err != nil {
		return err
	}

	brokers, err := kafkaConn.Brokers()
	if err != nil {
		log.Warn("get brokers error", zap.Any("error", err.Error()), zap.Any("connectionModel", model.ToLoggingString()))
		return fmt.Errorf("get conn brokers error %v", err)
	}
	brokersNum := uint32(len(brokers))

	controller, err := kafkaConn.Controller()
	if err != nil {
		log.Warn("get controller error", zap.Any("error", err.Error()), zap.Any("connectionModel", model.ToLoggingString()))
		return fmt.Errorf("get conn controller error %v", err)
	}

	client, err := createKafkaClient(model, theConnectedEndpoint)
	if err != nil {
		log.Warn("create kafka client error", zap.Any("error", err.Error()), zap.Any("connectionModel", model.ToLoggingString()))
		return fmt.Errorf("create kafka client failed %v", err)
	}
	resp, err := client.DescribeConfigs(ctx, &kafka.DescribeConfigsRequest{
		Resources: []kafka.DescribeConfigRequestResource{
			{
				ResourceType: kafka.ResourceTypeBroker,
				ResourceName: strconv.Itoa(controller.ID),
				ConfigNames:  []string{minInsyncReplicasConfigName},
			},
		},
	})

	if err != nil {
		log.Warn("describe kafka configs resp error", zap.Any("error", err.Error()), zap.Any("connectionModel", model.ToLoggingString()))
		return fmt.Errorf("describe kafka configs error %v", err)
	}
	return doValidateReplicationFactor(resp, model, brokersNum)
}

func doValidateReplicationFactor(
	resp *kafka.DescribeConfigsResponse, model *connectionModel, brokersNum uint32,
) error {
	log.Info("describe kafka configs resp success", zap.Any("resp", resp), zap.Any("connectionModel", model.ToLoggingString()))
	for _, v := range resp.Resources {
		if len(v.ConfigEntries) == 0 {
			continue
		}
		// Notice: searching all the entries instead of just use the first and only one(v.ConfigEntries[0]) is necessary
		// since KOP(Kafka-on-Pulsar) is not work as standard Kafka, it may return some configurations not specific in request.
		// Moreover, it doesn't return the configuration currently.
		for _, e := range v.ConfigEntries {
			if e.ConfigName != minInsyncReplicasConfigName {
				continue
			}
			err := doValidateReplicationFactorOnOneEntry(e, v.ResourceName, model, brokersNum)
			if err != nil {
				return err
			}
		}
	}
	log.Info("replication factor is meet the constraint", zap.Any("connectionModel", model.ToLoggingString()))
	return nil
}

func doValidateReplicationFactorOnOneEntry(
	configEntry kafka.DescribeConfigResponseConfigEntry, resourceName string, model *connectionModel, brokersNum uint32,
) error {
	minInsyncReplicasStr := configEntry.ConfigValue
	minInsyncReplicas, err := strconv.Atoi(minInsyncReplicasStr)
	if err != nil {
		return fmt.Errorf("atoi min.insync.replicas config failed %v", err)
	}
	log.Info("resource config info",
		zap.Any("resource name", resourceName),
		zap.Any(minInsyncReplicasConfigName, minInsyncReplicasStr),
		zap.Any("broker number", brokersNum),
		zap.Any("connectionModel", model.ToLoggingString()))

	if model.KafkaItem.ReplicationFactor < uint32(minInsyncReplicas) || model.KafkaItem.ReplicationFactor > brokersNum {
		log.Warn("replication factor is not meet the constraint",
			zap.Any("resource name", resourceName),
			zap.Any(minInsyncReplicasConfigName, minInsyncReplicasStr),
			zap.Any("broker number", brokersNum),
			zap.Any("connectionModel", model.ToLoggingString()))
		return fmt.Errorf("ReplicationFactor(%d) should >= %s(%d), and <= broker number(%d)",
			model.KafkaItem.ReplicationFactor, minInsyncReplicasConfigName, minInsyncReplicas, brokersNum)
	}
	return nil
}

func validateSchemaRegistryEndpointIfNeeded(endpoint string) error {
	if endpoint == "" {
		return nil
	}
	resp, err := http.Get(endpoint)
	if err != nil {
		log.Warn("connect schema registry endpoint failed",
			zap.Any("endpoint", endpoint),
			zap.Any("error", err))
		return fmt.Errorf("connect schema registry endpoint failed")
	}
	defer resp.Body.Close()

	text, err := io.ReadAll(resp.Body)
	if err != nil {
		log.Warn("parse schema registry response failed",
			zap.Any("endpoint", endpoint),
			zap.Any("error", err))
		return fmt.Errorf("parse schema registry response failed")
	}

	if string(text[:]) != "{}" {
		log.Error("Unexpected response from Schema Registry",
			zap.ByteString("response", text))
		return fmt.Errorf("upexpected schema registry response")
	}
	return nil
}

func createKafkaConnection(model *connectionModel) (conn *kafka.Conn, theConnectedEndpoint string, err error) {
	dialer := &kafka.Dialer{Timeout: defaultTimeout}
	dialer.TLS, dialer.SASLMechanism, err = makeTlsSaslCfg(model)
	if err != nil {
		return nil, "", err
	}

	endpointsString := strings.ReplaceAll(model.Url, " ", "")
	endpoints := strings.Split(endpointsString, ",")
	var errorStrings []string
	for _, v := range endpoints {
		log.Info("trying to connect to bootstrap endpoint ...",
			zap.Any("endpoint", v))
		conn, err = dialer.Dial("tcp", v)
		if err != nil {
			log.Warn("connect bootstrap endpoint failed",
				zap.Any("endpoint", v),
				zap.Any("error", err))
			errorStrings = append(errorStrings, fmt.Sprintf("connect to '%s' failed by error: %s ", v, err.Error()))
			continue
		}
		if conn == nil {
			log.Warn("empty bootstrap endpoint connection",
				zap.Any("endpoint", v))
			errorStrings = append(errorStrings, fmt.Sprintf("connect to '%s' failed since it return empty connection", v))
			continue
		}
		log.Info("successfully connect to bootstrap endpoint",
			zap.Any("endpoint", v))
		return conn, v, nil
	}
	log.Info("none of the kafka bootstrap endpoints can be connected",
		zap.Any("connectionModel", model.ToLoggingString()))
	return nil, "", fmt.Errorf("none of the kafka bootstrap endpoints can be connected:\n%s", strings.Join(errorStrings, "\n"))
}

func makeTlsSaslCfg(model *connectionModel) (tlsCfg *tls.Config, saslCfg sasl.Mechanism, err error) {
	if model.EnableTls { // #nosec G402
		tlsCfg = &tls.Config{
			MinVersion:         tls.VersionTLS12,
			NextProtos:         []string{"h2", "http/1.1"},
			InsecureSkipVerify: skipInsecurityVerify,
		}
	}

	saslCfg, err = makeSaslMechanism(model)
	if err != nil {
		log.Warn("failed to make sasl mechanism", zap.Any("connectionModel", model.ToLoggingString()))
		return nil, nil, fmt.Errorf("failed to make sasl mechanism, error: %v", err)
	}
	return tlsCfg, saslCfg, nil
}

func makeSaslMechanism(model *connectionModel) (mechanism sasl.Mechanism, err error) {
	switch model.AuthType {
	case "sasl-plain":
		return plain.Mechanism{Username: model.User, Password: model.Password}, nil
	case "sasl-scram-sha-256":
		return scram.Mechanism(scram.SHA256, model.User, model.Password)
	case "sasl-scram-sha-512":
		return scram.Mechanism(scram.SHA512, model.User, model.Password)
	case "":
		return nil, nil
	default:
		return nil, fmt.Errorf("unknown kafka auth type %s", model.AuthType)
	}
}

func createKafkaClient(model *connectionModel, connectableBootstrapEndpoint string) (client *kafka.Client, err error) {
	dialFunc := (&net.Dialer{}).DialContext
	transport := kafka.Transport{
		Dial: dialFunc,
	}

	transport.TLS, transport.SASL, err = makeTlsSaslCfg(model)
	if err != nil {
		return nil, err
	}

	client = &kafka.Client{
		Addr:      kafka.TCP(connectableBootstrapEndpoint),
		Timeout:   defaultTimeout,
		Transport: &transport,
	}

	return client, nil
}