package entity

import (
	"context"
	"fmt"
	"net/url"
	"regexp"
	"slices"
	"strconv"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy/serverless"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
	"github.com/tidbcloud/serverless-service/internal/utils"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

var validEvents = []string{
	"all dml",
	"all ddl",
	"insert",
	"update",
	"delete",
	"create schema",
	"drop schema",
	"create table",
	"drop table",
	"rename table",
	"truncate table",
	"alter table",
	"add table partition",
	"drop table partition",
	"truncate table partition",
	"create view",
	"drop view",
	"modify schema charset and collate",
	"recover table",
	"rebase auto id",
	"modify table comment",
	"modify table charset and collate",
	"exchange table partition",
	"reorganize table partition",
	"alter table partitioning",
	"remove table partitioning",
	"add column",
	"drop column",
	"modify column",
	"set default value",
	"add primary key",
	"drop primary key",
	"rename index",
	"alter index visibility",
	"alter ttl info",
	"alter ttl remove",
	"multi schema change",
}

var topicNameRegexp = regexp.MustCompile(`^[A-Za-z0-9_.-]+$`)

const (
	DateFormat                      = "20060102"
	DefaultNamePrefix               = "changefeed"
	EndpointLength                  = 2
	defaultCloudStorageCSVDelimiter = ","
	defaultCloudStorageCSVQuote     = "\""
	defaultCloudStorageCSVNull      = "\\N"
)

type CDCChangefeed struct {
	ID               string                  `gorm:"column:id;primaryKey;type:varchar(255);not null"`
	ChangefeedID     string                  `gorm:"column:changefeed_id;type:varchar(255);not null;index:idx_changefeed_id"`
	TenantID         string                  `gorm:"column:tenant_id;type:varchar(255);not null"`
	ClusterID        string                  `gorm:"column:cluster_id;type:varchar(255);not null;index:idx_cluster_id"`
	CloudProvider    string                  `gorm:"column:cloud_provider;type:varchar(255);not null"`
	Region           string                  `gorm:"column:region;type:varchar(255);not null"`
	Kubernetes       string                  `gorm:"column:kubernetes;type:varchar(255);not null;"`
	KeySpaceID       uint32                  `gorm:"column:keyspace_id;type:bigint(20);not null;"`
	State            constant.ChangfeedState `gorm:"column:state;type:varchar(32);not null"`
	Arch             constant.ChangefeedArch `gorm:"column:arch;type:varchar(32);not null;default:'replication_worker'"`
	RCU              uint32                  `gorm:"column:rcu;type:int unsigned;not null;default:3"`
	Name             string                  `gorm:"column:name;type:varchar(255);not null"`
	SinkType         constant.ChangefeedType `gorm:"column:sink_type;type:varchar(32);not null"`
	Kafka            *KafkaInfo              `gorm:"column:kafka;type:json;serializer:json;default:null"`
	MySQL            *MySQLInfo              `gorm:"column:mysql;type:json;serializer:json;default:null"`
	CloudStorageInfo *CloudStorageInfo       `gorm:"column:cloud_storage;type:json;serializer:json;default:null"`
	Filter           *CDCFilter              `gorm:"column:filter;type:json;serializer:json;default:null"`
	StartMode        constant.StartMode      `gorm:"column:start_mode;type:varchar(32)"`
	StartTSO         uint64                  `gorm:"column:start_tso;type:bigint(20);default:null"`
	TargetTSO        uint64                  `gorm:"column:target_tso;type:bigint(20);default:null"`
	CheckpointTSO    uint64                  `gorm:"column:checkpoint_tso;bigint(20);default:null"`
	Error            string                  `gorm:"column:error;type:text"`
	FailedTime       *time.Time              `gorm:"column:failed_time;type:timestamp;null;default:null"`
	CreatedAt        time.Time               `gorm:"column:created_at;type:timestamp;not null;autoCreateTime;index:idx_created_at"`
	UpdatedAt        *time.Time              `gorm:"column:updated_at;type:timestamp;null;default:null;autoUpdateTime"`
	DeletedAt        *time.Time              `gorm:"column:deleted_at;type:timestamp;null"`
	LastSyncAt       *time.Time              `gorm:"column:last_sync_at;type:timestamp;null;default:null"`
	RunningAt        *time.Time              `gorm:"column:running_at;type:timestamp;null;default:null"`
}

func (c *CDCChangefeed) GenerateName() {
	if c.Name == "" {
		c.Name = fmt.Sprintf("%s-%s-%s", DefaultNamePrefix, c.SinkType, time.Now().Format(DateFormat))
	}
}

func (c *CDCChangefeed) TableName() string {
	return "cdc_changefeeds"
}

type KafkaInfo struct {
	Network              KafkaNetwork         `json:"network,omitempty"`
	Broker               Broker               `json:"broker,omitempty"`
	Authentication       Authentication       `json:"authentication,omitempty"`
	DataFormat           DataFormat           `json:"data_format,omitempty"`
	TopicPartitionConfig TopicPartitionConfig `json:"topic_partition_config,omitempty"`
	ColumnSelectors      []ColumnSelector     `json:"column_selectors,omitempty"`
	OutputRawChangeEvent bool                 `json:"output_raw_change_event,omitempty"`
}

type MySQLInfo struct {
	Network        MySQLNetwork        `json:"network,omitempty"`
	Authentication MySQLAuthentication `json:"authentication,omitempty"`
}

type CloudStorageInfo struct {
	CloudStorage         *CloudStorage                      `json:"cloud_storage,omitempty"`
	DataFormat           CloudStorageDataFormat             `json:"data_format,omitempty"`
	DateSeparator        constant.CloudStorageDateSeparator `json:"date_separator,omitempty"`
	IntervalInSeconds    int                                `json:"interval_in_seconds,omitempty"`
	SizeInMiB            int                                `json:"size_in_mib,omitempty"`
	OutputRawChangeEvent bool                               `json:"output_raw_change_event,omitempty"`
	OutputColumnID       bool                               `json:"output_column_id,omitempty"`
	ColumnSelectors      []ColumnSelector                   `json:"column_selectors,omitempty"`
}

type CloudStorageDataFormat struct {
	Protocol        constant.CloudStorageProtocol `json:"protocol,omitempty"`
	CSVConfig       *CloudStorageCSVConfig        `json:"csv_config,omitempty"`
	CanalJSONConfig *CloudStorageCanalJSONCfg     `json:"canal_json_config,omitempty"`
}

type CloudStorageCSVConfig struct {
	Delimiter       *string                             `json:"delimiter,omitempty"`
	Quote           *string                             `json:"quote,omitempty"`
	NullValue       *string                             `json:"null_value,omitempty"`
	IncludeCommitTs bool                                `json:"include_commit_ts,omitempty"`
	Encoding        constant.CloudStorageBinaryEncoding `json:"encoding,omitempty"`
}

type CloudStorageCanalJSONCfg struct {
	EnableTiDBExtension bool `json:"enable_tidb_extension,omitempty"`
}

type MySQLNetwork struct {
	NetworkType             constant.MySQLNetworkType `json:"network_type,omitempty"`
	PublicEndpoint          string                    `json:"public_endpoint,omitempty"`
	PrivateLinkConnectionId string                    `json:"private_link_connection_id,omitempty"`
	PrivateLinkPort         int32                     `json:"private_link_port,omitempty"`
	PrivateLinkEndpoint     string                    `json:"private_link_endpoint,omitempty"`
}

type MySQLAuthentication struct {
	User      string `json:"user,omitempty"`
	Password  string `json:"password,omitempty"`
	EnableTLS bool   `json:"enable_tls,omitempty"`
}

func (m *MySQLInfo) Validate(ctx context.Context, encryptor facade.EncryptorFacade) bizerr.BizError {
	if m == nil {
		return bizerr.New(bizerr.ErrCodeParams, "mysql info cannot be nil")
	}
	// Validate Network
	switch m.Network.NetworkType {
	case constant.MySQLNetworkTypePublic:
		if !validateEndpoint(m.Network.PublicEndpoint) {
			return bizerr.New(bizerr.ErrCodeParams, "invalid MySQL endpoint: %s", m.Network.PublicEndpoint)
		}
	case constant.MySQLNetworkTypePrivateLink:
		if m.Network.PrivateLinkPort < 1 || m.Network.PrivateLinkPort > 65535 {
			return bizerr.New(bizerr.ErrCodeParams, "invalid private link port: %d", m.Network.PrivateLinkPort)
		}
	}
	return nil
}

func (k *KafkaInfo) Validate(ctx context.Context, encryptor facade.EncryptorFacade, isTest bool) bizerr.BizError {
	logger := log.WithContextL(ctx)
	if k == nil {
		return bizerr.New(bizerr.ErrCodeParams, "kafka info cannot be nil")
	}

	// Validate NetworkInfo
	switch k.Network.NetworkType {
	case constant.KafkaNetworkTypePublic:
		endpoints := strings.Split(k.Network.PublicEndpoints, ",")
		for _, endpoint := range endpoints {
			if !validateEndpoint(endpoint) {
				return bizerr.New(bizerr.ErrCodeParams, "invalid broker endpoint: %s", endpoint)
			}
		}
	}

	// Validate Authentication
	if k.Authentication.Password != "" {
		_, err := encryptor.EncryptBase64(k.Authentication.Password)
		if err != nil {
			logger.Error("failed to encrypt password", zap.Error(err))
			return bizerr.New(bizerr.ErrCodeUnknown, "failed to encrypt kafka password")
		}
	}

	if isTest {
		return nil
	}

	// Validate TopicPartitionConfig
	if !validateTopicName(k.TopicPartitionConfig.DefaultTopic) {
		return bizerr.New(bizerr.ErrCodeParams, "invalid topic name: %s", k.TopicPartitionConfig.DefaultTopic)
	}
	if k.TopicPartitionConfig.DispatchType == constant.TopicDispatchTypeByTable ||
		k.TopicPartitionConfig.DispatchType == constant.TopicDispatchTypeByDatabase {
		if !validateTopicNamePlaceholders(k.TopicPartitionConfig.TopicPrefix) {
			return bizerr.New(bizerr.ErrCodeParams, "invalid topic prefix: %s", k.TopicPartitionConfig.TopicPrefix)
		}
		if !validateTopicNamePlaceholders(k.TopicPartitionConfig.TopicSuffix) {
			return bizerr.New(bizerr.ErrCodeParams, "invalid topic suffix: %s", k.TopicPartitionConfig.TopicSuffix)
		}
		if !validateTopicNamePlaceholders(k.TopicPartitionConfig.Separator) {
			return bizerr.New(bizerr.ErrCodeParams, "invalid topic separator: %s", k.TopicPartitionConfig.Separator)
		}
	}

	for _, p := range k.TopicPartitionConfig.PartitionDispatchers {
		for _, matcher := range p.Matcher {
			if err := utils.ParseTableFilter(matcher); err != nil {
				logger.Error("invalid partition dispatchers matcher", zap.String("matcher", matcher), zap.Error(err))
				return bizerr.New(bizerr.ErrCodeParams, "invalid partition dispatchers matcher: %s", matcher)
			}
		}
	}

	// Validate DataFormat
	switch k.DataFormat.Protocol {
	case constant.CDCProtocolAvro:
		if k.TopicPartitionConfig.DispatchType != constant.TopicDispatchTypeByTable {
			return bizerr.New(
				bizerr.ErrCodeParams,
				"avro data format not support dispatch type: %s",
				string(k.TopicPartitionConfig.DispatchType),
			)
		}
		// check SchemaRegistry
		if k.DataFormat.AvroConfig.ConfluentSchemaRegistry != nil {
			schemaRegistry := k.DataFormat.AvroConfig.ConfluentSchemaRegistry
			if !strings.HasPrefix(schemaRegistry.SchemaRegistryEndpoints, "http://") &&
				!strings.HasPrefix(schemaRegistry.SchemaRegistryEndpoints, "https://") {
				return bizerr.New(
					bizerr.ErrCodeParams,
					"schema registry %s invalid",
					schemaRegistry.SchemaRegistryEndpoints,
				)
			}
			if schemaRegistry.EnableHTTPAuth {
				_, err := encryptor.EncryptBase64(schemaRegistry.Password)
				if err != nil {
					logger.Error("failed to encrypt schema registry password", zap.Error(err))
					return bizerr.New(bizerr.ErrCodeUnknown, "failed to encrypt schema registry password")
				}
			}
		}
		if k.DataFormat.AvroConfig.AWSGlueSchemaRegistry != nil {
			_, err := encryptor.EncryptBase64(k.DataFormat.AvroConfig.AWSGlueSchemaRegistry.SecretAccessKey)
			if err != nil {
				logger.Error("failed to encrypt schema registry secret access key", zap.Error(err))
				return bizerr.New(bizerr.ErrCodeUnknown, "failed to encrypt schema registry secret access key")
			}
		}
	}

	return validateColumnSelectors(ctx, k.ColumnSelectors)
}

func (c *CloudStorageInfo) Validate(ctx context.Context, cloudProvider string, isTest bool) bizerr.BizError {
	if c == nil {
		return bizerr.New(bizerr.ErrCodeParams, "cloud storage info cannot be nil")
	}
	if c.CloudStorage == nil {
		return bizerr.New(bizerr.ErrCodeParams, "cloud storage cannot be nil")
	}
	if c.CloudStorage.Type == constant.CloudStorageTypeTiDBCloud {
		return bizerr.New(bizerr.ErrCodeParams, "tidbcloud cloud storage is not supported yet in changefeed")
	}
	if err := validateColumnSelectors(ctx, c.ColumnSelectors); err != nil {
		return err
	}
	if c.CloudStorage.Type == constant.CloudStorageTypeGCS {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"gcs cloud storage is not supported yet, please contact us if you need it",
		)
	}
	if c.CloudStorage.Type == constant.CloudStorageTypeOSS && c.CloudStorage.OSS.RoleArn != nil {
		return bizerr.New(bizerr.ErrCodeParams, "oss role arn is not supported yet")
	}
	if err := c.CloudStorage.Validate(cloudProvider); err != nil {
		return err
	}

	if isTest {
		return nil
	}

	if c.IntervalInSeconds < 2 {
		return bizerr.New(bizerr.ErrCodeParams, "interval_in_seconds cannot be less than 2")
	}
	if c.SizeInMiB < 1 {
		return bizerr.New(bizerr.ErrCodeParams, "size_in_mib cannot be less than 1")
	}
	if c.IntervalInSeconds > 600 {
		return bizerr.New(bizerr.ErrCodeParams, "interval_in_seconds cannot be greater than 600")
	}
	if c.SizeInMiB > 512 {
		return bizerr.New(bizerr.ErrCodeParams, "size_in_mib cannot be greater than 512")
	}

	switch c.DataFormat.Protocol {
	case constant.CloudStorageProtocolCSV:
		if c.DataFormat.CSVConfig.Delimiter != nil && *c.DataFormat.CSVConfig.Delimiter == "" {
			return bizerr.New(
				bizerr.ErrCodeParams,
				"delimiter cannot be empty string, if you want to use default delimiter, please don't set it",
			)
		}
	}

	return nil
}

