package vo

import (
	"context"
	"time"

	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/console"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/v1beta1"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
	"github.com/tidbcloud/serverless-service/internal/utils"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"google.golang.org/protobuf/types/known/timestamppb"
)

// Changefeed represents a CDC changefeed in the domain layer.
type Changefeed struct {
	ClusterID     string                  `json:"cluster_id,omitempty"`
	ChangefeedID  string                  `json:"changefeed_id,omitempty"`
	State         constant.ChangfeedState `json:"state,omitempty"`
	Arch          constant.ChangefeedArch `json:"arch,omitempty"`
	RCU           uint32                  `json:"rcu,omitempty"`
	Name          string                  `json:"name,omitempty"`
	Sink          *SinkInfo               `json:"sink,omitempty"`
	Filter        *entity.CDCFilter       `json:"filter,omitempty"`
	StartPosition *StartPosition          `json:"start_position,omitempty"`
	Checkpoint    *CDCCheckpoint          `json:"checkpoint,omitempty"`
	Error         string                  `json:"error,omitempty"`
	CreateTime    *time.Time              `json:"create_time,omitempty"`
	UpdateTime    *time.Time              `json:"update_time,omitempty"`
	// CloudProvider / Region / Kubernetes locate the regional dataplane
	// hosting the changefeed. Consumed by the ops surface only; the
	// user-facing converters ignore them.
	CloudProvider string `json:"cloud_provider,omitempty"`
	Region        string `json:"region,omitempty"`
	Kubernetes    string `json:"kubernetes,omitempty"`
}

// StartPosition represents the start position configuration for a changefeed.
type StartPosition struct {
	Mode constant.StartMode `json:"mode,omitempty"`
	TSO  uint64             `json:"tso,omitempty"`
	Time *time.Time         `json:"time,omitempty"`
}

// Validate checks if the Changefeed has valid data.
func (c *Changefeed) Validate(
	ctx context.Context,
	encryptor facade.EncryptorFacade,
	cluster *tidbmgmtservicev1.Cluster,
) bizerr.BizError {
	// check name
	if len(c.Name) > 100 {
		return bizerr.New(bizerr.ErrCodeParams, "name length must be less than 100 characters")
	}

	// check sink info
	if err := c.Sink.Validate(ctx, encryptor, cluster, false); err != nil {
		return err
	}
	if err := entity.ValidateCloudStorageColumnSelectorsArch(c.Arch, c.Sink.CloudStorage); err != nil {
		return err
	}
	// check filter
	if err := c.Filter.Validate(ctx); err != nil {
		return err
	}

	// check filter and sink info(datafomat)
	var dataFormat constant.CDCProtocol
	switch c.Sink.Type {
	case constant.ChangefeedTypeKafka:
		dataFormat = c.Sink.Kafka.DataFormat.Protocol
	}
	if c.Filter.Mode == constant.TableModeForceSync {
		if dataFormat == constant.CDCProtocolAvro {
			// avro protocol must force-replicate disable https://github.com/pingcap/tiflow/pull/9479
			return bizerr.New(bizerr.ErrCodeParams, "avro data format does not support force sync mode")
		}
	}
	return nil
}

func (s *StartPosition) Validate(clusterCreateTime time.Time, safepoint uint64) bizerr.BizError {
	if s == nil {
		return bizerr.New(bizerr.ErrCodeParams, "start position cannot be nil")
	}
	currentTime := time.Now()
	switch s.Mode {
	case constant.StartModeFromTSO:
		if !utils.IsValidTSO(s.TSO) {
			return bizerr.New(bizerr.ErrCodeParams,
				"invalid TSO value: %d. See https://docs.pingcap.com/tidb/stable/tso/ for more details",
				s.TSO)
		}
		if s.TSO > utils.TimeToTSO(currentTime) {
			return bizerr.New(bizerr.ErrCodeParams, "TSO %d cannot be later than current tso: %d",
				s.TSO,
				utils.TimeToTSO(currentTime))
		}
		// if tso > safepoint, it means the tso must > cluster create time. because safepoint > cluster create time
		if s.TSO <= safepoint {
			return bizerr.New(bizerr.ErrCodeParams, "TSO %d must be greater than the current safepoint: %d",
				s.TSO,
				safepoint)
		}
	case constant.StartModeFromTime:
		utcTime := s.Time.UTC()
		if utcTime.After(currentTime) {
			return bizerr.New(bizerr.ErrCodeParams, "Time %s cannot be later than the current time: %s",
				s.Time,
				currentTime.UTC().Format(time.RFC3339))
		}
		if utils.TimeToTSO(utcTime) <= safepoint {
			return bizerr.New(bizerr.ErrCodeParams, "Time %s must be greater than the current safepoint: %d",
				s.Time,
				safepoint)
		}
	}
	return nil
}

// SinkInfo represents the sink configuration for a changefeed.
type SinkInfo struct {
	Type         constant.ChangefeedType  `json:"type,omitempty"`
	Kafka        *entity.KafkaInfo        `json:"kafka,omitempty"`
	MySQL        *entity.MySQLInfo        `json:"mysql,omitempty"`
	CloudStorage *entity.CloudStorageInfo `json:"cloud_storage,omitempty"`
}

// Validate onlyNetwork is for testConnection API which only check network
func (s *SinkInfo) Validate(
	ctx context.Context,
	encryptor facade.EncryptorFacade,
	cluster *tidbmgmtservicev1.Cluster,
	isTest bool,
) bizerr.BizError {
	if s == nil {
		return bizerr.New(bizerr.ErrCodeParams, "sink info cannot be nil")
	}
	switch s.Type {
	case constant.ChangefeedTypeKafka:
		return s.Kafka.Validate(ctx, encryptor, isTest)
	case constant.ChangefeedTypeMySQL:
		return s.MySQL.Validate(ctx, encryptor)
	case constant.ChangefeedTypeCloudStorage:
		return s.CloudStorage.Validate(ctx, cluster.CloudProvider, isTest)
	}
	return nil
}

// CDCCheckpoint represents the checkpoint of a changefeed.
type CDCCheckpoint struct {
	TSO  uint64     `json:"tso,omitempty"`
	Time *time.Time `json:"time,omitempty"`
}

type ListChangefeedsOptions struct {
	ClusterID string `json:"cluster_id,omitempty"`
	Offset    int64  `json:"offset,omitempty"`
	Limit     int32  `json:"limit,omitempty"`
	SinkType  string `json:"sink_type,omitempty"`
	OrderBy   string `json:"order_by,omitempty"`
}

