package impl

import (
	"context"
	"encoding/json"
	"fmt"
	"net/url"
	"strconv"
	"strings"

	"github.com/AlekSi/pointer"
	ticdc_v1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/dataflow-service-ng/app/config"
	"github.com/tidbcloud/dataflow-service-ng/app/models/biz_changefeed"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/dataflow-service-ng/app/utils"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	encryptor_wrapper "github.com/tidbcloud/dataflow-service-ng/pkg/encryptor-wrapper"
	regionalapi "github.com/tidbcloud/dataflow-service-ng/pkg/regional-api"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
)

type kafkaDownstreamAdaptor struct {
	cdcPreCheckCli regionalapi.TiCDCPreCheckClient
	encryptor      encryptor_wrapper.Encryptor
}

func NewKafkaDownstreamAdaptor(
	cdcPreCheckCli regionalapi.TiCDCPreCheckClient, encryptor encryptor_wrapper.Encryptor,
) biz_changefeed.DownstreamAdaptor {
	return &kafkaDownstreamAdaptor{
		cdcPreCheckCli: cdcPreCheckCli,
		encryptor:      encryptor,
	}
}

func (m *kafkaDownstreamAdaptor) TestConnection(
	ctx context.Context, cluster *utils.Cluster, downstreamParams interface{},
) bizerr.BizError {
	kafkaParams, bizErr := m.parseKafkaParams(downstreamParams)
	if bizErr != nil {
		return bizErr
	}
	if kafkaParams.GetBroker() == nil || kafkaParams.GetBroker().GetBrokerEndpoints() == "" {
		return bizerr.New(bizerr.ErrCodeParams, "broker is missing")
	}
	clusterIDStr := strconv.FormatUint(cluster.GetClusterID(), 10)
	region := &regionalapi.Region{SharedPoolID: cluster.GetSharedPoolId()}
	networkReq := &ticdc_v1.TestNetworkConnectionRequest{
		ClusterId: clusterIDStr,
		Type:      string(entity.TiCDCDownStreamTypeKafka),
		KafkaParam: &ticdc_v1.TestNetworkConnectionRequest_KafkaParameter{
			AuthType:          utils.TransferClusterProxyKafkaAuthType(kafkaParams.GetAuthentication().GetAuthType()),
			User:              kafkaParams.GetAuthentication().GetUserName(),
			Password:          kafkaParams.GetAuthentication().GetPassword(),
			BoostrapEndpoints: kafkaParams.GetBroker().GetBrokerEndpoints(),
			EnableTls:         kafkaParams.GetBroker().GetTlsEnable(),
		},
	}
	if err := m.cdcPreCheckCli.TestNetworkConnection(ctx, region, networkReq); err != nil {
		return bizerr.New(config.ErrCodeCDCFailedToConnectErr,
			config.ErrMsgCDCFailedToConnectErr,
			DownstreamDisplayNameMapping[dataflow.ChangefeedType_KAFKA],
			utils.FormatKafkaTestConnectionErrorContent(err))
	}
	return nil
}

