package console

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/url"
	"regexp"
	"strings"
	"time"

	"github.com/Masterminds/semver/v3"
	"github.com/aws/aws-sdk-go/aws/session"
	tfilter "github.com/pingcap/tidb/pkg/util/table-filter"
	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
	tiflowConfig "github.com/pingcap/tiflow/pkg/config"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/config"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/models/private_link"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/utils"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	internal_config "github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/internal/filtervalidator"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/s3"
	"github.com/tidbcloud/pkgv2/bizerr"
)

const (
	endpointPartsLength           = 2
	maxFilterRuleNum              = 100
	maxFilterRuleLength           = 50
	maxTotalTableFilterRuleLength = 5000
	maxEventFilterRuleNum         = 10

	privateLinkEndpointAddress = "%s:%d"

	CR                   = '\r'
	LF                   = '\n'
	DefaultFlushInterval = "1m"
	MinFlushInterval     = "2s"
	MaxFlushInterval     = "10m"
	DefaultFileSize      = 67108864 // bytes
	MinFileSize          = 1048576
	MaxFileSize          = 536870912

	fakeMySQLEndpoint = "mysql:3306"

	azureBlobStorageHostname = "blob.core.windows.net"
)

type versionConstraintType string

const (
	mysqlSinkConstraintType            versionConstraintType = "mysql sink"
	kafkaSinkConstraintType            versionConstraintType = "kafka sink"
	kafkaAuthTypeSinkConstraintType    versionConstraintType = "kafka auth type"
	avroFormatConstraintType           versionConstraintType = "avro format"
	debeziumFormatConstraintType       versionConstraintType = "debezium format"
	columnsDispatcherConstraintType    versionConstraintType = "columns dispatcher"
	indexValueDispatcherConstraintType versionConstraintType = "index value dispatcher"
	eventFilterConstraintType          versionConstraintType = "event filter"
	columnSelectorsConstraintType      versionConstraintType = "column selectors"
	pulsarSinkConstraintType           versionConstraintType = "pulsar sink"
	s3SinkConstraintType               versionConstraintType = "s3 sink"
	gcsSinkConstraintType              versionConstraintType = "gcs sink"
	azureBlobStorageConstraintType     versionConstraintType = "azure blob storage sink"
	outputRawChangeEventConstraintType versionConstraintType = "output-raw-change-event"
)

type ticdcFeature string

const featureCSVHeader ticdcFeature = "csv-header"

var (
	portRegexp = regexp.MustCompile(
		`^([0-9]{1,4}|[1-5][0-9]{4}|6[0-4][0-9]{3}|65[0-4][0-9]{2}|655[0-2][0-9]|6553[0-5])$`)
	nameRegexp            = regexp.MustCompile(`^[A-Za-z0-9_.-]+$`)
	topicDispatcherRegexp = regexp.MustCompile(`^[A-Za-z0-9_.-]*$`)
	identifierRegexp      = regexp.MustCompile(`^[a-zA-Z0-9_$]{1,64}$`)
	versionRegexp         = regexp.MustCompile(`^v\d+(\.\d+)*`)
)

var (
	mysqlSinkVersionConstraint, _ = semver.NewConstraint(">= v6.1.0")

	kafkaSinkVersionConstraint, _            = semver.NewConstraint(">= v6.1.3")
	kafkaAuthTypeVersionConstraint, _        = semver.NewConstraint(">= v6.1.0")
	avroFormatVersionConstraint, _           = semver.NewConstraint(">= v6.1.0")
	debeziumFormatVersionConstraint, _       = semver.NewConstraint(">= v8.1.0")
	columnsDispatcherVersionConstraint, _    = semver.NewConstraint(">= v7.5.0")
	indexValueDispatcherVersionConstraint, _ = semver.NewConstraint(">= v7.5.0")
	columnSelectorsVersionConstraint, _      = semver.NewConstraint(">= v7.5.0")

	pulsarSinkVersionConstraint, _ = semver.NewConstraint(">= v7.5.1")

	eventFilterVersionConstraint, _ = semver.NewConstraint("> v6.5.2")

	s3SinkVersionConstraint, _               = semver.NewConstraint(">= v7.1.0")
	gcsSinkVersionConstraint, _              = semver.NewConstraint(">= v7.1.1")
	azureBlobStorageSinkVersionConstraint, _ = semver.NewConstraint(">= v7.1.1")

	outputRawChangeEventV65Constraint, _ = semver.NewConstraint(">= v6.5.10")
	outputRawChangeEventV71Constraint, _ = semver.NewConstraint(">= v7.1.6")
	outputRawChangeEventV75Constraint, _ = semver.NewConstraint(">= v7.5.3")
	outputRawChangeEventConstraint, _    = semver.NewConstraint(">= v8.1.1")
)

type BizChangefeedDownstreamAdaptor interface {
	ParseCreateParams(ctx context.Context, req *dataflow.ConnectorCreateReq) (interface{}, bizerr.BizError)
	ValidateCreateParams(params interface{}, cluster *core.Cluster, ticdcVersion string, newArch bool) bizerr.BizError
	ParseEditParams(req *dataflow.ConnectorEditReq) interface{}
	ValidateEditParams(params interface{}, clusterVersion string, newArch bool) bizerr.BizError
	AssembleGetResponse(bizChangefeed *domain.BizChangefeed) (*dataflow.ConnectorCreateReq, bizerr.BizError)
}

type BizChangefeedDownstreamAdaptorFactory struct {
	mysqlAdaptor      BizChangefeedDownstreamAdaptor
	kafkaAdaptor      BizChangefeedDownstreamAdaptor
	pulsarAdaptor     BizChangefeedDownstreamAdaptor
	s3Adaptor         BizChangefeedDownstreamAdaptor
	gcsAdaptor        BizChangefeedDownstreamAdaptor
	azBlobAdaptor     BizChangefeedDownstreamAdaptor
	serverlessAdaptor BizChangefeedDownstreamAdaptor
}

func NewBizChangefeedDownstreamAdaptorFactory(
	sinkEndpointModel private_link.SinkEndpointModel, awsSession *session.Session,
) *BizChangefeedDownstreamAdaptorFactory {
	mysqlAdaptor := newMySQLBizChangefeedDownstreamAdaptor(sinkEndpointModel)
	kafkaAdaptor := newKafkaBizChangefeedDownstreamAdaptor(sinkEndpointModel)
	pulsarAdaptor := newPulsarBizChangefeedDownstreamAdaptor()
	s3Adaptor := NewS3BizChangefeedDownstreamAdaptor(awsSession)
	gcsAdaptor := NewGCSBizChangefeedDownstreamAdaptor()
	azBlobAdaptor := NewAzureBlobStorageBizChangefeedDownstreamAdaptor()
	serverlessAdaptor := newServerlessBizChangefeedDownstreamAdaptor(sinkEndpointModel)
	return &BizChangefeedDownstreamAdaptorFactory{
		mysqlAdaptor:      mysqlAdaptor,
		kafkaAdaptor:      kafkaAdaptor,
		pulsarAdaptor:     pulsarAdaptor,
		s3Adaptor:         s3Adaptor,
		gcsAdaptor:        gcsAdaptor,
		azBlobAdaptor:     azBlobAdaptor,
		serverlessAdaptor: serverlessAdaptor,
	}
}

func (m *BizChangefeedDownstreamAdaptorFactory) Get(
	connectorType dataflow.ConnectorType,
) (BizChangefeedDownstreamAdaptor, bool) {
	switch connectorType {
	case dataflow.ConnectorType_MYSQL:
		return m.mysqlAdaptor, true
	case dataflow.ConnectorType_KAFKA:
		return m.kafkaAdaptor, true
	case dataflow.ConnectorType_PULSAR:
		return m.pulsarAdaptor, true
	case dataflow.ConnectorType_S3:
		return m.s3Adaptor, true
	case dataflow.ConnectorType_GCS:
		return m.gcsAdaptor, true
	case dataflow.ConnectorType_AZURE_BLOB_STORAGE:
		return m.azBlobAdaptor, true
	case dataflow.ConnectorType_SERVERLESS_TIDB:
		return m.serverlessAdaptor, true
	default:
		return nil, false
	}
}

type mysqlBizChangefeedDownstreamAdaptor struct {
	sinkEndpointModel private_link.SinkEndpointModel
}

func newMySQLBizChangefeedDownstreamAdaptor(
	sinkEndpointModel private_link.SinkEndpointModel,
) BizChangefeedDownstreamAdaptor {
	return &mysqlBizChangefeedDownstreamAdaptor{
		sinkEndpointModel: sinkEndpointModel,
	}
}

func (m *mysqlBizChangefeedDownstreamAdaptor) ParseCreateParams(
	ctx context.Context, req *dataflow.ConnectorCreateReq,
) (interface{}, bizerr.BizError) {
	if err := m.assembleReqByPrivateLink(ctx, req); err != nil {
		return nil, err
	}
	return req.MysqlInfo, nil
}

func (m *mysqlBizChangefeedDownstreamAdaptor) ParseEditParams(
	req *dataflow.ConnectorEditReq,
) interface{} {
	return req.MysqlInfo
}