func (c *Changefeed) ConvertToV1beta1() *v1beta1.Changefeed {
	if c == nil {
		return nil
	}
	return &v1beta1.Changefeed{
		ClusterId:     c.ClusterID,
		ChangefeedId:  c.ChangefeedID,
		State:         convertCDCStateToV1beta1(c.State),
		Arch:          convertCDCArchToV1beta1(c.Arch),
		Rcu:           c.RCU,
		DisplayName:   c.Name,
		Sink:          convertSinkInfoToV1beta1(c.Sink),
		Filter:        convertCDCFilterToV1beta1(c.Filter),
		StartPosition: convertStartPositionToV1beta1(c.StartPosition),
		Checkpoint:    convertCDCCheckpointToV1beta1(c.Checkpoint),
		Error:         c.Error,
		CreateTime:    toProtoTimestamp(c.CreateTime),
		UpdateTime:    toProtoTimestamp(c.UpdateTime),
	}
}

func convertCDCArchToV1beta1(arch constant.ChangefeedArch) v1beta1.ChangefeedArch_Enum {
	switch arch {
	case constant.ChangefeedArchReplicationWorker:
		return v1beta1.ChangefeedArch_REPLICATION_WORKER
	case constant.ChangefeedArchTiCDCNewArch:
		return v1beta1.ChangefeedArch_TICDC_NEWARCH
	default:
		return v1beta1.ChangefeedArch_UNSPECIFIED
	}
}

func toProtoTimestamp(t *time.Time) *timestamppb.Timestamp {
	if t == nil {
		return nil
	}
	return timestamppb.New(*t)
}

func convertCDCStateToV1beta1(state constant.ChangfeedState) v1beta1.ChangefeedState_Enum {
	switch state {
	case constant.ChangfeedStateRunning:
		return v1beta1.ChangefeedState_RUNNING
	case constant.ChangefeedStateFailed:
		return v1beta1.ChangefeedState_RUNNING_FAILED
	case constant.ChangefeedStateCreating:
		return v1beta1.ChangefeedState_CREATING
	case constant.ChangefeedStateScaling:
		return v1beta1.ChangefeedState_SCALING
	case constant.ChangefeedStatePaused:
		return v1beta1.ChangefeedState_PAUSED
	case constant.ChangefeedStateDeleting:
		return v1beta1.ChangefeedState_DELETING
	case constant.ChangefeedStateDeleted:
		return v1beta1.ChangefeedState_DELETED
	case constant.ChangefeedStateWarning:
		return v1beta1.ChangefeedState_WARNING
	case constant.ChangefeedStateCreateFailed:
		return v1beta1.ChangefeedState_CREATE_FAILED
	default:
		return v1beta1.ChangefeedState_UNSPECIFIED
	}
}

func convertSinkInfoToV1beta1(s *SinkInfo) *v1beta1.SinkInfo {
	if s == nil {
		return nil
	}

	sinkInfo := &v1beta1.SinkInfo{}

	switch s.Type {
	case constant.ChangefeedTypeKafka:
		sinkInfo.Type = v1beta1.ChangefeedType_KAFKA
		sinkInfo.Changefeed = &v1beta1.SinkInfo_Kafka{
			Kafka: convertKafkaInfoToV1beta1(s.Kafka),
		}
	case constant.ChangefeedTypeMySQL:
		sinkInfo.Type = v1beta1.ChangefeedType_MYSQL
		sinkInfo.Changefeed = &v1beta1.SinkInfo_Mysql{
			Mysql: convertMySQLInfoToV1beta1(s.MySQL),
		}
	case constant.ChangefeedTypeCloudStorage:
		sinkInfo.Type = v1beta1.ChangefeedType_CLOUD_STORAGE
		sinkInfo.Changefeed = &v1beta1.SinkInfo_CloudStorage{
			CloudStorage: convertCloudStorageInfoToV1beta1(s.CloudStorage),
		}
	default:
		sinkInfo.Type = v1beta1.ChangefeedType_UNSPECIFIED
	}
	return sinkInfo
}