func validateColumnSelectors(ctx context.Context, selectors []ColumnSelector) bizerr.BizError {
	logger := log.WithContextL(ctx)
	for _, cs := range selectors {
		if len(cs.Matcher) == 0 {
			return bizerr.New(bizerr.ErrCodeParams, "column selector matcher cannot be empty")
		}
		if len(cs.Columns) == 0 {
			return bizerr.New(bizerr.ErrCodeParams, "column selector columns cannot be empty")
		}
		if err := utils.ParseColumnFilter(cs.Columns); err != nil {
			logger.Error("invalid column selector columns", zap.Strings("columns", cs.Columns), zap.Error(err))
			return bizerr.New(bizerr.ErrCodeParams, "invalid column selector columns")
		}
		for _, matcher := range cs.Matcher {
			if err := utils.ParseTableFilter(matcher); err != nil {
				logger.Error("invalid column selector matcher", zap.String("matcher", matcher), zap.Error(err))
				return bizerr.New(bizerr.ErrCodeParams, "invalid column selector matcher: %s", matcher)
			}
		}
	}
	return nil
}

// ValidateCloudStorageColumnSelectorsArch prevents a cloud-storage selector
// from being silently ignored by the legacy replication-worker architecture.
func ValidateCloudStorageColumnSelectorsArch(
	arch constant.ChangefeedArch,
	cloudStorage *CloudStorageInfo,
) bizerr.BizError {
	if cloudStorage != nil && len(cloudStorage.ColumnSelectors) > 0 &&
		arch != constant.ChangefeedArchTiCDCNewArch {
		return bizerr.New(bizerr.ErrCodeParams,
			"cloud storage column selectors are only supported by ticdc newarch changefeeds")
	}
	return nil
}