func (m *mysqlBizChangefeedDownstreamAdaptor) ValidateCreateParams(
	params interface{}, cluster *core.Cluster, ticdcVersion string, newArch bool,
) bizerr.BizError {
	if params == nil {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "mysql_connection_info")
	}
	mysqlInfo, ok := params.(*dataflow.MySQLInfo)
	if !ok || mysqlInfo == nil {
		return bizerr.New(bizerr.ErrCodeParams, "invalid mysql connection info")
	}
	if err := checkMySQLConnection(mysqlInfo.Connection); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check connection error")
	}
	if err := checkTableConfig(mysqlInfo.TableConfig); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check table config error")
	}
	if err := checkStartPosition(mysqlInfo.StartPosition); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check start position error")
	}
	if err := m.checkClusterVersion(mysqlInfo, ticdcVersion); err != nil {
		return err
	}
	return nil
}

func (m *mysqlBizChangefeedDownstreamAdaptor) ValidateEditParams(
	params interface{}, clusterVersion string, newArch bool,
) bizerr.BizError {
	if params == nil {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "mysql_connection_info")
	}
	mysqlInfo, ok := params.(*dataflow.MySQLInfo)
	if !ok || mysqlInfo == nil {
		return bizerr.New(bizerr.ErrCodeParams, "invalid mysql connection info")
	}
	// endpoint is not editable, password can be empty when editing the changefeed,
	// but must specify the connection username
	if mysqlInfo.Connection == nil || mysqlInfo.Connection.Username == "" {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "mysql_connection_info.username")
	}
	if err := checkTableConfig(mysqlInfo.TableConfig); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check table config error")
	}
	if err := m.checkClusterVersion(mysqlInfo, clusterVersion); err != nil {
		return err
	}
	return nil
}

func (m *mysqlBizChangefeedDownstreamAdaptor) checkClusterVersion(
	mysqlParams *dataflow.MySQLInfo, clusterVersion string,
) bizerr.BizError {
	if len(mysqlParams.GetTableConfig().GetEventFilterRule()) > 0 {
		if bizErr := checkClusterVersion(
			eventFilterVersionConstraint, eventFilterConstraintType, clusterVersion); bizErr != nil {
			return bizErr
		}
	}
	return checkClusterVersion(mysqlSinkVersionConstraint, mysqlSinkConstraintType, clusterVersion)
}

func (m *mysqlBizChangefeedDownstreamAdaptor) AssembleGetResponse(
	bizChangefeed *domain.BizChangefeed,
) (*dataflow.ConnectorCreateReq, bizerr.BizError) {
	networkInfo, err := decodeNetworkInfo(bizChangefeed.NetworkInfo)
	if err != nil {
		return nil, err
	}
	decoder := json.NewDecoder(strings.NewReader(bizChangefeed.DownstreamParameter))
	decoder.DisallowUnknownFields()
	mysqlInfo := new(dataflow.MySQLInfo)
	if err := decoder.Decode(mysqlInfo); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "parse mysql downstream parameter")
	}
	if mysqlInfo.Connection != nil {
		mysqlInfo.Connection.Password = ""
	}
	createReq := &dataflow.ConnectorCreateReq{
		ClusterId:     bizChangefeed.ClusterID,
		ProjectId:     bizChangefeed.ProjectID,
		OrgId:         bizChangefeed.TenantID,
		Name:          bizChangefeed.Name,
		ConnectorType: domain.TiCDCChangefeedDownStreamType2ConnectorType(bizChangefeed.DownstreamType),
		MysqlInfo:     mysqlInfo,
		NetworkInfo:   networkInfo,
	}
	return createReq, nil
}

func (m *mysqlBizChangefeedDownstreamAdaptor) assembleReqByPrivateLink(
	ctx context.Context, req *dataflow.ConnectorCreateReq,
) bizerr.BizError {
	if isNotPLPSC(req) {
		return nil
	}

	if len(req.NetworkInfo.Ports) == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "network ports can't be empty")
	}
	if len(req.NetworkInfo.Ports) > 1 {
		return bizerr.New(bizerr.ErrCodeParams, "can only specify one network port for mysql")
	}

	endpoint, err := getAndValidateSinkEndpoint(ctx, m.sinkEndpointModel, req.GetNetworkInfo().GetSinkEndpointId())
	if err != nil {
		return err
	}
	req.MysqlInfo.Connection.Endpoint = fmt.Sprintf(
		privateLinkEndpointAddress, endpoint.EndpointAddress, req.NetworkInfo.Ports[0])
	return nil
}

type kafkaBizChangefeedDownstreamAdaptor struct {
	sinkEndpointModel private_link.SinkEndpointModel
}

func newKafkaBizChangefeedDownstreamAdaptor(
	sinkEndpointModel private_link.SinkEndpointModel,
) BizChangefeedDownstreamAdaptor {
	return &kafkaBizChangefeedDownstreamAdaptor{
		sinkEndpointModel: sinkEndpointModel,
	}
}

func (m *kafkaBizChangefeedDownstreamAdaptor) ParseCreateParams(
	ctx context.Context, req *dataflow.ConnectorCreateReq,
) (interface{}, bizerr.BizError) {
	if err := m.assemblePrivateLinkKafkaReq(ctx, req); err != nil {
		return nil, err
	}
	return req.KafkaInfo, nil
}

func (m *kafkaBizChangefeedDownstreamAdaptor) ParseEditParams(
	req *dataflow.ConnectorEditReq,
) interface{} {
	return req.KafkaInfo
}

func (m *kafkaBizChangefeedDownstreamAdaptor) ValidateCreateParams(
	params interface{}, cluster *core.Cluster, ticdcVersion string, newArch bool,
) bizerr.BizError {
	if params == nil {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "kafka_connection_info")
	}
	kafkaInfo, ok := params.(*dataflow.KafkaInfo)
	if !ok {
		return bizerr.New(bizerr.ErrCodeParams, "invalid kafka connection info")
	}
	if isKafkaInfoEmpty(kafkaInfo) {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "kafka_connection_info")
	}
	if err := m.validateKafkaSubParams(kafkaInfo, false); err != nil {
		return err
	}
	if err := m.checkClusterVersion(kafkaInfo, ticdcVersion); err != nil {
		return err
	}
	return nil
}

func (m *kafkaBizChangefeedDownstreamAdaptor) ValidateEditParams(
	params interface{}, clusterVersion string, newArch bool,
) bizerr.BizError {
	kafkaInfo, ok := params.(*dataflow.KafkaInfo)
	if !ok || kafkaInfo == nil {
		return bizerr.New(bizerr.ErrCodeParams, "invalid kafka connection info")
	}

	if kafkaInfo.GetBroker() == nil || kafkaInfo.GetAuthentication() == nil ||
		kafkaInfo.GetTopicPartitionConfig() == nil || kafkaInfo.GetDataFormat() == nil ||
		kafkaInfo.GetTableConfig() == nil {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "kafka_info")
	}
	if err := m.validateKafkaSubParams(kafkaInfo, true); err != nil {
		return err
	}
	if err := m.checkClusterVersion(kafkaInfo, clusterVersion); err != nil {
		return err
	}
	return nil
}

func (m *kafkaBizChangefeedDownstreamAdaptor) validateKafkaSubParams(kafkaInfo *dataflow.KafkaInfo,
	isEdit bool,
) bizerr.BizError {
	// check broker
	if err := checkKafkaBroker(kafkaInfo.GetBroker(), isEdit); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check kafka broker error")
	}
	// check authentication
	if err := checkKafkaAuthentication(kafkaInfo.GetAuthentication(), isEdit); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check kafka authentication error")
	}
	// check start position
	if !isEdit {
		if err := checkStartPosition(kafkaInfo.GetStartPosition()); err != nil {
			return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check start position error")
		}
	}
	return m.validateKafkaSubCommonParams(kafkaInfo)
}

func (m *kafkaBizChangefeedDownstreamAdaptor) validateKafkaSubCommonParams(
	kafkaInfo *dataflow.KafkaInfo,
) bizerr.BizError {
	// check topic && partition config
	if err := checkKafkaTopicPartitionConfig(kafkaInfo.GetTopicPartitionConfig()); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check kafka topic partition config error")
	}
	// check data format
	if err := checkKafkaDataFormat(kafkaInfo.GetDataFormat(), kafkaInfo.GetTopicPartitionConfig()); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check kafka data format error")
	}
	// check column selectors
	if err := checkColumnSelectors(kafkaInfo.GetColumnSelectors()); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check column selectors error")
	}
	// check table config
	if err := checkTableConfig(kafkaInfo.GetTableConfig()); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check table config error")
	}
	return nil
}