func convertKafkaInfoToV1beta1(k *entity.KafkaInfo) *v1beta1.Kafka {
	if k == nil {
		return nil
	}

	result := &v1beta1.Kafka{}

	// NetworkInfo conversion
	result.Network = &v1beta1.Kafka_Network{}
	switch k.Network.NetworkType {
	case constant.KafkaNetworkTypePublic:
		result.Network.NetworkType = v1beta1.KafkaNetworkType_PUBLIC
		result.Network.Type = &v1beta1.Kafka_Network_PublicEndpoints{
			PublicEndpoints: k.Network.PublicEndpoints,
		}
	case constant.KafkaNetworkTypePrivateLink:
		result.Network.NetworkType = v1beta1.KafkaNetworkType_PRIVATE_LINK
		result.Network.Type = &v1beta1.Kafka_Network_PrivateLink{
			PrivateLink: &v1beta1.Kafka_PrivateLink{
				PrivateLinkConnectionId: k.Network.PrivateLinkConnectionId,
				Port:                    k.Network.PrivateLinkPort,
			},
		}
	default:
		result.Network.NetworkType = v1beta1.KafkaNetworkType_UNSPECIFIED
	}

	// Broker conversion
	var kafkaVersion v1beta1.KafkaVersion_Enum
	switch k.Broker.KafkaVersion {
	case constant.KafkaVersion2XX:
		kafkaVersion = v1beta1.KafkaVersion_VERSION_2XX
	case constant.KafkaVersion3XX:
		kafkaVersion = v1beta1.KafkaVersion_VERSION_3XX
	default:
		kafkaVersion = v1beta1.KafkaVersion_UNSPECIFIED
	}

	var compressionType v1beta1.KafkaCompressionType_Enum
	switch k.Broker.Compression {
	case constant.KafkaCompressionTypeNone:
		compressionType = v1beta1.KafkaCompressionType_NONE
	case constant.KafkaCompressionTypeGZIP:
		compressionType = v1beta1.KafkaCompressionType_GZIP
	case constant.KafkaCompressionTypeSnappy:
		compressionType = v1beta1.KafkaCompressionType_SNAPPY
	case constant.KafkaCompressionTypeLZ4:
		compressionType = v1beta1.KafkaCompressionType_LZ4
	case constant.KafkaCompressionTypeZSTD:
		compressionType = v1beta1.KafkaCompressionType_ZSTD
	default:
		compressionType = v1beta1.KafkaCompressionType_UNSPECIFIED
	}

	result.Broker = &v1beta1.Kafka_Broker{
		KafkaVersion: kafkaVersion,
		Compression:  compressionType,
	}

	// Authentication conversion
	var authType v1beta1.KafkaAuthType_Enum
	switch k.Authentication.AuthType {
	case constant.KafkaAuthTypeDISABLE:
		authType = v1beta1.KafkaAuthType_DISABLE
	case constant.KafkaAuthTypeSASLPlain:
		authType = v1beta1.KafkaAuthType_SASL_PLAIN
	case constant.KafkaAuthTypeSASLScram256:
		authType = v1beta1.KafkaAuthType_SASL_SCRAM_SHA_256
	case constant.KafkaAuthTypeSASLScram512:
		authType = v1beta1.KafkaAuthType_SASL_SCRAM_SHA_512
	default:
		authType = v1beta1.KafkaAuthType_UNSPECIFIED
	}

	result.Authentication = &v1beta1.Kafka_Authentication{
		AuthType:   authType,
		Username:   k.Authentication.UserName,
		EnableTls:  k.Authentication.EnableTLS,
		ServerName: k.Authentication.ServerName,
	}

	// DataFormat conversion
	var protocol v1beta1.KafkaProtocol_Enum
	switch k.DataFormat.Protocol {
	case constant.CDCProtocolCanalJSON:
		protocol = v1beta1.KafkaProtocol_CANAL_JSON
	case constant.CDCProtocolOpen:
		protocol = v1beta1.KafkaProtocol_OPEN_PROTOCOL
	case constant.CDCProtocolAvro:
		protocol = v1beta1.KafkaProtocol_AVRO
	case constant.CDCProtocolDebezium:
		protocol = v1beta1.KafkaProtocol_DEBEZIUM
	default:
		protocol = v1beta1.KafkaProtocol_UNSPECIFIED
	}

	dataFormat := &v1beta1.Kafka_DataFormat{
		Protocol:            protocol,
		EnableTidbExtension: k.DataFormat.EnableTiDBExtension,
	}

	// debezium config conversion
	if protocol == v1beta1.KafkaProtocol_DEBEZIUM && k.DataFormat.DebeziumConfig != nil {
		dataFormat.DebeziumConfig = &v1beta1.DebeziumConfig{
			DisableOutputOldValue: k.DataFormat.DebeziumConfig.DisableOutputOldValue,
			DisableSchema:         k.DataFormat.DebeziumConfig.DisableSchema,
		}
	}

	// AvroConfig conversion
	if k.DataFormat.AvroConfig != nil {
		var decimalMode v1beta1.DecimalHandlingMode_Enum
		switch k.DataFormat.AvroConfig.DecimalHandlingMode {
		case constant.DecimalHandlingModePrecise:
			decimalMode = v1beta1.DecimalHandlingMode_PRECISE
		case constant.DecimalHandlingModeString:
			decimalMode = v1beta1.DecimalHandlingMode_STRING
		default:
			decimalMode = v1beta1.DecimalHandlingMode_UNSPECIFIED
		}

		var bigintMode v1beta1.BigintUnsignedHandlingMode_Enum
		switch k.DataFormat.AvroConfig.BigintUnsignedHandlingMode {
		case constant.BigintUnsignedHandlingModeLong:
			bigintMode = v1beta1.BigintUnsignedHandlingMode_LONG
		case constant.BigintUnsignedHandlingModeString:
			bigintMode = v1beta1.BigintUnsignedHandlingMode_STRING
		default:
			bigintMode = v1beta1.BigintUnsignedHandlingMode_UNSPECIFIED
		}

		avroConfig := &v1beta1.AvroConfig{
			DecimalHandlingMode:        decimalMode,
			BigintUnsignedHandlingMode: bigintMode,
		}

		// SchemaRegistry conversion
		if k.DataFormat.AvroConfig.AWSGlueSchemaRegistry != nil {
			avroConfig.SchemaRegistry = &v1beta1.AvroConfig_AwsGlueSchemaRegistry{
				AwsGlueSchemaRegistry: &v1beta1.GlueSchemaRegistry{
					Region:      k.DataFormat.AvroConfig.AWSGlueSchemaRegistry.Region,
					Name:        k.DataFormat.AvroConfig.AWSGlueSchemaRegistry.Name,
					AccessKeyId: k.DataFormat.AvroConfig.AWSGlueSchemaRegistry.AccessKeyID,
				},
			}
		} else if k.DataFormat.AvroConfig.ConfluentSchemaRegistry != nil {
			avroConfig.SchemaRegistry = &v1beta1.AvroConfig_ConfluentSchemaRegistry{
				ConfluentSchemaRegistry: &v1beta1.ConfluentSchemaRegistry{
					Endpoint:       k.DataFormat.AvroConfig.ConfluentSchemaRegistry.SchemaRegistryEndpoints,
					EnableHttpAuth: k.DataFormat.AvroConfig.ConfluentSchemaRegistry.EnableHTTPAuth,
					Username:       k.DataFormat.AvroConfig.ConfluentSchemaRegistry.UserName,
				},
			}
		}
		dataFormat.AvroConfig = avroConfig
	}

	result.DataFormat = dataFormat

	// TopicPartitionConfig conversion
	var dispatchType v1beta1.TopicDispatchType_Enum
	switch k.TopicPartitionConfig.DispatchType {
	case constant.TopicDispatchTypeOneTopic:
		dispatchType = v1beta1.TopicDispatchType_ONE_TOPIC
	case constant.TopicDispatchTypeByTable:
		dispatchType = v1beta1.TopicDispatchType_BY_TABLE
	case constant.TopicDispatchTypeByDatabase:
		dispatchType = v1beta1.TopicDispatchType_BY_DATABASE
	default:
		dispatchType = v1beta1.TopicDispatchType_UNSPECIFIED
	}

	tpc := &v1beta1.Kafka_TopicPartitionConfig{
		DispatchType:      dispatchType,
		DefaultTopic:      k.TopicPartitionConfig.DefaultTopic,
		TopicPrefix:       k.TopicPartitionConfig.TopicPrefix,
		Separator:         k.TopicPartitionConfig.Separator,
		TopicSuffix:       k.TopicPartitionConfig.TopicSuffix,
		ReplicationFactor: k.TopicPartitionConfig.ReplicationFactor,
		PartitionNum:      k.TopicPartitionConfig.PartitionNum,
	}

	// PartitionDispatchers conversion
	if len(k.TopicPartitionConfig.PartitionDispatchers) > 0 {
		dispatchers := make(
			[]*v1beta1.Kafka_PartitionDispatcherConfig,
			0,
			len(k.TopicPartitionConfig.PartitionDispatchers),
		)
		for _, d := range k.TopicPartitionConfig.PartitionDispatchers {
			if d == nil {
				continue
			}

			var partitionType v1beta1.PartitionDispatcher_Enum
			switch d.PartitionType {
			case constant.PartitionDispatcherTable:
				partitionType = v1beta1.PartitionDispatcher_TABLE
			case constant.PartitionDispatcherIndexValue:
				partitionType = v1beta1.PartitionDispatcher_INDEX_VALUE
			case constant.PartitionDispatcherTS:
				partitionType = v1beta1.PartitionDispatcher_TS
			case constant.PartitionDispatcherColumns:
				partitionType = v1beta1.PartitionDispatcher_COLUMNS
			default:
				partitionType = v1beta1.PartitionDispatcher_UNSPECIFIED
			}

			dispatchers = append(dispatchers, &v1beta1.Kafka_PartitionDispatcherConfig{
				PartitionType: partitionType,
				Matcher:       d.Matcher,
				IndexName:     d.IndexName,
				Columns:       d.Columns,
			})
		}
		tpc.PartitionDispatchers = dispatchers
	}

	result.TopicPartitionConfig = tpc

	// ColumnSelectors conversion
	if len(k.ColumnSelectors) > 0 {
		selectors := make([]*v1beta1.Kafka_ColumnSelector, 0, len(k.ColumnSelectors))
		for _, c := range k.ColumnSelectors {
			selectors = append(selectors, &v1beta1.Kafka_ColumnSelector{
				Matcher: c.Matcher,
				Columns: c.Columns,
			})
		}
		result.ColumnSelectors = selectors
	}

	// OutputRawChangeEvent
	result.OutputRawChangeEvent = k.OutputRawChangeEvent
	return result
}