func validateTopicName(s string) bool {
	return topicNameRegexp.MatchString(s)
}

func validateTopicNamePlaceholders(s string) bool {
	if s == "" {
		return true
	}
	return topicNameRegexp.MatchString(s)
}

func validateEndpoint(s string) bool {
	if s == "" {
		return true
	}
	epList := strings.Split(s, ":")
	if len(epList) != EndpointLength {
		return false
	}
	// validate port number
	if len(epList[1]) > 1 && epList[1][0] == '0' {
		// e.g. "01"
		return false
	}
	port, err := strconv.ParseInt(epList[1], 10, 64)
	return err == nil && port >= 1 && port <= 65535
}

type KafkaNetwork struct {
	NetworkType             constant.KafkaNetworkType `json:"network_type,omitempty"`
	PrivateLinkConnectionId string                    `json:"private_link_connection_id,omitempty"`
	PublicEndpoints         string                    `json:"public_endpoints,omitempty"`
	PrivateLinkPort         int32                     `json:"private_link_port,omitempty"`
	PrivateLinkEndpoints    string                    `json:"private_link_endpoints,omitempty"`
}

type Broker struct {
	KafkaVersion constant.KafkaVersion         `json:"kafka_version,omitempty"`
	Compression  constant.KafkaCompressionType `json:"compression,omitempty"`
}

type Authentication struct {
	AuthType   constant.KafkaAuthType `json:"auth_type,omitempty"`
	UserName   string                 `json:"user_name,omitempty"`
	Password   string                 `json:"password,omitempty"`
	EnableTLS  bool                   `json:"enable_tls,omitempty"`
	ServerName string                 `json:"server_name,omitempty"`
}

type DataFormat struct {
	Protocol            constant.CDCProtocol `json:"protocol,omitempty"`
	EnableTiDBExtension bool                 `json:"enable_tidb_extension,omitempty"`
	AvroConfig          *AvroConfig          `json:"avro_config,omitempty"`
	DebeziumConfig      *DebeziumConfig      `json:"debezium_config,omitempty"`
}

type DebeziumConfig struct {
	DisableOutputOldValue bool `json:"disable_output_old_value,omitempty"`
	DisableSchema         bool `json:"disable_schema,omitempty"`
}

type AvroConfig struct {
	DecimalHandlingMode        constant.DecimalHandlingMode        `json:"decimal_handling_mode,omitempty"`
	BigintUnsignedHandlingMode constant.BigintUnsignedHandlingMode `json:"bigint_unsigned_handling_mode,omitempty"`
	ConfluentSchemaRegistry    *ConfluentSchemaRegistry            `json:"confluent_schema_registry,omitempty"`
	AWSGlueSchemaRegistry      *AWSGlueSchemaRegistry              `json:"aws_glue_schema_registry,omitempty"`
}

