package changefeed

import (
	"fmt"
	"strconv"
	"strings"

	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	changefeedopenbp "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/openapi/v1beta2"
	"github.com/tidbcloud/pkgv2/bizerr"
	"google.golang.org/protobuf/proto"
)

// Mapping from internal consoleChangefeed status to public-facing changefeed status.
var consoleStatus2PublicState = map[dataflow.Changefeed_Status]changefeedopenbp.Changefeed_State{
	dataflow.Changefeed_RUNNING:  changefeedopenbp.Changefeed_RUNNING,
	dataflow.Changefeed_FAILED:   changefeedopenbp.Changefeed_FAILED,
	dataflow.Changefeed_ERROR:    changefeedopenbp.Changefeed_ERROR,
	dataflow.Changefeed_CREATING: changefeedopenbp.Changefeed_CREATING,
	dataflow.Changefeed_PAUSING:  changefeedopenbp.Changefeed_PAUSING,
	dataflow.Changefeed_PAUSED:   changefeedopenbp.Changefeed_PAUSED,
	dataflow.Changefeed_DELETING: changefeedopenbp.Changefeed_DELETING,
	dataflow.Changefeed_DELETED:  changefeedopenbp.Changefeed_DELETED,
	dataflow.Changefeed_EDITING:  changefeedopenbp.Changefeed_EDITING,
	dataflow.Changefeed_WARNING:  changefeedopenbp.Changefeed_WARNING,
	dataflow.Changefeed_SCALING:  changefeedopenbp.Changefeed_SCALING,
	dataflow.Changefeed_RESUMING: changefeedopenbp.Changefeed_RESUMING,
}

func assert(flag bool, f string, args ...interface{}) {
	if !flag {
		panic(fmt.Errorf(f, args...))
	}
}

// convertConsoleToPublicChangefeed transforms an internal Changefeed object into a public Changefeed object.
func convertConsoleToPublicChangefeed(consoleChangefeed *dataflow.Changefeed) *changefeedopenbp.Changefeed {
	req := consoleChangefeed.GetRequest()
	assert(req != nil, "consoleChangefeed request should not be nil")

	out := &changefeedopenbp.Changefeed{
		Id:            strconv.FormatUint(consoleChangefeed.GetChangefeedId(), 10),
		TidbId:        strconv.FormatUint(req.GetClusterId(), 10),
		Name:          consoleChangefeed.Name,
		State:         consoleStatus2PublicState[consoleChangefeed.Status],
		CreateTime:    consoleChangefeed.CreatedAt,
		CheckpointTso: consoleChangefeed.CheckpointTso,

		NetworkInfo:        toOpenAPINetworkInfo(req.GetNetworkInfo()),
		CheckpointTs:       consoleChangefeed.GetCheckpointUtc(),
		ChangefeedCapacity: toOpenAPICCU(consoleChangefeed.GetChangefeedSpec()),
		DownstreamType:     downstreamTypeFromConsoleType(consoleChangefeed.GetChangefeedType()),
		TableConfig:        toOpenAPITableConfig(req, consoleChangefeed.GetChangefeedType()),
		StartPosition:      toOpenAPIStartPosition(req, consoleChangefeed.GetChangefeedType()),
		DownstreamConfig:   nil,
		Error:              consoleChangefeed.GetError(),
	}
	setOpenAPIDownstreamConfig(out, req, consoleChangefeed.GetChangefeedType())

	return out
}

// convertChangefeedToOpenAPIFilters extracts event filter rules from a Changefeed and converts them to the public API format.
func convertChangefeedToOpenAPIFilters(consoleChangefeed *dataflow.Changefeed) (*changefeedopenbp.EventFilters, error) {
	rules, err := getEventFilterRules(consoleChangefeed)
	if err != nil {
		return nil, err
	}

	response := &changefeedopenbp.EventFilters{
		EventFilters: make([]*changefeedopenbp.EventFilter, 0, len(rules)),
	}

	for _, rule := range rules {
		response.EventFilters = append(response.EventFilters, &changefeedopenbp.EventFilter{
			TableMatchers:                   rule.Matcher,
			IgnoredEvents:                   rule.IgnoreEvent,
			IgnoredSqlStatements:            rule.IgnoreSql,
			IgnoredInsertValueExpression:    rule.IgnoreInsertValueExpr,
			IgnoredUpdateOldValueExpression: rule.IgnoreUpdateOldValueExpr,
			IgnoredUpdateNewValueExpression: rule.IgnoreUpdateNewValueExpr,
			IgnoredDeleteValueExpression:    rule.IgnoreDeleteValueExpr,
		})
	}
	return response, nil
}

func toDataflowTableConfig(in *changefeedopenbp.TableConfig) *dataflow.TableConfig {
	if in == nil {
		return nil
	}

	var eventFilterRules []*dataflow.EventFilterRule
	if filters := in.GetEventFilters(); len(filters) > 0 {
		eventFilterRules = convertAPIFiltersToInternal(filters)
	}

	return &dataflow.TableConfig{
		FilterRule:      append([]string(nil), in.GetFilterRules()...),
		Mode:            toDataflowTableMode(in.GetMode()),
		CaseSensitive:   in.GetCaseSensitive(),
		EventFilterRule: eventFilterRules,
		BdrMode:         in.BdrMode,
	}
}

func toDataflowTableMode(in changefeedopenbp.TableConfig_Mode) dataflow.TableMode {
	switch in {
	case changefeedopenbp.TableConfig_IGNORE_NOT_SUPPORT_TABLE:
		return dataflow.TableMode_IGNORE_NOT_SUPPORT_TABLE
	case changefeedopenbp.TableConfig_FORCE_SYNC:
		return dataflow.TableMode_FORCE_SYNC
	default:
		return dataflow.TableMode_IGNORE_NOT_SUPPORT_TABLE
	}
}

func toDataflowStartPosition(in *changefeedopenbp.StartPosition) *dataflow.StartPosition {
	if in == nil {
		return nil
	}

	out := &dataflow.StartPosition{}
	switch in.GetMode() {
	case changefeedopenbp.StartPosition_FROM_TSO:
		out.Mode = dataflow.StartMode_FROM_TSO
		out.Tso = in.GetStartTso()
	case changefeedopenbp.StartPosition_FROM_UTC:
		out.Mode = dataflow.StartMode_FROM_UTC
		out.Utc = in.GetStartTimestamp()
	default:
		out.Mode = dataflow.StartMode_FROM_NOW
	}
	return out
}

// newChangefeedEditRequest creates a request to edit a consoleChangefeed by applying a new set of event filters.
// It creates a deep copy of the consoleChangefeed's configuration to avoid side effects.
func newChangefeedEditRequest(consoleChangefeed *dataflow.Changefeed,
	apiFilters []*changefeedopenbp.EventFilter,
) (*dataflow.ChangefeedEditReq, error) {
	if !isEventFilterSupported(consoleChangefeed.ChangefeedType) {
		return nil, bizerr.New(bizerr.ErrCodeParams, "this changefeed does not support event filters")
	}

	// Create a deep copy of the request object to prevent modifying the input parameter `consoleChangefeed`.
	reqCopy := proto.Clone(consoleChangefeed.GetRequest()).(*dataflow.ChangefeedCreateReq)

	// Convert public API filters to the internal format.
	internalFilterRules := convertAPIFiltersToInternal(apiFilters)

	// Set the new filter rules on the copied request object.
	setEventFilterRules(reqCopy, consoleChangefeed.ChangefeedType, internalFilterRules)

	// Build the final edit request.
	editReq := &dataflow.ChangefeedEditReq{
		ClusterId:      reqCopy.ClusterId,
		ChangefeedId:   consoleChangefeed.ChangefeedId,
		ChangefeedType: consoleChangefeed.ChangefeedType,
		Name:           consoleChangefeed.Name,
		KafkaInfo:      reqCopy.KafkaInfo,
		MysqlInfo:      reqCopy.MysqlInfo,
		S3Info:         reqCopy.S3Info,
		GcsInfo:        reqCopy.GcsInfo,
		PulsarInfo:     reqCopy.PulsarInfo,
	}

	return editReq, nil
}