func convertMySQLInfoToV1beta1(m *entity.MySQLInfo) *v1beta1.MySQL {
	if m == nil {
		return nil
	}
	mysql := &v1beta1.MySQL{
		Authentication: &v1beta1.MySQL_Authentication{
			UserName:  m.Authentication.User,
			EnableTls: m.Authentication.EnableTLS,
		},
		Network: &v1beta1.MySQL_Network{},
	}

	switch m.Network.NetworkType {
	case constant.MySQLNetworkTypePublic:
		mysql.Network.NetworkType = v1beta1.MySQLNetworkType_PUBLIC
		mysql.Network.Type = &v1beta1.MySQL_Network_PublicEndpoint{
			PublicEndpoint: m.Network.PublicEndpoint,
		}
	case constant.MySQLNetworkTypePrivateLink:
		mysql.Network.NetworkType = v1beta1.MySQLNetworkType_PRIVATE_LINK
		mysql.Network.Type = &v1beta1.MySQL_Network_PrivateLink{
			PrivateLink: &v1beta1.MySQL_PrivateLink{
				PrivateLinkConnectionId: m.Network.PrivateLinkConnectionId,
				Port:                    m.Network.PrivateLinkPort,
			},
		}
	}

	return mysql
}

func convertCloudStorageInfoToV1beta1(c *entity.CloudStorageInfo) *v1beta1.CloudStorage {
	if c == nil || c.CloudStorage == nil {
		return nil
	}
	storage := CloudStorageToCommon(c.CloudStorage.Type, c.CloudStorage)

	out := &v1beta1.CloudStorage{
		Storage:              storage,
		IntervalInSeconds:    uint32(c.IntervalInSeconds),
		SizeInMib:            uint32(c.SizeInMiB),
		OutputColumnId:       c.OutputColumnID,
		OutputRawChangeEvent: c.OutputRawChangeEvent,
	}
	for _, selector := range c.ColumnSelectors {
		out.ColumnSelectors = append(out.ColumnSelectors, &v1beta1.CloudStorage_ColumnSelector{
			Matcher: selector.Matcher,
			Columns: selector.Columns,
		})
	}

	// Date separator
	switch c.DateSeparator {
	case constant.DateSeparatorNone:
		out.DateSeparator = v1beta1.CloudStorage_DateSeparator_NONE
	case constant.DateSeparatorYear:
		out.DateSeparator = v1beta1.CloudStorage_DateSeparator_YEAR
	case constant.DateSeparatorMonth:
		out.DateSeparator = v1beta1.CloudStorage_DateSeparator_MONTH
	case constant.DateSeparatorDay:
		out.DateSeparator = v1beta1.CloudStorage_DateSeparator_DAY
	default:
		out.DateSeparator = v1beta1.CloudStorage_DateSeparator_UNSPECIFIED
	}

	// Data format
	df := &v1beta1.CloudStorage_DataFormat{}
	switch c.DataFormat.Protocol {
	case constant.CloudStorageProtocolCanalJSON:
		df.Protocol = v1beta1.CloudStorageProtocol_CANAL_JSON
		if c.DataFormat.CanalJSONConfig != nil {
			df.Configuration = &v1beta1.CloudStorage_DataFormat_CanalJsonConfig{
				CanalJsonConfig: &v1beta1.CloudStorage_CanalJsonConfig{
					EnableTidbExtension: c.DataFormat.CanalJSONConfig.EnableTiDBExtension,
				},
			}
		}
	case constant.CloudStorageProtocolCSV:
		df.Protocol = v1beta1.CloudStorageProtocol_CSV
		if c.DataFormat.CSVConfig != nil {
			var encoding v1beta1.CloudStorage_BinaryEncoding_Enum
			switch c.DataFormat.CSVConfig.Encoding {
			case constant.CloudStorageBinaryEncodingBase64:
				encoding = v1beta1.CloudStorage_BinaryEncoding_BASE64
			case constant.CloudStorageBinaryEncodingHex:
				encoding = v1beta1.CloudStorage_BinaryEncoding_HEX
			default:
				encoding = v1beta1.CloudStorage_BinaryEncoding_UNSPECIFIED
			}
			df.Configuration = &v1beta1.CloudStorage_DataFormat_CsvConfig{
				CsvConfig: &v1beta1.CloudStorage_CSVConfig{
					Delimiter:       c.DataFormat.CSVConfig.Delimiter,
					Quote:           c.DataFormat.CSVConfig.Quote,
					NullValue:       c.DataFormat.CSVConfig.NullValue,
					IncludeCommitTs: c.DataFormat.CSVConfig.IncludeCommitTs,
					Encoding:        encoding,
				},
			}
		}
	default:
		df.Protocol = v1beta1.CloudStorageProtocol_UNSPECIFIED
	}
	out.DataFormat = df
	return out
}