type ConfluentSchemaRegistry struct {
	SchemaRegistryEndpoints string `json:"schema_registry_endpoints,omitempty"`
	EnableHTTPAuth          bool   `json:"enable_http_auth,omitempty"`
	UserName                string `json:"user_name,omitempty"`
	Password                string `json:"password,omitempty"`
}

type AWSGlueSchemaRegistry struct {
	Region          string `json:"region,omitempty"`
	Name            string `json:"name,omitempty"`
	AccessKeyID     string `json:"access_key_id,omitempty"`
	SecretAccessKey string `json:"secret_access_key,omitempty"`
}
type TopicPartitionConfig struct {
	DispatchType         constant.TopicDispatchType   `json:"dispatch_type,omitempty"`
	DefaultTopic         string                       `json:"default_topic,omitempty"`
	TopicPrefix          string                       `json:"topic_prefix,omitempty"`
	Separator            string                       `json:"separator,omitempty"`
	TopicSuffix          string                       `json:"topic_suffix,omitempty"`
	ReplicationFactor    uint32                       `json:"replication_factor,omitempty"`
	PartitionNum         uint32                       `json:"partition_num,omitempty"`
	PartitionDispatchers []*PartitionDispatcherConfig `json:"partition_dispatchers,omitempty"`
}

type PartitionDispatcherConfig struct {
	PartitionType constant.PartitionDispatcher `json:"partition_type,omitempty"`
	Matcher       []string                     `json:"matcher,omitempty"`
	IndexName     string                       `json:"index_name,omitempty"`
	Columns       []string                     `json:"columns,omitempty"`
}

type ColumnSelector struct {
	Matcher []string `json:"matcher,omitempty"`
	Columns []string `json:"columns,omitempty"`
}

type CDCFilter struct {
	FilterRule       []string           `json:"filter_rule,omitempty"`
	Mode             constant.TableMode `json:"mode,omitempty"`
	EventFilterRule  []*EventFilterRule `json:"event_filter_rule,omitempty"`
	CaseSensitive    bool               `json:"case_sensitive,omitempty"`
	IgnoreTxnStartTs []uint64           `json:"ignore_txn_start_ts,omitempty"`
	BDRMode          *bool              `json:"bdr_mode,omitempty"`
}

func (f *CDCFilter) Validate(ctx context.Context) bizerr.BizError {
	logger := log.WithContextL(ctx)
	if f == nil {
		return bizerr.New(bizerr.ErrCodeParams, "filter cannot be nil")
	}
	for _, f := range f.FilterRule {
		if err := utils.ParseTableFilter(f); err != nil {
			logger.Error("invalid filter", zap.String("filter", f), zap.Error(err))
			return bizerr.New(bizerr.ErrCodeParams, "invalid filter rule: %s", f)
		}
	}
	for _, ef := range f.EventFilterRule {
		for _, matcher := range ef.Matcher {
			if err := utils.ParseTableFilter(matcher); err != nil {
				logger.Error("invalid event filter matcher", zap.String("matcher", matcher), zap.Error(err))
				return bizerr.New(bizerr.ErrCodeParams, "invalid matcher: '%s' in event filter", matcher)
			}
		}
		for _, ignoreEvent := range ef.IgnoreEvent {
			if !slices.Contains(validEvents, ignoreEvent) {
				logger.Error("invalid event filter ignore event", zap.String("ignore event", ignoreEvent))
				return bizerr.New(bizerr.ErrCodeParams, "invalid ignore_event: %s in event filter", ignoreEvent)
			}
		}
	}
	return nil
}

type EventFilterRule struct {
	Matcher                  []string `json:"matcher,omitempty"`
	IgnoreEvent              []string `json:"ignore_event,omitempty"`
	IgnoreSQL                []string `json:"ignore_sql,omitempty"`
	IgnoreInsertValueExpr    string   `json:"ignore_insert_value_expr,omitempty"`
	IgnoreUpdateNewValueExpr string   `json:"ignore_update_new_value_expr,omitempty"`
	IgnoreUpdateOldValueExpr string   `json:"ignore_update_old_value_expr,omitempty"`
	IgnoreDeleteValueExpr    string   `json:"ignore_delete_value_expr,omitempty"`
}

type ListChangefeedsFilter struct {
	ClusterID       string
	CloudProvider   string
	Region          string
	Kubernetes      string
	Arch            constant.ChangefeedArch
	Limit           *int32
	Offset          *int64
	State           constant.ChangfeedState
	SinkType        string
	OrderBy         string
	StatesNotIn     []constant.ChangfeedState
	StatesIn        []constant.ChangfeedState
	CheckpointTSOLT uint64
	FailedTimeGTE   *time.Time
}

func (c *CDCChangefeed) FillWithCluster(cluster *tidbmgmtservicev1.Cluster) {
	c.CloudProvider = cluster.CloudProvider
	c.TenantID = strconv.FormatUint(cluster.OrganizationId, 10)
	c.Region = cluster.Region
	c.Kubernetes = cluster.GetServerlessCluster().GetCluster().GetKubernetes()
	c.KeySpaceID = cluster.GetServerlessCluster().GetCluster().GetKeyspaceId()
}

