package utils

import (
	"errors"
	"fmt"
	"strings"

	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	"github.com/tidbcloud/pkgv2/bizerr"
)

const (
	privateLinkConnectionErrorTips = "If you select Private Link or Private Service Connect as the Connectivity Method, " +
		"ensure that the Kafka brokers' 'advertised.listeners' configuration follows the " +
		"'Kafka Advertised Listener Pattern'. If an endpoint was created with a different pattern for the Kafka service," +
		" you have two options: 1) Update the Kafka brokers' 'advertised.listeners' to match the pattern if it is not " +
		"connected by other Kafka clients. 2) Wait approximately 30 minutes for the endpoint to expire if no changefeeds" +
		" are connected to the Kafka service, then create a new one with the expected pattern."
)

const (
	KafkaProtocolTypeCanalJson    = "canal-json"
	KafkaProtocolTypeOpenProtocol = "open-protocol"
	KafkaProtocolTypeAvro         = "avro"
	KafkaProtocolDebezium         = "debezium"
	KafkaProtocolDebeziumAvro     = "debezium-avro"

	ClusterProxyKafkaAuthTypePlain       = "sasl-plain"
	ClusterProxyKafkaAuthTypeScramSha256 = "sasl-scram-sha-256"
	ClusterProxyKafkaAuthTypeScramSha512 = "sasl-scram-sha-512"

	KafkaAuthTypePlain       = "PLAIN"
	KafkaAuthTypeScramSha256 = "SCRAM-SHA-256"
	KafkaAuthTypeScramSha512 = "SCRAM-SHA-512"

	KafkaVersion1xx = "1.0.0"
	KafkaVersion2xx = "2.0.0"
	KafkaVersion3xx = "3.0.0"

	PartitionTs         = "ts"
	PartitionTable      = "table"
	PartitionIndexValue = "index-value"
	PartitionColumns    = "columns"

	DecimalHandlingModePrecise       = "precise"
	DecimalHandlingModeString        = "string"
	BigintUnsignedHandlingModeLong   = "long"
	BigintUnsignedHandlingModeString = "string"

	CompressionGzip   = "gzip"
	CompressionSnappy = "snappy"
	CompressionLz4    = "lz4"
	CompressionZSTD   = "zstd"
)

// FormatKafkaTestConnectionErrorContent
// Currently we got two types of error:
// 1. Client.Timeout exceeded while awaiting headers => test connection timeout
// 2. Request fail, 400 => test connection failed
// 3. BizError => test connection failed with specific error message
func FormatKafkaTestConnectionErrorContent(err error) string {
	content := err.Error()
	var bzErr bizerr.BizError
	formatedErr := false
	if ok := errors.As(err, &bzErr); ok {
		content = bzErr.ErrMsg()
		formatedErr = true
	}
	if strings.Contains(content, "Client.Timeout exceeded while awaiting headers") {
		return "connect to downstream timeout. " +
			"Please retry later. If retry can not solve the problem, please contact technical support."
	}

	testFailedPrefix := "Request fail, 400-"
	if strings.HasPrefix(content, testFailedPrefix) {
		content = strings.TrimPrefix(content, testFailedPrefix)
		formatedErr = true
	}
	if formatedErr {
		parts := strings.Split(content, "\n")
		reason := strings.TrimRight(parts[0], ".") + "."
		if len(parts) == 1 {
			return reason
		}

		list := ""
		for i := 1; i < len(parts); i++ {
			p := strings.TrimSpace(parts[i])
			if p != "" {
				list = list + fmt.Sprintf("<li>%s</li>", p)
			}
		}
		if list == "" {
			return reason
		}
		return fmt.Sprintf("%s<ul>%s</ul>%s", reason, list, privateLinkConnectionErrorTips)
	}

	// unknown error
	return content
}

