package streaming

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"

	"github.com/tidbcloud/service-proxy/pkg/checker/core"
)

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
)

func init() {
	core.DefaultRegistry.Register("kafka", func(req *core.CheckRequest) (core.Checker, error) {
		if req.Kafka == nil {
			return nil, errors.New("kafka params required")
		}
		return NewKafkaChecker(req.Kafka), nil
	})
}

// KafkaChecker checks Kafka broker connectivity.
type KafkaChecker struct {
	params *core.KafkaParams
}

// NewKafkaChecker creates a new Kafka checker.
func NewKafkaChecker(params *core.KafkaParams) *KafkaChecker {
	return &KafkaChecker{params: params}
}

// Type returns the checker type.
func (c *KafkaChecker) Type() string {
	return "kafka"
}

// safeLogFields returns zap fields for logging without sensitive information.
func (c *KafkaChecker) safeLogFields() []zap.Field {
	return []zap.Field{
		zap.String("brokers", c.params.Brokers),
		zap.String("authType", c.params.AuthType),
	}
}

// Check performs Kafka connectivity check.
func (c *KafkaChecker) Check(ctx context.Context) error {
	if ctx == nil {
		return errors.New("context is required")
	}
	if strings.TrimSpace(c.params.Brokers) == "" {
		return fmt.Errorf("brokers is required")
	}
	if err := ctx.Err(); err != nil {
		return err
	}
	isKafkaBrokersReachabilityTest := c.params.ReplicationFactor == 0 && c.params.SchemaRegistryEndpoints == ""
	if isKafkaBrokersReachabilityTest {
		return c.testAllBrokersReachability(ctx)
	} else {
		return c.validateReplicationConfiguration(ctx)
	}
}

func (c *KafkaChecker) testAllBrokersReachability(ctx context.Context) error {
	kafkaConn, _, err := c.createKafkaConnection(ctx)
	if err != nil {
		return err
	}
	defer kafkaConn.Close()
	if err := setConnDeadline(ctx, kafkaConn); err != nil {
		return err
	}
	brokers, err := kafkaConn.Brokers()
	if err != nil {
		log.Warn("get conn brokers error", append([]zap.Field{zap.Error(err)}, c.safeLogFields()...)...)
		return fmt.Errorf("failed to get brokers: %w", err)
	}

	if len(brokers) == 0 {
		log.Warn("no brokers found", c.safeLogFields()...)
		return fmt.Errorf("no brokers found")
	}

	return c.checkBrokersReachability(ctx, brokers)
}

func (c *KafkaChecker) checkBrokersReachability(ctx context.Context, brokers []kafka.Broker) error {
	errorStrings := []string{"The following Kafka brokers can not be connected:"}
	for _, v := range brokers {
		if err := ctx.Err(); err != nil {
			return err
		}
		err := testSingleBrokerReachability(ctx, v.Host, v.Port)
		if err != nil {
			if ctxErr := ctx.Err(); ctxErr != nil {
				return ctxErr
			}
			errorStrings = append(errorStrings, fmt.Sprintf("broker: %s, connection error: %s", net.JoinHostPort(v.Host, strconv.Itoa(v.Port)), err.Error()))
			log.Warn("can not connect to broker",
				zap.String("host", v.Host),
				zap.Int("port", v.Port),
				zap.Error(err))
			continue
		}
		log.Info("successfully connect to broker",
			zap.String("host", v.Host),
			zap.Int("port", v.Port))
	}

	if len(errorStrings) > 1 {
		log.Warn("not all brokers can be connected", c.safeLogFields()...)
		return errors.New(strings.Join(errorStrings, "\n"))
	}
	log.Info("all brokers can be connected", c.safeLogFields()...)

	return nil
}