func (c *CDCChangefeed) Desensitization(ctx context.Context, encryptor facade.EncryptorFacade) bizerr.BizError {
	logger := log.WithContextL(ctx)
	if c == nil {
		return bizerr.New(bizerr.ErrCodeParams, "cdc changefeed cannot be nil")
	}
	// Desensitize Kafka Authentication
	if c.Kafka != nil && c.Kafka.Authentication.Password != "" {
		passwordSecret, err := encryptor.EncryptBase64(c.Kafka.Authentication.Password)
		if err != nil {
			logger.Error("failed to encrypt kafka password", zap.Error(err))
			return bizerr.New(bizerr.ErrCodeUnknown, "failed to encrypt kafka password")
		}
		c.Kafka.Authentication.Password = passwordSecret
	}
	// Desensitize Schema Registry Authentication
	if c.Kafka != nil && c.Kafka.DataFormat.AvroConfig != nil {
		if c.Kafka.DataFormat.AvroConfig.ConfluentSchemaRegistry != nil &&
			c.Kafka.DataFormat.AvroConfig.ConfluentSchemaRegistry.Password != "" {
			secret, err := encryptor.EncryptBase64(c.Kafka.DataFormat.AvroConfig.ConfluentSchemaRegistry.Password)
			if err != nil {
				logger.Error("failed to encrypt schema registry password", zap.Error(err))
				return bizerr.New(bizerr.ErrCodeUnknown, "failed to encrypt schema registry password")
			}
			c.Kafka.DataFormat.AvroConfig.ConfluentSchemaRegistry.Password = secret
		}
		if c.Kafka.DataFormat.AvroConfig.AWSGlueSchemaRegistry != nil {
			secret, err := encryptor.EncryptBase64(c.Kafka.DataFormat.AvroConfig.AWSGlueSchemaRegistry.SecretAccessKey)
			if err != nil {
				logger.Error("failed to encrypt schema registry secret access key", zap.Error(err))
				return bizerr.New(bizerr.ErrCodeUnknown, "failed to encrypt schema registry secret access key")
			}
			c.Kafka.DataFormat.AvroConfig.AWSGlueSchemaRegistry.SecretAccessKey = secret
		}
	}
	// Desensitize MySQL Connection
	if c.MySQL != nil && c.MySQL.Authentication.Password != "" {
		passwordSecret, err := encryptor.EncryptBase64(c.MySQL.Authentication.Password)
		if err != nil {
			logger.Error("failed to encrypt mysql password", zap.Error(err))
			return bizerr.New(bizerr.ErrCodeUnknown, "failed to encrypt mysql password")
		}
		c.MySQL.Authentication.Password = passwordSecret
	}
	// Desensitize Cloud Storage
	if c.CloudStorageInfo != nil {
		cs := c.CloudStorageInfo.CloudStorage
		switch cs.Type {
		case constant.CloudStorageTypeS3:
			if cs.S3.AccessKey != nil {
				secret, err := encryptor.EncryptBase64(cs.S3.AccessKey.SecretAccessKey)
				if err != nil {
					logger.Error("failed to encrypt s3 secret access key", zap.Error(err))
					return bizerr.New(bizerr.ErrCodeUnknown, "failed to encrypt s3 secret access key")
				}
				cs.S3.AccessKey.SecretAccessKey = secret
			}
		case constant.CloudStorageTypeGCS:
			if cs.GCS.ServiceAccountKey != "" {
				secret, err := encryptor.EncryptBase64(cs.GCS.ServiceAccountKey)
				if err != nil {
					logger.Error("failed to encrypt gcs service account key", zap.Error(err))
					return bizerr.New(bizerr.ErrCodeUnknown, "failed to encrypt gcs service account key")
				}
				cs.GCS.ServiceAccountKey = secret
			}
		case constant.CloudStorageTypeAzureBlob:
			if cs.AzureBlob.SasToken != "" {
				secret, err := encryptor.EncryptBase64(cs.AzureBlob.SasToken)
				if err != nil {
					logger.Error("failed to encrypt azure blob sas token", zap.Error(err))
					return bizerr.New(bizerr.ErrCodeUnknown, "failed to encrypt azure blob sas token")
				}
				cs.AzureBlob.SasToken = secret
			}
		case constant.CloudStorageTypeOSS:
			if cs.OSS.AccessKey != nil {
				secret, err := encryptor.EncryptBase64(cs.OSS.AccessKey.AccessKeySecret)
				if err != nil {
					logger.Error("failed to encrypt oss secret access key", zap.Error(err))
					return bizerr.New(bizerr.ErrCodeUnknown, "failed to encrypt oss secret access key")
				}
				cs.OSS.AccessKey.AccessKeySecret = secret
			}
		default:
			return bizerr.New(bizerr.ErrCodeParams, "unsupported cloud storage type: %s", cs.Type)
		}
	}
	return nil
}

type CDCCluster struct {
	ClusterID       string                   `gorm:"column:cluster_id;type:varchar(255);primaryKey;not null"`
	TenantID        string                   `gorm:"column:tenant_id;type:varchar(255);not null"`
	CloudProvider   string                   `gorm:"column:cloud_provider;type:varchar(255);not null"`
	Region          string                   `gorm:"column:region;type:varchar(255);not null"`
	Kubernetes      string                   `gorm:"column:kubernetes;type:varchar(255);not null;"`
	KeySpaceID      uint32                   `gorm:"column:keyspace_id;type:bigint(20);not null;"`
	ClusterStatusId uint64                   `gorm:"column:cluster_status_id;type:bigint(20);not null"`
	ChangefeedCount int                      `gorm:"column:changefeed_count;type:bigint(20);not null; default:1"`
	CreatedAt       *time.Time               `gorm:"column:created_at;type:timestamp;not null;autoCreateTime"`
	UpdatedAt       *time.Time               `gorm:"column:updated_at;type:timestamp;default:null;autoUpdateTime"`
	State           constant.CDCClusterState `gorm:"column:state;type:varchar(32);not null"`
}

func (c *CDCCluster) TableName() string {
	return "cdc_cluster"
}

type ChangefeedCountByCluster struct {
	ClusterID string
	Count     int
}