func convertCDCFilterToV1beta1(f *entity.CDCFilter) *v1beta1.ChangefeedFilter {
	if f == nil {
		return nil
	}
	filter := &v1beta1.ChangefeedFilter{
		FilterRule:       f.FilterRule,
		CaseSensitive:    f.CaseSensitive,
		IgnoreTxnStartTs: f.IgnoreTxnStartTs,
		BdrMode:          f.BDRMode,
	}

	switch f.Mode {
	case constant.TableModeIgnoreNotSupportTable:
		filter.Mode = v1beta1.TableMode_IGNORE_NOT_SUPPORT_TABLE
	case constant.TableModeForceSync:
		filter.Mode = v1beta1.TableMode_FORCE_SYNC
	}

	if len(f.EventFilterRule) > 0 {
		efr := make([]*v1beta1.EventFilterRule, 0, len(f.EventFilterRule))
		for _, e := range f.EventFilterRule {
			efr = append(efr, &v1beta1.EventFilterRule{
				Matcher:     e.Matcher,
				IgnoreEvent: e.IgnoreEvent,
			})
		}
		filter.EventFilterRule = efr
	}
	return filter
}

func convertStartPositionToV1beta1(s *StartPosition) *v1beta1.StartPosition {
	if s == nil {
		return nil
	}
	switch s.Mode {
	case constant.StartModeFromNow:
		return &v1beta1.StartPosition{
			Mode: v1beta1.StartMode_FROM_NOW,
		}
	case constant.StartModeFromTSO:
		return &v1beta1.StartPosition{
			Mode: v1beta1.StartMode_FROM_TSO,
			Tso:  s.TSO,
		}
	case constant.StartModeFromTime:
		return &v1beta1.StartPosition{
			Mode: v1beta1.StartMode_FROM_TIME,
			Time: toProtoTimestamp(s.Time),
		}
	}
	return nil
}

func convertCDCCheckpointToV1beta1(c *CDCCheckpoint) *v1beta1.Checkpoint {
	if c == nil {
		return nil
	}
	return &v1beta1.Checkpoint{
		Tso:  c.TSO,
		Time: toProtoTimestamp(c.Time),
	}
}

func (c *Changefeed) ToEntity(
	ctx context.Context,
	encryptor facade.EncryptorFacade,
) (*entity.CDCChangefeed, bizerr.BizError) {
	ent := &entity.CDCChangefeed{
		ID:               entity.GenerateIDWithPrefix(constant.CDCChangefeedIDPrefix),
		ClusterID:        c.ClusterID,
		State:            c.State,
		Arch:             c.Arch,
		RCU:              c.RCU,
		Name:             c.Name,
		SinkType:         c.Sink.Type,
		Kafka:            c.Sink.Kafka,
		MySQL:            c.Sink.MySQL,
		CloudStorageInfo: c.Sink.CloudStorage,
		Filter:           c.Filter,
	}

	// startPosition
	if c.StartPosition != nil {
		switch c.StartPosition.Mode {
		case constant.StartModeFromNow:
			ent.StartMode = constant.StartModeFromNow
		case constant.StartModeFromTSO:
			ent.StartMode = constant.StartModeFromTSO
			ent.StartTSO = c.StartPosition.TSO
		case constant.StartModeFromTime:
			ent.StartMode = constant.StartModeFromTime
			ent.StartTSO = utils.TimeToTSO(*c.StartPosition.Time)
		}
	}

	// Desensitization
	if err := ent.Desensitization(ctx, encryptor); err != nil {
		return nil, err
	}
	return ent, nil
}

func FromEntity(ctx context.Context, ent *entity.CDCChangefeed) *Changefeed {
	conn := &Changefeed{
		ChangefeedID: ent.ID,
		ClusterID:    ent.ClusterID,
		State:        ent.State,
		Arch:         ent.Arch,
		RCU:          ent.RCU,
		Name:         ent.Name,
		Sink: &SinkInfo{
			Type: ent.SinkType,
		},
		Filter:        ent.Filter,
		Error:         ent.Error,
		CreateTime:    &ent.CreatedAt,
		UpdateTime:    ent.UpdatedAt,
		CloudProvider: ent.CloudProvider,
		Region:        ent.Region,
		Kubernetes:    ent.Kubernetes,
	}

	// sinkinfo
	switch ent.SinkType {
	case constant.ChangefeedTypeKafka:
		conn.Sink.Kafka = ent.Kafka
	case constant.ChangefeedTypeMySQL:
		conn.Sink.MySQL = ent.MySQL
	case constant.ChangefeedTypeCloudStorage:
		conn.Sink.CloudStorage = ent.CloudStorageInfo
	}

	// StartPosition
	startPos := &StartPosition{
		Mode: ent.StartMode,
		TSO:  ent.StartTSO,
	}
	switch ent.StartMode {
	case constant.StartModeFromTime:
		time := utils.TSOToTime(ent.StartTSO)
		startPos.Time = &time
	}
	conn.StartPosition = startPos

	// check point
	if ent.CheckpointTSO > 0 {
		time := utils.TSOToTime(ent.CheckpointTSO)
		conn.Checkpoint = &CDCCheckpoint{
			TSO:  ent.CheckpointTSO,
			Time: &time,
		}
	}
	return conn
}

func (c *Changefeed) ConvertToConsole() *console.Changefeed {
	if c == nil {
		return nil
	}
	return &console.Changefeed{
		ClusterId:     c.ClusterID,
		ChangefeedId:  c.ChangefeedID,
		State:         convertCDCStateToConsole(c.State),
		Arch:          convertCDCArchToConsole(c.Arch),
		Rcu:           c.RCU,
		DisplayName:   c.Name,
		Sink:          convertSinkInfoToConsole(c.Sink),
		Filter:        convertCDCFilterToConsole(c.Filter),
		StartPosition: convertStartPositionToConsole(c.StartPosition),
		Checkpoint:    convertCDCCheckpointToConsole(c.Checkpoint),
		Error:         c.Error,
		CreateTime:    toProtoTimestamp(c.CreateTime),
		UpdateTime:    toProtoTimestamp(c.UpdateTime),
	}
}