func (m *kafkaBizChangefeedDownstreamAdaptor) checkClusterVersion(
	kafkaParams *dataflow.KafkaInfo, clusterVersion string,
) bizerr.BizError {
	if kafkaParams.Authentication != nil && kafkaParams.Authentication.AuthType != dataflow.KafkaAuthType_DISABLE {
		if bizErr := checkClusterVersion(
			kafkaAuthTypeVersionConstraint, kafkaAuthTypeSinkConstraintType, clusterVersion); bizErr != nil {
			return bizErr
		}
	}
	if len(kafkaParams.GetTableConfig().GetEventFilterRule()) > 0 {
		if bizErr := checkClusterVersion(
			eventFilterVersionConstraint, eventFilterConstraintType, clusterVersion); bizErr != nil {
			return bizErr
		}
	}
	if bizErr := m.checkPartitionDispatcherVersion(
		kafkaParams.GetTopicPartitionConfig().GetPartitionDispatchers(), clusterVersion); bizErr != nil {
		return bizErr
	}
	if bizErr := m.checkDataFormatVersion(kafkaParams.GetDataFormat(), clusterVersion); bizErr != nil {
		return bizErr
	}
	if len(kafkaParams.ColumnSelectors) > 0 {
		if bizErr := checkClusterVersion(
			columnSelectorsVersionConstraint, columnSelectorsConstraintType, clusterVersion); bizErr != nil {
			return bizErr
		}
	}
	return checkClusterVersion(kafkaSinkVersionConstraint, kafkaSinkConstraintType, clusterVersion)
}

func (m *kafkaBizChangefeedDownstreamAdaptor) checkPartitionDispatcherVersion(
	partitionDispatchers []*dataflow.PartitionDispatcherConfig, clusterVersion string,
) bizerr.BizError {
	for _, partitionDispatcher := range partitionDispatchers {
		if partitionDispatcher.GetPartitionType() == dataflow.PartitionDispatcher_INDEX_VALUE &&
			partitionDispatcher.GetIndexName() != "" {
			if bizErr := checkClusterVersion(
				indexValueDispatcherVersionConstraint,
				indexValueDispatcherConstraintType,
				clusterVersion); bizErr != nil {
				return bizErr
			}
		}
		if partitionDispatcher.GetPartitionType() == dataflow.PartitionDispatcher_COLUMNS {
			if bizErr := checkClusterVersion(
				columnsDispatcherVersionConstraint, columnsDispatcherConstraintType, clusterVersion); bizErr != nil {
				return bizErr
			}
			for _, item := range partitionDispatcher.GetColumns() {
				if !validateMySQLIdentifier(item) {
					return bizerr.New(bizerr.ErrCodeParams, "dispatcher column name is invalid %s", item)
				}
			}
		}
	}
	return nil
}

func (m *kafkaBizChangefeedDownstreamAdaptor) checkDataFormatVersion(
	dataFormat *dataflow.DataFormat, clusterVersion string,
) bizerr.BizError {
	if dataFormat == nil {
		return nil
	}
	if dataFormat.GetProtocol() == dataflow.Protocol_AVRO {
		if bizErr := checkClusterVersion(
			avroFormatVersionConstraint, avroFormatConstraintType, clusterVersion); bizErr != nil {
			return bizErr
		}
	} else if dataFormat.GetProtocol() == dataflow.Protocol_DEBEZIUM {
		if bizErr := checkClusterVersion(
			debeziumFormatVersionConstraint, debeziumFormatConstraintType, clusterVersion); bizErr != nil {
			return bizErr
		}
	}
	if dataFormat.OutputRawChangeEvent {
		if bizErr := checkOutputRawChangeEventVersion(clusterVersion); bizErr != nil {
			return bizErr
		}
	}
	return nil
}

func (m *kafkaBizChangefeedDownstreamAdaptor) assemblePrivateLinkKafkaReq(
	ctx context.Context, req *dataflow.ConnectorCreateReq,
) bizerr.BizError {
	if isNotPLPSC(req) {
		return nil
	}

	endpoint, err := getAndValidateSinkEndpoint(ctx, m.sinkEndpointModel, req.GetNetworkInfo().GetSinkEndpointId())
	if err != nil {
		return err
	}
	var brokerEndpoints []string
	for _, port := range req.NetworkInfo.Ports {
		brokerEndpoints = append(
			brokerEndpoints, fmt.Sprintf(privateLinkEndpointAddress, endpoint.EndpointAddress, port))
	}
	req.KafkaInfo.Broker.BrokerEndpoints = strings.Join(brokerEndpoints, ",")
	return nil
}

func (m *kafkaBizChangefeedDownstreamAdaptor) AssembleGetResponse(
	bizChangefeed *domain.BizChangefeed,
) (*dataflow.ConnectorCreateReq, bizerr.BizError) {
	networkInfo, err := decodeNetworkInfo(bizChangefeed.NetworkInfo)
	if err != nil {
		return nil, err
	}
	decoder := json.NewDecoder(strings.NewReader(bizChangefeed.DownstreamParameter))
	decoder.DisallowUnknownFields()
	kafkaInfo := new(dataflow.KafkaInfo)
	if err := decoder.Decode(kafkaInfo); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "parse kafka downstream parameter")
	}
	if kafkaInfo.GetAuthentication() != nil {
		kafkaInfo.Authentication.Password = ""
	}
	if kafkaInfo.GetDataFormat() != nil && kafkaInfo.GetDataFormat().GetSchemaRegistry() != nil {
		kafkaInfo.DataFormat.SchemaRegistry.Password = ""
	}

	createReq := &dataflow.ConnectorCreateReq{
		ClusterId:     bizChangefeed.ClusterID,
		ProjectId:     bizChangefeed.ProjectID,
		OrgId:         bizChangefeed.TenantID,
		Name:          bizChangefeed.Name,
		ConnectorType: domain.TiCDCChangefeedDownStreamType2ConnectorType(bizChangefeed.DownstreamType),
		KafkaInfo:     kafkaInfo,
		NetworkInfo:   networkInfo,
	}

	return createReq, nil
}

type pulsarBizChangefeedDownstreamAdaptor struct{}

func newPulsarBizChangefeedDownstreamAdaptor() BizChangefeedDownstreamAdaptor {
	return &pulsarBizChangefeedDownstreamAdaptor{}
}

func (m *pulsarBizChangefeedDownstreamAdaptor) ParseCreateParams(
	ctx context.Context, req *dataflow.ConnectorCreateReq,
) (interface{}, bizerr.BizError) {
	return req.PulsarInfo, nil
}

func (m *pulsarBizChangefeedDownstreamAdaptor) ValidateCreateParams(
	params interface{}, cluster *core.Cluster, ticdcVersion string, newArch bool,
) bizerr.BizError {
	pulsarInfo, ok := params.(*dataflow.PulsarInfo)
	if !ok || pulsarInfo == nil {
		return bizerr.New(bizerr.ErrCodeParams, "invalid pulsar info")
	}

	if !checkPulsarEndpoint(pulsarInfo.Endpoint) {
		return bizerr_wrapper.NewInvalidInputError(config.ErrInvalidParam, "pulsar endpoint", pulsarInfo.Endpoint)
	}
	if err := m.validatePulsarSubParams(pulsarInfo, false); err != nil {
		return err
	}
	if err := m.checkClusterVersion(pulsarInfo, ticdcVersion); err != nil {
		return err.Chain("check cluster version")
	}
	return nil
}

func (m *pulsarBizChangefeedDownstreamAdaptor) validatePulsarSubParams(
	pulsarInfo *dataflow.PulsarInfo, isEdit bool,
) bizerr.BizError {
	if err := checkPulsarDestinationProtocol(pulsarInfo.DestinationProtocol); err != nil {
		return err
	}
	if err := checkPulsarAuthentication(pulsarInfo.Authentication, isEdit); err != nil {
		return err.Chain("check authentication")
	}
	if err := checkPulsarDataFormat(pulsarInfo.DataFormat); err != nil {
		return err.Chain("check data format")
	}
	if err := checkPulsarTopicDistribution(pulsarInfo.TopicDistribution); err != nil {
		return err.Chain("check topic distribution")
	}
	if err := checkPartitionDispatchers(pulsarInfo.PartitionDispatchers); err != nil {
		return err.Chain("check partition dispatchers")
	}
	if err := checkTableConfig(pulsarInfo.TableConfig); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check table config")
	}
	if err := checkStartPosition(pulsarInfo.StartPosition); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check start position")
	}
	if err := checkPulsarAdvancedSetting(pulsarInfo.AdvancedSetting); err != nil {
		return err.Chain("check advanced setting")
	}
	return nil
}

func (m *pulsarBizChangefeedDownstreamAdaptor) checkClusterVersion(
	pulsarInfo *dataflow.PulsarInfo, clusterVersion string,
) bizerr.BizError {
	if pulsarInfo.GetDataFormat().GetOutputRawChangeEvent() {
		if bizErr := checkOutputRawChangeEventVersion(clusterVersion); bizErr != nil {
			return bizErr
		}
	}
	return checkClusterVersion(pulsarSinkVersionConstraint, pulsarSinkConstraintType, clusterVersion)
}

func (m *pulsarBizChangefeedDownstreamAdaptor) ParseEditParams(
	req *dataflow.ConnectorEditReq,
) interface{} {
	return req.PulsarInfo
}

func (m *pulsarBizChangefeedDownstreamAdaptor) ValidateEditParams(
	params interface{}, clusterVersion string, newArch bool,
) bizerr.BizError {
	pulsarInfo, ok := params.(*dataflow.PulsarInfo)
	if !ok || pulsarInfo == nil {
		return bizerr.New(bizerr.ErrCodeParams, "invalid pulsar info")
	}

	if err := m.validatePulsarSubParams(pulsarInfo, true); err != nil {
		return err
	}
	if err := m.checkClusterVersion(pulsarInfo, clusterVersion); err != nil {
		return err.Chain("check cluster version")
	}
	return nil
}

