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/console"
	"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"
	"github.com/tidbcloud/tiinf/sdk/auditlog"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/timestamppb"
)

var consoleCDCTracer = otel.Tracer("cdc.console")

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

func NewChangefeedConsoleService(
	cdc service.CDCService,
	account facade.AccountFacade,
	clusterProxyProvider facade.ClusterProxyProviderFacade,
	mgmt facade.MgmtSvcFacade,
) console.ChangefeedServiceServer {
	return &changefeedConsoleService{
		cdc:                  cdc,
		account:              account,
		clusterProxyProvider: clusterProxyProvider,
		mgmt:                 mgmt,
	}
}

func (s *changefeedConsoleService) CreateChangefeed(
	ctx context.Context,
	req *console.CreateChangefeedReq,
) (resp *console.Changefeed, err error) {
	ctx, span := consoleCDCTracer.Start(ctx, "changefeedConsoleService.CreateChangefeed")
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))
	var changefeedVo *vo.Changefeed
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareCDCErrorMessage(ctx, err)
		go recordCreateChangefeedAuditLog(ctx, req, changefeedVo, err)
	}()
	span.SetAttributes(attribute.String("clusterId", req.ClusterId))
	logger.Info("start to create changefeed")

	changefeedVo, err = buildConsoleChangefeedVo(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.ConvertToConsole(), nil
}

func (s *changefeedConsoleService) EditChangefeed(
	ctx context.Context,
	req *console.EditChangefeedReq,
) (resp *console.Changefeed, err error) {
	ctx, span := consoleCDCTracer.Start(ctx, "changefeedConsoleService.EditChangefeed")
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", req.ClusterId), zap.String("changefeed_id", req.ChangefeedId))
	var changefeedVo *vo.Changefeed
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareCDCErrorMessage(ctx, err)
		go recordEditChangefeedAuditLog(ctx, req, changefeedVo, err)
	}()
	span.SetAttributes(
		attribute.String("clusterId", req.ClusterId),
		attribute.String("changefeedId", req.ChangefeedId),
	)
	logger.Info("start to edit changefeed")

	changefeedVo, err = buildConsoleChangefeedForUpdate(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.ConvertToConsole(), nil
}