func (c *CDCChangefeed) ToChangefeed(
	ctx context.Context,
	encryptor facade.EncryptorFacade,
) (_ *serverless.ChangefeedConfig, err bizerr.BizError) {
	// sink uri
	var sinkUri string
	switch c.SinkType {
	case constant.ChangefeedTypeKafka:
		if c.Kafka.TopicPartitionConfig.DefaultTopic == "" {
			return nil, bizerr.New(bizerr.ErrCodeParams, "kafka default topic cannot be empty")
		}
		endpoints := c.Kafka.Network.PublicEndpoints
		if c.Kafka.Network.NetworkType == constant.KafkaNetworkTypePrivateLink {
			endpoints = c.Kafka.Network.PrivateLinkEndpoints
			// use server name as the endpoint if provided.
			// this is only a workaround for server name indication because ticdc does not support specifying server name now.
			// the best way is specify endpoints as url and specify server name separately.
			if c.Kafka.Authentication.ServerName != "" {
				endpoints = fmt.Sprintf("%s:%d", c.Kafka.Authentication.ServerName, c.Kafka.Network.PrivateLinkPort)
			}
		}
		sinkUri = fmt.Sprintf("kafka://%s/%s",
			endpoints,
			c.Kafka.TopicPartitionConfig.DefaultTopic)
	case constant.ChangefeedTypeMySQL:
		var password string
		if c.MySQL.Authentication.Password != "" {
			pswd, err := encryptor.DecryptBase64(c.MySQL.Authentication.Password)
			if err != nil {
				return nil, bizerr.New(bizerr.ErrCodeParams, "failed to decrypt mysql password: %v", err)
			}
			password = pswd
		}
		endpoint := c.MySQL.Network.PublicEndpoint
		if c.MySQL.Network.NetworkType == constant.MySQLNetworkTypePrivateLink {
			endpoint = c.MySQL.Network.PrivateLinkEndpoint
		}
		sinkUri = fmt.Sprintf("mysql://%s:%s@%s",
			url.QueryEscape(c.MySQL.Authentication.User),
			url.QueryEscape(password),
			// endpoint should not contains special characters
			endpoint,
		)
	case constant.ChangefeedTypeCloudStorage:
		sinkUri, err = c.CloudStorageSinkUri(ctx, encryptor)
		if err != nil {
			return nil, err
		}
	default:
		return nil, bizerr.New(bizerr.ErrCodeParams, "unsupported sink type: %s", c.SinkType)
	}

	// filter
	var eventFilters []*serverless.EventFilter
	if len(c.Filter.EventFilterRule) > 0 {
		eventFilters = make([]*serverless.EventFilter, 0, len(c.Filter.EventFilterRule))
		for _, ef := range c.Filter.EventFilterRule {
			sef := &serverless.EventFilter{
				Matcher:                  ef.Matcher,
				IgnoreDeleteValueExpr:    &ef.IgnoreDeleteValueExpr,
				IgnoreInsertValueExpr:    &ef.IgnoreInsertValueExpr,
				IgnoreUpdateNewValueExpr: &ef.IgnoreUpdateNewValueExpr,
				IgnoreUpdateOldValueExpr: &ef.IgnoreUpdateOldValueExpr,
				IgnoreEvent:              ef.IgnoreEvent,
				IgnoreSQL:                ef.IgnoreSQL,
			}
			eventFilters = append(eventFilters, sef)
		}
	}

	// sink config
	sinkConfig := &serverless.SinkConfig{}
	switch c.SinkType {
	case constant.ChangefeedTypeKafka:
		if err := c.KafkaSinkConfig(sinkConfig, encryptor); err != nil {
			return nil, err
		}
	case constant.ChangefeedTypeMySQL:
		if err := c.MySQLSinkConfig(sinkConfig, encryptor); err != nil {
			return nil, err
		}
	case constant.ChangefeedTypeCloudStorage:
		if err := c.CloudStorageSinkConfig(sinkConfig, encryptor); err != nil {
			return nil, err
		}
	default:
		return nil, bizerr.New(bizerr.ErrCodeParams, "unsupported sink type: %s", c.SinkType)
	}

	// start TSO
	var startTSO *uint64
	if c.StartTSO > 0 {
		startTSO = &c.StartTSO
	}

	ignoreTxnStartTS := make([]*uint64, 0, len(c.Filter.IgnoreTxnStartTs))
	for i := range c.Filter.IgnoreTxnStartTs {
		ignoreTxnStartTS = append(ignoreTxnStartTS, &c.Filter.IgnoreTxnStartTs[i])
	}

	cf := &serverless.ChangefeedConfig{
		// use db primary key as changefeed id
		ID:      aws.String(c.ID),
		StartTS: startTSO,
		SinkURI: aws.String(sinkUri),
		ReplicaConfig: &serverless.ReplicaConfig{
			Filter: &serverless.FilterConfig{
				Rules:            c.Filter.FilterRule,
				EventFilters:     eventFilters,
				IgnoreTxnStartTS: ignoreTxnStartTS,
			},
			CaseSensitive:         aws.Bool(c.Filter.CaseSensitive),
			IgnoreIneligibleTable: aws.Bool(c.Filter.Mode == constant.TableModeIgnoreNotSupportTable),
			ForceReplicate:        aws.Bool(c.Filter.Mode == constant.TableModeForceSync),
			BDRMode:               c.Filter.BDRMode,
			Sink:                  sinkConfig,
		},
	}
	return cf, nil
}

func (c *CDCChangefeed) ToChangefeedForUpdate(
	ctx context.Context,
	encryptor facade.EncryptorFacade,
) (*serverless.ChangefeedConfig, bizerr.BizError) {
	cff, err := c.ToChangefeed(ctx, encryptor)
	if err != nil {
		return nil, err
	}
	// update need not to change the ID and StartTS
	cff.ID = nil
	cff.StartTS = nil
	return cff, nil
}

func (c *CDCChangefeed) CloudStorageSinkUri(
	ctx context.Context,
	encryptor facade.EncryptorFacade,
) (string, bizerr.BizError) {
	cs := c.CloudStorageInfo.CloudStorage
	switch cs.Type {
	case constant.CloudStorageTypeS3:
		if cs.S3.AccessKey != nil {
			key, err := encryptor.DecryptBase64(cs.S3.AccessKey.SecretAccessKey)
			if err != nil {
				return "", bizerr.New(bizerr.ErrCodeUnknown, "failed to decrypt s3 secret access key: %v", err)
			}
			cs.S3.AccessKey.SecretAccessKey = key
		}
	case constant.CloudStorageTypeGCS:
		if cs.GCS.ServiceAccountKey != "" {
			key, err := encryptor.DecryptBase64(cs.GCS.ServiceAccountKey)
			if err != nil {
				return "", bizerr.New(bizerr.ErrCodeUnknown, "failed to decrypt gcs service account key: %v", err)
			}
			cs.GCS.ServiceAccountKey = key
		}
	case constant.CloudStorageTypeAzureBlob:
		if cs.AzureBlob.SasToken != "" {
			key, err := encryptor.DecryptBase64(cs.AzureBlob.SasToken)
			if err != nil {
				return "", bizerr.New(bizerr.ErrCodeUnknown, "failed to decrypt azure blob sas token: %v", err)
			}
			cs.AzureBlob.SasToken = key
		}
	case constant.CloudStorageTypeOSS:
		if cs.OSS.AccessKey != nil {
			key, err := encryptor.DecryptBase64(cs.OSS.AccessKey.AccessKeySecret)
			if err != nil {
				return "", bizerr.New(bizerr.ErrCodeUnknown, "failed to decrypt oss secret access key: %v", err)
			}
			cs.OSS.AccessKey.AccessKeySecret = key
		}
	default:
		return "", bizerr.New(bizerr.ErrCodeParams, "unsupported cloud storage type: %s", cs.Type)
	}
	cloudStorageUri, err := cs.ToUri(ctx, c.Region, c.CloudProvider)
	if err != nil {
		return "", err
	}
	// ticdc cloud storage sink does not support enable-tidb-extension in sinkConfig
	if c.CloudStorageInfo.DataFormat.Protocol == constant.CloudStorageProtocolCanalJSON {
		if c.CloudStorageInfo.DataFormat.CanalJSONConfig.EnableTiDBExtension {
			cloudStorageUri += "&enable-tidb-extension=true"
		}
	}
	return cloudStorageUri, nil
}

