package cdc_svc

import (
	"context"

	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/tracing"
	"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/domain/service"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	"github.com/tidbcloud/serverless-service/internal/utils"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
	"google.golang.org/protobuf/types/known/timestamppb"
)

var v1beta1Tracer = otel.Tracer("cdc.v1beta1")

type cdcV1beta1Service struct {
	cdc                  service.CDCService
	account              facade.AccountFacade
	clusterProxyProvider facade.ClusterProxyProviderFacade
	mgmt                 facade.MgmtSvcFacade
	v1beta1.UnimplementedChangefeedServiceServer
}

func NewCDCV1beta1Service(
	cdc service.CDCService,
	account facade.AccountFacade,
	clusterProxyProvider facade.ClusterProxyProviderFacade,
	mgmt facade.MgmtSvcFacade,
) v1beta1.ChangefeedServiceServer {
	return &cdcV1beta1Service{
		cdc:                  cdc,
		account:              account,
		clusterProxyProvider: clusterProxyProvider,
		mgmt:                 mgmt,
	}
}

func (s *cdcV1beta1Service) CreateChangefeed(
	ctx context.Context,
	req *v1beta1.CreateChangefeedReq,
) (_ *v1beta1.Changefeed, err error) {
	ctx, span := v1beta1Tracer.Start(ctx, "cdcV1beta1Service.CreateChangefeed")
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareCDCErrorMessage(ctx, err)
	}()
	span.SetAttributes(
		attribute.String("clusterId", req.ClusterId),
	)
	logger.Info("start to create changefeed")

	changefeedVo, err := buildChangefeedVo(req)
	if err != nil {
		logger.Error("failed to build changefeed VO", zap.Error(err))
		return nil, err
	}
	changefeedVo, err = s.cdc.CreateChangefeed(ctx, changefeedVo)
	if err != nil {
		logger.Error("failed to create changefeed", zap.Error(err))
		return nil, err
	}

	return changefeedVo.ConvertToV1beta1(), nil
}

func (s *cdcV1beta1Service) ScaleChangefeed(
	ctx context.Context,
	req *v1beta1.ScaleChangefeedReq,
) (_ *v1beta1.Changefeed, err error) {
	ctx, span := v1beta1Tracer.Start(ctx, "cdcV1beta1Service.ScaleChangefeed")
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", req.ClusterId), zap.String("changefeed_id", req.ChangefeedId))
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareCDCErrorMessage(ctx, err)
	}()
	span.SetAttributes(attribute.String("clusterId", req.ClusterId), attribute.String("changefeedId", req.ChangefeedId))
	logger.Info("start to scale changefeed")

	changefeedVo, err := s.cdc.ScaleChangefeed(ctx, req.ClusterId, req.ChangefeedId, req.Rcu)
	if err != nil {
		logger.Error("failed to scale changefeed", zap.Error(err))
		return nil, err
	}
	return changefeedVo.ConvertToV1beta1(), nil
}

// buildChangefeedVo will check:
// 1. is enum value valid
// 2. is required field set
// 3. the validate will not be checked
func buildChangefeedVo(req *v1beta1.CreateChangefeedReq) (*vo.Changefeed, bizerr.BizError) {
	if req == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "empty request")
	}
	changefeed := &vo.Changefeed{
		ClusterID: req.ClusterId,
		Arch:      buildCDCArchFromV1beta1(req.Arch),
		RCU:       req.Rcu,
		Name:      req.DisplayName,
	}

	sink, err := buildSinkInfoVo(req.Sink, req.ClusterId, false)
	if err != nil {
		return nil, err
	}
	changefeed.Sink = sink

	filter, err := buildCDCFilterVo(req.Filter)
	if err != nil {
		return nil, err
	}
	changefeed.Filter = filter

	startPosition, err := buildStartPositionVo(req.StartPosition)
	if err != nil {
		return nil, err
	}
	changefeed.StartPosition = startPosition

	return changefeed, nil
}

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

// buildChangefeedForUpdate will check:
// 1. is enum value valid
// 2. is required field set
// 3. the validate will not be checked
func buildChangefeedForUpdate(req *v1beta1.EditChangefeedReq) (*vo.Changefeed, bizerr.BizError) {
	if req == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "empty request")
	}
	changefeed := &vo.Changefeed{
		ClusterID:    req.ClusterId,
		ChangefeedID: req.ChangefeedId,
		Name:         req.DisplayName,
	}

	sink, err := buildSinkInfoVo(req.Sink, req.ClusterId, false)
	if err != nil {
		return nil, err
	}
	changefeed.Sink = sink

	filter, err := buildCDCFilterVo(req.Filter)
	if err != nil {
		return nil, err
	}
	changefeed.Filter = filter

	return changefeed, nil
}