// --- Helper Functions for Event Filter Logic ---

// getTableConfigHolder is a helper to retrieve an interface that can access the TableConfig.
// This reduces code duplication by abstracting the specific sink type.
func getTableConfigHolder(req *dataflow.ChangefeedCreateReq,
	cType dataflow.ChangefeedType,
) (interface{ GetTableConfig() *dataflow.TableConfig }, error) {
	switch cType {
	case dataflow.ChangefeedType_KAFKA:
		return req.GetKafkaInfo(), nil
	case dataflow.ChangefeedType_TIDB, dataflow.ChangefeedType_MYSQL:
		return req.GetMysqlInfo(), nil
	case dataflow.ChangefeedType_S3:
		return req.GetS3Info(), nil
	case dataflow.ChangefeedType_GCS:
		return req.GetGcsInfo(), nil
	case dataflow.ChangefeedType_PULSAR:
		return req.GetPulsarInfo(), nil
	default:
		return nil, bizerr.New(bizerr.ErrCodeParams, "this changefeed does not support event filters")
	}
}

// getEventFilterRules extracts the filter rules from a consoleChangefeed using a helper.
func getEventFilterRules(consoleChangefeed *dataflow.Changefeed) ([]*dataflow.EventFilterRule, error) {
	holder, err := getTableConfigHolder(consoleChangefeed.GetRequest(), consoleChangefeed.ChangefeedType)
	if err != nil {
		return nil, err
	}
	if holder == nil || holder.GetTableConfig() == nil {
		return nil, nil // No config, so no rules.
	}
	return holder.GetTableConfig().EventFilterRule, nil
}

// setEventFilterRules sets the filter rules on the correct sink info object within the create request.
func setEventFilterRules(req *dataflow.ChangefeedCreateReq,
	cType dataflow.ChangefeedType, rules []*dataflow.EventFilterRule,
) {
	holder, err := getTableConfigHolder(req, cType)
	if err != nil {
		// This should not happen if checked before, but handle defensively.
		panic(fmt.Sprintf("setEventFilterRules called with unsupported type: %v", cType))
	}
	if holder.GetTableConfig() == nil {
		// If TableConfig is nil, we can't set rules. This indicates a programming error or inconsistent state.
		// Depending on requirements, one might need to initialize the TableConfig struct here.
		return
	}
	holder.GetTableConfig().EventFilterRule = rules
}

// isEventFilterSupported checks if a given consoleChangefeed type supports event filtering.
func isEventFilterSupported(cType dataflow.ChangefeedType) bool {
	switch cType {
	case dataflow.ChangefeedType_KAFKA,
		dataflow.ChangefeedType_TIDB, dataflow.ChangefeedType_MYSQL,
		dataflow.ChangefeedType_S3,
		dataflow.ChangefeedType_GCS,
		dataflow.ChangefeedType_PULSAR:
		return true
	default:
		return false
	}
}

// convertAPIFiltersToInternal converts a slice of public API filters to the internal format.
func convertAPIFiltersToInternal(filters []*changefeedopenbp.EventFilter) []*dataflow.EventFilterRule {
	internalRules := make([]*dataflow.EventFilterRule, 0, len(filters))
	for _, filter := range filters {
		internalRules = append(internalRules, &dataflow.EventFilterRule{
			Matcher:                  filter.TableMatchers,
			IgnoreEvent:              filter.IgnoredEvents,
			IgnoreSql:                filter.IgnoredSqlStatements,
			IgnoreInsertValueExpr:    filter.IgnoredInsertValueExpression,
			IgnoreUpdateNewValueExpr: filter.IgnoredUpdateNewValueExpression,
			IgnoreUpdateOldValueExpr: filter.IgnoredUpdateOldValueExpression,
			IgnoreDeleteValueExpr:    filter.IgnoredDeleteValueExpression,
		})
	}
	return internalRules
}

func toOpenAPINetworkInfo(in *dataflow.NetworkInfo) *changefeedopenbp.NetworkInfo {
	if in == nil {
		return nil
	}
	out := &changefeedopenbp.NetworkInfo{
		NetworkType: toOpenAPINetworkType(in.GetNetworkType()),
		Ports:       append([]uint32(nil), in.GetPorts()...),
	}
	if sinkEndpointID := in.GetSinkEndpointId(); sinkEndpointID != 0 {
		out.SinkEndpointId = &sinkEndpointID
	}
	return out
}

func toOpenAPINetworkType(in dataflow.NetworkType) changefeedopenbp.NetworkInfo_NetworkType {
	switch in {
	case dataflow.NetworkType_NetworkType_Public:
		return changefeedopenbp.NetworkInfo_NETWORK_TYPE_PUBLIC
	case dataflow.NetworkType_NetworkType_VPC_Peering:
		return changefeedopenbp.NetworkInfo_NETWORK_TYPE_VPC_PEERING
	case dataflow.NetworkType_NetworkType_Private_Link:
		return changefeedopenbp.NetworkInfo_NETWORK_TYPE_PRIVATE_LINK
	default:
		return changefeedopenbp.NetworkInfo_NETWORK_TYPE_UNSPECIFIED
	}
}

func toOpenAPITableConfig(req *dataflow.ChangefeedCreateReq, consoleType dataflow.ChangefeedType) *changefeedopenbp.TableConfig {
	internal := sinkTableConfig(req, consoleType)
	if internal == nil {
		return nil
	}

	out := &changefeedopenbp.TableConfig{
		FilterRules:   append([]string(nil), internal.GetFilterRule()...),
		Mode:          toOpenAPITableMode(internal.GetMode()),
		CaseSensitive: internal.GetCaseSensitive(),
		EventFilters:  nil, // fill it later
		BdrMode:       internal.BdrMode,
	}
	if rules := internal.GetEventFilterRule(); len(rules) > 0 {
		out.EventFilters = make([]*changefeedopenbp.EventFilter, 0, len(rules))
		for _, rule := range rules {
			out.EventFilters = append(out.EventFilters, &changefeedopenbp.EventFilter{
				TableMatchers:                   append([]string(nil), rule.GetMatcher()...),
				IgnoredEvents:                   append([]string(nil), rule.GetIgnoreEvent()...),
				IgnoredSqlStatements:            append([]string(nil), rule.GetIgnoreSql()...),
				IgnoredInsertValueExpression:    rule.GetIgnoreInsertValueExpr(),
				IgnoredUpdateNewValueExpression: rule.GetIgnoreUpdateNewValueExpr(),
				IgnoredUpdateOldValueExpression: rule.GetIgnoreUpdateOldValueExpr(),
				IgnoredDeleteValueExpression:    rule.GetIgnoreDeleteValueExpr(),
			})
		}
	}
	return out
}

func toOpenAPITableMode(in dataflow.TableMode) changefeedopenbp.TableConfig_Mode {
	switch in {
	case dataflow.TableMode_FORCE_SYNC:
		return changefeedopenbp.TableConfig_FORCE_SYNC
	case dataflow.TableMode_IGNORE_NOT_SUPPORT_TABLE:
		fallthrough
	default:
		return changefeedopenbp.TableConfig_IGNORE_NOT_SUPPORT_TABLE
	}
}

