package changefeed

import (
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	changefeedopenbp "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/openapi/v1beta2"
)

func toOpenAPIPulsarDownstreamConfig(in *dataflow.PulsarInfo) *changefeedopenbp.PulsarDownstreamConfig {
	if in == nil {
		return nil
	}
	out := &changefeedopenbp.PulsarDownstreamConfig{
		DestinationProtocol:  toOpenAPIPulsarDestinationProtocol(in.GetDestinationProtocol()),
		Endpoint:             in.GetEndpoint(),
		Authentication:       toOpenAPIPulsarAuthentication(in.GetAuthentication()),
		Compression:          toOpenAPIPulsarCompression(in.GetCompressionType()),
		DataFormat:           toOpenAPIPulsarDataFormat(in.GetDataFormat()),
		TopicDistribution:    toOpenAPIPulsarTopicDistribution(in.GetTopicDistribution()),
		PartitionDispatchers: toOpenAPIPulsarPartitionDispatchers(in.GetPartitionDispatchers()),
		AdvancedSetting:      toOpenAPIPulsarAdvancedSetting(in.GetAdvancedSetting()),
	}
	return out
}

func toDataflowPulsarInfo(cf *changefeedopenbp.Changefeed) *dataflow.PulsarInfo {
	cfg := cf.GetPulsar()
	if cfg == nil {
		return nil
	}
	out := toDataflowPulsarConfig(cfg)
	out.TableConfig = toDataflowTableConfig(cf.GetTableConfig())
	out.StartPosition = toDataflowStartPosition(cf.GetStartPosition())
	return out
}

func toDataflowPulsarConfig(in *changefeedopenbp.PulsarDownstreamConfig) *dataflow.PulsarInfo {
	if in == nil {
		return nil
	}
	out := &dataflow.PulsarInfo{
		DestinationProtocol:  toDataflowPulsarDestinationProtocol(in.GetDestinationProtocol()),
		Endpoint:             in.GetEndpoint(),
		Authentication:       toDataflowPulsarAuthentication(in.GetAuthentication()),
		CompressionType:      toDataflowPulsarCompression(in.GetCompression()),
		DataFormat:           toDataflowPulsarDataFormat(in.GetDataFormat()),
		TopicDistribution:    toDataflowPulsarTopicDistribution(in.GetTopicDistribution()),
		PartitionDispatchers: toDataflowPulsarPartitionDispatchers(in.GetPartitionDispatchers()),
		AdvancedSetting:      toDataflowPulsarAdvancedSetting(in.GetAdvancedSetting()),
	}
	return out
}

func toOpenAPIPulsarPartitionDispatchers(
	in []*dataflow.PartitionDispatcherConfig,
) []*changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_PartitionDispatcherConfig {
	if len(in) == 0 {
		return nil
	}
	out := make(
		[]*changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_PartitionDispatcherConfig,
		0,
		len(in),
	)
	for _, dispatcher := range in {
		out = append(out, &changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_PartitionDispatcherConfig{
			PartitionType: toOpenAPIKafkaPartitionDispatcher(dispatcher.GetPartitionType()),
			Matcher:       append([]string(nil), dispatcher.GetMatcher()...),
			IndexName:     dispatcher.GetIndexName(),
			Columns:       append([]string(nil), dispatcher.GetColumns()...),
		})
	}
	return out
}

func toDataflowPulsarPartitionDispatchers(
	in []*changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_PartitionDispatcherConfig,
) []*dataflow.PartitionDispatcherConfig {
	if len(in) == 0 {
		return nil
	}
	out := make([]*dataflow.PartitionDispatcherConfig, 0, len(in))
	for _, dispatcher := range in {
		out = append(out, &dataflow.PartitionDispatcherConfig{
			PartitionType: toDataflowKafkaPartitionDispatcher(dispatcher.GetPartitionType()),
			Matcher:       append([]string(nil), dispatcher.GetMatcher()...),
			IndexName:     dispatcher.GetIndexName(),
			Columns:       append([]string(nil), dispatcher.GetColumns()...),
		})
	}
	return out
}

func toOpenAPIPulsarDestinationProtocol(
	in dataflow.PulsarDestinationProtocol,
) changefeedopenbp.PulsarDownstreamConfig_DestinationProtocol {
	switch in {
	case dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR_SSL:
		return changefeedopenbp.PulsarDownstreamConfig_PULSAR_SSL
	default:
		return changefeedopenbp.PulsarDownstreamConfig_PULSAR
	}
}

func toDataflowPulsarDestinationProtocol(
	in changefeedopenbp.PulsarDownstreamConfig_DestinationProtocol,
) dataflow.PulsarDestinationProtocol {
	switch in {
	case changefeedopenbp.PulsarDownstreamConfig_PULSAR_SSL:
		return dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR_SSL
	default:
		return dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR
	}
}

func toOpenAPIPulsarAuthentication(
	in *dataflow.PulsarAuthentication,
) *changefeedopenbp.PulsarDownstreamConfig_Authentication {
	if in == nil {
		return nil
	}
	authType := changefeedopenbp.PulsarDownstreamConfig_Authentication_BASIC
	if in.GetAuthType() == dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_TOKEN {
		authType = changefeedopenbp.PulsarDownstreamConfig_Authentication_TOKEN
	}
	return &changefeedopenbp.PulsarDownstreamConfig_Authentication{
		AuthType: authType,
		Username: in.GetUserName(),
	}
}