func buildSinkInfoVo(sink *v1beta1.SinkInfo, clusterId string, isTest bool) (*vo.SinkInfo, bizerr.BizError) {
	if sink == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "empty sink")
	}
	sinkInfo := &vo.SinkInfo{}
	switch sink.Type {
	case v1beta1.ChangefeedType_KAFKA:
		sinkInfo.Type = constant.ChangefeedTypeKafka
		kafka, err := buildKafkaInfoVo(sink.GetKafka(), isTest)
		if err != nil {
			return nil, err
		}
		sinkInfo.Kafka = kafka
	case v1beta1.ChangefeedType_MYSQL:
		sinkInfo.Type = constant.ChangefeedTypeMySQL
		mysql, err := buildMySQLInfoVo(sink.GetMysql())
		if err != nil {
			return nil, err
		}
		sinkInfo.MySQL = mysql
	case v1beta1.ChangefeedType_CLOUD_STORAGE:
		sinkInfo.Type = constant.ChangefeedTypeCloudStorage
		cloudStorage, err := buildCloudStorageInfoFromV1beta1(sink.GetCloudStorage(), clusterId, isTest)
		if err != nil {
			return nil, err
		}
		sinkInfo.CloudStorage = cloudStorage
	default:
		return nil, bizerr.New(bizerr.ErrCodeParams, "unsupported sink type:%s", sink.Type.String())
	}
	return sinkInfo, nil
}

func buildCloudStorageInfoFromV1beta1(
	cloudStorage *v1beta1.CloudStorage,
	clusterId string,
	isTest bool,
) (*entity.CloudStorageInfo, bizerr.BizError) {
	if cloudStorage == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cloud storage is required")
	}
	if cloudStorage.Storage == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cloud storage is required")
	}
	storage, _, err := vo.CommonToCloudStorage(cloudStorage.Storage, clusterId)
	if err != nil {
		return nil, err
	}
	info := &entity.CloudStorageInfo{
		CloudStorage:    storage,
		ColumnSelectors: make([]entity.ColumnSelector, 0, len(cloudStorage.ColumnSelectors)),
	}
	for _, selector := range cloudStorage.ColumnSelectors {
		if selector == nil {
			continue
		}
		info.ColumnSelectors = append(info.ColumnSelectors, entity.ColumnSelector{
			Matcher: selector.Matcher,
			Columns: selector.Columns,
		})
	}
	if isTest {
		return info, nil
	}

	if cloudStorage.DataFormat == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cloud storage data format is required")
	}

	// interval in seconds and size in mib
	info.IntervalInSeconds = int(cloudStorage.IntervalInSeconds)
	info.SizeInMiB = int(cloudStorage.SizeInMib)
	if info.IntervalInSeconds == 0 {
		info.IntervalInSeconds = 5
	}
	if info.SizeInMiB == 0 {
		info.SizeInMiB = 64
	}
	// date separator
	switch cloudStorage.DateSeparator {
	case v1beta1.CloudStorage_DateSeparator_UNSPECIFIED:
		info.DateSeparator = constant.DateSeparatorDay
	case v1beta1.CloudStorage_DateSeparator_NONE:
		info.DateSeparator = constant.DateSeparatorNone
	case v1beta1.CloudStorage_DateSeparator_YEAR:
		info.DateSeparator = constant.DateSeparatorYear
	case v1beta1.CloudStorage_DateSeparator_MONTH:
		info.DateSeparator = constant.DateSeparatorMonth
	case v1beta1.CloudStorage_DateSeparator_DAY:
		info.DateSeparator = constant.DateSeparatorDay
	default:
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"unsupported cloud storage date separator: %s",
			cloudStorage.DateSeparator.String(),
		)
	}

	// data format
	dataFormat := entity.CloudStorageDataFormat{}
	switch cloudStorage.DataFormat.Protocol {
	case v1beta1.CloudStorageProtocol_CANAL_JSON:
		dataFormat.Protocol = constant.CloudStorageProtocolCanalJSON
		dataFormat.CanalJSONConfig = &entity.CloudStorageCanalJSONCfg{
			EnableTiDBExtension: cloudStorage.DataFormat.GetCanalJsonConfig().GetEnableTidbExtension(),
		}
	case v1beta1.CloudStorageProtocol_CSV:
		dataFormat.Protocol = constant.CloudStorageProtocolCSV
		dataFormat.CSVConfig = &entity.CloudStorageCSVConfig{
			Encoding: constant.CloudStorageBinaryEncodingBase64,
		}
		csvConfig := cloudStorage.DataFormat.GetCsvConfig()
		if csvConfig != nil {
			var encoding constant.CloudStorageBinaryEncoding
			switch csvConfig.Encoding {
			case v1beta1.CloudStorage_BinaryEncoding_HEX:
				encoding = constant.CloudStorageBinaryEncodingHex
			case v1beta1.CloudStorage_BinaryEncoding_UNSPECIFIED:
				encoding = constant.CloudStorageBinaryEncodingBase64
			case v1beta1.CloudStorage_BinaryEncoding_BASE64:
				encoding = constant.CloudStorageBinaryEncodingBase64
			default:
				return nil, bizerr.New(
					bizerr.ErrCodeParams,
					"unsupported cloud storage csv encoding: %s",
					csvConfig.Encoding.String(),
				)
			}
			dataFormat.CSVConfig = &entity.CloudStorageCSVConfig{
				Delimiter:       csvConfig.Delimiter,
				Quote:           csvConfig.Quote,
				NullValue:       csvConfig.NullValue,
				IncludeCommitTs: csvConfig.IncludeCommitTs,
				Encoding:        encoding,
			}
		}
	default:
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"unsupported cloud storage protocol: %s",
			cloudStorage.DataFormat.Protocol.String(),
		)
	}
	info.DataFormat = dataFormat
	// output raw change event
	info.OutputRawChangeEvent = cloudStorage.OutputRawChangeEvent
	// output column id (adds source column IDs to CSV schema files)
	info.OutputColumnID = cloudStorage.OutputColumnId
	return info, nil
}