func toOpenAPIStartPosition(req *dataflow.ChangefeedCreateReq, consoleType dataflow.ChangefeedType) *changefeedopenbp.StartPosition {
	internal := sinkStartPosition(req, consoleType)
	if internal == nil {
		return nil
	}

	out := &changefeedopenbp.StartPosition{Mode: toOpenAPIStartMode(internal.GetMode())}
	switch internal.GetMode() {
	case dataflow.StartMode_FROM_TSO:
		out.Position = &changefeedopenbp.StartPosition_StartTso{StartTso: internal.GetTso()}
	case dataflow.StartMode_FROM_UTC:
		out.Position = &changefeedopenbp.StartPosition_StartTimestamp{StartTimestamp: internal.GetUtc()}
	}
	return out
}

func toOpenAPIStartMode(in dataflow.StartMode) changefeedopenbp.StartPosition_Mode {
	switch in {
	case dataflow.StartMode_FROM_TSO:
		return changefeedopenbp.StartPosition_FROM_TSO
	case dataflow.StartMode_FROM_UTC:
		return changefeedopenbp.StartPosition_FROM_UTC
	case dataflow.StartMode_FROM_NOW:
		fallthrough
	default:
		return changefeedopenbp.StartPosition_FROM_NOW
	}
}

func setOpenAPIDownstreamConfig(out *changefeedopenbp.Changefeed, req *dataflow.ChangefeedCreateReq, consoleType dataflow.ChangefeedType) {
	switch consoleType {
	case dataflow.ChangefeedType_KAFKA:
		out.DownstreamConfig = &changefeedopenbp.Changefeed_Kafka{Kafka: toOpenAPIKafkaDownstreamConfig(req.GetKafkaInfo())}
	case dataflow.ChangefeedType_MYSQL, dataflow.ChangefeedType_TIDB:
		out.DownstreamConfig = &changefeedopenbp.Changefeed_Mysql{Mysql: toOpenAPIMySQLDownstreamConfig(effectiveMySQLInfo(req, consoleType))}
	case dataflow.ChangefeedType_S3:
		out.DownstreamConfig = &changefeedopenbp.Changefeed_S3{S3: toOpenAPIS3DownstreamConfig(req.GetS3Info())}
	case dataflow.ChangefeedType_GCS:
		out.DownstreamConfig = &changefeedopenbp.Changefeed_Gcs{Gcs: toOpenAPIGcsDownstreamConfig(req.GetGcsInfo())}
	case dataflow.ChangefeedType_PULSAR:
		out.DownstreamConfig = &changefeedopenbp.Changefeed_Pulsar{Pulsar: toOpenAPIPulsarDownstreamConfig(req.GetPulsarInfo())}
	}
}

func toOpenAPIKafkaDownstreamConfig(in *dataflow.KafkaInfo) *changefeedopenbp.KafkaDownstreamConfig {
	if in == nil {
		return nil
	}

	out := &changefeedopenbp.KafkaDownstreamConfig{
		Broker:               toOpenAPIKafkaBroker(in.GetBroker()),
		Authentication:       toOpenAPIKafkaAuthentication(in.GetAuthentication()),
		DataFormat:           toOpenAPIKafkaDataFormat(in.GetDataFormat()),
		TopicPartitionConfig: toOpenAPIKafkaTopicPartitionConfig(in.GetTopicPartitionConfig()),
		ColumnSelectors:      nil, // fill it later
	}
	if selectors := in.GetColumnSelectors(); len(selectors) > 0 {
		out.ColumnSelectors = make([]*changefeedopenbp.KafkaDownstreamConfig_ColumnSelector, 0, len(selectors))
		for _, selector := range selectors {
			out.ColumnSelectors = append(out.ColumnSelectors, &changefeedopenbp.KafkaDownstreamConfig_ColumnSelector{
				Matcher: append([]string(nil), selector.GetMatcher()...),
				Columns: append([]string(nil), selector.GetColumns()...),
			})
		}
	}
	return out
}

func toOpenAPIKafkaBroker(in *dataflow.Broker) *changefeedopenbp.KafkaDownstreamConfig_Broker {
	if in == nil {
		return nil
	}
	return &changefeedopenbp.KafkaDownstreamConfig_Broker{
		Version:            toOpenAPIKafkaVersion(in.GetKafkaVersion()),
		BrokerEndpoints:    in.GetBrokerEndpoints(),
		UseTls:             in.GetTlsEnable(),
		InsecureSkipVerify: in.GetInsecureSkipVerify(),
		Compression:        toOpenAPIKafkaCompression(in.GetCompression()),
		KafkaClientId:      in.GetKafkaClientId(),
	}
}

func toOpenAPIKafkaAuthentication(in *dataflow.Authentication) *changefeedopenbp.KafkaDownstreamConfig_Authentication {
	if in == nil {
		return nil
	}
	username := in.GetUserName()
	return &changefeedopenbp.KafkaDownstreamConfig_Authentication{
		AuthType: toOpenAPIKafkaAuthType(in.GetAuthType()),
		Username: stringPtr(username),
		Password: nil, // redact sentitive info
	}
}

func toOpenAPIKafkaDataFormat(in *dataflow.DataFormat) *changefeedopenbp.KafkaDownstreamConfig_DataFormat {
	if in == nil {
		return nil
	}

	out := &changefeedopenbp.KafkaDownstreamConfig_DataFormat{
		Protocol:             toOpenAPIKafkaProtocol(in.GetProtocol()),
		EnableTidbExtension:  in.GetEnableTidbExtension(),
		OutputRawChangeEvent: in.GetOutputRawChangeEvent(),
		Config:               nil, // fill it later based on protocol or data format
	}
	if avro := in.GetSchemaRegistry(); avro != nil || in.GetProtocol() == dataflow.Protocol_AVRO {
		out.Config = &changefeedopenbp.KafkaDownstreamConfig_DataFormat_AvroConfig_{
			AvroConfig: &changefeedopenbp.KafkaDownstreamConfig_DataFormat_AvroConfig{
				DecimalHandlingMode:        toOpenAPIKafkaAvroDecimalHandlingMode(in.GetDecimalHandlingMode()),
				BigintUnsignedHandlingMode: toOpenAPIKafkaAvroBigintHandlingMode(in.GetBigintUnsignedHandlingMode()),
				SchemaRegistry:             toOpenAPIKafkaAvroSchemaRegistry(in.GetSchemaRegistry()),
			},
		}
	}
	if debezium := in.GetDebeziumConfig(); debezium != nil || in.GetProtocol() == dataflow.Protocol_DEBEZIUM {
		out.Config = &changefeedopenbp.KafkaDownstreamConfig_DataFormat_DebeziumConfig_{
			DebeziumConfig: &changefeedopenbp.KafkaDownstreamConfig_DataFormat_DebeziumConfig{
				OutputOldValue: debezium.GetOutputOldValue(),
				DisableSchema:  debezium.GetDisableSchema(),
			},
		}
	}
	return out
}

func toOpenAPIKafkaAvroSchemaRegistry(in *dataflow.SchemaRegistry) *changefeedopenbp.KafkaDownstreamConfig_DataFormat_AvroConfig_SchemaRegistry {
	if in == nil {
		return nil
	}
	return &changefeedopenbp.KafkaDownstreamConfig_DataFormat_AvroConfig_SchemaRegistry{
		SchemaRegistryEndpoints: in.GetSchemaRegistryEndpoints(),
		EnableBasicAuth:         in.GetEnableHttpAuth(),
		BasicAuthUsername:       in.GetUserName(),
		BasicAuthPassword:       "", // redact sentitive info
	}
}