func toDataflowPulsarAuthentication(
	in *changefeedopenbp.PulsarDownstreamConfig_Authentication,
) *dataflow.PulsarAuthentication {
	if in == nil {
		return nil
	}
	authType := dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC
	if in.GetAuthType() == changefeedopenbp.PulsarDownstreamConfig_Authentication_TOKEN {
		authType = dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_TOKEN
	}
	return &dataflow.PulsarAuthentication{
		AuthType: authType,
		UserName: in.GetUsername(),
		Password: in.GetPassword(),
		Token:    in.GetToken(),
	}
}

func toOpenAPIPulsarCompression(
	in dataflow.PulsarCompressionType,
) changefeedopenbp.PulsarDownstreamConfig_CompressionType {
	switch in {
	case dataflow.PulsarCompressionType_PULSAR_COMPRESSION_LZ4:
		return changefeedopenbp.PulsarDownstreamConfig_LZ4
	case dataflow.PulsarCompressionType_PULSAR_COMPRESSION_ZLIB:
		return changefeedopenbp.PulsarDownstreamConfig_ZLIB
	case dataflow.PulsarCompressionType_PULSAR_COMPRESSION_ZSTD:
		return changefeedopenbp.PulsarDownstreamConfig_ZSTD
	default:
		return changefeedopenbp.PulsarDownstreamConfig_NONE
	}
}

func toDataflowPulsarCompression(
	in changefeedopenbp.PulsarDownstreamConfig_CompressionType,
) dataflow.PulsarCompressionType {
	switch in {
	case changefeedopenbp.PulsarDownstreamConfig_LZ4:
		return dataflow.PulsarCompressionType_PULSAR_COMPRESSION_LZ4
	case changefeedopenbp.PulsarDownstreamConfig_ZLIB:
		return dataflow.PulsarCompressionType_PULSAR_COMPRESSION_ZLIB
	case changefeedopenbp.PulsarDownstreamConfig_ZSTD:
		return dataflow.PulsarCompressionType_PULSAR_COMPRESSION_ZSTD
	default:
		return dataflow.PulsarCompressionType_PULSAR_COMPRESSION_NONE
	}
}

func toOpenAPIPulsarDataFormat(in *dataflow.PulsarDataFormat) *changefeedopenbp.PulsarDownstreamConfig_DataFormat {
	if in == nil {
		return nil
	}
	return &changefeedopenbp.PulsarDownstreamConfig_DataFormat{
		Protocol:            changefeedopenbp.PulsarDownstreamConfig_DataFormat_CANAL_JSON,
		EnableTidbExtension: in.GetEnableTidbExtension(),
	}
}

func toDataflowPulsarDataFormat(in *changefeedopenbp.PulsarDownstreamConfig_DataFormat) *dataflow.PulsarDataFormat {
	if in == nil {
		return nil
	}
	return &dataflow.PulsarDataFormat{
		Protocol:            dataflow.Protocol_CANAL_JSON,
		EnableTidbExtension: in.GetEnableTidbExtension(),
	}
}

func toOpenAPIPulsarTopicDistribution(
	in *dataflow.PulsarTopicDistribution,
) *changefeedopenbp.PulsarDownstreamConfig_TopicDistribution {
	if in == nil {
		return nil
	}
	return &changefeedopenbp.PulsarDownstreamConfig_TopicDistribution{
		Tenant:       in.GetPulsarTenant(),
		Namespace:    in.GetPulsarNamespace(),
		DefaultTopic: in.GetDefaultTopic(),
		DispatchType: toOpenAPIKafkaTopicDispatchType(in.GetDispatchType()),
		TopicPrefix:  in.GetTopicPrefix(),
		Separator:    in.GetSeparator(),
		TopicSuffix:  in.GetTopicSuffix(),
	}
}

func toDataflowPulsarTopicDistribution(
	in *changefeedopenbp.PulsarDownstreamConfig_TopicDistribution,
) *dataflow.PulsarTopicDistribution {
	if in == nil {
		return nil
	}
	return &dataflow.PulsarTopicDistribution{
		PulsarTenant:    in.GetTenant(),
		PulsarNamespace: in.GetNamespace(),
		DefaultTopic:    in.GetDefaultTopic(),
		DispatchType:    toDataflowKafkaDispatchType(in.GetDispatchType()),
		TopicPrefix:     in.GetTopicPrefix(),
		Separator:       in.GetSeparator(),
		TopicSuffix:     in.GetTopicSuffix(),
	}
}

func toOpenAPIPulsarAdvancedSetting(
	in *dataflow.PulsarAdvancedSetting,
) *changefeedopenbp.PulsarDownstreamConfig_AdvancedSetting {
	if in == nil {
		return nil
	}
	return &changefeedopenbp.PulsarDownstreamConfig_AdvancedSetting{
		BatchingMaxMessages:     in.GetBatchingMaxMessages(),
		BatchingMaxPublishDelay: in.GetBatchingMaxPublishDelay(),
		ConnectionTimeout:       in.GetConnectionTimeout(),
		OperationTimeout:        in.GetOperationTimeout(),
		SendTimeout:             in.GetSendTimeout(),
	}
}

func toDataflowPulsarAdvancedSetting(
	in *changefeedopenbp.PulsarDownstreamConfig_AdvancedSetting,
) *dataflow.PulsarAdvancedSetting {
	if in == nil {
		return nil
	}
	return &dataflow.PulsarAdvancedSetting{
		BatchingMaxMessages:     in.GetBatchingMaxMessages(),
		BatchingMaxPublishDelay: in.GetBatchingMaxPublishDelay(),
		ConnectionTimeout:       in.GetConnectionTimeout(),
		OperationTimeout:        in.GetOperationTimeout(),
		SendTimeout:             in.GetSendTimeout(),
	}
}