func (m *pulsarBizChangefeedDownstreamAdaptor) AssembleGetResponse(
	bizChangefeed *domain.BizChangefeed,
) (*dataflow.ConnectorCreateReq, bizerr.BizError) {
	networkInfo, err := decodeNetworkInfo(bizChangefeed.NetworkInfo)
	if err != nil {
		return nil, err
	}
	decoder := json.NewDecoder(strings.NewReader(bizChangefeed.DownstreamParameter))
	decoder.DisallowUnknownFields()
	pulsarInfo := new(dataflow.PulsarInfo)
	if err := decoder.Decode(pulsarInfo); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "parse pulsar downstream parameter")
	}
	if pulsarInfo.GetAuthentication() != nil {
		pulsarInfo.Authentication.Password = ""
		pulsarInfo.Authentication.Token = ""
	}
	createReq := &dataflow.ConnectorCreateReq{
		ClusterId:     bizChangefeed.ClusterID,
		ProjectId:     bizChangefeed.ProjectID,
		OrgId:         bizChangefeed.TenantID,
		Name:          bizChangefeed.Name,
		ConnectorType: domain.TiCDCChangefeedDownStreamType2ConnectorType(bizChangefeed.DownstreamType),
		PulsarInfo:    pulsarInfo,
		NetworkInfo:   networkInfo,
	}
	return createReq, nil
}

type s3BizChangefeedDownstreamAdaptor struct {
	session *session.Session
}

func NewS3BizChangefeedDownstreamAdaptor(sess *session.Session) BizChangefeedDownstreamAdaptor {
	return &s3BizChangefeedDownstreamAdaptor{
		session: sess,
	}
}

func (m *s3BizChangefeedDownstreamAdaptor) ParseCreateParams(
	ctx context.Context, req *dataflow.ConnectorCreateReq,
) (interface{}, bizerr.BizError) {
	return req.S3Info, nil
}

func (m *s3BizChangefeedDownstreamAdaptor) ValidateCreateParams(
	params interface{}, cluster *core.Cluster, ticdcVersion string, newArch bool,
) bizerr.BizError {
	s3Info, ok := params.(*dataflow.S3Info)
	if !ok || s3Info == nil {
		return bizerr.New(bizerr.ErrCodeParams, "invalid s3 info")
	}

	if err := checkS3Endpoint(s3Info.Endpoint, false); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check endpoint")
	}
	if err := checkS3Region(s3Info.Endpoint.S3Uri, cluster.RegionName, m.session); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check s3 region")
	}
	if err := checkS3DataFormat(s3Info.DataFormat); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check data format")
	}
	if err := checkTableConfig(s3Info.TableConfig); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check table config")
	}
	if err := checkStartPosition(s3Info.StartPosition); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check start position")
	}
	s3Info.StorageFlushConfig = initStorageFlushConfig(s3Info.GetStorageFlushConfig())
	if err := checkStorageFlushConfig(s3Info.StorageFlushConfig); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check storage flush config")
	}

	if err := m.checkClusterVersion(s3Info, ticdcVersion, newArch); err != nil {
		return err
	}
	return nil
}

func (m *s3BizChangefeedDownstreamAdaptor) ParseEditParams(req *dataflow.ConnectorEditReq) interface{} {
	return req.S3Info
}

func (m *s3BizChangefeedDownstreamAdaptor) ValidateEditParams(
	params interface{}, clusterVersion string, newArch bool,
) bizerr.BizError {
	s3Info, ok := params.(*dataflow.S3Info)
	if !ok || s3Info == nil {
		return bizerr.New(bizerr.ErrCodeParams, "invalid s3 info")
	}

	// s3 uri can not edit, and secret key can be empty
	if err := checkS3Endpoint(s3Info.Endpoint, true); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check endpoint")
	}

	if err := checkS3DataFormat(s3Info.DataFormat); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check data format")
	}
	if err := checkTableConfig(s3Info.TableConfig); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check table config")
	}
	if err := checkStorageFlushConfig(s3Info.StorageFlushConfig); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check storage flush config")
	}

	if err := m.checkClusterVersion(s3Info, clusterVersion, newArch); err != nil {
		return err
	}
	return nil
}

func (m *s3BizChangefeedDownstreamAdaptor) checkClusterVersion(
	s3Info *dataflow.S3Info, clusterVersion string, newArch bool,
) bizerr.BizError {
	if len(s3Info.GetTableConfig().GetEventFilterRule()) > 0 {
		if bizErr := checkClusterVersion(
			eventFilterVersionConstraint, eventFilterConstraintType, clusterVersion); bizErr != nil {
			return bizErr
		}
	}
	if s3Info.GetDataFormat().GetOutputRawChangeEvent() {
		if bizErr := checkOutputRawChangeEventVersion(clusterVersion); bizErr != nil {
			return bizErr
		}
	}
	if s3Info.GetDataFormat().GetOutputFieldHeader() {
		if bizErr := checkTiCDCClusterVersion(featureCSVHeader, clusterVersion, newArch); bizErr != nil {
			return bizErr
		}
	}
	return checkClusterVersion(s3SinkVersionConstraint, s3SinkConstraintType, clusterVersion)
}

func (m *s3BizChangefeedDownstreamAdaptor) AssembleGetResponse(
	bizChangefeed *domain.BizChangefeed,
) (*dataflow.ConnectorCreateReq, bizerr.BizError) {
	decoder := json.NewDecoder(strings.NewReader(bizChangefeed.DownstreamParameter))
	decoder.DisallowUnknownFields()
	s3Info := new(dataflow.S3Info)
	if err := decoder.Decode(s3Info); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "parse s3 downstream parameter")
	}
	if s3Info.Endpoint != nil {
		s3Info.Endpoint.SecretKey = ""
	}
	createReq := &dataflow.ConnectorCreateReq{
		ClusterId:     bizChangefeed.ClusterID,
		ProjectId:     bizChangefeed.ProjectID,
		OrgId:         bizChangefeed.TenantID,
		Name:          bizChangefeed.Name,
		ConnectorType: domain.TiCDCChangefeedDownStreamType2ConnectorType(bizChangefeed.DownstreamType),
		S3Info:        s3Info,
	}
	return createReq, nil
}

type gcsBizChangefeedDownstreamAdaptor struct{}

func NewGCSBizChangefeedDownstreamAdaptor() BizChangefeedDownstreamAdaptor {
	return &gcsBizChangefeedDownstreamAdaptor{}
}

func (m *gcsBizChangefeedDownstreamAdaptor) ParseCreateParams(
	ctx context.Context, req *dataflow.ConnectorCreateReq,
) (interface{}, bizerr.BizError) {
	return req.GcsInfo, nil
}

//nolint:dupl
func (m *gcsBizChangefeedDownstreamAdaptor) ValidateCreateParams(
	params interface{}, cluster *core.Cluster, ticdcVersion string, newArch bool,
) bizerr.BizError {
	gcsInfo, ok := params.(*dataflow.GcsInfo)
	if !ok || gcsInfo == nil {
		return bizerr.New(bizerr.ErrCodeParams, "invalid gcs info")
	}

	if err := checkGCSEndpoint(gcsInfo.Endpoint); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check endpoint")
	}
	if err := checkS3DataFormat(gcsInfo.DataFormat); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check data format")
	}
	if err := checkTableConfig(gcsInfo.TableConfig); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check table config")
	}
	if err := checkStartPosition(gcsInfo.StartPosition); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check start position")
	}
	gcsInfo.StorageFlushConfig = initStorageFlushConfig(gcsInfo.GetStorageFlushConfig())
	if err := checkStorageFlushConfig(gcsInfo.StorageFlushConfig); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check storage flush config")
	}

	if err := m.checkClusterVersion(gcsInfo, ticdcVersion, newArch); err != nil {
		return err
	}
	return nil
}

func (m *gcsBizChangefeedDownstreamAdaptor) ParseEditParams(req *dataflow.ConnectorEditReq) interface{} {
	return req.GcsInfo
}

func (m *gcsBizChangefeedDownstreamAdaptor) ValidateEditParams(
	params interface{}, clusterVersion string, newArch bool,
) bizerr.BizError {
	gcsInfo, ok := params.(*dataflow.GcsInfo)
	if !ok || gcsInfo == nil {
		return bizerr.New(bizerr.ErrCodeParams, "invalid gcs info")
	}

	if err := checkS3DataFormat(gcsInfo.DataFormat); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check data format")
	}
	if err := checkTableConfig(gcsInfo.TableConfig); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check table config")
	}
	if err := checkStorageFlushConfig(gcsInfo.StorageFlushConfig); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check storage flush config")
	}

	if err := m.checkClusterVersion(gcsInfo, clusterVersion, newArch); err != nil {
		return err
	}
	return nil
}