func toOpenAPIKafkaTopicPartitionConfig(in *dataflow.TopicPartitionConfig) *changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig {
	if in == nil {
		return nil
	}
	out := &changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig{
		DispatchType:         toOpenAPIKafkaTopicDispatchType(in.GetDispatchType()),
		DefaultTopic:         in.GetDefaultTopic(),
		ReplicationFactor:    in.GetReplicationFactor(),
		PartitionNum:         in.GetPartitionNum(),
		TopicPrefix:          nil, // fill it later
		Separator:            nil, // fill it later
		TopicSuffix:          nil, // fill it later
		PartitionDispatchers: nil, // fill it later
	}
	if prefix := in.GetTopicPrefix(); prefix != "" {
		out.TopicPrefix = stringPtr(prefix)
	}
	if separator := in.GetSeparator(); separator != "" {
		out.Separator = stringPtr(separator)
	}
	if suffix := in.GetTopicSuffix(); suffix != "" {
		out.TopicSuffix = stringPtr(suffix)
	}
	if dispatchers := in.GetPartitionDispatchers(); len(dispatchers) > 0 {
		out.PartitionDispatchers = make([]*changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_PartitionDispatcherConfig, 0, len(dispatchers))
		for _, dispatcher := range dispatchers {
			out.PartitionDispatchers = append(out.PartitionDispatchers, &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 toOpenAPIKafkaTopicDispatchType(in dataflow.DispatchType) changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_DispatchType {
	switch in {
	case dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE:
		return changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_DISPATCH_TYPE_BY_TABLE
	case dataflow.DispatchType_DISPATCH_TYPE_BY_DATABASE:
		return changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_DISPATCH_TYPE_BY_DATABASE
	case dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC:
		fallthrough
	default:
		return changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_DISPATCH_TYPE_ONE_TOPIC
	}
}

func toOpenAPIKafkaPartitionDispatcher(in dataflow.PartitionDispatcher) changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_PartitionDispatcher {
	switch in {
	case dataflow.PartitionDispatcher_INDEX_VALUE:
		return changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_INDEX_VALUE
	case dataflow.PartitionDispatcher_TS:
		return changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_TIMESTAMP
	case dataflow.PartitionDispatcher_COLUMNS:
		return changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_COLUMNS
	case dataflow.PartitionDispatcher_TABLE:
		fallthrough
	default:
		return changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_TABLE
	}
}

func toOpenAPIKafkaVersion(in dataflow.KafkaVersion) changefeedopenbp.KafkaDownstreamConfig_KafkaVersion {
	switch in {
	case dataflow.KafkaVersion_KAFKA_VERSION_1XX:
		return changefeedopenbp.KafkaDownstreamConfig_KAFKA_VERSION_1XX
	case dataflow.KafkaVersion_KAFKA_VERSION_2XX:
		return changefeedopenbp.KafkaDownstreamConfig_KAFKA_VERSION_2XX
	case dataflow.KafkaVersion_KAFKA_VERSION_3XX:
		return changefeedopenbp.KafkaDownstreamConfig_KAFKA_VERSION_3XX
	case dataflow.KafkaVersion_KAFKA_VERSION_0XX:
		fallthrough
	default:
		return changefeedopenbp.KafkaDownstreamConfig_KAFKA_VERSION_UNSPECIFIED
	}
}

func toOpenAPIKafkaCompression(in dataflow.CompressionType) changefeedopenbp.KafkaDownstreamConfig_CompressionType {
	switch in {
	case dataflow.CompressionType_GZIP:
		return changefeedopenbp.KafkaDownstreamConfig_GZIP
	case dataflow.CompressionType_SNAPPY:
		return changefeedopenbp.KafkaDownstreamConfig_SNAPPY
	case dataflow.CompressionType_LZ4:
		return changefeedopenbp.KafkaDownstreamConfig_LZ4
	case dataflow.CompressionType_ZSTD:
		return changefeedopenbp.KafkaDownstreamConfig_ZSTD
	case dataflow.CompressionType_NONE:
		fallthrough
	default:
		return changefeedopenbp.KafkaDownstreamConfig_NONE
	}
}

func toOpenAPIKafkaAuthType(in dataflow.KafkaAuthType) changefeedopenbp.KafkaDownstreamConfig_Authentication_AuthType {
	switch in {
	case dataflow.KafkaAuthType_SASL_PLAIN:
		return changefeedopenbp.KafkaDownstreamConfig_Authentication_SASL_PLAIN
	case dataflow.KafkaAuthType_SASL_SCRAM_SHA_256:
		return changefeedopenbp.KafkaDownstreamConfig_Authentication_SASL_SCRAM_SHA_256
	case dataflow.KafkaAuthType_SASL_SCRAM_SHA_512:
		return changefeedopenbp.KafkaDownstreamConfig_Authentication_SASL_SCRAM_SHA_512
	case dataflow.KafkaAuthType_DISABLE:
		fallthrough
	default:
		return changefeedopenbp.KafkaDownstreamConfig_Authentication_DISABLE
	}
}

func toOpenAPIKafkaProtocol(in dataflow.Protocol) changefeedopenbp.KafkaDownstreamConfig_DataFormat_Protocol {
	switch in {
	case dataflow.Protocol_OPEN_PROTOCOL:
		return changefeedopenbp.KafkaDownstreamConfig_DataFormat_PROTOCOL_OPEN_PROTOCOL
	case dataflow.Protocol_AVRO:
		return changefeedopenbp.KafkaDownstreamConfig_DataFormat_PROTOCOL_AVRO
	case dataflow.Protocol_DEBEZIUM:
		return changefeedopenbp.KafkaDownstreamConfig_DataFormat_PROTOCOL_DEBEZIUM
	case dataflow.Protocol_DEBEZIUM_AVRO:
		return changefeedopenbp.KafkaDownstreamConfig_DataFormat_PROTOCOL_DEBEZIUM_AVRO
	case dataflow.Protocol_CANAL_JSON:
		fallthrough
	default:
		return changefeedopenbp.KafkaDownstreamConfig_DataFormat_PROTOCOL_CANAL_JSON
	}
}

func toOpenAPIKafkaAvroDecimalHandlingMode(in dataflow.DecimalHandlingMode) changefeedopenbp.KafkaDownstreamConfig_DataFormat_AvroConfig_DecimalHandlingMode {
	switch in {
	case dataflow.DecimalHandlingMode_DECIMAL_HANDLING_MODE_STRING:
		return changefeedopenbp.KafkaDownstreamConfig_DataFormat_AvroConfig_DECIMAL_HANDLING_MODE_STRING
	case dataflow.DecimalHandlingMode_DECIMAL_HANDLING_MODE_PRECISE:
		fallthrough
	default:
		return changefeedopenbp.KafkaDownstreamConfig_DataFormat_AvroConfig_DECIMAL_HANDLING_MODE_PRECISE
	}
}

func toOpenAPIKafkaAvroBigintHandlingMode(in dataflow.BigintUnsignedHandlingMode) changefeedopenbp.KafkaDownstreamConfig_DataFormat_AvroConfig_BigintUnsignedHandlingMode {
	switch in {
	case dataflow.BigintUnsignedHandlingMode_BIGINT_UNSIGNED_HANDLING_MODE_STRING:
		return changefeedopenbp.KafkaDownstreamConfig_DataFormat_AvroConfig_BIGINT_UNSIGNED_HANDLING_MODE_STRING
	case dataflow.BigintUnsignedHandlingMode_BIGINT_UNSIGNED_HANDLING_MODE_LONG:
		fallthrough
	default:
		return changefeedopenbp.KafkaDownstreamConfig_DataFormat_AvroConfig_BIGINT_UNSIGNED_HANDLING_MODE_LONG
	}
}

func toOpenAPIMySQLDownstreamConfig(in *dataflow.MySQLInfo) *changefeedopenbp.MySQLDownstreamConfig {
	if in == nil {
		return nil
	}
	return &changefeedopenbp.MySQLDownstreamConfig{
		Connection: toOpenAPIConnection(in.GetConnection()),
	}
}

func toOpenAPIConnection(in *dataflow.Connection) *changefeedopenbp.Connection {
	if in == nil {
		return nil
	}
	return &changefeedopenbp.Connection{
		Endpoint: in.GetEndpoint(),
		Username: in.GetUsername(),
		Password: "", // redact sentitive info
	}
}

func toOpenAPIS3DownstreamConfig(in *dataflow.S3Info) *changefeedopenbp.S3DownstreamConfig {
	if in == nil {
		return nil
	}
	return &changefeedopenbp.S3DownstreamConfig{
		Endpoint:           toOpenAPIS3Endpoint(in.GetEndpoint()),
		DataFormat:         toOpenAPIS3DataFormat(in.GetDataFormat()),
		StorageFlushConfig: toOpenAPIStorageFlushConfig(in.GetStorageFlushConfig()),
	}
}

func toOpenAPIGcsDownstreamConfig(in *dataflow.GcsInfo) *changefeedopenbp.GcsDownstreamConfig {
	if in == nil {
		return nil
	}
	return &changefeedopenbp.GcsDownstreamConfig{
		Endpoint:           toOpenAPIGcsEndpoint(in.GetEndpoint()),
		DataFormat:         toOpenAPIS3DataFormat(in.GetDataFormat()),
		StorageFlushConfig: toOpenAPIStorageFlushConfig(in.GetStorageFlushConfig()),
	}
}

func toOpenAPIS3Endpoint(in *dataflow.S3Endpoint) *changefeedopenbp.S3DownstreamConfig_Endpoint {
	if in == nil {
		return nil
	}
	return &changefeedopenbp.S3DownstreamConfig_Endpoint{
		S3Uri:             in.GetS3Uri(),
		AuthType:          toOpenAPIS3AuthType(in.GetAuthType()),
		AccessKey:         in.AccessKey,
		SecretKey:         "", // redact sentitive info
		IamAssumedRoleArn: in.GetIamAssumedRoleArn(),
		CustomEndpoint:    in.GetEndpoint(),
		Provider:          toOpenAPIS3Provider(in.GetProvider()),
	}
}

func toOpenAPIGcsEndpoint(in *dataflow.GcsEndpoint) *changefeedopenbp.GcsDownstreamConfig_Endpoint {
	if in == nil {
		return nil
	}
	return &changefeedopenbp.GcsDownstreamConfig_Endpoint{GcsUri: in.GetGcsUri()}
}

func toOpenAPIS3AuthType(in dataflow.S3AuthType) changefeedopenbp.S3DownstreamConfig_Endpoint_S3AuthType {
	switch in {
	case dataflow.S3AuthType_S3_AUTH_TYPE_ASSUME_ROLE:
		return changefeedopenbp.S3DownstreamConfig_Endpoint_S3_AUTH_TYPE_ASSUME_ROLE
	case dataflow.S3AuthType_S3_AUTH_TYPE_AK_SK:
		fallthrough
	default:
		return changefeedopenbp.S3DownstreamConfig_Endpoint_S3_AUTH_TYPE_AK_SK
	}
}

func toOpenAPIS3Provider(in dataflow.S3Provider) changefeedopenbp.S3DownstreamConfig_Endpoint_Provider {
	switch in {
	case dataflow.S3Provider_S3_PROVIDER_ALIBABA_OSS:
		return changefeedopenbp.S3DownstreamConfig_Endpoint_ALIBABA_OSS
	default:
		return changefeedopenbp.S3DownstreamConfig_Endpoint_AWS
	}
}

func toOpenAPIS3DataFormat(in *dataflow.S3Dataformat) *changefeedopenbp.S3DataFormat {
	if in == nil {
		return nil
	}
	out := &changefeedopenbp.S3DataFormat{
		Protocol:             toOpenAPIS3Protocol(in.GetProtocol()),
		Terminator:           in.GetTerminator(),
		DateSeparator:        toOpenAPIDateSeparator(in.GetDateSeparator()),
		EnableTidbExtension:  in.GetEnableTidbExtension(),
		Delimiter:            in.GetDelimiter(),
		Quote:                in.GetQuote(),
		Null:                 in.GetNull(),
		IncludeCommitTs:      in.GetIncludeCommitTs(),
		BinaryEncodingMethod: toOpenAPIBinaryEncodingMethod(in.GetBinaryEncodingMethod()),
		OutputRawChangeEvent: in.GetOutputRawChangeEvent(),
		OutputFieldHeader:    in.GetOutputFieldHeader(),
	}
	return out
}

func toOpenAPIStorageFlushConfig(in *dataflow.StorageFlushConfig) *changefeedopenbp.StorageFlushConfig {
	if in == nil {
		return nil
	}
	return &changefeedopenbp.StorageFlushConfig{
		Interval: in.GetFlushInterval(),
		FileSize: in.GetFileSize(),
	}
}

func toOpenAPIS3Protocol(in dataflow.S3Protocol) changefeedopenbp.S3DataFormat_Protocol {
	switch in {
	case dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON:
		return changefeedopenbp.S3DataFormat_PROTOCOL_CANAL_JSON
	case dataflow.S3Protocol_S3_PROTOCOL_CSV:
		fallthrough
	default:
		return changefeedopenbp.S3DataFormat_PROTOCOL_CSV
	}
}

func toOpenAPIDateSeparator(in string) changefeedopenbp.S3DataFormat_DateSeparator {
	switch strings.ToUpper(in) {
	case "NONE":
		return changefeedopenbp.S3DataFormat_NONE
	case "YEAR":
		return changefeedopenbp.S3DataFormat_YEAR
	case "MONTH":
		return changefeedopenbp.S3DataFormat_MONTH
	case "DAY":
		return changefeedopenbp.S3DataFormat_DAY
	default:
		return changefeedopenbp.S3DataFormat_NONE
	}
}

func toOpenAPIBinaryEncodingMethod(in dataflow.BinaryEncodingMethod) changefeedopenbp.S3DataFormat_BinaryEncodingMethod {
	switch in {
	case dataflow.BinaryEncodingMethod_HEX:
		return changefeedopenbp.S3DataFormat_HEX
	case dataflow.BinaryEncodingMethod_BASE64:
		fallthrough
	default:
		return changefeedopenbp.S3DataFormat_BASE64
	}
}

func toDataflowNetworkInfo(in *changefeedopenbp.NetworkInfo) (*dataflow.NetworkInfo, error) {
	if in == nil {
		return nil, nil
	}
	networkType, err := toDataflowNetworkType(in.GetNetworkType())
	if err != nil {
		return nil, err
	}
	out := &dataflow.NetworkInfo{NetworkType: networkType}
	if sinkEndpointID := in.GetSinkEndpointId(); sinkEndpointID != 0 {
		out.SinkEndpointId = &sinkEndpointID
	}
	if ports := in.GetPorts(); len(ports) > 0 {
		out.Ports = ports
	}
	return out, nil
}

func toDataflowNetworkType(in changefeedopenbp.NetworkInfo_NetworkType) (dataflow.NetworkType, error) {
	switch in {
	case changefeedopenbp.NetworkInfo_NETWORK_TYPE_PUBLIC, changefeedopenbp.NetworkInfo_NETWORK_TYPE_UNSPECIFIED:
		return dataflow.NetworkType_NetworkType_Public, nil
	case changefeedopenbp.NetworkInfo_NETWORK_TYPE_VPC_PEERING:
		return dataflow.NetworkType_NetworkType_VPC_Peering, nil
	case changefeedopenbp.NetworkInfo_NETWORK_TYPE_PRIVATE_LINK:
		return dataflow.NetworkType_NetworkType_Private_Link, nil
	default:
		return 0, bizerr.New(bizerr.ErrCodeParams, "unsupported network_info.network_type %q", in.String())
	}
}

func toDataflowKafkaInfo(cf *changefeedopenbp.Changefeed) *dataflow.KafkaInfo {
	cfg := cf.GetKafka()
	if cfg == nil {
		return nil
	}

	info := &dataflow.KafkaInfo{
		Broker:               toDataflowKafkaBroker(cfg.GetBroker()),
		Authentication:       toDataflowKafkaAuthentication(cfg.GetAuthentication()),
		DataFormat:           toDataflowKafkaDataFormat(cfg.GetDataFormat()),
		TopicPartitionConfig: toDataflowKafkaTopicPartitionConfig(cfg.GetTopicPartitionConfig()),
		TableConfig:          toDataflowTableConfig(cf.GetTableConfig()),
		StartPosition:        toDataflowStartPosition(cf.GetStartPosition()),
		ColumnSelectors:      toDataflowKafkaColumnSelectors(cfg.GetColumnSelectors()),
	}
	return info
}

func toDataflowMySQLInfo(cf *changefeedopenbp.Changefeed) *dataflow.MySQLInfo {
	cfg := cf.GetMysql()
	if cfg == nil {
		return nil
	}
	info := &dataflow.MySQLInfo{
		Connection:      toDataflowConnection(cfg.GetConnection()),
		TableConfig:     toDataflowTableConfig(cf.GetTableConfig()),
		StartPosition:   toDataflowStartPosition(cf.GetStartPosition()),
		PrivatelinkInfo: nil,
	}
	if networkInfo := cf.GetNetworkInfo(); networkInfo != nil {
		if endpointID := networkInfo.GetSinkEndpointId(); endpointID != 0 {
			info.PrivatelinkInfo = &dataflow.PrivatelinkInfo{DfPrivatelinkEndpointId: endpointID}
		}
	}
	return info
}

func toDataflowS3Info(cf *changefeedopenbp.Changefeed) *dataflow.S3Info {
	cfg := cf.GetS3()
	if cfg == nil {
		return nil
	}
	info := &dataflow.S3Info{
		Endpoint:           toDataflowS3Endpoint(cfg.GetEndpoint()),
		DataFormat:         toDataflowS3DataFormat(cfg.GetDataFormat()),
		StorageFlushConfig: toDataflowStorageFlushConfig(cfg.GetStorageFlushConfig()),
		TableConfig:        toDataflowTableConfig(cf.GetTableConfig()),
		StartPosition:      toDataflowStartPosition(cf.GetStartPosition()),
	}
	return info
}

func toDataflowGcsInfo(cf *changefeedopenbp.Changefeed) *dataflow.GcsInfo {
	cfg := cf.GetGcs()
	if cfg == nil {
		return nil
	}
	info := &dataflow.GcsInfo{
		Endpoint:           toDataflowGcsEndpoint(cfg.GetEndpoint()),
		DataFormat:         toDataflowS3DataFormat(cfg.GetDataFormat()),
		StorageFlushConfig: toDataflowStorageFlushConfig(cfg.GetStorageFlushConfig()),
		TableConfig:        toDataflowTableConfig(cf.GetTableConfig()),
		StartPosition:      toDataflowStartPosition(cf.GetStartPosition()),
	}
	return info
}

func toDataflowConnection(in *changefeedopenbp.Connection) *dataflow.Connection {
	if in == nil {
		return nil
	}
	return &dataflow.Connection{
		Endpoint: in.GetEndpoint(),
		Username: in.GetUsername(),
		Password: in.GetPassword(),
	}
}

func toDataflowS3Endpoint(in *changefeedopenbp.S3DownstreamConfig_Endpoint) *dataflow.S3Endpoint {
	if in == nil {
		return nil
	}
	return &dataflow.S3Endpoint{
		S3Uri:             in.GetS3Uri(),
		AccessKey:         in.GetAccessKey(),
		SecretKey:         in.GetSecretKey(),
		AuthType:          toDataflowS3AuthType(in.GetAuthType()),
		IamAssumedRoleArn: in.GetIamAssumedRoleArn(),
		Endpoint:          in.GetCustomEndpoint(),
		Provider:          toDataflowS3Provider(in.GetProvider()),
	}
}

func toDataflowGcsEndpoint(in *changefeedopenbp.GcsDownstreamConfig_Endpoint) *dataflow.GcsEndpoint {
	if in == nil {
		return nil
	}
	return &dataflow.GcsEndpoint{GcsUri: in.GetGcsUri()}
}

func toDataflowS3AuthType(in changefeedopenbp.S3DownstreamConfig_Endpoint_S3AuthType) dataflow.S3AuthType {
	switch in {
	case changefeedopenbp.S3DownstreamConfig_Endpoint_S3_AUTH_TYPE_ASSUME_ROLE:
		return dataflow.S3AuthType_S3_AUTH_TYPE_ASSUME_ROLE
	case changefeedopenbp.S3DownstreamConfig_Endpoint_S3_AUTH_TYPE_AK_SK:
		fallthrough
	default:
		return dataflow.S3AuthType_S3_AUTH_TYPE_AK_SK
	}
}

func toDataflowS3Provider(in changefeedopenbp.S3DownstreamConfig_Endpoint_Provider) dataflow.S3Provider {
	switch in {
	case changefeedopenbp.S3DownstreamConfig_Endpoint_ALIBABA_OSS:
		return dataflow.S3Provider_S3_PROVIDER_ALIBABA_OSS
	default:
		return dataflow.S3Provider_S3_PROVIDER_AWS
	}
}

func toDataflowS3DataFormat(in *changefeedopenbp.S3DataFormat) *dataflow.S3Dataformat {
	if in == nil {
		return nil
	}

	return &dataflow.S3Dataformat{
		Protocol:             toDataflowS3Protocol(in.GetProtocol()),
		Terminator:           in.GetTerminator(),
		DateSeparator:        toDataflowS3DateSeparator(in.GetDateSeparator()),
		EnableTidbExtension:  in.GetEnableTidbExtension(),
		Delimiter:            in.GetDelimiter(),
		Quote:                in.GetQuote(),
		Null:                 in.GetNull(),
		IncludeCommitTs:      in.GetIncludeCommitTs(),
		BinaryEncodingMethod: toDataflowS3BinaryEncodingMethod(in.GetBinaryEncodingMethod()),
		OutputRawChangeEvent: in.GetOutputRawChangeEvent(),
		OutputFieldHeader:    in.GetOutputFieldHeader(),
	}
}

func toDataflowStorageFlushConfig(in *changefeedopenbp.StorageFlushConfig) *dataflow.StorageFlushConfig {
	if in == nil {
		return nil
	}
	return &dataflow.StorageFlushConfig{
		FlushInterval: in.GetInterval(),
		FileSize:      in.GetFileSize(),
	}
}

func toDataflowS3Protocol(in changefeedopenbp.S3DataFormat_Protocol) dataflow.S3Protocol {
	switch in {
	case changefeedopenbp.S3DataFormat_PROTOCOL_CANAL_JSON:
		return dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON
	case changefeedopenbp.S3DataFormat_PROTOCOL_CSV, changefeedopenbp.S3DataFormat_PROTOCOL_UNSPECIFIED:
		fallthrough
	default:
		return dataflow.S3Protocol_S3_PROTOCOL_CSV
	}
}

func toDataflowS3DateSeparator(in changefeedopenbp.S3DataFormat_DateSeparator) string {
	switch in {
	case changefeedopenbp.S3DataFormat_NONE:
		return "none"
	case changefeedopenbp.S3DataFormat_YEAR:
		return "year"
	case changefeedopenbp.S3DataFormat_MONTH:
		return "month"
	case changefeedopenbp.S3DataFormat_DAY:
		return "day"
	default:
		return "none"
	}
}

func toDataflowS3BinaryEncodingMethod(in changefeedopenbp.S3DataFormat_BinaryEncodingMethod) dataflow.BinaryEncodingMethod {
	switch in {
	case changefeedopenbp.S3DataFormat_HEX:
		return dataflow.BinaryEncodingMethod_HEX
	case changefeedopenbp.S3DataFormat_BASE64, changefeedopenbp.S3DataFormat_BINARY_ENCODING_METHOD_UNSPECIFIED:
		fallthrough
	default:
		return dataflow.BinaryEncodingMethod_BASE64
	}
}

func toDataflowKafkaColumnSelectors(in []*changefeedopenbp.KafkaDownstreamConfig_ColumnSelector) []*dataflow.ColumnSelector {
	if len(in) == 0 {
		return nil
	}
	out := make([]*dataflow.ColumnSelector, 0, len(in))
	for _, selector := range in {
		if selector == nil {
			continue
		}
		out = append(out, &dataflow.ColumnSelector{
			Matcher: append([]string(nil), selector.GetMatcher()...),
			Columns: append([]string(nil), selector.GetColumns()...),
		})
	}
	return out
}

func toDataflowKafkaDispatchType(in changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_DispatchType) dataflow.DispatchType {
	switch in {
	case changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_DISPATCH_TYPE_BY_TABLE:
		return dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE
	case changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_DISPATCH_TYPE_BY_DATABASE:
		return dataflow.DispatchType_DISPATCH_TYPE_BY_DATABASE
	case changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_DISPATCH_TYPE_ONE_TOPIC,
		changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_DISPATCH_TYPE_UNSPECIFIED:
		fallthrough
	default:
		return dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC
	}
}

func toDataflowKafkaPartitionDispatcher(in changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_PartitionDispatcher) dataflow.PartitionDispatcher {
	switch in {
	case changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_INDEX_VALUE:
		return dataflow.PartitionDispatcher_INDEX_VALUE
	case changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_TIMESTAMP:
		return dataflow.PartitionDispatcher_TS
	case changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_COLUMNS:
		return dataflow.PartitionDispatcher_COLUMNS
	case changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_TABLE,
		changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_PARTITION_DISPATCHER_UNSPECIFIED:
		fallthrough
	default:
		return dataflow.PartitionDispatcher_TABLE
	}
}

func toDataflowKafkaVersion(in changefeedopenbp.KafkaDownstreamConfig_KafkaVersion) dataflow.KafkaVersion {
	switch in {
	case changefeedopenbp.KafkaDownstreamConfig_KAFKA_VERSION_1XX:
		return dataflow.KafkaVersion_KAFKA_VERSION_1XX
	case changefeedopenbp.KafkaDownstreamConfig_KAFKA_VERSION_2XX:
		return dataflow.KafkaVersion_KAFKA_VERSION_2XX
	case changefeedopenbp.KafkaDownstreamConfig_KAFKA_VERSION_3XX:
		return dataflow.KafkaVersion_KAFKA_VERSION_3XX
	default:
		return dataflow.KafkaVersion_KAFKA_VERSION_0XX
	}
}

func toDataflowKafkaCompression(in changefeedopenbp.KafkaDownstreamConfig_CompressionType) dataflow.CompressionType {
	switch in {
	case changefeedopenbp.KafkaDownstreamConfig_GZIP:
		return dataflow.CompressionType_GZIP
	case changefeedopenbp.KafkaDownstreamConfig_SNAPPY:
		return dataflow.CompressionType_SNAPPY
	case changefeedopenbp.KafkaDownstreamConfig_LZ4:
		return dataflow.CompressionType_LZ4
	case changefeedopenbp.KafkaDownstreamConfig_ZSTD:
		return dataflow.CompressionType_ZSTD
	case changefeedopenbp.KafkaDownstreamConfig_NONE,
		changefeedopenbp.KafkaDownstreamConfig_COMPRESSION_TYPE_UNSPECIFIED:
		fallthrough
	default:
		return dataflow.CompressionType_NONE
	}
}

func toDataflowKafkaAuthType(in changefeedopenbp.KafkaDownstreamConfig_Authentication_AuthType) dataflow.KafkaAuthType {
	switch in {
	case changefeedopenbp.KafkaDownstreamConfig_Authentication_SASL_PLAIN:
		return dataflow.KafkaAuthType_SASL_PLAIN
	case changefeedopenbp.KafkaDownstreamConfig_Authentication_SASL_SCRAM_SHA_256:
		return dataflow.KafkaAuthType_SASL_SCRAM_SHA_256
	case changefeedopenbp.KafkaDownstreamConfig_Authentication_SASL_SCRAM_SHA_512:
		return dataflow.KafkaAuthType_SASL_SCRAM_SHA_512
	case changefeedopenbp.KafkaDownstreamConfig_Authentication_DISABLE,
		changefeedopenbp.KafkaDownstreamConfig_Authentication_AUTH_TYPE_UNSPECIFIED:
		fallthrough
	default:
		return dataflow.KafkaAuthType_DISABLE
	}
}

func toDataflowKafkaProtocol(in changefeedopenbp.KafkaDownstreamConfig_DataFormat_Protocol) dataflow.Protocol {
	switch in {
	case changefeedopenbp.KafkaDownstreamConfig_DataFormat_PROTOCOL_OPEN_PROTOCOL:
		return dataflow.Protocol_OPEN_PROTOCOL
	case changefeedopenbp.KafkaDownstreamConfig_DataFormat_PROTOCOL_AVRO:
		return dataflow.Protocol_AVRO
	case changefeedopenbp.KafkaDownstreamConfig_DataFormat_PROTOCOL_DEBEZIUM:
		return dataflow.Protocol_DEBEZIUM
	case changefeedopenbp.KafkaDownstreamConfig_DataFormat_PROTOCOL_DEBEZIUM_AVRO:
		return dataflow.Protocol_DEBEZIUM_AVRO
	case changefeedopenbp.KafkaDownstreamConfig_DataFormat_PROTOCOL_CANAL_JSON,
		changefeedopenbp.KafkaDownstreamConfig_DataFormat_PROTOCOL_UNSPECIFIED:
		fallthrough
	default:
		return dataflow.Protocol_CANAL_JSON
	}
}

func toDataflowAvroDecimalHandlingMode(in changefeedopenbp.KafkaDownstreamConfig_DataFormat_AvroConfig_DecimalHandlingMode) dataflow.DecimalHandlingMode {
	switch in {
	case changefeedopenbp.KafkaDownstreamConfig_DataFormat_AvroConfig_DECIMAL_HANDLING_MODE_STRING:
		return dataflow.DecimalHandlingMode_DECIMAL_HANDLING_MODE_STRING
	case changefeedopenbp.KafkaDownstreamConfig_DataFormat_AvroConfig_DECIMAL_HANDLING_MODE_PRECISE:
		fallthrough
	default:
		return dataflow.DecimalHandlingMode_DECIMAL_HANDLING_MODE_PRECISE
	}
}

func toDataflowAvroBigintUnsignedHandlingMode(in changefeedopenbp.KafkaDownstreamConfig_DataFormat_AvroConfig_BigintUnsignedHandlingMode) dataflow.BigintUnsignedHandlingMode {
	switch in {
	case changefeedopenbp.KafkaDownstreamConfig_DataFormat_AvroConfig_BIGINT_UNSIGNED_HANDLING_MODE_STRING:
		return dataflow.BigintUnsignedHandlingMode_BIGINT_UNSIGNED_HANDLING_MODE_STRING
	case changefeedopenbp.KafkaDownstreamConfig_DataFormat_AvroConfig_BIGINT_UNSIGNED_HANDLING_MODE_LONG:
		fallthrough
	default:
		return dataflow.BigintUnsignedHandlingMode_BIGINT_UNSIGNED_HANDLING_MODE_LONG
	}
}

func toDataflowKafkaBroker(in *changefeedopenbp.KafkaDownstreamConfig_Broker) *dataflow.Broker {
	if in == nil {
		return nil
	}
	return &dataflow.Broker{
		KafkaVersion:       toDataflowKafkaVersion(in.GetVersion()),
		BrokerEndpoints:    in.GetBrokerEndpoints(),
		TlsEnable:          in.GetUseTls(),
		InsecureSkipVerify: in.GetInsecureSkipVerify(),
		Compression:        toDataflowKafkaCompression(in.GetCompression()),
		KafkaClientId:      in.GetKafkaClientId(),
	}
}

func toDataflowKafkaAuthentication(in *changefeedopenbp.KafkaDownstreamConfig_Authentication) *dataflow.Authentication {
	if in == nil {
		return nil
	}
	return &dataflow.Authentication{
		AuthType: toDataflowKafkaAuthType(in.GetAuthType()),
		UserName: in.GetUsername(),
		Password: in.GetPassword(),
	}
}

func toDataflowKafkaDataFormat(in *changefeedopenbp.KafkaDownstreamConfig_DataFormat) *dataflow.DataFormat {
	if in == nil {
		return nil
	}
	out := &dataflow.DataFormat{
		Protocol:             toDataflowKafkaProtocol(in.GetProtocol()),
		EnableTidbExtension:  in.GetEnableTidbExtension(),
		OutputRawChangeEvent: in.GetOutputRawChangeEvent(),
		// for avro
		DecimalHandlingMode:        0,   // fill it later
		BigintUnsignedHandlingMode: 0,   // fill it later
		SchemaRegistry:             nil, // fill it later
		// for debezium
		DebeziumConfig: nil, // fill it later
	}
	switch cfg := in.GetConfig().(type) {
	case *changefeedopenbp.KafkaDownstreamConfig_DataFormat_AvroConfig_:
		if cfg.AvroConfig != nil {
			out.DecimalHandlingMode = toDataflowAvroDecimalHandlingMode(cfg.AvroConfig.GetDecimalHandlingMode())
			out.BigintUnsignedHandlingMode = toDataflowAvroBigintUnsignedHandlingMode(cfg.AvroConfig.GetBigintUnsignedHandlingMode())
			out.SchemaRegistry = toDataflowKafkaAvroSchemaRegistry(cfg.AvroConfig.GetSchemaRegistry())
		}
	case *changefeedopenbp.KafkaDownstreamConfig_DataFormat_DebeziumConfig_:
		if cfg.DebeziumConfig != nil {
			out.DebeziumConfig = &dataflow.DebeziumConfig{
				OutputOldValue: cfg.DebeziumConfig.GetOutputOldValue(),
				DisableSchema:  cfg.DebeziumConfig.GetDisableSchema(),
			}
		}
	}
	return out
}

func toDataflowKafkaAvroSchemaRegistry(in *changefeedopenbp.KafkaDownstreamConfig_DataFormat_AvroConfig_SchemaRegistry) *dataflow.SchemaRegistry {
	if in == nil {
		return nil
	}
	return &dataflow.SchemaRegistry{
		SchemaRegistryEndpoints: in.GetSchemaRegistryEndpoints(),
		EnableHttpAuth:          in.GetEnableBasicAuth(),
		UserName:                in.GetBasicAuthUsername(),
		Password:                in.GetBasicAuthPassword(),
	}
}

func toDataflowKafkaTopicPartitionConfig(in *changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig) *dataflow.TopicPartitionConfig {
	if in == nil {
		return nil
	}
	out := &dataflow.TopicPartitionConfig{
		DispatchType:      toDataflowKafkaDispatchType(in.GetDispatchType()),
		DefaultTopic:      in.GetDefaultTopic(),
		TopicPrefix:       in.GetTopicPrefix(),
		Separator:         in.GetSeparator(),
		TopicSuffix:       in.GetTopicSuffix(),
		ReplicationFactor: in.GetReplicationFactor(),
		PartitionNum:      in.GetPartitionNum(),
		PartitionDispatcher: func() dataflow.PartitionDispatcher {
			dispatchers := in.GetPartitionDispatchers()
			if len(dispatchers) == 0 {
				return dataflow.PartitionDispatcher_TABLE
			}
			return toDataflowKafkaPartitionDispatcher(dispatchers[0].GetPartitionType())
		}(),
		PartitionDispatchers: nil, // fill it later
	}
	if dispatchers := in.GetPartitionDispatchers(); len(dispatchers) > 0 {
		out.PartitionDispatchers = make([]*dataflow.PartitionDispatcherConfig, 0, len(dispatchers))
		for _, dispatcher := range dispatchers {
			out.PartitionDispatchers = append(out.PartitionDispatchers, &dataflow.PartitionDispatcherConfig{
				PartitionType: toDataflowKafkaPartitionDispatcher(dispatcher.GetPartitionType()),
				Matcher:       append([]string(nil), dispatcher.GetMatcher()...),
				IndexName:     dispatcher.GetIndexName(),
				Columns:       append([]string(nil), dispatcher.GetColumns()...),
			})
		}
	}
	return out
}

func sinkTableConfig(req *dataflow.ChangefeedCreateReq, consoleType dataflow.ChangefeedType) *dataflow.TableConfig {
	if req == nil {
		return nil
	}
	switch consoleType {
	case dataflow.ChangefeedType_KAFKA:
		return req.GetKafkaInfo().GetTableConfig()
	case dataflow.ChangefeedType_MYSQL, dataflow.ChangefeedType_TIDB:
		return effectiveMySQLInfo(req, consoleType).GetTableConfig()
	case dataflow.ChangefeedType_S3:
		return req.GetS3Info().GetTableConfig()
	case dataflow.ChangefeedType_GCS:
		return req.GetGcsInfo().GetTableConfig()
	case dataflow.ChangefeedType_PULSAR:
		return req.GetPulsarInfo().GetTableConfig()
	default:
		return nil
	}
}

func sinkStartPosition(req *dataflow.ChangefeedCreateReq, consoleType dataflow.ChangefeedType) *dataflow.StartPosition {
	if req == nil {
		return nil
	}
	switch consoleType {
	case dataflow.ChangefeedType_KAFKA:
		return req.GetKafkaInfo().GetStartPosition()
	case dataflow.ChangefeedType_MYSQL, dataflow.ChangefeedType_TIDB:
		return effectiveMySQLInfo(req, consoleType).GetStartPosition()
	case dataflow.ChangefeedType_S3:
		return req.GetS3Info().GetStartPosition()
	case dataflow.ChangefeedType_GCS:
		return req.GetGcsInfo().GetStartPosition()
	case dataflow.ChangefeedType_PULSAR:
		return req.GetPulsarInfo().GetStartPosition()
	default:
		return nil
	}
}

func effectiveMySQLInfo(req *dataflow.ChangefeedCreateReq, consoleType dataflow.ChangefeedType) *dataflow.MySQLInfo {
	if req == nil {
		return nil
	}
	switch consoleType {
	case dataflow.ChangefeedType_MYSQL, dataflow.ChangefeedType_TIDB:
		fallthrough
	default:
		return req.GetMysqlInfo()
	}
}