func buildKafkaInfoVo(kafka *v1beta1.Kafka, isTest bool) (*entity.KafkaInfo, bizerr.BizError) {
	if kafka == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "kafka info is required")
	}
	if kafka.Authentication == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "authentication info is required")
	}
	if kafka.Network == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "network info is required")
	}

	kafkaInfo := &entity.KafkaInfo{}

	// Network info
	switch kafka.Network.NetworkType {
	case v1beta1.KafkaNetworkType_PUBLIC:
		if kafka.Network.GetPublicEndpoints() == "" {
			return nil, bizerr.New(bizerr.ErrCodeParams, "empty broker endpoints")
		}
		kafkaInfo.Network = entity.KafkaNetwork{
			NetworkType:     constant.KafkaNetworkTypePublic,
			PublicEndpoints: kafka.Network.GetPublicEndpoints(),
		}
	case v1beta1.KafkaNetworkType_PRIVATE_LINK:
		if kafka.Network.GetPrivateLink().GetPrivateLinkConnectionId() == "" {
			return nil, bizerr.New(bizerr.ErrCodeParams, "private link id is required")
		}
		kafkaInfo.Network = entity.KafkaNetwork{
			NetworkType:             constant.KafkaNetworkTypePrivateLink,
			PrivateLinkConnectionId: kafka.Network.GetPrivateLink().PrivateLinkConnectionId,
			PrivateLinkPort:         kafka.Network.GetPrivateLink().Port,
		}
	default:
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"unsupported kafka network type: %s",
			kafka.Network.NetworkType.String(),
		)
	}

	// Authentication
	kafkaInfo.Authentication = entity.Authentication{
		UserName:  kafka.Authentication.Username,
		Password:  kafka.Authentication.Password,
		EnableTLS: kafka.Authentication.EnableTls,
	}

	if kafka.Authentication.ServerName != "" {
		if !kafka.Authentication.EnableTls {
			return nil, bizerr.New(bizerr.ErrCodeParams, "server name can only be set when TLS is enabled")
		}
		kafkaInfo.Authentication.ServerName = kafka.Authentication.ServerName
	}

	// Set auth type
	switch kafka.Authentication.AuthType {
	case v1beta1.KafkaAuthType_DISABLE:
		kafkaInfo.Authentication.AuthType = constant.KafkaAuthTypeDISABLE
	case v1beta1.KafkaAuthType_SASL_PLAIN:
		kafkaInfo.Authentication.AuthType = constant.KafkaAuthTypeSASLPlain
		if kafka.Authentication.Username == "" || kafka.Authentication.Password == "" {
			return nil, bizerr.New(
				bizerr.ErrCodeParams,
				"username and password must be set for SASL_PLAIN authentication",
			)
		}
	case v1beta1.KafkaAuthType_SASL_SCRAM_SHA_256:
		kafkaInfo.Authentication.AuthType = constant.KafkaAuthTypeSASLScram256
		if kafka.Authentication.Username == "" || kafka.Authentication.Password == "" {
			return nil, bizerr.New(
				bizerr.ErrCodeParams,
				"username and password must be set for SASL_SCRAM_SHA_256 authentication",
			)
		}
	case v1beta1.KafkaAuthType_SASL_SCRAM_SHA_512:
		kafkaInfo.Authentication.AuthType = constant.KafkaAuthTypeSASLScram512
		if kafka.Authentication.Username == "" || kafka.Authentication.Password == "" {
			return nil, bizerr.New(
				bizerr.ErrCodeParams,
				"username and password must be set for SASL_SCRAM_SHA_512 authentication",
			)
		}
	default:
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"unsupported kafka auth type: %s",
			kafka.Authentication.AuthType.String(),
		)
	}

	// Partition Num and replication factor is optional for test
	if kafka.TopicPartitionConfig != nil {
		kafkaInfo.TopicPartitionConfig = entity.TopicPartitionConfig{
			ReplicationFactor: kafka.TopicPartitionConfig.ReplicationFactor,
			PartitionNum:      kafka.TopicPartitionConfig.PartitionNum,
		}
	}

	if isTest {
		return kafkaInfo, nil
	}

	if kafka.Broker == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "broker info is required")
	}
	if kafka.TopicPartitionConfig == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "topic partition config is required")
	}
	if kafka.DataFormat == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "data format info is required")
	}

	// Broker
	switch kafka.Broker.KafkaVersion {
	case v1beta1.KafkaVersion_VERSION_2XX:
		kafkaInfo.Broker.KafkaVersion = constant.KafkaVersion2XX
	case v1beta1.KafkaVersion_VERSION_3XX:
		kafkaInfo.Broker.KafkaVersion = constant.KafkaVersion3XX
	default:
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"unsupported kafka version: %s",
			kafka.Broker.KafkaVersion.String(),
		)
	}

	// Set compression type
	switch kafka.Broker.Compression {
	case v1beta1.KafkaCompressionType_NONE:
		kafkaInfo.Broker.Compression = constant.KafkaCompressionTypeNone
	case v1beta1.KafkaCompressionType_GZIP:
		kafkaInfo.Broker.Compression = constant.KafkaCompressionTypeGZIP
	case v1beta1.KafkaCompressionType_SNAPPY:
		kafkaInfo.Broker.Compression = constant.KafkaCompressionTypeSnappy
	case v1beta1.KafkaCompressionType_LZ4:
		kafkaInfo.Broker.Compression = constant.KafkaCompressionTypeLZ4
	case v1beta1.KafkaCompressionType_ZSTD:
		kafkaInfo.Broker.Compression = constant.KafkaCompressionTypeZSTD
	default:
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"unsupported kafka compression type: %s",
			kafka.Broker.Compression.String(),
		)
	}

	// Data format
	dataFormat := entity.DataFormat{
		EnableTiDBExtension: kafka.DataFormat.EnableTidbExtension,
	}

	// Set protocol
	switch kafka.DataFormat.Protocol {
	case v1beta1.KafkaProtocol_CANAL_JSON:
		dataFormat.Protocol = constant.CDCProtocolCanalJSON
	case v1beta1.KafkaProtocol_OPEN_PROTOCOL:
		dataFormat.Protocol = constant.CDCProtocolOpen
	case v1beta1.KafkaProtocol_DEBEZIUM:
		dataFormat.Protocol = constant.CDCProtocolDebezium
		dataFormat.DebeziumConfig = &entity.DebeziumConfig{
			DisableOutputOldValue: kafka.DataFormat.GetDebeziumConfig().GetDisableOutputOldValue(),
			DisableSchema:         kafka.DataFormat.GetDebeziumConfig().GetDisableSchema(),
		}
	case v1beta1.KafkaProtocol_AVRO:
		dataFormat.Protocol = constant.CDCProtocolAvro
		if kafka.DataFormat.AvroConfig == nil {
			return nil, bizerr.New(bizerr.ErrCodeParams, "Avro config must be set for Avro protocol")
		}
		avroConfig := &entity.AvroConfig{}
		// Set decimal handling mode
		switch kafka.DataFormat.AvroConfig.DecimalHandlingMode {
		case v1beta1.DecimalHandlingMode_PRECISE:
			avroConfig.DecimalHandlingMode = constant.DecimalHandlingModePrecise
		case v1beta1.DecimalHandlingMode_STRING:
			avroConfig.DecimalHandlingMode = constant.DecimalHandlingModeString
		default:
			return nil, bizerr.New(
				bizerr.ErrCodeParams,
				"unsupported decimal handling mode: %s",
				kafka.DataFormat.AvroConfig.DecimalHandlingMode.String(),
			)
		}
		// Set bigint unsigned handling mode
		switch kafka.DataFormat.AvroConfig.BigintUnsignedHandlingMode {
		case v1beta1.BigintUnsignedHandlingMode_LONG:
			avroConfig.BigintUnsignedHandlingMode = constant.BigintUnsignedHandlingModeLong
		case v1beta1.BigintUnsignedHandlingMode_STRING:
			avroConfig.BigintUnsignedHandlingMode = constant.BigintUnsignedHandlingModeString
		default:
			return nil, bizerr.New(
				bizerr.ErrCodeParams,
				"unsupported bigint unsigned handling mode: %s",
				kafka.DataFormat.AvroConfig.BigintUnsignedHandlingMode.String(),
			)
		}
		// Set schema registry
		if kafka.DataFormat.AvroConfig.GetConfluentSchemaRegistry() != nil {
			avroConfig.ConfluentSchemaRegistry = &entity.ConfluentSchemaRegistry{
				SchemaRegistryEndpoints: kafka.DataFormat.AvroConfig.GetConfluentSchemaRegistry().Endpoint,
				EnableHTTPAuth:          kafka.DataFormat.AvroConfig.GetConfluentSchemaRegistry().EnableHttpAuth,
				UserName:                kafka.DataFormat.AvroConfig.GetConfluentSchemaRegistry().Username,
				Password:                kafka.DataFormat.AvroConfig.GetConfluentSchemaRegistry().Password,
			}
			if avroConfig.ConfluentSchemaRegistry.SchemaRegistryEndpoints == "" {
				return nil, bizerr.New(bizerr.ErrCodeParams, "schema registry endpoints must be set")
			}
			if avroConfig.ConfluentSchemaRegistry.EnableHTTPAuth &&
				(avroConfig.ConfluentSchemaRegistry.UserName == "" || avroConfig.ConfluentSchemaRegistry.Password == "") {
				return nil, bizerr.New(
					bizerr.ErrCodeParams,
					"schema registry username and password must be set when enable http auth",
				)
			}
		} else if kafka.DataFormat.AvroConfig.GetAwsGlueSchemaRegistry() != nil {
			avroConfig.AWSGlueSchemaRegistry = &entity.AWSGlueSchemaRegistry{
				Region:          kafka.DataFormat.AvroConfig.GetAwsGlueSchemaRegistry().Region,
				Name:            kafka.DataFormat.AvroConfig.GetAwsGlueSchemaRegistry().Name,
				AccessKeyID:     kafka.DataFormat.AvroConfig.GetAwsGlueSchemaRegistry().AccessKeyId,
				SecretAccessKey: kafka.DataFormat.AvroConfig.GetAwsGlueSchemaRegistry().SecretAccessKey,
			}
			if avroConfig.AWSGlueSchemaRegistry.Region == "" {
				return nil, bizerr.New(bizerr.ErrCodeParams, "AWS Glue Schema Registry region must be set")
			}
			if avroConfig.AWSGlueSchemaRegistry.Name == "" {
				return nil, bizerr.New(bizerr.ErrCodeParams, "AWS Glue Schema Registry name must be set")
			}
			if avroConfig.AWSGlueSchemaRegistry.AccessKeyID == "" || avroConfig.AWSGlueSchemaRegistry.SecretAccessKey == "" {
				return nil, bizerr.New(bizerr.ErrCodeParams, "AWS Glue Schema Registry access key and secret access key must be set")
			}
		} else {
			return nil, bizerr.New(bizerr.ErrCodeParams, "schema registry config must be set for Avro protocol")
		}
		dataFormat.AvroConfig = avroConfig
	default:
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"unsupported CDC protocol: %s",
			kafka.DataFormat.Protocol.String(),
		)
	}
	kafkaInfo.DataFormat = dataFormat

	// Topic partition config
	topicConfig := entity.TopicPartitionConfig{
		DefaultTopic:      kafka.TopicPartitionConfig.DefaultTopic,
		TopicPrefix:       kafka.TopicPartitionConfig.TopicPrefix,
		Separator:         kafka.TopicPartitionConfig.Separator,
		TopicSuffix:       kafka.TopicPartitionConfig.TopicSuffix,
		ReplicationFactor: kafka.TopicPartitionConfig.ReplicationFactor,
		PartitionNum:      kafka.TopicPartitionConfig.PartitionNum,
	}

	if topicConfig.ReplicationFactor <= 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "replication factor must be greater than 0")
	}
	if topicConfig.PartitionNum <= 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "partition number must be greater than 0")
	}
	if topicConfig.DefaultTopic == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "topic must be set")
	}

	// Set dispatch type
	switch kafka.TopicPartitionConfig.DispatchType {
	case v1beta1.TopicDispatchType_ONE_TOPIC:
		topicConfig.DispatchType = constant.TopicDispatchTypeOneTopic
	case v1beta1.TopicDispatchType_BY_TABLE:
		topicConfig.DispatchType = constant.TopicDispatchTypeByTable
	case v1beta1.TopicDispatchType_BY_DATABASE:
		topicConfig.DispatchType = constant.TopicDispatchTypeByDatabase
	default:
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"unsupported topic dispatch type: %s",
			kafka.TopicPartitionConfig.DispatchType.String(),
		)
	}

	// Partition dispatchers
	if len(kafka.TopicPartitionConfig.PartitionDispatchers) == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "at least one partition dispatcher must be set")
	}
	dispatchers := make(
		[]*entity.PartitionDispatcherConfig,
		0,
		len(kafka.TopicPartitionConfig.PartitionDispatchers),
	)
	for _, d := range kafka.TopicPartitionConfig.PartitionDispatchers {
		if d == nil {
			continue
		}

		if len(d.Matcher) == 0 {
			return nil, bizerr.New(bizerr.ErrCodeParams, "matcher must be set for partition dispatcher")
		}

		dispatcher := &entity.PartitionDispatcherConfig{
			Matcher:   d.Matcher,
			IndexName: d.IndexName,
			Columns:   d.Columns,
		}

		// Set partition type
		switch d.PartitionType {
		case v1beta1.PartitionDispatcher_TABLE:
			dispatcher.PartitionType = constant.PartitionDispatcherTable
		case v1beta1.PartitionDispatcher_INDEX_VALUE:
			dispatcher.PartitionType = constant.PartitionDispatcherIndexValue
		case v1beta1.PartitionDispatcher_TS:
			dispatcher.PartitionType = constant.PartitionDispatcherTS
		case v1beta1.PartitionDispatcher_COLUMNS:
			dispatcher.PartitionType = constant.PartitionDispatcherColumns
			if len(d.Columns) == 0 {
				return nil, bizerr.New(bizerr.ErrCodeParams, "columns must be set for COLUMNS partition dispatcher")
			}
		default:
			return nil, bizerr.New(
				bizerr.ErrCodeParams,
				"unsupported partition dispatcher type: %s",
				d.PartitionType.String(),
			)
		}

		dispatchers = append(dispatchers, dispatcher)
	}
	topicConfig.PartitionDispatchers = dispatchers
	kafkaInfo.TopicPartitionConfig = topicConfig

	// Column selectors
	if len(kafka.ColumnSelectors) > 0 {
		selectors := make([]entity.ColumnSelector, 0, len(kafka.ColumnSelectors))
		for _, c := range kafka.ColumnSelectors {
			if c == nil {
				continue
			}
			selectors = append(selectors, entity.ColumnSelector{
				Matcher: c.Matcher,
				Columns: c.Columns,
			})
		}
		kafkaInfo.ColumnSelectors = selectors
	}

	// out put raw change events
	kafkaInfo.OutputRawChangeEvent = kafka.OutputRawChangeEvent
	return kafkaInfo, nil
}