func (m *kafkaDownstreamAdaptor) PreCheckChangefeedParameters(
	ctx context.Context, cluster *utils.Cluster, downstreamParams interface{},
) bizerr.BizError {
	kafkaParams, bizErr := m.parseKafkaParams(downstreamParams)
	if bizErr != nil {
		return bizErr
	}
	startTSO, bizErr := parseStartTSO(kafkaParams.GetStartPosition())
	if bizErr != nil {
		return bizErr
	}

	var schemaRegistry string
	if utils.IsAvroProtocol(kafkaParams.GetDataFormat().Protocol) {
		schemaRegistry, bizErr = m.assembleSchemaRegistryURI(kafkaParams.GetDataFormat().GetSchemaRegistry())
		if bizErr != nil {
			return bizErr
		}
	}
	clusterIDStr := strconv.FormatUint(cluster.GetClusterID(), 10)
	region := &regionalapi.Region{SharedPoolID: cluster.GetSharedPoolId()}
	changefeedParamReq := &ticdc_v1.ValidateChangefeedParametersRequest{
		ClusterId: clusterIDStr,
		Type:      string(entity.TiCDCDownStreamTypeKafka),
		StartTso:  startTSO,
		KafkaParam: &ticdc_v1.ValidateChangefeedParametersRequest_KafkaParameter{
			AuthType: utils.TransferClusterProxyKafkaAuthType(
				kafkaParams.GetAuthentication().GetAuthType()),
			User:                    kafkaParams.GetAuthentication().GetUserName(),
			Password:                kafkaParams.GetAuthentication().GetPassword(),
			Url:                     kafkaParams.GetBroker().GetBrokerEndpoints(),
			EnableTls:               kafkaParams.GetBroker().GetTlsEnable(),
			Topic:                   kafkaParams.GetTopicPartitionConfig().GetDefaultTopic(),
			ReplicationFactor:       kafkaParams.GetTopicPartitionConfig().GetReplicationFactor(),
			PartitionNum:            kafkaParams.GetTopicPartitionConfig().GetPartitionNum(),
			SchemaRegistryEndpoints: schemaRegistry,
		},
	}
	if err := m.cdcPreCheckCli.ValidateChangefeedParameters(ctx, region, changefeedParamReq); err != nil {
		return bizerr.New(config.ErrCodeCDCFailedToConnectErr,
			config.ErrMsgCDCFailedToConnectErr,
			DownstreamDisplayNameMapping[dataflow.ChangefeedType_KAFKA],
			utils.FormatKafkaTestConnectionErrorContent(err))
	}
	return nil
}

func (m *kafkaDownstreamAdaptor) EncryptPasswordInPlace(
	ctx context.Context, downstreamParams interface{},
) bizerr.BizError {
	kafkaParams, bizErr := m.parseKafkaParams(downstreamParams)
	if bizErr != nil {
		return bizErr
	}

	if kafkaParams.GetAuthentication().GetPassword() != "" {
		if bizErr = m.encryptor.EncryptBase64Inplace(&kafkaParams.GetAuthentication().Password); bizErr != nil {
			return bizErr.Chain("encrypt password of authentication")
		}
	}
	if kafkaParams.GetDataFormat().GetSchemaRegistry().GetPassword() != "" {
		if bizErr = m.encryptor.EncryptBase64Inplace(
			&kafkaParams.GetDataFormat().GetSchemaRegistry().Password); bizErr != nil {
			return bizErr.Chain("encrypt password of schema registry")
		}
	}
	return nil
}

func (m *kafkaDownstreamAdaptor) AssembleChangefeedConfig(
	ctx context.Context, cluster *utils.Cluster, bizCf *entity.BizChangefeed,
) (*domain.TiCDCChangefeedConfig, bizerr.BizError) {
	kafkaParams, bizErr := m.parseKafkaParamsStr(bizCf.DownstreamParams)
	if bizErr != nil {
		return nil, bizErr
	}

	uriConfig, bizErr := m.assembleURIConfigData(ctx, kafkaParams, bizCf.NetworkType)
	if bizErr != nil {
		return nil, bizErr.Chain("assemble uri config")
	}
	changefeedConfig, bizErr := m.assembleChangefeedConfigData(ctx, kafkaParams)
	if bizErr != nil {
		return nil, bizErr.Chain("assemble changefeed config")
	}
	return &domain.TiCDCChangefeedConfig{
		URIConfig: uriConfig,
		Config:    changefeedConfig,
	}, nil
}