func (c *CDCChangefeed) KafkaSinkConfig(sink *serverless.SinkConfig, encryptor facade.EncryptorFacade) bizerr.BizError {
	kafkaConfig := &serverless.KafkaConfig{
		AutoCreateTopic: aws.Bool(true),
	}
	sink.KafkaConfig = kafkaConfig

	// Broker
	switch c.Kafka.Broker.KafkaVersion {
	case constant.KafkaVersion2XX:
		kafkaConfig.KafkaVersion = aws.String("2.0.0")
	case constant.KafkaVersion3XX:
		kafkaConfig.KafkaVersion = aws.String("3.0.0")
	default:
		return bizerr.New(bizerr.ErrCodeParams, "unsupported kafka version: %s", c.Kafka.Broker.KafkaVersion)
	}
	if c.Kafka.Broker.Compression != constant.KafkaCompressionTypeNone {
		kafkaConfig.Compression = aws.String(string(c.Kafka.Broker.Compression))
	}

	// Auth
	switch c.Kafka.Authentication.AuthType {
	case constant.KafkaAuthTypeSASLPlain, constant.KafkaAuthTypeSASLScram256, constant.KafkaAuthTypeSASLScram512:
		kafkaConfig.SASLMechanism = aws.String(string(c.Kafka.Authentication.AuthType))
		userName := c.Kafka.Authentication.UserName
		password, err := encryptor.DecryptBase64(c.Kafka.Authentication.Password)
		if err != nil {
			return bizerr.New(bizerr.ErrCodeParams, "failed to decrypt kafka password: %v", err)
		}
		kafkaConfig.SASLUser = &userName
		kafkaConfig.SASLPassword = aws.String(password)
		kafkaConfig.EnableTLS = aws.Bool(c.Kafka.Authentication.EnableTLS)
		if c.Kafka.Authentication.EnableTLS {
			// default to skip server name verify
			kafkaConfig.InsecureSkipVerify = aws.Bool(true)
		}
	}

	// DataFormat
	sink.Protocol = aws.String(string(c.Kafka.DataFormat.Protocol))
	switch c.Kafka.DataFormat.Protocol {
	case constant.CDCProtocolAvro:
		kafkaConfig.CodecConfig = &serverless.CodecConfig{
			EnableTidbExtension:     aws.Bool(c.Kafka.DataFormat.EnableTiDBExtension),
			AvroDecimalHandlingMode: aws.String(string(c.Kafka.DataFormat.AvroConfig.DecimalHandlingMode)),
			AvroBigintUnsignedHandlingMode: aws.String(
				string(c.Kafka.DataFormat.AvroConfig.BigintUnsignedHandlingMode),
			),
		}
		if c.Kafka.DataFormat.AvroConfig.ConfluentSchemaRegistry != nil {
			schemaRegistryUri, err := buildSchemaRegistryURI(
				c.Kafka.DataFormat.AvroConfig.ConfluentSchemaRegistry,
				encryptor,
			)
			if err != nil {
				return err
			}
			sink.SchemaRegistry = aws.String(schemaRegistryUri)
		}
		if c.Kafka.DataFormat.AvroConfig.AWSGlueSchemaRegistry != nil {
			awsGlue := c.Kafka.DataFormat.AvroConfig.AWSGlueSchemaRegistry
			secretAccessKey, err := encryptor.DecryptBase64(awsGlue.SecretAccessKey)
			if err != nil {
				return bizerr.New(
					bizerr.ErrCodeParams,
					"failed to decrypt schema registry secret access key with error: %s",
					err.Error(),
				)
			}
			kafkaConfig.GlueSchemaRegistryConfig = &serverless.GlueSchemaRegistryConfig{
				RegistryName:    aws.String(awsGlue.Name),
				Region:          aws.String(awsGlue.Region),
				AccessKey:       aws.String(awsGlue.AccessKeyID),
				SecretAccessKey: aws.String(secretAccessKey),
			}
		}
	case constant.CDCProtocolCanalJSON:
		kafkaConfig.CodecConfig = &serverless.CodecConfig{
			EnableTidbExtension: aws.Bool(c.Kafka.DataFormat.EnableTiDBExtension),
		}
	case constant.CDCProtocolDebezium:
		outputOldValue := true
		disableSchema := false
		if c.Kafka.DataFormat.DebeziumConfig != nil {
			outputOldValue = !c.Kafka.DataFormat.DebeziumConfig.DisableOutputOldValue
			disableSchema = c.Kafka.DataFormat.DebeziumConfig.DisableSchema
		}
		sink.DebeziumConfig = &serverless.DebeziumConfig{
			OutputOldValue: aws.Bool(outputOldValue),
		}
		sink.DebeziumDisableSchema = aws.Bool(disableSchema)
	case constant.CDCProtocolOpen:
	}

	// TopicPartitionConfig
	kafkaConfig.PartitionNum = aws.Int32(int32(c.Kafka.TopicPartitionConfig.PartitionNum))
	kafkaConfig.ReplicationFactor = aws.Int16(int16(c.Kafka.TopicPartitionConfig.ReplicationFactor))
	sink.Dispatchers = buildTopicDispatchers(&c.Kafka.TopicPartitionConfig)

	sink.ColumnSelectors = buildColumnSelectors(c.Kafka.ColumnSelectors)

	// OutputRawChangeEvent
	kafkaConfig.OutputRawChangeEvent = aws.Bool(c.Kafka.OutputRawChangeEvent)
	return nil
}