func (m *gcsBizChangefeedDownstreamAdaptor) checkClusterVersion(
	gcsInfo *dataflow.GcsInfo, clusterVersion string, newArch bool,
) bizerr.BizError {
	if len(gcsInfo.GetTableConfig().GetEventFilterRule()) > 0 {
		if bizErr := checkClusterVersion(
			eventFilterVersionConstraint, eventFilterConstraintType, clusterVersion); bizErr != nil {
			return bizErr
		}
	}
	if gcsInfo.GetDataFormat().GetOutputRawChangeEvent() {
		if bizErr := checkOutputRawChangeEventVersion(clusterVersion); bizErr != nil {
			return bizErr
		}
	}
	if gcsInfo.GetDataFormat().GetOutputFieldHeader() {
		if bizErr := checkTiCDCClusterVersion(featureCSVHeader, clusterVersion, newArch); bizErr != nil {
			return bizErr
		}
	}
	return checkClusterVersion(gcsSinkVersionConstraint, gcsSinkConstraintType, clusterVersion)
}

func (m *gcsBizChangefeedDownstreamAdaptor) AssembleGetResponse(
	bizChangefeed *domain.BizChangefeed,
) (*dataflow.ConnectorCreateReq, bizerr.BizError) {
	decoder := json.NewDecoder(strings.NewReader(bizChangefeed.DownstreamParameter))
	decoder.DisallowUnknownFields()
	gcsInfo := new(dataflow.GcsInfo)
	if err := decoder.Decode(gcsInfo); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "parse gcs downstream parameter")
	}
	createReq := &dataflow.ConnectorCreateReq{
		ClusterId:     bizChangefeed.ClusterID,
		ProjectId:     bizChangefeed.ProjectID,
		OrgId:         bizChangefeed.TenantID,
		Name:          bizChangefeed.Name,
		ConnectorType: domain.TiCDCChangefeedDownStreamType2ConnectorType(bizChangefeed.DownstreamType),
		GcsInfo:       gcsInfo,
	}
	return createReq, nil
}

type azureBlobStorageBizChangefeedDownstreamAdaptor struct{}

func NewAzureBlobStorageBizChangefeedDownstreamAdaptor() BizChangefeedDownstreamAdaptor {
	return &azureBlobStorageBizChangefeedDownstreamAdaptor{}
}

func (m *azureBlobStorageBizChangefeedDownstreamAdaptor) ParseCreateParams(
	ctx context.Context, req *dataflow.ConnectorCreateReq,
) (interface{}, bizerr.BizError) {
	return req.AzureBlobStorageInfo, nil
}

//nolint:dupl
func (m *azureBlobStorageBizChangefeedDownstreamAdaptor) ValidateCreateParams(
	params interface{}, cluster *core.Cluster, ticdcVersion string, newArch bool,
) bizerr.BizError {
	azBlobInfo, ok := params.(*dataflow.AzureBlobStorageInfo)
	if !ok || azBlobInfo == nil {
		return bizerr.New(bizerr.ErrCodeParams, "invalid azure blob storage info")
	}

	if err := checkAzureBlobStorageEndpoint(azBlobInfo.Endpoint); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check endpoint")
	}
	if err := checkS3DataFormat(azBlobInfo.DataFormat); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check data format")
	}
	if err := checkTableConfig(azBlobInfo.TableConfig); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check table config")
	}
	if err := checkStartPosition(azBlobInfo.StartPosition); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check start position")
	}
	azBlobInfo.StorageFlushConfig = initStorageFlushConfig(azBlobInfo.GetStorageFlushConfig())
	if err := checkStorageFlushConfig(azBlobInfo.StorageFlushConfig); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check storage flush config")
	}

	if err := m.checkClusterVersion(azBlobInfo, ticdcVersion, newArch); err != nil {
		return err
	}
	return nil
}

func (m *azureBlobStorageBizChangefeedDownstreamAdaptor) ParseEditParams(req *dataflow.ConnectorEditReq) interface{} {
	return req.AzureBlobStorageInfo
}

func (m *azureBlobStorageBizChangefeedDownstreamAdaptor) ValidateEditParams(
	params interface{}, clusterVersion string, newArch bool,
) bizerr.BizError {
	azBlobInfo, ok := params.(*dataflow.AzureBlobStorageInfo)
	if !ok || azBlobInfo == nil {
		return bizerr.New(bizerr.ErrCodeParams, "invalid azure blob storage info")
	}

	if err := checkS3DataFormat(azBlobInfo.DataFormat); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check data format")
	}
	if err := checkTableConfig(azBlobInfo.TableConfig); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check table config")
	}
	if err := checkStorageFlushConfig(azBlobInfo.StorageFlushConfig); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "check storage flush config")
	}

	if err := m.checkClusterVersion(azBlobInfo, clusterVersion, newArch); err != nil {
		return err
	}
	return nil
}

func (m *azureBlobStorageBizChangefeedDownstreamAdaptor) checkClusterVersion(
	azBlobInfo *dataflow.AzureBlobStorageInfo, clusterVersion string, newArch bool,
) bizerr.BizError {
	if len(azBlobInfo.GetTableConfig().GetEventFilterRule()) > 0 {
		if bizErr := checkClusterVersion(
			eventFilterVersionConstraint, eventFilterConstraintType, clusterVersion); bizErr != nil {
			return bizErr
		}
	}
	if azBlobInfo.GetDataFormat().GetOutputRawChangeEvent() {
		if bizErr := checkOutputRawChangeEventVersion(clusterVersion); bizErr != nil {
			return bizErr
		}
	}
	if azBlobInfo.GetDataFormat().GetOutputFieldHeader() {
		if bizErr := checkTiCDCClusterVersion(featureCSVHeader, clusterVersion, newArch); bizErr != nil {
			return bizErr
		}
	}
	return checkClusterVersion(azureBlobStorageSinkVersionConstraint, azureBlobStorageConstraintType, clusterVersion)
}

func (m *azureBlobStorageBizChangefeedDownstreamAdaptor) AssembleGetResponse(
	bizChangefeed *domain.BizChangefeed,
) (*dataflow.ConnectorCreateReq, bizerr.BizError) {
	decoder := json.NewDecoder(strings.NewReader(bizChangefeed.DownstreamParameter))
	decoder.DisallowUnknownFields()
	azBlobInfo := new(dataflow.AzureBlobStorageInfo)
	if err := decoder.Decode(azBlobInfo); err != nil {
		return nil, bizerr.WrapErrorWithCode(
			err, bizerr.ErrCodeUnknown, "parse azure blob storage downstream parameter")
	}
	if azBlobInfo.Endpoint != nil {
		azBlobInfo.Endpoint.SasToken = ""
	}
	createReq := &dataflow.ConnectorCreateReq{
		ClusterId:            bizChangefeed.ClusterID,
		ProjectId:            bizChangefeed.ProjectID,
		OrgId:                bizChangefeed.TenantID,
		Name:                 bizChangefeed.Name,
		ConnectorType:        domain.TiCDCChangefeedDownStreamType2ConnectorType(bizChangefeed.DownstreamType),
		AzureBlobStorageInfo: azBlobInfo,
	}
	return createReq, nil
}

type serverlessBizChangefeedDownstreamAdaptor struct {
	mysqlAdaptor BizChangefeedDownstreamAdaptor
}

func newServerlessBizChangefeedDownstreamAdaptor(
	sinkEndpointModel private_link.SinkEndpointModel,
) BizChangefeedDownstreamAdaptor {
	return &serverlessBizChangefeedDownstreamAdaptor{
		mysqlAdaptor: newMySQLBizChangefeedDownstreamAdaptor(sinkEndpointModel),
	}
}

func (m *serverlessBizChangefeedDownstreamAdaptor) ParseCreateParams(
	ctx context.Context, req *dataflow.ConnectorCreateReq,
) (interface{}, bizerr.BizError) {
	return req.ServerlessTidbInfo, nil
}

func (m *serverlessBizChangefeedDownstreamAdaptor) ValidateCreateParams(
	params interface{}, cluster *core.Cluster, ticdcVersion string, newArch bool,
) bizerr.BizError {
	serverlessInfo, ok := params.(*dataflow.ServerlessTidbInfo)
	if !ok || serverlessInfo == nil {
		return bizerr.New(bizerr.ErrCodeParams, "invalid serverless tidb info")
	}

	if serverlessInfo.ClusterId == 0 {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "serverless_tidb_info")
	}

	if serverlessInfo.GetMysqlInfo().GetConnection() != nil {
		// the endpoint of serverless tidb is empty.
		// set a fake valid endpoint to pass the validation.
		endpoint := serverlessInfo.MysqlInfo.Connection.Endpoint
		serverlessInfo.MysqlInfo.Connection.Endpoint = fakeMySQLEndpoint
		defer func() { serverlessInfo.MysqlInfo.Connection.Endpoint = endpoint }()
	}

	if err := m.mysqlAdaptor.ValidateCreateParams(
		serverlessInfo.MysqlInfo, cluster, ticdcVersion, newArch); err != nil {
		return err
	}
	if serverlessInfo.MysqlInfo.PrivatelinkInfo == nil ||
		serverlessInfo.MysqlInfo.PrivatelinkInfo.DfPrivatelinkEndpointId == 0 {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "privatelink_info")
	}
	return nil
}