func (m *kafkaDownstreamAdaptor) assembleURIConfigData(
	ctx context.Context, kafkaParams *dataflow.KafkaInfo,
	networkType entity.ChangefeedDownstreamNetworkType,
) (*domain.TiCDCChangefeedKafkaURIConfigData, bizerr.BizError) {
	uriConfig := &domain.TiCDCChangefeedKafkaURIConfigData{
		BrokerEndpoints:   strings.ReplaceAll(kafkaParams.GetBroker().GetBrokerEndpoints(), " ", ""),
		EnableTLS:         kafkaParams.GetBroker().GetTlsEnable(),
		Protocol:          utils.TransferProtocol(kafkaParams.GetDataFormat().GetProtocol()),
		DefaultTopic:      kafkaParams.GetTopicPartitionConfig().GetDefaultTopic(),
		ReplicationFactor: kafkaParams.GetTopicPartitionConfig().GetReplicationFactor(),
		PartitionNum:      kafkaParams.GetTopicPartitionConfig().GetPartitionNum(),
	}
	kafkaVersion := utils.TransferKafkaVersion(kafkaParams.GetBroker().GetKafkaVersion())
	if kafkaVersion == "" {
		return nil, bizerr.New(
			bizerr.ErrCodeParams, "unknown kafka version %s", kafkaParams.GetBroker().GetKafkaVersion().String())
	}
	uriConfig.KafkaVersion = kafkaVersion
	compression := utils.TransferKafkaCompression(kafkaParams.GetBroker().GetCompression())
	if compression != "" {
		uriConfig.Compression = pointer.ToString(compression)
	}
	// Kafka client ID is optional: only forward it to the kernel when the caller
	// explicitly set it, otherwise let the kernel use its own default.
	if clientID := kafkaParams.GetBroker().GetKafkaClientId(); clientID != "" {
		uriConfig.KafkaClientID = pointer.ToString(clientID)
	}

	if bizErr := m.assembleURIAuthentication(uriConfig, kafkaParams.GetAuthentication()); bizErr != nil {
		return nil, bizErr
	}
	m.assembleURIDataFormat(uriConfig, kafkaParams.GetDataFormat())
	uriConfig.AutoCreateTopic = pointer.ToBool(true)
	if kafkaParams.GetBroker().GetTlsEnable() {
		if IsPrivateLink(networkType) || kafkaParams.GetBroker().InsecureSkipVerify {
			uriConfig.InsecureSkipVerify = pointer.ToBool(true)
		}
	}
	return uriConfig, nil
}

func (m *kafkaDownstreamAdaptor) assembleChangefeedConfigData(
	ctx context.Context, kafkaParams *dataflow.KafkaInfo,
) (*domain.TiCDCChangefeedConfigData, bizerr.BizError) {
	startTSO, bizErr := parseStartTSO(kafkaParams.GetStartPosition())
	if bizErr != nil {
		return nil, bizErr
	}

	changefeedConfig := &domain.TiCDCChangefeedConfigData{
		StartTs: &startTSO,
		ReplicaConfig: &clusterproxy.ReplicaConfig{
			CaseSensitive:  pointer.ToBool(kafkaParams.GetTableConfig().GetCaseSensitive()),
			EnableOldValue: pointer.ToBool(true),
			Sink: &clusterproxy.SinkConfigV2{
				DispatchRules:   buildTopicDispatchers(kafkaParams.GetTopicPartitionConfig()),
				ColumnSelectors: convertColumnSelectors(kafkaParams.GetColumnSelectors()),
				KafkaSink: &clusterproxy.KafkaConfig{
					OutputRawChangeEvent: pointer.ToBool(kafkaParams.GetDataFormat().OutputRawChangeEvent),
				},
			},
			Filter: &clusterproxy.FilterConfig{},
		},
	}

	assembleChangefeedSinkConfigTableConfig(kafkaParams.TableConfig, changefeedConfig.ReplicaConfig)
	if bizErr = m.assembleChangefeedSinkConfigDataFormat(
		kafkaParams.GetDataFormat(), changefeedConfig.ReplicaConfig.Sink); bizErr != nil {
		return nil, bizErr.Chain("assemble data format")
	}
	if utils.IsAvroProtocol(kafkaParams.GetDataFormat().GetProtocol()) {
		// avro protocol must force-replicate disable  https://github.com/pingcap/tiflow/pull/9479
		changefeedConfig.ReplicaConfig.ForceReplicate = pointer.ToBool(false)
	}
	return changefeedConfig, nil
}