func buildMySQLInfoVo(mysql *v1beta1.MySQL) (*entity.MySQLInfo, bizerr.BizError) {
	if mysql == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "mysql info is required")
	}
	if mysql.Authentication == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "auth info is required")
	}
	if mysql.Authentication.UserName == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "mysql user is required")
	}
	if mysql.Authentication.Password == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "mysql password is required")
	}
	if mysql.Network == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "network info is required")
	}

	mysqlInfo := &entity.MySQLInfo{
		Authentication: entity.MySQLAuthentication{
			User:      mysql.Authentication.UserName,
			Password:  mysql.Authentication.Password,
			EnableTLS: mysql.Authentication.EnableTls,
		},
	}

	switch mysql.Network.NetworkType {
	case v1beta1.MySQLNetworkType_PUBLIC:
		if mysql.Network.GetPublicEndpoint() == "" {
			return nil, bizerr.New(bizerr.ErrCodeParams, "mysql endpoint is required")
		}
		mysqlInfo.Network = entity.MySQLNetwork{
			NetworkType:    constant.MySQLNetworkTypePublic,
			PublicEndpoint: mysql.Network.GetPublicEndpoint(),
		}
	case v1beta1.MySQLNetworkType_PRIVATE_LINK:
		if mysql.Network.GetPrivateLink().GetPrivateLinkConnectionId() == "" {
			return nil, bizerr.New(bizerr.ErrCodeParams, "private link id is required")
		}
		if mysql.Network.GetPrivateLink().GetPort() == 0 {
			return nil, bizerr.New(bizerr.ErrCodeParams, "private link port is required")
		}
		mysqlInfo.Network = entity.MySQLNetwork{
			NetworkType:             constant.MySQLNetworkTypePrivateLink,
			PrivateLinkConnectionId: mysql.Network.GetPrivateLink().PrivateLinkConnectionId,
			PrivateLinkPort:         mysql.Network.GetPrivateLink().Port,
		}
	default:
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"unsupported mysql network type: %s",
			mysql.Network.NetworkType.String(),
		)
	}

	return mysqlInfo, nil
}

