package common

import (
	"crypto/tls"
	"fmt"
	"net"
	"net/url"
	"strings"
	"time"

	"github.com/IBM/sarama"
	. "github.com/tidbcloud/dataflow-service-e2e/internal/model"
)

// ConsumeKafkaEvent reads a run-unique topic from its oldest offset and only
// succeeds after observing the requested marker in an actual event payload.
func ConsumeKafkaEvent(entity *KafkaEntity, topic, marker string) error {
	brokers := strings.Split(entity.BootstrapEndpoints, ",")
	for i := range brokers {
		brokers[i] = strings.TrimSpace(brokers[i])
	}
	if len(brokers) == 0 || brokers[0] == "" {
		return fmt.Errorf("Kafka bootstrap endpoints are empty")
	}
	cfg := sarama.NewConfig()
	cfg.Version = sarama.V2_0_0_0
	cfg.Consumer.Return.Errors = true
	if entity.AuthType == DataflowKafkaAuthTypeSASLPLAIN {
		cfg.Net.SASL.Enable, cfg.Net.SASL.User, cfg.Net.SASL.Password, cfg.Net.SASL.Mechanism = true, entity.User, entity.Password, sarama.SASLTypePlaintext
	}
	if entity.TLSEnabled {
		serverName := brokers[0]
		if host, _, err := net.SplitHostPort(serverName); err == nil {
			serverName = host
		} else if parsed, parseErr := url.Parse("//" + serverName); parseErr == nil && parsed.Hostname() != "" {
			serverName = parsed.Hostname()
		}
		cfg.Net.TLS.Enable = true
		cfg.Net.TLS.Config = &tls.Config{MinVersion: tls.VersionTLS12, ServerName: serverName}
	}
	consumer, err := sarama.NewConsumer(brokers, cfg)
	if err != nil {
		return fmt.Errorf("create Kafka consumer: %w", err)
	}
	defer consumer.Close()
	partitions, err := consumer.Partitions(topic)
	if err != nil {
		return err
	}
	deadline := time.Now().Add(90 * time.Second)
	for _, partition := range partitions {
		pc, err := consumer.ConsumePartition(topic, partition, sarama.OffsetOldest)
		if err != nil {
			return err
		}
		for time.Now().Before(deadline) {
			select {
			case msg := <-pc.Messages():
				if msg != nil && strings.Contains(string(msg.Value), marker) {
					_ = pc.Close()
					return nil
				}
			case readErr := <-pc.Errors():
				if readErr != nil {
					_ = pc.Close()
					return readErr
				}
			case <-time.After(time.Second):
			}
		}
		_ = pc.Close()
	}
	return fmt.Errorf("Kafka topic %q did not contain marker %q", topic, marker)
}