func (m *kafkaDownstreamAdaptor) assembleChangefeedSinkConfigDataFormat(
	dataFormat *dataflow.DataFormat, sinkConfig *clusterproxy.SinkConfigV2,
) bizerr.BizError {
	sinkConfig.Protocol = utils.TransferProtocol(dataFormat.GetProtocol())
	switch dataFormat.GetProtocol() {
	case dataflow.Protocol_AVRO, dataflow.Protocol_DEBEZIUM_AVRO:
		if bizErr := m.assembleChangefeedSinkConfigAvro(dataFormat.GetSchemaRegistry(), sinkConfig); bizErr != nil {
			return bizErr.Chain("assemble avro config")
		}
	case dataflow.Protocol_DEBEZIUM:
		m.assembleChangefeedSinkConfigDebezium(dataFormat.GetDebeziumConfig(), sinkConfig)
	}
	return nil
}

func (m *kafkaDownstreamAdaptor) assembleChangefeedSinkConfigAvro(
	schemaRegistry *dataflow.SchemaRegistry, sinkConfig *clusterproxy.SinkConfigV2,
) bizerr.BizError {
	if schemaRegistry == nil {
		return bizerr.New(bizerr.ErrCodeParams, "schema registry is empty")
	}
	schemaRegistryUri, bizErr := m.assembleSchemaRegistryURI(schemaRegistry)
	if bizErr != nil {
		return bizErr
	}
	sinkConfig.SchemaRegistry = schemaRegistryUri
	return nil
}

func (m *kafkaDownstreamAdaptor) assembleSchemaRegistryURI(
	schemaRegistry *dataflow.SchemaRegistry,
) (string, bizerr.BizError) {
	schemaURL, err := url.Parse(schemaRegistry.GetSchemaRegistryEndpoints())
	if err != nil {
		return "", bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "parse schema registry endpoint failed")
	}
	if schemaURL.Host == "" {
		return "", bizerr.New(bizerr.ErrCodeParams, "schema registry endpoint is empty")
	}
	if schemaRegistry.GetEnableHttpAuth() {
		var password string
		var bizErr bizerr.BizError
		if schemaRegistry.GetPassword() != "" {
			password, bizErr = m.encryptor.DecryptBase64(schemaRegistry.GetPassword())
			if bizErr != nil {
				return "", bizerr.WrapError(err, "failed decrypt schema registry password")
			}
		}
		schemaURL.User = url.UserPassword(schemaRegistry.GetUserName(), password)
		return schemaURL.String(), nil
	}
	return schemaRegistry.GetSchemaRegistryEndpoints(), nil
}

func (m *kafkaDownstreamAdaptor) assembleChangefeedSinkConfigDebezium(
	debeziumConfig *dataflow.DebeziumConfig, sinkConfig *clusterproxy.SinkConfigV2,
) {
	sinkConfig.DebeziumConfig = &clusterproxy.DebeziumConfig{
		// TODO DebeziumDisableSchema
		OutputOldValue: pointer.ToBool(true),
	}
	if debeziumConfig != nil {
		// TODO DebeziumDisableSchema
		sinkConfig.DebeziumConfig.OutputOldValue = pointer.ToBool(debeziumConfig.OutputOldValue)
	}
}