func (c *CDCChangefeed) CloudStorageSinkConfig(
	sink *serverless.SinkConfig,
	encryptor facade.EncryptorFacade,
) bizerr.BizError {
	if c.CloudStorageInfo == nil {
		return bizerr.New(bizerr.ErrCodeParams, "cloud storage info cannot be nil")
	}
	if c.CloudStorageInfo.CloudStorage == nil {
		return bizerr.New(bizerr.ErrCodeParams, "cloud storage cannot be nil")
	}

	// Cloud storage config
	cloudStorageConfig := &serverless.CloudStorageConfig{
		FlushInterval:        aws.String(fmt.Sprintf("%ds", c.CloudStorageInfo.IntervalInSeconds)),
		FileSize:             aws.Int(c.CloudStorageInfo.SizeInMiB * 1024 * 1024),
		OutputRawChangeEvent: aws.Bool(c.CloudStorageInfo.OutputRawChangeEvent),
		OutputColumnID:       aws.Bool(c.CloudStorageInfo.OutputColumnID),
	}
	sink.CloudStorageConfig = cloudStorageConfig
	sink.ColumnSelectors = buildColumnSelectors(c.CloudStorageInfo.ColumnSelectors)

	// Date separator
	sink.DateSeparator = aws.String(string(c.CloudStorageInfo.DateSeparator))

	// Data format
	sink.Protocol = aws.String(string(c.CloudStorageInfo.DataFormat.Protocol))
	switch c.CloudStorageInfo.DataFormat.Protocol {
	case constant.CloudStorageProtocolCSV:
		csvCfg := c.CloudStorageInfo.DataFormat.CSVConfig
		// The public API treats these fields as optional, but TiCDC's update
		// path validates the resulting CSV object without applying create-time
		// defaults. Materialize them here so edits of existing changefeeds do
		// not turn omitted values into invalid empty strings downstream.
		delimiter := csvCfg.Delimiter
		if delimiter == nil {
			delimiter = aws.String(defaultCloudStorageCSVDelimiter)
		}
		quote := csvCfg.Quote
		if quote == nil {
			quote = aws.String(defaultCloudStorageCSVQuote)
		}
		nullValue := csvCfg.NullValue
		if nullValue == nil {
			nullValue = aws.String(defaultCloudStorageCSVNull)
		}
		encoding := csvCfg.Encoding
		if encoding == "" {
			encoding = constant.CloudStorageBinaryEncodingBase64
		}
		sink.CSV = &serverless.CSVConfig{
			Delimiter:            delimiter,
			Quote:                quote,
			Null:                 nullValue,
			IncludeCommitTS:      aws.Bool(csvCfg.IncludeCommitTs),
			BinaryEncodingMethod: aws.String(string(encoding)),
		}
	}
	return nil
}

func buildColumnSelectors(selectors []ColumnSelector) []*serverless.ColumnSelector {
	if len(selectors) == 0 {
		return nil
	}
	result := make([]*serverless.ColumnSelector, 0, len(selectors))
	for _, selector := range selectors {
		result = append(result, &serverless.ColumnSelector{
			Columns: selector.Columns,
			Matcher: selector.Matcher,
		})
	}
	return result
}

func buildSchemaRegistryURI(
	schemaRegistry *ConfluentSchemaRegistry,
	encryptor facade.EncryptorFacade,
) (string, bizerr.BizError) {
	schemaURL, err := url.Parse(schemaRegistry.SchemaRegistryEndpoints)
	if err != nil {
		return "", bizerr.New(
			bizerr.ErrCodeParams,
			"invalid schema registry endpoint: %s",
			schemaRegistry.SchemaRegistryEndpoints,
		)
	}
	if schemaURL.Host == "" {
		return "", bizerr.New(
			bizerr.ErrCodeParams,
			"empty host of schema registry endpoint: %s",
			schemaRegistry.SchemaRegistryEndpoints,
		)
	}
	if schemaRegistry.EnableHTTPAuth {
		password, err := encryptor.DecryptBase64(schemaRegistry.Password)
		if err != nil {
			return "", bizerr.New(
				bizerr.ErrCodeParams,
				"failed to decrypt schema registry password with error: %s",
				err.Error(),
			)
		}
		schemaURL.User = url.UserPassword(schemaRegistry.UserName, password)
		return schemaURL.String(), nil
	}
	return schemaRegistry.SchemaRegistryEndpoints, nil
}

// CDC kernel allow users to define multiple Dispatch rules, each rule can define its own topic rule.
// But in the cloud, we only allow user define one topic rule that apply to all tables.
//
// CDC kernel doc for topic dispatching  https://docs.pingcap.com/zh/tidb/stable/ticdc-sink-to-kafka#topic-%E5%88%86%E5%8F%91%E5%99%A8
func buildTopicDispatchers(tpconfig *TopicPartitionConfig) []*serverless.DispatcherConfig {
	var topicValue string
	// config := info.GetTopicPartitionConfig()
	dispatchType := tpconfig.DispatchType
	prefix := tpconfig.TopicPrefix
	suffix := tpconfig.TopicSuffix
	separator := tpconfig.Separator
	switch dispatchType {
	case constant.TopicDispatchTypeOneTopic:
		topicValue = ""
	case constant.TopicDispatchTypeByTable:
		// topic expression pattern: [prefix]{schema}[middle]{table}[suffix]
		topicValue = fmt.Sprintf("%s{schema}%s{table}%s", prefix, separator, suffix)
	case constant.TopicDispatchTypeByDatabase:
		// topic expression pattern: [prefix]{schema}[suffix]
		topicValue = fmt.Sprintf("%s{schema}%s", prefix, suffix)
	}

	var dispatchConfigs []*serverless.DispatcherConfig
	// use the new API for topic dispatching if partition dispatchers is not empty
	if len(tpconfig.PartitionDispatchers) > 0 {
		for _, partitionCfg := range tpconfig.PartitionDispatchers {
			config := &serverless.DispatcherConfig{
				Topic:     aws.String(topicValue),
				Partition: aws.String(string(partitionCfg.PartitionType)),
				Matcher:   partitionCfg.Matcher,
			}
			switch partitionCfg.PartitionType {
			case constant.PartitionDispatcherIndexValue:
				config.Index = aws.String(partitionCfg.IndexName)
			case constant.PartitionDispatcherColumns:
				config.Columns = partitionCfg.Columns
			}
			dispatchConfigs = append(dispatchConfigs, config)
		}
	}
	// append a *.* matcher for the topic dispatch
	dispatchConfigs = append(dispatchConfigs, &serverless.DispatcherConfig{
		Topic:   aws.String(topicValue),
		Matcher: []string{"*.*"},
	})
	return dispatchConfigs
}

func (c *CDCChangefeed) MySQLSinkConfig(sink *serverless.SinkConfig, encryptor facade.EncryptorFacade) bizerr.BizError {
	mysqlConfig := &serverless.MySQLConfig{}
	// ssl ca path depends on the operation system of the uderlying cdc server.
	// currently, use the redhat default ca path.
	if c.MySQL.Authentication.EnableTLS {
		mysqlConfig.SSLCa = aws.String("/etc/pki/tls/certs/ca-bundle.crt")
	}
	sink.MySQLConfig = mysqlConfig
	return nil
}

type ChangefeedStateCount struct {
	State constant.ChangfeedState
	Count int
}