func buildCDCFilterVo(f *v1beta1.ChangefeedFilter) (*entity.CDCFilter, bizerr.BizError) {
	if f == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "empty filter")
	}
	if len(f.FilterRule) == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "filter rule must be set")
	}
	filter := &entity.CDCFilter{
		FilterRule:       f.FilterRule,
		CaseSensitive:    f.CaseSensitive,
		IgnoreTxnStartTs: f.IgnoreTxnStartTs,
		BDRMode:          f.BdrMode,
	}
	switch f.Mode {
	case v1beta1.TableMode_IGNORE_NOT_SUPPORT_TABLE:
		filter.Mode = constant.TableModeIgnoreNotSupportTable
	case v1beta1.TableMode_FORCE_SYNC:
		filter.Mode = constant.TableModeForceSync
	default:
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid table mode: %s", f.Mode.String())
	}
	if len(f.EventFilterRule) > 0 {
		efr := make([]*entity.EventFilterRule, 0, len(f.EventFilterRule))
		for _, e := range f.EventFilterRule {
			if e == nil {
				continue
			}
			efr = append(efr, &entity.EventFilterRule{
				Matcher:     e.Matcher,
				IgnoreEvent: e.IgnoreEvent,
			})
		}
		filter.EventFilterRule = efr
	}
	return filter, nil
}