func (m *kafkaDownstreamAdaptor) assembleURIAuthentication(
	uriConfig *domain.TiCDCChangefeedKafkaURIConfigData, kafkaAuth *dataflow.Authentication,
) bizerr.BizError {
	authType := utils.TransferKafkaAuthType(kafkaAuth.GetAuthType())
	if authType == "" {
		return nil
	}

	password := ""
	if kafkaAuth.GetPassword() != "" {
		decryptedPassword, bizErr := m.encryptor.DecryptBase64(kafkaAuth.GetPassword())
		if bizErr != nil {
			return bizErr.Chain("decrypt password")
		}
		password = decryptedPassword
	}

	uriConfig.SASLMechanism = pointer.ToString(authType)
	if kafkaAuth.GetAuthType() == dataflow.KafkaAuthType_SASL_PLAIN {
		uriConfig.SASLUser = pointer.ToString(url.QueryEscape(kafkaAuth.GetUserName()))
		uriConfig.SASLPassword = pointer.ToString(url.QueryEscape(password))
	} else {
		uriConfig.SASLUser = pointer.ToString(kafkaAuth.GetUserName())
		uriConfig.SASLPassword = pointer.ToString(password)
	}
	return nil
}

func (m *kafkaDownstreamAdaptor) assembleURIDataFormat(
	uriConfig *domain.TiCDCChangefeedKafkaURIConfigData, kafkaDataFormat *dataflow.DataFormat,
) {
	if utils.IsAvroProtocol(kafkaDataFormat.GetProtocol()) {
		uriConfig.AvroDecimalHandlingMode = pointer.ToString(
			utils.TransferDecimalHandlingMode(kafkaDataFormat.GetDecimalHandlingMode()))
		uriConfig.AvroBigintUnsignedHandlingMode = pointer.ToString(
			utils.TransferBigintUnsignedHandlingMode(kafkaDataFormat.GetBigintUnsignedHandlingMode()))
	}
	if kafkaDataFormat.GetProtocol() != dataflow.Protocol_OPEN_PROTOCOL {
		uriConfig.EnableTiDBExtension = pointer.ToBool(kafkaDataFormat.EnableTidbExtension)
	}
}

func (m *kafkaDownstreamAdaptor) CompleteEditParams(
	ctx context.Context, editDownstreamParams interface{}, currentDownstreamParams string,
) bizerr.BizError {
	editKafkaParams, bizErr := m.parseKafkaParams(editDownstreamParams)
	if bizErr != nil {
		return bizErr
	}
	currentKafkaParams, bizErr := m.parseKafkaParamsStr(currentDownstreamParams)
	if bizErr != nil {
		return bizErr
	}

	editKafkaParams.GetBroker().BrokerEndpoints = currentKafkaParams.GetBroker().GetBrokerEndpoints()
	if editKafkaParams.Authentication.Password == "" {
		editKafkaParams.Authentication.Password = currentKafkaParams.GetAuthentication().GetPassword()
	} else {
		if bizErr = m.encryptor.EncryptBase64Inplace(&editKafkaParams.Authentication.Password); bizErr != nil {
			return bizErr.Chain("encrypt password of authentication")
		}
	}
	schemaRegistry := editKafkaParams.GetDataFormat().GetSchemaRegistry()
	if schemaRegistry != nil {
		if schemaRegistry.Password == "" {
			schemaRegistry.Password = currentKafkaParams.GetDataFormat().GetSchemaRegistry().GetPassword()
		} else {
			if bizErr = m.encryptor.EncryptBase64Inplace(&schemaRegistry.Password); bizErr != nil {
				return bizErr.Chain("encrypt password of schema registry")
			}
		}
	}

	return nil
}

func (m *kafkaDownstreamAdaptor) parseKafkaParams(downstreamParams interface{}) (*dataflow.KafkaInfo, bizerr.BizError) {
	kafkaParams, ok := downstreamParams.(*dataflow.KafkaInfo)
	if !ok || kafkaParams == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "downstreamParams is not *dataflow.KafkaInfo type")
	}
	return kafkaParams, nil
}

func (m *kafkaDownstreamAdaptor) parseKafkaParamsStr(downstreamParams string) (*dataflow.KafkaInfo, bizerr.BizError) {
	decoder := json.NewDecoder(strings.NewReader(downstreamParams))
	decoder.DisallowUnknownFields()
	kafkaParams := new(dataflow.KafkaInfo)
	if err := decoder.Decode(kafkaParams); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "unmarshal kafka params error")
	}
	return kafkaParams, nil
}