func (s *changefeedConsoleService) ScaleChangefeed(
	ctx context.Context,
	req *console.ScaleChangefeedReq,
) (resp *console.Changefeed, err error) {
	ctx, span := consoleCDCTracer.Start(ctx, "changefeedConsoleService.ScaleChangefeed")
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", req.ClusterId), zap.String("changefeed_id", req.ChangefeedId))
	var changefeedVo *vo.Changefeed
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareCDCErrorMessage(ctx, err)
		go recordScaleChangefeedAuditLog(ctx, req, changefeedVo, 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.ConvertToConsole(), nil
}

func (s *changefeedConsoleService) ListChangefeedSpecifications(
	ctx context.Context,
	req *console.ListChangefeedSpecificationsReq,
) (_ *console.ListChangefeedSpecificationsResp, err error) {
	ctx, span := consoleCDCTracer.Start(ctx, "changefeedConsoleService.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([]*console.ChangefeedSpecification, 0, len(specs))
	for _, spec := range specs {
		items = append(items, spec.ConvertToConsole())
	}
	return &console.ListChangefeedSpecificationsResp{
		Items: items,
		Total: uint32(len(items)),
	}, nil
}

func (s *changefeedConsoleService) ListChangefeeds(
	ctx context.Context,
	req *console.ListChangefeedsReq,
) (_ *console.Changefeeds, err error) {
	ctx, span := consoleCDCTracer.Start(ctx, "changefeedConsoleService.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.NewPagerNumberPager()
	if err := pager.Normalize(req); err != nil {
		return nil, err
	}

	sinkType := ""
	switch req.ChangefeedType {
	case console.ChangefeedType_KAFKA:
		sinkType = string(constant.ChangefeedTypeKafka)
	case console.ChangefeedType_MYSQL:
		sinkType = string(constant.ChangefeedTypeMySQL)
	}

	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
	}

	res := make([]*console.Changefeed, 0, len(cvs))
	for _, cv := range cvs {
		res = append(res, cv.ConvertToConsole())
	}
	return &console.Changefeeds{
		Changefeeds:   res,
		NextPageToken: pager.NextPageToken(total),
		TotalSize:     total,
	}, nil
}

func (s *changefeedConsoleService) GetChangefeed(
	ctx context.Context,
	req *console.ChangefeedID,
) (_ *console.Changefeed, err error) {
	ctx, span := consoleCDCTracer.Start(ctx, "changefeedConsoleService.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.ConvertToConsole(), nil
}

func (s *changefeedConsoleService) StartChangefeed(
	ctx context.Context,
	req *console.ChangefeedID,
) (resp *console.Changefeed, err error) {
	ctx, span := consoleCDCTracer.Start(ctx, "changefeedConsoleService.StartChangefeed")
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", req.ClusterId), zap.String("changefeed_id", req.ChangefeedId))
	var cv *vo.Changefeed
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareCDCErrorMessage(ctx, err)
		go recordChangefeedIDAuditLog(ctx, auditlog.EventStartChangefeed, req, cv, 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.ConvertToConsole(), nil
}

func (s *changefeedConsoleService) StopChangefeed(
	ctx context.Context,
	req *console.ChangefeedID,
) (resp *console.Changefeed, err error) {
	ctx, span := consoleCDCTracer.Start(ctx, "changefeedConsoleService.StopChangefeed")
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", req.ClusterId), zap.String("changefeed_id", req.ChangefeedId))
	var cv *vo.Changefeed
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareCDCErrorMessage(ctx, err)
		go recordChangefeedIDAuditLog(ctx, auditlog.EventStopChangefeed, req, cv, 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.ConvertToConsole(), nil
}

func (s *changefeedConsoleService) DeleteChangefeed(
	ctx context.Context,
	req *console.ChangefeedID,
) (resp *console.Changefeed, err error) {
	ctx, span := consoleCDCTracer.Start(ctx, "changefeedConsoleService.DeleteChangefeed")
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", req.ClusterId), zap.String("changefeed_id", req.ChangefeedId))
	var cv *vo.Changefeed
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareCDCErrorMessage(ctx, err)
		go recordChangefeedIDAuditLog(ctx, auditlog.EventDeleteChangefeed, req, cv, 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.ConvertToConsole(), nil
}

func (s *changefeedConsoleService) TestChangefeed(
	ctx context.Context,
	req *console.TestChangefeedReq,
) (_ *console.TestChangefeedResp, err error) {
	ctx, span := consoleCDCTracer.Start(ctx, "changefeedConsoleService.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 := buildSinkInfoVoFromConsole(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 &console.TestChangefeedResp{}, err
}

func (s *changefeedConsoleService) DescribeSchemaTable(
	ctx context.Context,
	req *console.DescribeSchemaTableReq,
) (_ *console.DescribeSchemaTableResp, err error) {
	ctx, span := consoleCDCTracer.Start(ctx, "changefeedConsoleService.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 &console.DescribeSchemaTableResp{
		ClusterId:          req.ClusterId,
		HandleKeyTables:    handleKeySchema,
		NonHandleKeyTables: nonHandleKeySchema,
		GcSafePoint:        timestamppb.New(utils.TSOToTime(safePoint)),
	}, nil
}

func buildConsoleChangefeedVo(req *console.CreateChangefeedReq) (*vo.Changefeed, bizerr.BizError) {
	if req == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "empty request")
	}
	cf := &vo.Changefeed{
		ClusterID: req.ClusterId,
		Arch:      buildCDCArchFromConsole(req.Arch),
		RCU:       req.Rcu,
		Name:      req.DisplayName,
	}
	sink, err := buildSinkInfoVoFromConsole(req.Sink, req.ClusterId, false)
	if err != nil {
		return nil, err
	}
	cf.Sink = sink

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

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

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

func buildConsoleChangefeedForUpdate(req *console.EditChangefeedReq) (*vo.Changefeed, bizerr.BizError) {
	if req == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "empty request")
	}
	cf := &vo.Changefeed{
		ClusterID:    req.ClusterId,
		ChangefeedID: req.ChangefeedId,
		Name:         req.DisplayName,
	}
	sink, err := buildSinkInfoVoFromConsole(req.Sink, req.ClusterId, false)
	if err != nil {
		return nil, err
	}
	cf.Sink = sink

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

func buildSinkInfoVoFromConsole(sink *console.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 console.ChangefeedType_KAFKA:
		sinkInfo.Type = constant.ChangefeedTypeKafka
		kafka, err := buildKafkaInfoFromConsole(sink.GetKafka(), isTest)
		if err != nil {
			return nil, err
		}
		sinkInfo.Kafka = kafka
	case console.ChangefeedType_MYSQL:
		sinkInfo.Type = constant.ChangefeedTypeMySQL
		mysql, err := buildMySQLInfoFromConsole(sink.GetMysql())
		if err != nil {
			return nil, err
		}
		sinkInfo.MySQL = mysql
	case console.ChangefeedType_CLOUD_STORAGE:
		sinkInfo.Type = constant.ChangefeedTypeCloudStorage
		cloudStorage, err := buildCloudStorageInfoFromConsole(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 buildCloudStorageInfoFromConsole(
	cloudStorage *console.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 console.CloudStorage_DateSeparator_UNSPECIFIED:
		info.DateSeparator = constant.DateSeparatorDay
	case console.CloudStorage_DateSeparator_NONE:
		info.DateSeparator = constant.DateSeparatorNone
	case console.CloudStorage_DateSeparator_YEAR:
		info.DateSeparator = constant.DateSeparatorYear
	case console.CloudStorage_DateSeparator_MONTH:
		info.DateSeparator = constant.DateSeparatorMonth
	case console.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 console.CloudStorageProtocol_CANAL_JSON:
		dataFormat.Protocol = constant.CloudStorageProtocolCanalJSON
		dataFormat.CanalJSONConfig = &entity.CloudStorageCanalJSONCfg{
			EnableTiDBExtension: cloudStorage.DataFormat.GetCanalJsonConfig().GetEnableTidbExtension(),
		}
	case console.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 console.CloudStorage_BinaryEncoding_HEX:
				encoding = constant.CloudStorageBinaryEncodingHex
			case console.CloudStorage_BinaryEncoding_UNSPECIFIED:
				encoding = constant.CloudStorageBinaryEncodingBase64
			case console.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: add source column IDs to CSV schema files so a downstream
	// consumer can map columns by ID across DDL changes.
	info.OutputColumnID = cloudStorage.OutputColumnId
	return info, nil
}

func buildKafkaInfoFromConsole(kafka *console.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 console.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 console.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,
	}
	switch kafka.Authentication.AuthType {
	case console.KafkaAuthType_DISABLE:
		kafkaInfo.Authentication.AuthType = constant.KafkaAuthTypeDISABLE
	case console.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 console.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 console.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 console.KafkaVersion_VERSION_2XX:
		kafkaInfo.Broker.KafkaVersion = constant.KafkaVersion2XX
	case console.KafkaVersion_VERSION_3XX:
		kafkaInfo.Broker.KafkaVersion = constant.KafkaVersion3XX
	default:
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"unsupported kafka version: %s",
			kafka.Broker.KafkaVersion.String(),
		)
	}
	switch kafka.Broker.Compression {
	case console.KafkaCompressionType_NONE:
		kafkaInfo.Broker.Compression = constant.KafkaCompressionTypeNone
	case console.KafkaCompressionType_GZIP:
		kafkaInfo.Broker.Compression = constant.KafkaCompressionTypeGZIP
	case console.KafkaCompressionType_SNAPPY:
		kafkaInfo.Broker.Compression = constant.KafkaCompressionTypeSnappy
	case console.KafkaCompressionType_LZ4:
		kafkaInfo.Broker.Compression = constant.KafkaCompressionTypeLZ4
	case console.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,
	}
	switch kafka.DataFormat.Protocol {
	case console.KafkaProtocol_CANAL_JSON:
		dataFormat.Protocol = constant.CDCProtocolCanalJSON
	case console.KafkaProtocol_OPEN_PROTOCOL:
		dataFormat.Protocol = constant.CDCProtocolOpen
	case console.KafkaProtocol_DEBEZIUM:
		dataFormat.Protocol = constant.CDCProtocolDebezium
		dataFormat.DebeziumConfig = &entity.DebeziumConfig{
			DisableOutputOldValue: kafka.DataFormat.GetDebeziumConfig().GetDisableOutputOldValue(),
			DisableSchema:         kafka.DataFormat.GetDebeziumConfig().GetDisableSchema(),
		}
	case console.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 console.DecimalHandlingMode_PRECISE:
			avroConfig.DecimalHandlingMode = constant.DecimalHandlingModePrecise
		case console.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 console.BigintUnsignedHandlingMode_LONG:
			avroConfig.BigintUnsignedHandlingMode = constant.BigintUnsignedHandlingModeLong
		case console.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 console.TopicDispatchType_ONE_TOPIC:
		topicConfig.DispatchType = constant.TopicDispatchTypeOneTopic
	case console.TopicDispatchType_BY_TABLE:
		topicConfig.DispatchType = constant.TopicDispatchTypeByTable
	case console.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 console.PartitionDispatcher_TABLE:
			dispatcher.PartitionType = constant.PartitionDispatcherTable
		case console.PartitionDispatcher_INDEX_VALUE:
			dispatcher.PartitionType = constant.PartitionDispatcherIndexValue
		case console.PartitionDispatcher_TS:
			dispatcher.PartitionType = constant.PartitionDispatcherTS
		case console.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 buildMySQLInfoFromConsole(mysql *console.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 console.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 console.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 buildFilterFromConsole(f *console.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 console.TableMode_IGNORE_NOT_SUPPORT_TABLE:
		filter.Mode = constant.TableModeIgnoreNotSupportTable
	case console.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 buildStartPositionFromConsole(s *console.StartPosition) (*vo.StartPosition, bizerr.BizError) {
	if s == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "empty start position")
	}
	switch s.Mode {
	case console.StartMode_FROM_NOW:
		return &vo.StartPosition{
			Mode: constant.StartModeFromNow,
		}, nil
	case console.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 console.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())
	}
}