func buildStartPositionVo(s *v1beta1.StartPosition) (*vo.StartPosition, bizerr.BizError) {
	if s == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "empty start position")
	}
	switch s.Mode {
	case v1beta1.StartMode_FROM_NOW:
		return &vo.StartPosition{
			Mode: constant.StartModeFromNow,
		}, nil
	case v1beta1.StartMode_FROM_TSO:
		if s.Tso == 0 {
			return nil, bizerr.New(bizerr.ErrCodeParams, "tso must be set for FROM_TSO start mode")
		}
		return &vo.StartPosition{
			Mode: constant.StartModeFromTSO,
			TSO:  s.Tso,
		}, nil
	case v1beta1.StartMode_FROM_TIME:
		if s.Time == nil {
			return nil, bizerr.New(bizerr.ErrCodeParams, "time must be set for FROM_TIME start mode")
		}
		time := s.Time.AsTime()
		return &vo.StartPosition{
			Mode: constant.StartModeFromTime,
			Time: &time,
		}, nil
	default:
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid start mode:%s", s.Mode.String())
	}
}

func (s *cdcV1beta1Service) ListChangefeeds(
	ctx context.Context,
	req *v1beta1.ListChangefeedsReq,
) (_ *v1beta1.Changefeeds, err error) {
	ctx, span := v1beta1Tracer.Start(ctx, "cdcV1beta1Service.ListChangefeeds")
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareCDCErrorMessage(ctx, err)
	}()
	span.SetAttributes(
		attribute.String("clusterId", req.ClusterId),
	)
	logger.Info("start to list changefeeds")

	pager := utils.NewAipPager()
	err = pager.Normalize(req)
	if err != nil {
		return nil, err
	}

	// check sink type
	sinkType := ""
	switch req.ChangefeedType {
	case v1beta1.ChangefeedType_KAFKA:
		sinkType = string(constant.ChangefeedTypeKafka)
	}

	// list changefeeds
	cvs, total, err := s.cdc.ListChangefeeds(ctx, &vo.ListChangefeedsOptions{
		ClusterID: req.ClusterId,
		Limit:     int32(pager.PageSize()),
		Offset:    int64(pager.Offset()),
		SinkType:  sinkType,
	})
	if err != nil {
		logger.Error("list changefeeds failed", zap.Error(err))
		return nil, err
	}

	// build response
	changefeeds := make([]*v1beta1.Changefeed, 0, len(cvs))
	for _, cv := range cvs {
		changefeeds = append(changefeeds, cv.ConvertToV1beta1())
	}
	return &v1beta1.Changefeeds{
		Changefeeds:   changefeeds,
		NextPageToken: pager.NextPageToken(total),
		TotalSize:     total,
	}, nil
}