func testSingleBrokerReachability(ctx context.Context, host string, port int) error {
	if ctx == nil {
		return errors.New("context is required")
	}
	if err := ctx.Err(); err != nil {
		return err
	}
	address := net.JoinHostPort(host, strconv.Itoa(port))
	dialer := &net.Dialer{Timeout: defaultTimeout}
	conn, err := dialer.DialContext(ctx, "tcp", address)
	if err != nil {
		return err
	}
	// Connection is considered reachable once established; close immediately to avoid fd leak.
	defer conn.Close()
	return nil
}

func (c *KafkaChecker) validateReplicationConfiguration(ctx context.Context) error {
	if err := c.validateReplicationFactorIfNeeded(ctx); err != nil {
		return err
	}
	if err := validateSchemaRegistryEndpointIfNeeded(ctx, c.params.SchemaRegistryEndpoints); err != nil {
		return err
	}
	return nil
}

func (c *KafkaChecker) validateReplicationFactorIfNeeded(ctx context.Context) error {
	if c.params.ReplicationFactor == 0 {
		return nil
	}
	kafkaConn, theConnectedEndpoint, err := c.createKafkaConnection(ctx)
	if err != nil {
		return err
	}
	defer kafkaConn.Close()

	if err := setConnDeadline(ctx, kafkaConn); err != nil {
		return err
	}
	brokers, err := kafkaConn.Brokers()
	if err != nil {
		log.Warn("get brokers error", append([]zap.Field{zap.Error(err)}, c.safeLogFields()...)...)
		return fmt.Errorf("failed to get brokers: %w", err)
	}
	brokersNum := uint32(len(brokers))

	if err := setConnDeadline(ctx, kafkaConn); err != nil {
		return err
	}
	controller, err := kafkaConn.Controller()
	if err != nil {
		log.Warn("get controller error", append([]zap.Field{zap.Error(err)}, c.safeLogFields()...)...)
		return fmt.Errorf("failed to get controller: %w", err)
	}

	client, err := c.createKafkaClient(theConnectedEndpoint)
	if err != nil {
		log.Warn("create kafka client error", append([]zap.Field{zap.Error(err)}, c.safeLogFields()...)...)
		return fmt.Errorf("failed to create kafka client: %w", 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", append([]zap.Field{zap.Error(err)}, c.safeLogFields()...)...)
		return fmt.Errorf("failed to describe kafka configs: %w", err)
	}
	return c.doValidateReplicationFactor(resp, brokersNum)
}

func setConnDeadline(ctx context.Context, conn *kafka.Conn) error {
	if ctx == nil {
		return errors.New("context is required")
	}
	if err := ctx.Err(); err != nil {
		return err
	}
	deadline := time.Now().Add(defaultTimeout)
	if ctxDeadline, ok := ctx.Deadline(); ok && ctxDeadline.Before(deadline) {
		deadline = ctxDeadline
	}
	return conn.SetDeadline(deadline)
}

func (c *KafkaChecker) doValidateReplicationFactor(resp *kafka.DescribeConfigsResponse, brokersNum uint32) error {
	log.Info("describe kafka configs resp success", zap.Any("resp", resp))
	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 := c.doValidateReplicationFactorOnOneEntry(e, v.ResourceName, brokersNum)
			if err != nil {
				return err
			}
		}
	}
	log.Info("replication factor is meet the constraint", c.safeLogFields()...)
	return nil
}