func (m *serverlessBizChangefeedDownstreamAdaptor) ParseEditParams(req *dataflow.ConnectorEditReq) interface{} {
	return req.ServerlessTidbInfo
}

func (m *serverlessBizChangefeedDownstreamAdaptor) ValidateEditParams(
	params interface{}, clusterVersion string, newArch bool,
) bizerr.BizError {
	serverlessInfo, ok := params.(*dataflow.ServerlessTidbInfo)
	if !ok || serverlessInfo == nil {
		return bizerr.New(bizerr.ErrCodeParams, "invalid serverless tidb info")
	}

	if serverlessInfo.ClusterId == 0 {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "serverless_tidb_info")
	}
	if err := m.mysqlAdaptor.ValidateEditParams(serverlessInfo.MysqlInfo, clusterVersion, newArch); err != nil {
		return err
	}
	if serverlessInfo.MysqlInfo.PrivatelinkInfo == nil ||
		serverlessInfo.MysqlInfo.PrivatelinkInfo.DfPrivatelinkEndpointId == 0 {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "privatelink_info")
	}
	return nil
}

func (m *serverlessBizChangefeedDownstreamAdaptor) AssembleGetResponse(
	bizChangefeed *domain.BizChangefeed,
) (*dataflow.ConnectorCreateReq, bizerr.BizError) {
	decoder := json.NewDecoder(strings.NewReader(bizChangefeed.DownstreamParameter))
	decoder.DisallowUnknownFields()
	serverlessInfo := new(dataflow.ServerlessTidbInfo)
	if err := decoder.Decode(serverlessInfo); err != nil {
		return nil, bizerr.WrapErrorWithCode(
			err, bizerr.ErrCodeUnknown, "parse serverless tidb downstream parameter")
	}
	if serverlessInfo.MysqlInfo.Connection != nil {
		serverlessInfo.MysqlInfo.Connection.Password = ""
	}
	createReq := &dataflow.ConnectorCreateReq{
		ClusterId:          bizChangefeed.ClusterID,
		ProjectId:          bizChangefeed.ProjectID,
		OrgId:              bizChangefeed.TenantID,
		Name:               bizChangefeed.Name,
		ConnectorType:      domain.TiCDCChangefeedDownStreamType2ConnectorType(bizChangefeed.DownstreamType),
		ServerlessTidbInfo: serverlessInfo,
	}
	return createReq, nil
}

func checkTableConfig(tableConfig *dataflow.TableConfig) error {
	if tableConfig == nil {
		return errors.New("empty table config")
	}
	if err := checkTableFilter(tableConfig.GetFilterRule()); err != nil {
		return err
	}
	if err := checkEventFilter(tableConfig.GetEventFilterRule(), tableConfig.GetFilterRule()); err != nil {
		return err
	}
	return nil
}

func checkTableFilter(filters []string) error {
	if len(filters) == 0 {
		return nil
	}
	if len(filters) > maxFilterRuleNum {
		return fmt.Errorf("max table filter rules number %d", maxFilterRuleNum)
	}
	totalLen := 0
	for _, rule := range filters {
		totalLen += len(rule)
	}
	if totalLen > maxTotalTableFilterRuleLength {
		return fmt.Errorf("max total table filter rules length %d", maxTotalTableFilterRuleLength)
	}

	if _, err := tfilter.Parse(filters); err != nil {
		return fmt.Errorf("filter rule invalid, %s", err.Error())
	}
	return nil
}

func checkEventFilter(eventFilters []*dataflow.EventFilterRule, filters []string) error {
	if len(eventFilters) == 0 {
		return nil
	}
	if len(eventFilters) > maxEventFilterRuleNum {
		return fmt.Errorf("max event filter rules number %d", maxEventFilterRuleNum)
	}

	eventFilterRules := make([]*tiflowConfig.EventFilterRule, 0, len(eventFilters))
	for _, efRule := range eventFilters {
		for _, match := range efRule.GetMatcher() {
			if len(match) > maxFilterRuleLength {
				return fmt.Errorf("max table matching rules length %d", maxFilterRuleLength)
			}
		}
		eventFilterRules = append(eventFilterRules, eventFilterRuleDataflow2Tiflow(efRule))
	}

	err := filtervalidator.Validate(&tiflowConfig.ReplicaConfig{
		Filter: &tiflowConfig.FilterConfig{
			EventFilters: eventFilterRules,
			Rules:        filters,
		},
	})
	if err != nil {
		return fmt.Errorf("event filter rule invalid, %s", err.Error())
	}
	return nil
}

func checkMySQLConnection(connection *dataflow.Connection) error {
	if connection == nil {
		return errors.New("empty connection")
	}
	if !validateEndpoint(connection.GetEndpoint()) {
		return fmt.Errorf("connection endpoint %s invalid", connection.GetEndpoint())
	}
	if connection.GetUsername() == "" {
		return errors.New("empty username")
	}
	if connection.GetPassword() == "" {
		return errors.New("empty password")
	}
	return nil
}

func validateEndpoint(s string) bool {
	epList := strings.Split(s, ":")
	if len(epList) != endpointPartsLength {
		return false
	}
	// validate port number
	return portRegexp.MatchString(epList[1])
}

func checkStartPosition(startPosition *dataflow.StartPosition) error {
	if startPosition == nil {
		return errors.New("empty start position")
	}

	switch startPosition.GetMode() {
	case dataflow.StartMode_FROM_NOW:
		break
	case dataflow.StartMode_FROM_TSO:
		if startPosition.GetTso() == 0 {
			return errors.New("empty start position tso")
		}
	case dataflow.StartMode_FROM_UTC:
		if startPosition.GetUtc() == "" {
			return errors.New("empty start position utc")
		}
		_, err := time.Parse(time.RFC3339, startPosition.GetUtc())
		if err != nil {
			return err
		}
	default:
		return fmt.Errorf("unknown start position %d", startPosition.GetMode())
	}

	return nil
}

func eventFilterRuleDataflow2Tiflow(rule *dataflow.EventFilterRule) *tiflowConfig.EventFilterRule {
	ignoreEvents := make([]bf.EventType, 0, len(rule.GetIgnoreEvent()))
	for _, event := range rule.GetIgnoreEvent() {
		ignoreEvents = append(ignoreEvents, bf.EventType(event))
	}
	return &tiflowConfig.EventFilterRule{
		Matcher:                  rule.GetMatcher(),
		IgnoreEvent:              ignoreEvents,
		IgnoreSQL:                rule.GetIgnoreSql(),
		IgnoreInsertValueExpr:    rule.GetIgnoreInsertValueExpr(),
		IgnoreUpdateNewValueExpr: rule.GetIgnoreUpdateNewValueExpr(),
		IgnoreUpdateOldValueExpr: rule.GetIgnoreUpdateOldValueExpr(),
		IgnoreDeleteValueExpr:    rule.GetIgnoreDeleteValueExpr(),
	}
}

func isKafkaInfoEmpty(kafkaInfo *dataflow.KafkaInfo) bool {
	return kafkaInfo == nil || kafkaInfo.GetBroker() == nil || kafkaInfo.GetAuthentication() == nil ||
		kafkaInfo.GetTopicPartitionConfig() == nil || kafkaInfo.GetDataFormat() == nil ||
		kafkaInfo.GetTableConfig() == nil || kafkaInfo.GetStartPosition() == nil
}

func checkKafkaBroker(broker *dataflow.Broker, isEdit bool) error {
	if dataflow.KafkaVersion_KAFKA_VERSION_1XX != broker.GetKafkaVersion() &&
		dataflow.KafkaVersion_KAFKA_VERSION_2XX != broker.GetKafkaVersion() &&
		dataflow.KafkaVersion_KAFKA_VERSION_3XX != broker.GetKafkaVersion() {
		return fmt.Errorf("unknown kafka version %s", broker.GetKafkaVersion().String())
	}
	if !isEdit {
		if broker.GetBrokerEndpoints() == "" {
			return errors.New("empty broker endpoints")
		} else {
			endpoints := strings.Split(broker.GetBrokerEndpoints(), ",")
			for _, endpoint := range endpoints {
				if !validateEndpoint(endpoint) {
					return fmt.Errorf("broker endpoint %s invalid", endpoint)
				}
			}
		}
	}
	return nil
}

func checkKafkaAuthentication(authentication *dataflow.Authentication, isEdit bool) error {
	if _, exist := dataflow.KafkaAuthType_name[int32(authentication.GetAuthType())]; !exist {
		return fmt.Errorf("unknown auth type %s", authentication.GetAuthType())
	}
	if dataflow.KafkaAuthType_DISABLE != authentication.GetAuthType() {
		if authentication.GetUserName() == "" {
			return errors.New("empty kafka auth username")
		}
		if !isEdit && authentication.GetPassword() == "" {
			return errors.New("empty kafka auth password")
		}
	}
	return nil
}