func (s *cdcV1beta1Service) GetChangefeed(
	ctx context.Context,
	req *v1beta1.ChangefeedID,
) (_ *v1beta1.Changefeed, err error) {
	ctx, span := v1beta1Tracer.Start(ctx, "cdcV1beta1Service.GetChangefeed")
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", req.ClusterId), zap.String("changefeed_id", req.ChangefeedId))
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareCDCErrorMessage(ctx, err)
	}()
	span.SetAttributes(
		attribute.String("clusterId", req.ClusterId),
		attribute.String("changefeedId", req.ChangefeedId),
	)
	logger.Info("start to get changefeed")
	cv, err := s.cdc.GetChangefeed(ctx, req.ClusterId, req.ChangefeedId)
	if err != nil {
		logger.Error("failed to get changefeed", zap.Error(err))
		return nil, err
	}
	return cv.ConvertToV1beta1(), nil
}

func (s *cdcV1beta1Service) StartChangefeed(
	ctx context.Context,
	req *v1beta1.ChangefeedID,
) (_ *v1beta1.Changefeed, err error) {
	ctx, span := v1beta1Tracer.Start(ctx, "cdcV1beta1Service.StartChangefeed")
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", req.ClusterId), zap.String("changefeed_id", req.ChangefeedId))
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareCDCErrorMessage(ctx, err)
	}()
	span.SetAttributes(
		attribute.String("clusterId", req.ClusterId),
		attribute.String("changefeedId", req.ChangefeedId),
	)
	logger.Info("start to start changefeed")
	cv, err := s.cdc.StartChangefeed(ctx, req.ClusterId, req.ChangefeedId)
	if err != nil {
		logger.Error("failed to start changefeed", zap.Error(err))
		return nil, err
	}
	return cv.ConvertToV1beta1(), nil
}

func (s *cdcV1beta1Service) StopChangefeed(
	ctx context.Context,
	req *v1beta1.ChangefeedID,
) (_ *v1beta1.Changefeed, err error) {
	ctx, span := v1beta1Tracer.Start(ctx, "cdcV1beta1Service.StopChangefeed")
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", req.ClusterId), zap.String("changefeed_id", req.ChangefeedId))
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareCDCErrorMessage(ctx, err)
	}()
	span.SetAttributes(
		attribute.String("clusterId", req.ClusterId),
		attribute.String("changefeedId", req.ChangefeedId),
	)
	logger.Info("start to stop changefeed")
	cv, err := s.cdc.StopChangefeed(ctx, req.ClusterId, req.ChangefeedId)
	if err != nil {
		logger.Error("failed to stop changefeed", zap.Error(err))
		return nil, err
	}
	return cv.ConvertToV1beta1(), nil
}