func (c *KafkaChecker) doValidateReplicationFactorOnOneEntry(
	configEntry kafka.DescribeConfigResponseConfigEntry,
	resourceName string,
	brokersNum uint32,
) error {
	minInsyncReplicasStr := configEntry.ConfigValue
	minInsyncReplicas, err := strconv.Atoi(minInsyncReplicasStr)
	if err != nil {
		return fmt.Errorf("failed to parse min.insync.replicas: %w", err)
	}
	log.Info("resource config info",
		zap.String("resource name", resourceName),
		zap.String(minInsyncReplicasConfigName, minInsyncReplicasStr),
		zap.Uint32("broker number", brokersNum),
		zap.Uint32("replicationFactor", c.params.ReplicationFactor))

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

func validateSchemaRegistryEndpointIfNeeded(ctx context.Context, endpoint string) error {
	if endpoint == "" {
		return nil
	}
	if ctx == nil {
		return errors.New("context is required")
	}
	if err := ctx.Err(); err != nil {
		return err
	}

	client := &http.Client{
		Timeout: defaultTimeout,
		Transport: &http.Transport{
			TLSClientConfig: &tls.Config{
				InsecureSkipVerify: skipInsecurityVerify,
				MinVersion:         tls.VersionTLS12,
			},
		},
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
	if err != nil {
		return fmt.Errorf("build schema registry request: %w", err)
	}
	resp, err := client.Do(req)
	if err != nil {
		log.Warn("connect schema registry endpoint failed",
			zap.String("endpoint", endpoint),
			zap.Error(err))
		return fmt.Errorf("connect schema registry endpoint failed: %w", err)
	}
	defer resp.Body.Close()

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

	if string(text) != "{}" {
		log.Error("unexpected schema registry response",
			zap.ByteString("response", text))
		return fmt.Errorf("unexpected schema registry response: %s", string(text))
	}
	return nil
}

func (c *KafkaChecker) createKafkaConnection(ctx context.Context) (conn *kafka.Conn, theConnectedEndpoint string, err error) {
	if ctx == nil {
		return nil, "", errors.New("context is required")
	}
	dialer := &kafka.Dialer{Timeout: defaultTimeout}
	dialer.TLS, dialer.SASLMechanism, err = c.makeTlsSaslCfg()
	if err != nil {
		return nil, "", err
	}

	endpointsString := strings.ReplaceAll(c.params.Brokers, " ", "")
	endpoints := strings.Split(endpointsString, ",")
	var errorStrings []string
	for _, v := range endpoints {
		if v == "" {
			continue
		}
		if err := ctx.Err(); err != nil {
			return nil, "", err
		}
		log.Info("trying to connect to bootstrap endpoint ...",
			zap.String("endpoint", v))
		conn, err = dialer.DialContext(ctx, "tcp", v)
		if err != nil {
			if ctxErr := ctx.Err(); ctxErr != nil {
				return nil, "", ctxErr
			}
			log.Warn("connect bootstrap endpoint failed",
				zap.String("endpoint", v),
				zap.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.String("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.String("endpoint", v))
		return conn, v, nil
	}
	log.Info("none of the kafka bootstrap endpoints can be connected", c.safeLogFields()...)
	return nil, "", fmt.Errorf("none of the kafka bootstrap endpoints can be connected: %s", strings.Join(errorStrings, ";"))
}

func (c *KafkaChecker) makeTlsSaslCfg() (tlsCfg *tls.Config, saslCfg sasl.Mechanism, err error) {
	if c.params.EnableTLS {
		tlsCfg = &tls.Config{
			MinVersion:         tls.VersionTLS12,
			NextProtos:         []string{"h2", "http/1.1"},
			InsecureSkipVerify: skipInsecurityVerify,
		}
	}

	saslCfg, err = c.makeSaslMechanism()
	if err != nil {
		log.Warn("failed to make sasl mechanism", c.safeLogFields()...)
		return nil, nil, fmt.Errorf("failed to create sasl mechanism: %w", err)
	}
	return tlsCfg, saslCfg, nil
}

func (c *KafkaChecker) makeSaslMechanism() (mechanism sasl.Mechanism, err error) {
	switch core.KafkaAuthType(c.params.AuthType) {
	case core.KafkaAuthTypeSASLPlain:
		return plain.Mechanism{Username: c.params.User, Password: c.params.Password}, nil
	case core.KafkaAuthTypeSASLSHA256:
		return scram.Mechanism(scram.SHA256, c.params.User, c.params.Password)
	case core.KafkaAuthTypeSASLSHA512:
		return scram.Mechanism(scram.SHA512, c.params.User, c.params.Password)
	case core.KafkaAuthTypeNone:
		return nil, nil
	default:
		return nil, fmt.Errorf("unknown kafka auth type: %s", c.params.AuthType)
	}
}

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

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

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

	return client, nil
}