func checkKafkaTopicPartitionConfig(tpCfg *dataflow.TopicPartitionConfig) error {
	if _, exist := dataflow.DispatchType_name[int32(tpCfg.GetDispatchType())]; !exist {
		return errors.New("unknown topic dispatch type")
	}

	// validate default topic name
	if !validateKafkaTopic(tpCfg.GetDefaultTopic()) {
		return fmt.Errorf("invalid default topic name %s", tpCfg.GetDefaultTopic())
	}

	// it's the new API for topic dispatching, we should check the partition dispatchers
	if tpCfg.PartitionDispatchers != nil {
		if err := checkPartitionDispatchers(tpCfg.PartitionDispatchers); err != nil {
			return err
		}
	} else {
		if _, exist := dataflow.PartitionDispatcher_name[int32(tpCfg.GetPartitionDispatcher())]; !exist {
			return fmt.Errorf("unknown partition dispatcher %s", tpCfg.GetPartitionDispatcher())
		}
	}

	if tpCfg.GetPartitionNum() <= 0 {
		return fmt.Errorf("invalid topic dispatcher partition num %d", tpCfg.GetPartitionNum())
	}
	return nil
}

func checkPartitionDispatchers(pds []*dataflow.PartitionDispatcherConfig) bizerr.BizError {
	if len(pds) == 0 {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "partition dispatchers")
	}
	for _, pdc := range pds {
		if len(pdc.GetMatcher()) == 0 {
			return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "matcher of partition dispatcher")
		}
		if _, exist := dataflow.PartitionDispatcher_name[int32(pdc.GetPartitionType())]; !exist {
			return bizerr.New(
				bizerr.ErrCodeParams, "unknown partition dispatcher %s", pdc.GetPartitionType().String())
		}
		if pdc.GetPartitionType() == dataflow.PartitionDispatcher_COLUMNS && len(pdc.GetColumns()) == 0 {
			return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "columns of partition dispatcher")
		}
	}
	return nil
}

func checkKafkaDataFormat(dataFormat *dataflow.DataFormat, tpCfg *dataflow.TopicPartitionConfig) error {
	if _, exist := dataflow.Protocol_name[int32(dataFormat.GetProtocol())]; !exist {
		return fmt.Errorf("unknown data format %s", dataFormat.GetProtocol())
	}
	switch dataFormat.GetProtocol() {
	case dataflow.Protocol_AVRO:
		return checkKafkaAvroDataFormat(dataFormat, tpCfg)
	}
	return nil
}

func checkKafkaAvroDataFormat(dataFormat *dataflow.DataFormat, tpCfg *dataflow.TopicPartitionConfig) error {
	if dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE != tpCfg.GetDispatchType() {
		return fmt.Errorf(
			"avro data format not support current topic dispatch type %s", tpCfg.GetDispatchType().String())
	}

	if _, exist := dataflow.DecimalHandlingMode_name[int32(dataFormat.GetDecimalHandlingMode())]; !exist {
		return fmt.Errorf("unknown DecimalHandlingMode  %s", dataFormat.GetDecimalHandlingMode())
	}
	if _, exist := dataflow.BigintUnsignedHandlingMode_name[int32(dataFormat.GetBigintUnsignedHandlingMode())]; !exist {
		return fmt.Errorf("unknown BigintUnsignedHandlingMode  %s", dataFormat.GetBigintUnsignedHandlingMode())
	}
	if err := checkKafkaSchemaRegistry(dataFormat.GetSchemaRegistry()); err != nil {
		return err
	}
	return nil
}

func checkKafkaSchemaRegistry(schemaRegistry *dataflow.SchemaRegistry) error {
	if schemaRegistry == nil {
		return errors.New("empty schema registry")
	}
	// SchemaRegistryEndpoints: http(s)://127.0.0.1:1231
	endpointStrList := strings.Split(schemaRegistry.GetSchemaRegistryEndpoints(), "//")
	if len(endpointStrList) != endpointPartsLength {
		return fmt.Errorf("schema registry %s invalid", schemaRegistry.GetSchemaRegistryEndpoints())
	}
	if schemaRegistry.GetEnableHttpAuth() {
		if schemaRegistry.GetUserName() == "" {
			return errors.New("empty schema registry auth username")
		}
		if schemaRegistry.GetPassword() == "" {
			return errors.New("empty schema registry auth password")
		}
	}
	return nil
}

func checkColumnSelectors(columnSelectors []*dataflow.ColumnSelector) error {
	for _, columnSelector := range columnSelectors {
		if _, err := tfilter.Parse(columnSelector.Matcher); err != nil {
			return fmt.Errorf("matcher %v is invalid: %w", columnSelector.Matcher, err)
		}
		if _, err := tfilter.ParseColumnFilter(columnSelector.Columns); err != nil {
			return fmt.Errorf("columns %v is invalid: %w", columnSelector.Columns, err)
		}
	}
	return nil
}

func validateKafkaTopic(s string) bool {
	return nameRegexp.MatchString(s)
}

func checkPulsarEndpoint(pulsarEndpoint string) bool {
	endpoints := strings.Split(pulsarEndpoint, ",")
	for _, endpoint := range endpoints {
		if !validateEndpoint(endpoint) {
			return false
		}
	}
	return true
}

func checkPulsarDestinationProtocol(protocol dataflow.PulsarDestinationProtocol) bizerr.BizError {
	switch protocol {
	case dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR,
		dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR_SSL:
		return nil
	default:
		return bizerr.New(bizerr.ErrCodeParams, "unsupported protocol %s", protocol.String())
	}
}

func checkPulsarAuthentication(authentication *dataflow.PulsarAuthentication, isEdit bool) bizerr.BizError {
	if authentication == nil {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "pulsar authentication")
	}
	switch authentication.AuthType {
	case dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC:
		if authentication.UserName == "" {
			return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "username")
		}
		if !isEdit && authentication.Password == "" {
			return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "password")
		}
		return nil
	case dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_TOKEN:
		if !isEdit && authentication.Token == "" {
			return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "token")
		}
		return nil
	default:
		return bizerr.New(bizerr.ErrCodeParams, "unsupported auth type %s", authentication.String())
	}
}

func checkPulsarDataFormat(dataFormat *dataflow.PulsarDataFormat) bizerr.BizError {
	if dataFormat == nil {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "pulsar data format")
	}
	if dataFormat.Protocol != dataflow.Protocol_CANAL_JSON {
		return bizerr.New(bizerr.ErrCodeParams, "unsupported data format protocol %s", dataFormat.Protocol.String())
	}
	return nil
}

func checkPulsarTopicDistribution(td *dataflow.PulsarTopicDistribution) bizerr.BizError {
	if td == nil {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "pulsar topic distribution")
	}
	if td.PulsarNamespace == "" {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "pulsar namespace")
	}
	if td.PulsarTenant == "" {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "pulsar tenant")
	}
	if td.DefaultTopic == "" {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "pulsar default topic")
	}
	if _, exist := dataflow.DispatchType_name[int32(td.DispatchType)]; !exist {
		return bizerr.New(bizerr.ErrCodeParams, "unsupported dispatch type %s", td.DispatchType.String())
	}
	if !topicDispatcherRegexp.MatchString(td.TopicPrefix) {
		return bizerr_wrapper.NewInvalidInputError(config.ErrInvalidParam, "topic prefix", td.TopicPrefix)
	}
	if !topicDispatcherRegexp.MatchString(td.Separator) {
		return bizerr_wrapper.NewInvalidInputError(config.ErrInvalidParam, "topic separator", td.Separator)
	}
	if !topicDispatcherRegexp.MatchString(td.TopicSuffix) {
		return bizerr_wrapper.NewInvalidInputError(config.ErrInvalidParam, "topic suffix", td.TopicSuffix)
	}
	return nil
}

func checkPulsarAdvancedSetting(setting *dataflow.PulsarAdvancedSetting) bizerr.BizError {
	if setting == nil {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "pulsar advanced setting")
	}
	if setting.ConnectionTimeout == 0 {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "pulsar connection timeout")
	}
	if setting.OperationTimeout == 0 {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "pulsar operation timeout")
	}
	if setting.SendTimeout == 0 {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "pulsar send timeout")
	}
	if setting.BatchingMaxMessages == 0 {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "pulsar batching max messages")
	}
	if setting.BatchingMaxPublishDelay == 0 {
		return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "pulsar batching max publish delay")
	}
	return nil
}

func checkClusterVersion(
	constraint *semver.Constraints, constraintType versionConstraintType, clusterVersion string,
) bizerr.BizError {
	versionWithoutSuffix := utils.NormalizeVersion(clusterVersion)
	v, err := semver.NewVersion(versionWithoutSuffix)
	if err != nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "cluster version is invalid, err: %s", err.Error())
	}
	if !constraint.Check(v) {
		return bizerr.New(
			config.ErrCodeClusterVersionNotSupport,
			config.ErrMsgClusterVersionNotSupport, clusterVersion, constraintType, constraint.String())
	}
	return nil
}

func checkTiCDCClusterVersion(feature ticdcFeature, ticdcVersion string, newArch bool) bizerr.BizError {
	versionWithoutSuffix := utils.NormalizeVersion(ticdcVersion)
	if internal_config.Global().ChangefeedConfig.IsFeatureEnabled(string(feature), versionWithoutSuffix, newArch) {
		return nil
	}
	return bizerr.New(
		config.ErrCodeClusterVersionNotSupport, config.ErrMsgTiCDCClusterVersionNotSupport, versionWithoutSuffix, feature)
}