func convertCDCArchToConsole(arch constant.ChangefeedArch) console.ChangefeedArch_Enum {
	switch arch {
	case constant.ChangefeedArchReplicationWorker:
		return console.ChangefeedArch_REPLICATION_WORKER
	case constant.ChangefeedArchTiCDCNewArch:
		return console.ChangefeedArch_TICDC_NEWARCH
	default:
		return console.ChangefeedArch_UNSPECIFIED
	}
}

func convertCDCStateToConsole(state constant.ChangfeedState) console.ChangefeedState_Enum {
	switch state {
	case constant.ChangfeedStateRunning:
		return console.ChangefeedState_RUNNING
	case constant.ChangefeedStateFailed:
		return console.ChangefeedState_RUNNING_FAILED
	case constant.ChangefeedStateCreating:
		return console.ChangefeedState_CREATING
	case constant.ChangefeedStateScaling:
		return console.ChangefeedState_SCALING
	case constant.ChangefeedStatePaused:
		return console.ChangefeedState_PAUSED
	case constant.ChangefeedStateDeleting:
		return console.ChangefeedState_DELETING
	case constant.ChangefeedStateDeleted:
		return console.ChangefeedState_DELETED
	case constant.ChangefeedStateWarning:
		return console.ChangefeedState_WARNING
	case constant.ChangefeedStateCreateFailed:
		return console.ChangefeedState_CREATE_FAILED
	default:
		return console.ChangefeedState_UNSPECIFIED
	}
}

func convertSinkInfoToConsole(s *SinkInfo) *console.SinkInfo {
	if s == nil {
		return nil
	}

	sinkInfo := &console.SinkInfo{}
	switch s.Type {
	case constant.ChangefeedTypeKafka:
		sinkInfo.Type = console.ChangefeedType_KAFKA
		sinkInfo.Changefeed = &console.SinkInfo_Kafka{
			Kafka: convertKafkaInfoToConsole(s.Kafka),
		}
	case constant.ChangefeedTypeMySQL:
		sinkInfo.Type = console.ChangefeedType_MYSQL
		sinkInfo.Changefeed = &console.SinkInfo_Mysql{
			Mysql: convertMySQLInfoToConsole(s.MySQL),
		}
	case constant.ChangefeedTypeCloudStorage:
		sinkInfo.Type = console.ChangefeedType_CLOUD_STORAGE
		sinkInfo.Changefeed = &console.SinkInfo_CloudStorage{
			CloudStorage: convertCloudStorageInfoToConsole(s.CloudStorage),
		}
	default:
		sinkInfo.Type = console.ChangefeedType_UNSPECIFIED
	}
	return sinkInfo
}