func TransferClusterProxyKafkaAuthType(authType dataflow.KafkaAuthType) string {
	switch authType {
	case dataflow.KafkaAuthType_SASL_PLAIN:
		return ClusterProxyKafkaAuthTypePlain
	case dataflow.KafkaAuthType_SASL_SCRAM_SHA_256:
		return ClusterProxyKafkaAuthTypeScramSha256
	case dataflow.KafkaAuthType_SASL_SCRAM_SHA_512:
		return ClusterProxyKafkaAuthTypeScramSha512
	}
	return ""
}

func TransferKafkaAuthType(authType dataflow.KafkaAuthType) string {
	switch authType {
	case dataflow.KafkaAuthType_SASL_PLAIN:
		return KafkaAuthTypePlain
	case dataflow.KafkaAuthType_SASL_SCRAM_SHA_256:
		return KafkaAuthTypeScramSha256
	case dataflow.KafkaAuthType_SASL_SCRAM_SHA_512:
		return KafkaAuthTypeScramSha512
	}
	return ""
}

func TransferProtocol(protocol dataflow.Protocol) string {
	switch protocol {
	case dataflow.Protocol_CANAL_JSON:
		return KafkaProtocolTypeCanalJson
	case dataflow.Protocol_OPEN_PROTOCOL:
		return KafkaProtocolTypeOpenProtocol
	case dataflow.Protocol_AVRO:
		return KafkaProtocolTypeAvro
	case dataflow.Protocol_DEBEZIUM:
		return KafkaProtocolDebezium
	case dataflow.Protocol_DEBEZIUM_AVRO:
		return KafkaProtocolDebeziumAvro
	}
	return ""
}

func IsAvroProtocol(protocol dataflow.Protocol) bool {
	return protocol == dataflow.Protocol_AVRO || protocol == dataflow.Protocol_DEBEZIUM_AVRO
}

func TransferKafkaVersion(kafkaVersion dataflow.KafkaVersion) string {
	switch kafkaVersion {
	case dataflow.KafkaVersion_KAFKA_VERSION_1XX:
		return KafkaVersion1xx
	case dataflow.KafkaVersion_KAFKA_VERSION_2XX:
		return KafkaVersion2xx
	case dataflow.KafkaVersion_KAFKA_VERSION_3XX:
		return KafkaVersion3xx
	default:
		return ""
	}
}

func TransferDecimalHandlingMode(mode dataflow.DecimalHandlingMode) string {
	switch mode {
	case dataflow.DecimalHandlingMode_DECIMAL_HANDLING_MODE_PRECISE:
		return DecimalHandlingModePrecise
	case dataflow.DecimalHandlingMode_DECIMAL_HANDLING_MODE_STRING:
		return DecimalHandlingModeString
	}
	return ""
}

func TransferBigintUnsignedHandlingMode(mode dataflow.BigintUnsignedHandlingMode) string {
	switch mode {
	case dataflow.BigintUnsignedHandlingMode_BIGINT_UNSIGNED_HANDLING_MODE_STRING:
		return BigintUnsignedHandlingModeString
	case dataflow.BigintUnsignedHandlingMode_BIGINT_UNSIGNED_HANDLING_MODE_LONG:
		return BigintUnsignedHandlingModeLong
	}
	return ""
}

func TransferPartitionDispatcher(partition dataflow.PartitionDispatcher) string {
	switch partition {
	case dataflow.PartitionDispatcher_TABLE:
		return PartitionTable
	case dataflow.PartitionDispatcher_INDEX_VALUE:
		return PartitionIndexValue
	case dataflow.PartitionDispatcher_TS:
		return PartitionTs
	case dataflow.PartitionDispatcher_COLUMNS:
		return PartitionColumns
	}
	return ""
}

func TransferKafkaCompression(compression dataflow.CompressionType) string {
	switch compression {
	case dataflow.CompressionType_GZIP:
		return CompressionGzip
	case dataflow.CompressionType_SNAPPY:
		return CompressionSnappy
	case dataflow.CompressionType_LZ4:
		return CompressionLz4
	case dataflow.CompressionType_ZSTD:
		return CompressionZSTD
	default:
		return ""
	}
}