func decodeNetworkInfo(info string) (*dataflow.NetworkInfo, bizerr.BizError) {
	// the network info maybe empty when the changefeed is created before the network info is introduced
	if info == "" {
		return new(dataflow.NetworkInfo), nil
	}
	decoder := json.NewDecoder(strings.NewReader(info))
	decoder.DisallowUnknownFields()
	networkInfo := new(dataflow.NetworkInfo)
	if err := decoder.Decode(networkInfo); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "parse network info")
	}
	return networkInfo, nil
}

func isNotPLPSC(req *dataflow.ConnectorCreateReq) bool {
	return req.GetNetworkInfo() == nil ||
		(req.GetNetworkInfo().GetNetworkType() != dataflow.NetworkType_NetworkType_Private_Link &&
			req.GetNetworkInfo().GetNetworkType() != dataflow.NetworkType_NetworkType_Private_Service_Connect)
}

//nolint:nestif
func checkS3Endpoint(s3Endpoint *dataflow.S3Endpoint, isEdit bool) error {
	if s3Endpoint == nil {
		return errors.New("empty s3 endpoint")
	}

	if !isEdit {
		if s3Endpoint.GetS3Uri() == "" {
			return errors.New("empty s3 endpoint uri")
		}
		if !strings.HasPrefix(s3Endpoint.GetS3Uri(), "s3://") {
			return errors.New("invalid s3 uri, must begin with s3")
		}
	}

	if s3Endpoint.GetAuthType() == dataflow.S3AuthType_S3_AUTH_TYPE_ASSUME_ROLE {
		if s3Endpoint.GetIamAssumedRoleArn() == "" {
			return errors.New("empty role arn")
		}
	} else if s3Endpoint.GetAuthType() == dataflow.S3AuthType_S3_AUTH_TYPE_AK_SK {
		if s3Endpoint.GetAccessKey() == "" {
			return errors.New("empty s3 access key")
		}
		if s3Endpoint.GetSecretKey() == "" && !isEdit {
			return errors.New("empty s3 secret key")
		}
	} else {
		return fmt.Errorf("unsupported s3 auth type %s", s3Endpoint.GetAuthType())
	}
	return nil
}

func checkGCSEndpoint(gcsEndpoint *dataflow.GcsEndpoint) error {
	if gcsEndpoint == nil {
		return errors.New("empty gcs endpoint")
	}

	if gcsEndpoint.GetGcsUri() == "" {
		return errors.New("empty gcs endpoint uri")
	}
	if !strings.HasPrefix(gcsEndpoint.GetGcsUri(), "gs://") &&
		!strings.HasPrefix(gcsEndpoint.GetGcsUri(), "gcs://") {
		return errors.New("invalid gcs uri, must begin with gs or gcs")
	}
	return nil
}

func checkAzureBlobStorageEndpoint(absEndpoint *dataflow.AzureBlobStorageEndpoint) error {
	if absEndpoint == nil {
		return errors.New("empty azure blob storage endpoint")
	}
	if absEndpoint.GetBlobUri() == "" {
		return errors.New("empty azure blob storage endpoint uri")
	}

	u, err := url.Parse(absEndpoint.GetBlobUri())
	if err != nil {
		return fmt.Errorf("parse azure blob storage uri error: %w", err)
	}
	if u.Scheme != "https" {
		return errors.New("invalid azure blob storage uri, must begin with https")
	}
	host := u.Hostname()
	i := strings.Index(host, ".")
	if i == -1 {
		return errors.New("invalid azure blob storage uri, can't find storage account")
	}
	if host[i+1:] != azureBlobStorageHostname {
		return errors.New("invalid azure blob storage uri, can't find azure blob storage hostname")
	}
	if strings.Trim(u.Path, "/") == "" {
		return errors.New("invalid azure blob storage uri, can't find container")
	}

	if absEndpoint.GetSasToken() == "" {
		return errors.New("empty azure blob storage sas token")
	}
	return nil
}

func checkS3Region(s3URL string, clusterRegion string, se *session.Session) error {
	u, err := s3.NewS3URL(s3URL)
	if err != nil {
		return fmt.Errorf("s3 url:%s is invalid: %w", s3URL, err)
	}
	region, err := u.GetBucketRegionWithSession(context.Background(), se)
	if err != nil {
		return fmt.Errorf("get bucket region error: %w", err)
	}
	if region != clusterRegion {
		return fmt.Errorf(config.ErrMsgCDCAWSRegionNotSameWithCluster, region, clusterRegion)
	}
	return nil
}

//nolint:nestif
func checkS3DataFormat(dataFormat *dataflow.S3Dataformat) error {
	if dataFormat == nil {
		return errors.New("empty s3 data format")
	}
	if dataflow.S3Protocol_S3_PROTOCOL_CSV == dataFormat.GetProtocol() {
		if len(dataFormat.GetQuote()) > 1 {
			return errors.New("csv config quote contains more than one character")
		}
		if len(dataFormat.GetQuote()) == 1 {
			quote := dataFormat.GetQuote()[0]
			if quote == CR || quote == LF {
				return errors.New("csv config quote cannot be line break character")
			}
		}

		// validate delimiter
		if len(dataFormat.GetDelimiter()) == 0 {
			return errors.New("csv config delimiter cannot be empty")
		}
		if strings.ContainsRune(dataFormat.GetDelimiter(), CR) ||
			strings.ContainsRune(dataFormat.GetDelimiter(), LF) {
			return errors.New("csv config delimiter contains line break characters")
		}
		if len(dataFormat.GetQuote()) > 0 && strings.Contains(dataFormat.GetDelimiter(), dataFormat.GetQuote()) {
			return errors.New("csv config quote and delimiter cannot be the same")
		}
	}

	return nil
}

func checkStorageFlushConfig(storageFlushConfig *dataflow.StorageFlushConfig) error {
	if storageFlushConfig == nil {
		return errors.New("empty storage flush config")
	}
	interval, err := time.ParseDuration(storageFlushConfig.GetFlushInterval())
	if err != nil {
		return fmt.Errorf("parse duration %s failed %s", storageFlushConfig.GetFlushInterval(), err.Error())
	}
	minInterval, _ := time.ParseDuration(MinFlushInterval)
	maxInterval, _ := time.ParseDuration(MaxFlushInterval)
	if interval < minInterval || interval > maxInterval {
		return fmt.Errorf("flush inteval should be in range [%s,%s]", MinFlushInterval, MaxFlushInterval)
	}
	if storageFlushConfig.GetFileSize() < MinFileSize || storageFlushConfig.GetFileSize() > MaxFileSize {
		return fmt.Errorf("flush file size should be in range [%d,%d]", MinFileSize, MaxFileSize)
	}
	return nil
}

func initStorageFlushConfig(storageFlushConfig *dataflow.StorageFlushConfig) *dataflow.StorageFlushConfig {
	if storageFlushConfig == nil {
		storageFlushConfig = &dataflow.StorageFlushConfig{}
	}
	if storageFlushConfig.GetFlushInterval() == "" {
		storageFlushConfig.FlushInterval = DefaultFlushInterval
	}
	if storageFlushConfig.GetFileSize() == 0 {
		storageFlushConfig.FileSize = DefaultFileSize
	}
	return storageFlushConfig
}

func validateMySQLIdentifier(s string) bool {
	return identifierRegexp.MatchString(s)
}

// checkOutputRawChangeEventVersion compatible TiCDC versions are >= v6.5.10 or >= v7.1.6 or >= v7.5.3 or >= v8.1.1
func checkOutputRawChangeEventVersion(clusterVersion string) bizerr.BizError {
	if strings.HasPrefix(clusterVersion, "v6.5") {
		return checkClusterVersion(
			outputRawChangeEventV65Constraint, outputRawChangeEventConstraintType, clusterVersion)
	} else if strings.HasPrefix(clusterVersion, "v7.1") {
		return checkClusterVersion(
			outputRawChangeEventV71Constraint, outputRawChangeEventConstraintType, clusterVersion)
	} else if strings.HasPrefix(clusterVersion, "v7.5") {
		return checkClusterVersion(
			outputRawChangeEventV75Constraint, outputRawChangeEventConstraintType, clusterVersion)
	} else {
		return checkClusterVersion(
			outputRawChangeEventConstraint, outputRawChangeEventConstraintType, clusterVersion)
	}
}

func getAndValidateSinkEndpoint(
	ctx context.Context, sinkEndpointModel private_link.SinkEndpointModel, sinkEndpointID uint64,
) (*entity.BizPrivateLinkEndpoint, bizerr.BizError) {
	req := &domain.QuerySinkEndpointReq{
		ID: sinkEndpointID,
	}
	resp, bizErr := sinkEndpointModel.QuerySinkEndpoint(ctx, req)
	if bizErr != nil {
		return nil, bizErr.Chain("query biz private link error")
	}

	endpoint := resp.SinkEndpoint
	if endpoint.Status != entity.BizPLEndpointStatusAvailable {
		return nil, bizerr.New(
			bizerr.ErrCodeParams, "sink endpoint status is %s, please wait it available", endpoint.Status)
	}
	return endpoint, nil
}