// There are two things we need know before make changes to the topic dispatching logic
//
//  1. CDC kernel allow users to define multiple Dispatch rules, each rule can define its own matching tables
//     but in the cloud, we only allow user define one rule that apply to all tables.
//
//  2. There is a compatible issue of Dispatch Rule definition. Dispatch Rule has three fields:
//     a) matcher, defines the matching tables;
//     b) topic, defines which topic to dispatch the change data;
//     c) partition, defines how to partition topic
//     For the topic filed:
//     a) < v6.5.7, if we need to send change data to only one topic, we should leave topic filed empty;
//     if we need to send to different topics based on database or table, we should put topic expression in it.
//     b) >= v6.5.7, the topic filed can be static(only one topic) or dynamical(based on database or table), also compatible to empty.
//
// CDC kernel doc for topic dispatching  https://docs.pingcap.com/zh/tidb/stable/ticdc-sink-to-kafka#topic-%E5%88%86%E5%8F%91%E5%99%A8
func buildTopicDispatchers(topicConfig *dataflow.TopicPartitionConfig) []*clusterproxy.DispatchRuleV2 {
	var topicValue string
	dispatchType := topicConfig.DispatchType
	prefix := topicConfig.GetTopicPrefix()
	suffix := topicConfig.GetTopicSuffix()
	if dispatchType == dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC {
		topicValue = ""
	} else if dispatchType == dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE {
		// topic expression pattern: [prefix]{schema}[middle]{table}[suffix]
		// FIXME: both CDC kernel and frontend allow middle empty, but we set "_" as default before,
		// FIXME: just keep it as default for compatible. We can change frontend to make it consistent.
		middle := "_"
		if topicConfig.GetSeparator() != "" {
			middle = topicConfig.GetSeparator()
		}
		topicValue = fmt.Sprintf("%s{schema}%s{table}%s", prefix, middle, suffix)
	} else if dispatchType == dataflow.DispatchType_DISPATCH_TYPE_BY_DATABASE {
		// topic expression pattern: [prefix]{schema}[suffix]
		topicValue = fmt.Sprintf("%s{schema}%s", prefix, suffix)
	}

	var dispatchRules []*clusterproxy.DispatchRuleV2
	// use the new API for topic dispatching if partition dispatchers is not empty
	if len(topicConfig.GetPartitionDispatchers()) > 0 {
		for _, partitionCfg := range topicConfig.PartitionDispatchers {
			rule := &clusterproxy.DispatchRuleV2{
				Topic:     topicValue,
				Partition: utils.TransferPartitionDispatcher(partitionCfg.GetPartitionType()),
				Matcher:   partitionCfg.Matcher,
				IndexName: partitionCfg.IndexName,
				Columns:   partitionCfg.Columns,
			}
			dispatchRules = append(dispatchRules, rule)
		}
	} else {
		dispatchRule := &clusterproxy.DispatchRuleV2{
			Topic:     topicValue,
			Partition: utils.TransferPartitionDispatcher(topicConfig.GetPartitionDispatcher()),
			Matcher:   []string{"*.*"},
		}
		dispatchRules = append(dispatchRules, dispatchRule)
	}
	return dispatchRules
}

func convertColumnSelectors(columnSelectors []*dataflow.ColumnSelector) []*clusterproxy.ColumnSelector {
	cpColumnSelectors := make([]*clusterproxy.ColumnSelector, 0, len(columnSelectors))
	for _, columnSelector := range columnSelectors {
		cpColumnSelectors = append(cpColumnSelectors, &clusterproxy.ColumnSelector{
			Matcher: columnSelector.Matcher,
			Columns: columnSelector.Columns,
		})
	}
	return cpColumnSelectors
}

func IsPrivateLink(networkType entity.ChangefeedDownstreamNetworkType) bool {
	return networkType == entity.ChangefeedDownstreamNetworkTypePrivateLink
}