func convertKafkaInfoToConsole(k *entity.KafkaInfo) *console.Kafka {
	if k == nil {
		return nil
	}
	result := &console.Kafka{}

	// Network
	result.Network = &console.Kafka_Network{}
	switch k.Network.NetworkType {
	case constant.KafkaNetworkTypePublic:
		result.Network.NetworkType = console.KafkaNetworkType_PUBLIC
		result.Network.Type = &console.Kafka_Network_PublicEndpoints{
			PublicEndpoints: k.Network.PublicEndpoints,
		}
	case constant.KafkaNetworkTypePrivateLink:
		result.Network.NetworkType = console.KafkaNetworkType_PRIVATE_LINK
		result.Network.Type = &console.Kafka_Network_PrivateLink{
			PrivateLink: &console.Kafka_PrivateLink{
				PrivateLinkConnectionId: k.Network.PrivateLinkConnectionId,
				Port:                    k.Network.PrivateLinkPort,
			},
		}
	default:
		result.Network.NetworkType = console.KafkaNetworkType_UNSPECIFIED
	}

	// Broker
	var kafkaVersion console.KafkaVersion_Enum
	switch k.Broker.KafkaVersion {
	case constant.KafkaVersion2XX:
		kafkaVersion = console.KafkaVersion_VERSION_2XX
	case constant.KafkaVersion3XX:
		kafkaVersion = console.KafkaVersion_VERSION_3XX
	default:
		kafkaVersion = console.KafkaVersion_UNSPECIFIED
	}
	var compressionType console.KafkaCompressionType_Enum
	switch k.Broker.Compression {
	case constant.KafkaCompressionTypeNone:
		compressionType = console.KafkaCompressionType_NONE
	case constant.KafkaCompressionTypeGZIP:
		compressionType = console.KafkaCompressionType_GZIP
	case constant.KafkaCompressionTypeSnappy:
		compressionType = console.KafkaCompressionType_SNAPPY
	case constant.KafkaCompressionTypeLZ4:
		compressionType = console.KafkaCompressionType_LZ4
	case constant.KafkaCompressionTypeZSTD:
		compressionType = console.KafkaCompressionType_ZSTD
	default:
		compressionType = console.KafkaCompressionType_UNSPECIFIED
	}
	result.Broker = &console.Kafka_Broker{
		KafkaVersion: kafkaVersion,
		Compression:  compressionType,
	}

	// Auth
	var authType console.KafkaAuthType_Enum
	switch k.Authentication.AuthType {
	case constant.KafkaAuthTypeDISABLE:
		authType = console.KafkaAuthType_DISABLE
	case constant.KafkaAuthTypeSASLPlain:
		authType = console.KafkaAuthType_SASL_PLAIN
	case constant.KafkaAuthTypeSASLScram256:
		authType = console.KafkaAuthType_SASL_SCRAM_SHA_256
	case constant.KafkaAuthTypeSASLScram512:
		authType = console.KafkaAuthType_SASL_SCRAM_SHA_512
	default:
		authType = console.KafkaAuthType_UNSPECIFIED
	}
	result.Authentication = &console.Kafka_Authentication{
		AuthType:  authType,
		Username:  k.Authentication.UserName,
		EnableTls: k.Authentication.EnableTLS,
	}

	// DataFormat
	var protocol console.KafkaProtocol_Enum
	switch k.DataFormat.Protocol {
	case constant.CDCProtocolCanalJSON:
		protocol = console.KafkaProtocol_CANAL_JSON
	case constant.CDCProtocolOpen:
		protocol = console.KafkaProtocol_OPEN_PROTOCOL
	case constant.CDCProtocolAvro:
		protocol = console.KafkaProtocol_AVRO
	case constant.CDCProtocolDebezium:
		protocol = console.KafkaProtocol_DEBEZIUM
	default:
		protocol = console.KafkaProtocol_UNSPECIFIED
	}
	dataFormat := &console.Kafka_DataFormat{
		Protocol:            protocol,
		EnableTidbExtension: k.DataFormat.EnableTiDBExtension,
	}
	if protocol == console.KafkaProtocol_DEBEZIUM && k.DataFormat.DebeziumConfig != nil {
		dataFormat.DebeziumConfig = &console.DebeziumConfig{
			DisableOutputOldValue: k.DataFormat.DebeziumConfig.DisableOutputOldValue,
			DisableSchema:         k.DataFormat.DebeziumConfig.DisableSchema,
		}
	}
	if k.DataFormat.AvroConfig != nil {
		var decimalMode console.DecimalHandlingMode_Enum
		switch k.DataFormat.AvroConfig.DecimalHandlingMode {
		case constant.DecimalHandlingModePrecise:
			decimalMode = console.DecimalHandlingMode_PRECISE
		case constant.DecimalHandlingModeString:
			decimalMode = console.DecimalHandlingMode_STRING
		default:
			decimalMode = console.DecimalHandlingMode_UNSPECIFIED
		}
		var bigintMode console.BigintUnsignedHandlingMode_Enum
		switch k.DataFormat.AvroConfig.BigintUnsignedHandlingMode {
		case constant.BigintUnsignedHandlingModeLong:
			bigintMode = console.BigintUnsignedHandlingMode_LONG
		case constant.BigintUnsignedHandlingModeString:
			bigintMode = console.BigintUnsignedHandlingMode_STRING
		default:
			bigintMode = console.BigintUnsignedHandlingMode_UNSPECIFIED
		}
		avroCfg := &console.AvroConfig{
			DecimalHandlingMode:        decimalMode,
			BigintUnsignedHandlingMode: bigintMode,
		}
		if k.DataFormat.AvroConfig.AWSGlueSchemaRegistry != nil {
			avroCfg.SchemaRegistry = &console.AvroConfig_AwsGlueSchemaRegistry{
				AwsGlueSchemaRegistry: &console.GlueSchemaRegistry{
					Region:      k.DataFormat.AvroConfig.AWSGlueSchemaRegistry.Region,
					Name:        k.DataFormat.AvroConfig.AWSGlueSchemaRegistry.Name,
					AccessKeyId: k.DataFormat.AvroConfig.AWSGlueSchemaRegistry.AccessKeyID,
				},
			}
		} else if k.DataFormat.AvroConfig.ConfluentSchemaRegistry != nil {
			avroCfg.SchemaRegistry = &console.AvroConfig_ConfluentSchemaRegistry{
				ConfluentSchemaRegistry: &console.ConfluentSchemaRegistry{
					Endpoint:       k.DataFormat.AvroConfig.ConfluentSchemaRegistry.SchemaRegistryEndpoints,
					EnableHttpAuth: k.DataFormat.AvroConfig.ConfluentSchemaRegistry.EnableHTTPAuth,
					Username:       k.DataFormat.AvroConfig.ConfluentSchemaRegistry.UserName,
				},
			}
		}
		dataFormat.AvroConfig = avroCfg
	}
	result.DataFormat = dataFormat

	// TopicPartitionConfig
	var dispatchType console.TopicDispatchType_Enum
	switch k.TopicPartitionConfig.DispatchType {
	case constant.TopicDispatchTypeOneTopic:
		dispatchType = console.TopicDispatchType_ONE_TOPIC
	case constant.TopicDispatchTypeByTable:
		dispatchType = console.TopicDispatchType_BY_TABLE
	case constant.TopicDispatchTypeByDatabase:
		dispatchType = console.TopicDispatchType_BY_DATABASE
	default:
		dispatchType = console.TopicDispatchType_UNSPECIFIED
	}
	tpc := &console.Kafka_TopicPartitionConfig{
		DispatchType:      dispatchType,
		DefaultTopic:      k.TopicPartitionConfig.DefaultTopic,
		TopicPrefix:       k.TopicPartitionConfig.TopicPrefix,
		Separator:         k.TopicPartitionConfig.Separator,
		TopicSuffix:       k.TopicPartitionConfig.TopicSuffix,
		ReplicationFactor: k.TopicPartitionConfig.ReplicationFactor,
		PartitionNum:      k.TopicPartitionConfig.PartitionNum,
	}
	if len(k.TopicPartitionConfig.PartitionDispatchers) > 0 {
		for _, d := range k.TopicPartitionConfig.PartitionDispatchers {
			if d == nil {
				continue
			}
			var pType console.PartitionDispatcher_Enum
			switch d.PartitionType {
			case constant.PartitionDispatcherTable:
				pType = console.PartitionDispatcher_TABLE
			case constant.PartitionDispatcherIndexValue:
				pType = console.PartitionDispatcher_INDEX_VALUE
			case constant.PartitionDispatcherTS:
				pType = console.PartitionDispatcher_TS
			case constant.PartitionDispatcherColumns:
				pType = console.PartitionDispatcher_COLUMNS
			default:
				pType = console.PartitionDispatcher_UNSPECIFIED
			}
			tpc.PartitionDispatchers = append(tpc.PartitionDispatchers, &console.Kafka_PartitionDispatcherConfig{
				PartitionType: pType,
				Matcher:       d.Matcher,
				IndexName:     d.IndexName,
				Columns:       d.Columns,
			})
		}
	}
	result.TopicPartitionConfig = tpc

	// ColumnSelectors
	if len(k.ColumnSelectors) > 0 {
		for _, csel := range k.ColumnSelectors {
			result.ColumnSelectors = append(result.ColumnSelectors, &console.Kafka_ColumnSelector{
				Matcher: csel.Matcher,
				Columns: csel.Columns,
			})
		}
	}

	result.OutputRawChangeEvent = k.OutputRawChangeEvent
	return result
}