func (s *cdcV1beta1Service) EditChangefeed(
	ctx context.Context,
	req *v1beta1.EditChangefeedReq,
) (_ *v1beta1.Changefeed, err error) {
	ctx, span := v1beta1Tracer.Start(ctx, "cdcV1beta1Service.EditChangefeed")
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", req.ClusterId), zap.String("changefeed_id", req.ChangefeedId))
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareCDCErrorMessage(ctx, err)
	}()
	span.SetAttributes(
		attribute.String("clusterId", req.ClusterId),
		attribute.String("changefeedId", req.ChangefeedId),
	)
	logger.Info("start to edit changefeed")

	changefeedVo, err := buildChangefeedForUpdate(req)
	if err != nil {
		logger.Error("failed to build changefeed VO", zap.Error(err))
		return nil, err
	}
	changefeedVo, err = s.cdc.EditChangefeed(ctx, changefeedVo)
	if err != nil {
		logger.Error("failed to edit changefeed", zap.Error(err))
		return nil, err
	}

	return changefeedVo.ConvertToV1beta1(), nil
}

func (s *cdcV1beta1Service) DeleteChangefeed(
	ctx context.Context,
	req *v1beta1.ChangefeedID,
) (_ *v1beta1.Changefeed, err error) {
	ctx, span := v1beta1Tracer.Start(ctx, "cdcV1beta1Service.DeleteChangefeed")
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", req.ClusterId), zap.String("changefeed_id", req.ChangefeedId))
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareCDCErrorMessage(ctx, err)
	}()
	span.SetAttributes(
		attribute.String("clusterId", req.ClusterId),
		attribute.String("changefeedId", req.ChangefeedId),
	)
	logger.Info("start to delete changefeed")
	cv, err := s.cdc.DeleteChangefeed(ctx, req.ClusterId, req.ChangefeedId)
	if err != nil {
		logger.Error("failed to delete changefeed", zap.Error(err))
		return nil, err
	}
	return cv.ConvertToV1beta1(), nil
}

func (s *cdcV1beta1Service) TestChangefeed(
	ctx context.Context,
	req *v1beta1.TestChangefeedReq,
) (_ *emptypb.Empty, err error) {
	ctx, span := v1beta1Tracer.Start(ctx, "cdcV1beta1Service.TestChangefeed")
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareCDCErrorMessage(ctx, err)
	}()
	span.SetAttributes(
		attribute.String("clusterId", req.ClusterId),
	)
	logger.Info("start to test changefeed")

	if req.ClusterId == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster ID is required")
	}
	sinkInfo, err := buildSinkInfoVo(req.Sink, req.ClusterId, true)
	if err != nil {
		logger.Error("failed to build sink info VO", zap.Error(err))
		return nil, err
	}

	err = s.cdc.TestChangefeed(ctx, &vo.Changefeed{
		ClusterID: req.ClusterId,
		Sink:      sinkInfo,
	})
	return &emptypb.Empty{}, err
}

func (s *cdcV1beta1Service) DescribeSchemaTable(
	ctx context.Context,
	req *v1beta1.DescribeSchemaTableReq,
) (_ *v1beta1.DescribeSchemaTableResp, err error) {
	ctx, span := v1beta1Tracer.Start(ctx, "cdcV1beta1Service.DescribeSchemaTable")
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareCDCErrorMessage(ctx, err)
	}()
	span.SetAttributes(
		attribute.String("clusterId", req.ClusterId),
	)
	logger.Info("start to describe schema table")

	if len(req.FilterRules) == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "filter rule is required")
	}

	handleKeySchema, nonHandleKeySchema, safePoint, err := s.cdc.DescribeSchemaTable(
		ctx,
		req.ClusterId,
		req.FilterRules,
		req.CaseSensitive,
	)
	if err != nil {
		logger.Error("failed to describe schema table", zap.Error(err))
		return nil, err
	}

	return &v1beta1.DescribeSchemaTableResp{
		ClusterId:          req.ClusterId,
		HandleKeyTables:    handleKeySchema,
		NonHandleKeyTables: nonHandleKeySchema,
		GcSafePoint:        timestamppb.New(utils.TSOToTime(safePoint)),
	}, nil
}

func (s *cdcV1beta1Service) ListChangefeedSpecifications(
	ctx context.Context,
	req *v1beta1.ListChangefeedSpecificationsReq,
) (_ *v1beta1.ListChangefeedSpecificationsResp, err error) {
	ctx, span := v1beta1Tracer.Start(ctx, "cdcV1beta1Service.ListChangefeedSpecifications")
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareCDCErrorMessage(ctx, err)
	}()
	span.SetAttributes(attribute.String("clusterId", req.ClusterId))

	specs, err := s.cdc.ListChangefeedSpecifications(ctx, req.ClusterId)
	if err != nil {
		logger.Error("failed to list changefeed specifications", zap.Error(err))
		return nil, err
	}
	items := make([]*v1beta1.ChangefeedSpecification, 0, len(specs))
	for _, spec := range specs {
		items = append(items, spec.ConvertToV1beta1())
	}
	return &v1beta1.ListChangefeedSpecificationsResp{
		Items: items,
		Total: uint32(len(items)),
	}, nil
}