func convertMySQLInfoToConsole(m *entity.MySQLInfo) *console.MySQL {
	if m == nil {
		return nil
	}
	mysql := &console.MySQL{
		Authentication: &console.MySQL_Authentication{
			UserName:  m.Authentication.User,
			EnableTls: m.Authentication.EnableTLS,
		},
		Network: &console.MySQL_Network{},
	}
	switch m.Network.NetworkType {
	case constant.MySQLNetworkTypePublic:
		mysql.Network.NetworkType = console.MySQLNetworkType_PUBLIC
		mysql.Network.Type = &console.MySQL_Network_PublicEndpoint{
			PublicEndpoint: m.Network.PublicEndpoint,
		}
	case constant.MySQLNetworkTypePrivateLink:
		mysql.Network.NetworkType = console.MySQLNetworkType_PRIVATE_LINK
		mysql.Network.Type = &console.MySQL_Network_PrivateLink{
			PrivateLink: &console.MySQL_PrivateLink{
				PrivateLinkConnectionId: m.Network.PrivateLinkConnectionId,
				Port:                    m.Network.PrivateLinkPort,
			},
		}
	default:
		mysql.Network.NetworkType = console.MySQLNetworkType_UNSPECIFIED
	}
	return mysql
}

func convertCloudStorageInfoToConsole(c *entity.CloudStorageInfo) *console.CloudStorage {
	if c == nil || c.CloudStorage == nil {
		return nil
	}
	storage := CloudStorageToCommon(c.CloudStorage.Type, c.CloudStorage)

	out := &console.CloudStorage{
		Storage:              storage,
		IntervalInSeconds:    uint32(c.IntervalInSeconds),
		SizeInMib:            uint32(c.SizeInMiB),
		OutputColumnId:       c.OutputColumnID,
		OutputRawChangeEvent: c.OutputRawChangeEvent,
	}
	for _, selector := range c.ColumnSelectors {
		out.ColumnSelectors = append(out.ColumnSelectors, &console.CloudStorage_ColumnSelector{
			Matcher: selector.Matcher,
			Columns: selector.Columns,
		})
	}

	// Date separator
	switch c.DateSeparator {
	case constant.DateSeparatorNone:
		out.DateSeparator = console.CloudStorage_DateSeparator_NONE
	case constant.DateSeparatorYear:
		out.DateSeparator = console.CloudStorage_DateSeparator_YEAR
	case constant.DateSeparatorMonth:
		out.DateSeparator = console.CloudStorage_DateSeparator_MONTH
	case constant.DateSeparatorDay:
		out.DateSeparator = console.CloudStorage_DateSeparator_DAY
	default:
		out.DateSeparator = console.CloudStorage_DateSeparator_UNSPECIFIED
	}

	// Data format
	df := &console.CloudStorage_DataFormat{}
	switch c.DataFormat.Protocol {
	case constant.CloudStorageProtocolCanalJSON:
		df.Protocol = console.CloudStorageProtocol_CANAL_JSON
		if c.DataFormat.CanalJSONConfig != nil {
			df.Configuration = &console.CloudStorage_DataFormat_CanalJsonConfig{
				CanalJsonConfig: &console.CloudStorage_CanalJsonConfig{
					EnableTidbExtension: c.DataFormat.CanalJSONConfig.EnableTiDBExtension,
				},
			}
		}
	case constant.CloudStorageProtocolCSV:
		df.Protocol = console.CloudStorageProtocol_CSV
		if c.DataFormat.CSVConfig != nil {
			var encoding console.CloudStorage_BinaryEncoding_Enum
			switch c.DataFormat.CSVConfig.Encoding {
			case constant.CloudStorageBinaryEncodingBase64:
				encoding = console.CloudStorage_BinaryEncoding_BASE64
			case constant.CloudStorageBinaryEncodingHex:
				encoding = console.CloudStorage_BinaryEncoding_HEX
			default:
				encoding = console.CloudStorage_BinaryEncoding_UNSPECIFIED
			}
			df.Configuration = &console.CloudStorage_DataFormat_CsvConfig{
				CsvConfig: &console.CloudStorage_CSVConfig{
					Delimiter:       c.DataFormat.CSVConfig.Delimiter,
					Quote:           c.DataFormat.CSVConfig.Quote,
					NullValue:       c.DataFormat.CSVConfig.NullValue,
					IncludeCommitTs: c.DataFormat.CSVConfig.IncludeCommitTs,
					Encoding:        encoding,
				},
			}
		}
	default:
		df.Protocol = console.CloudStorageProtocol_UNSPECIFIED
	}
	out.DataFormat = df
	return out
}

func convertCDCFilterToConsole(f *entity.CDCFilter) *console.ChangefeedFilter {
	if f == nil {
		return nil
	}
	filter := &console.ChangefeedFilter{
		FilterRule:       f.FilterRule,
		CaseSensitive:    f.CaseSensitive,
		IgnoreTxnStartTs: f.IgnoreTxnStartTs,
		BdrMode:          f.BDRMode,
	}
	switch f.Mode {
	case constant.TableModeIgnoreNotSupportTable:
		filter.Mode = console.TableMode_IGNORE_NOT_SUPPORT_TABLE
	case constant.TableModeForceSync:
		filter.Mode = console.TableMode_FORCE_SYNC
	default:
		filter.Mode = console.TableMode_UNSPECIFIED
	}
	if len(f.EventFilterRule) > 0 {
		for _, e := range f.EventFilterRule {
			filter.EventFilterRule = append(filter.EventFilterRule, &console.EventFilterRule{
				Matcher:     e.Matcher,
				IgnoreEvent: e.IgnoreEvent,
			})
		}
	}
	return filter
}

func convertStartPositionToConsole(s *StartPosition) *console.StartPosition {
	if s == nil {
		return nil
	}
	switch s.Mode {
	case constant.StartModeFromNow:
		return &console.StartPosition{Mode: console.StartMode_FROM_NOW}
	case constant.StartModeFromTSO:
		return &console.StartPosition{Mode: console.StartMode_FROM_TSO, Tso: s.TSO}
	case constant.StartModeFromTime:
		return &console.StartPosition{Mode: console.StartMode_FROM_TIME, Time: toProtoTimestamp(s.Time)}
	default:
		return &console.StartPosition{Mode: console.StartMode_UNSPECIFIED}
	}
}

func convertCDCCheckpointToConsole(c *CDCCheckpoint) *console.Checkpoint {
	if c == nil {
		return nil
	}
	return &console.Checkpoint{
		Tso:  c.TSO,
		Time: toProtoTimestamp(c.Time),
	}
}